erp-nutzfahrzeuge/backend/app/services/file_service.py

"""File service: upload, retrieve, list, delete, and validation for vehicle files.

All file operations use UPLOAD_DIR from config for storage.
Supports images (jpg, png, webp), documents (pdf, doc, docx), and scans (jpg, png).
Enforces MAX_FILE_SIZE_MB from config (default 50MB, but endpoint enforces 20MB).
"""

from __future__ import annotations

import logging
import uuid
from pathlib import Path
from typing import Optional

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.models.file import File
from app.utils.thumbnails import generate_thumbnail, is_image_mime_type

logger = logging.getLogger(__name__)

# Allowed MIME types for upload
ALLOWED_MIME_TYPES = {
    # Images
    "image/jpeg",
    "image/png",
    "image/webp",
    # Documents
    "application/pdf",
    "application/msword",
    "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
}

# Map MIME types to file extensions for validation
MIME_TO_EXTENSIONS = {
    "image/jpeg": {"jpg", "jpeg"},
    "image/png": {"png"},
    "image/webp": {"webp"},
    "application/pdf": {"pdf"},
    "application/msword": {"doc"},
    "application/vnd.openxmlformats-officedocument.wordprocessingml.document": {"docx"},
}


def validate_mime_type(mime_type: str, filename: str) -> bool:
    """Validate that the MIME type is allowed and matches the file extension.

    Args:
        mime_type: The MIME type from the upload.
        filename: The original filename to check extension consistency.

    Returns:
        True if the MIME type is allowed and extension matches.
    """
    mime_type = mime_type.lower().strip()
    if mime_type not in ALLOWED_MIME_TYPES:
        return False

    # Check extension consistency
    ext = Path(filename).suffix.lower().lstrip(".")
    allowed_extensions = MIME_TO_EXTENSIONS.get(mime_type, set())
    if ext not in allowed_extensions:
        return False

    return True


def validate_file_size(file_size: int, max_size_mb: int = 20) -> bool:
    """Validate that the file size is within the allowed limit.

    Args:
        file_size: File size in bytes.
        max_size_mb: Maximum file size in MB (default 20MB for uploads).

    Returns:
        True if the file size is within the limit.
    """
    max_size_bytes = max_size_mb * 1024 * 1024
    return file_size <= max_size_bytes


async def upload_file(
    db: AsyncSession,
    vehicle_id: uuid.UUID,
    file_content: bytes,
    original_filename: str,
    mime_type: str,
) -> File:
    """Upload a file for a vehicle.

    Validates MIME type and file size, stores the file in UPLOAD_DIR,
    generates a thumbnail for images, and creates a DB record.

    Raises:
        ValueError: If MIME type or file size validation fails.
    """
    # Validate MIME type
    if not validate_mime_type(mime_type, original_filename):
        raise ValueError(
            f"Unsupported MIME type: {mime_type} for file: {original_filename}"
        )

    # Validate file size (20MB limit for uploads)
    file_size = len(file_content)
    if not validate_file_size(file_size, max_size_mb=20):
        raise ValueError(f"File size {file_size} bytes exceeds 20MB limit")

    # Generate stored filename
    ext = Path(original_filename).suffix.lower()
    stored_filename = f"{uuid.uuid4().hex}{ext}"

    # Create vehicle-specific upload directory
    upload_dir = Path(settings.UPLOAD_DIR) / str(vehicle_id)
    upload_dir.mkdir(parents=True, exist_ok=True)

    file_path = upload_dir / stored_filename

    # Write file to disk
    with open(file_path, "wb") as f:
        f.write(file_content)

    logger.info("File uploaded: %s -> %s", original_filename, file_path)

    # Generate thumbnail for images
    thumbnail_path = None
    if is_image_mime_type(mime_type):
        thumbnail_dir = upload_dir / "thumbnails"
        thumbnail_path = generate_thumbnail(
            source_path=file_path,
            thumbnail_dir=thumbnail_dir,
            stored_filename=stored_filename,
        )

    # Create DB record
    file_record = File(
        vehicle_id=vehicle_id,
        original_filename=original_filename,
        stored_filename=stored_filename,
        file_path=str(file_path),
        mime_type=mime_type,
        file_size=file_size,
        thumbnail_path=thumbnail_path,
    )
    db.add(file_record)
    await db.flush()
    await db.refresh(file_record)

    return file_record


async def get_file(
    db: AsyncSession,
    vehicle_id: uuid.UUID,
    file_id: uuid.UUID,
) -> Optional[File]:
    """Get a single file by ID and vehicle ID."""
    stmt = select(File).where(
        File.id == file_id,
        File.vehicle_id == vehicle_id,
    )
    result = await db.execute(stmt)
    return result.scalar_one_or_none()


async def list_files(
    db: AsyncSession,
    vehicle_id: uuid.UUID,
    page: int = 1,
    page_size: int = 20,
) -> tuple[list[File], int]:
    """List files for a vehicle with pagination.

    Returns (files, total_count).
    """
    # Count total files for this vehicle
    count_stmt = select(func.count(File.id)).where(File.vehicle_id == vehicle_id)
    count_result = await db.execute(count_stmt)
    total = count_result.scalar_one()

    # Fetch paginated files
    offset = (page - 1) * page_size
    data_stmt = (
        select(File)
        .where(File.vehicle_id == vehicle_id)
        .order_by(File.created_at.desc())
        .offset(offset)
        .limit(page_size)
    )
    result = await db.execute(data_stmt)
    files = list(result.scalars().all())

    return files, total


async def delete_file(
    db: AsyncSession,
    vehicle_id: uuid.UUID,
    file_id: uuid.UUID,
) -> Optional[File]:
    """Delete a file: remove from disk and DB.

    Returns the deleted file record, or None if not found.
    """
    file_record = await get_file(db, vehicle_id, file_id)
    if file_record is None:
        return None

    # Remove file from disk
    file_path = Path(file_record.file_path)
    if file_path.exists():
        try:
            file_path.unlink()
            logger.info("File deleted from disk: %s", file_path)
        except OSError as exc:
            logger.error("Failed to delete file from disk: %s", exc)

    # Remove thumbnail from disk if it exists
    if file_record.thumbnail_path:
        thumb_path = Path(file_record.thumbnail_path)
        if thumb_path.exists():
            try:
                thumb_path.unlink()
                logger.info("Thumbnail deleted from disk: %s", thumb_path)
            except OSError as exc:
                logger.error("Failed to delete thumbnail from disk: %s", exc)

    # Remove from DB
    await db.delete(file_record)
    await db.flush()

    return file_record