app/utils/file_utils.py

"""Author RAG Chatbot SaaS — File Utilities.

SHA-256 hashing, MIME type detection by magic bytes, size checking.
RULE: Always validate by magic bytes, NEVER by file extension alone.
"""

import hashlib
import os
import zipfile
from pathlib import Path

import structlog

from app.config import get_settings
from app.exceptions.ingestion import (
    EmptyFileError,
    FileTooLargeError,
    UnsupportedFormatError,
)

logger = structlog.get_logger(__name__)
cfg = get_settings()

# MIME types we accept → maps to our extension labels
_SUPPORTED_MIME_TYPES: dict[str, str] = {
    "application/pdf": "pdf",
    "application/epub+zip": "epub",
    "application/vnd.openxmlformats-officedocument.wordprocessingml.document": "docx",
    "text/plain": "txt",
    "text/x-tex": "txt",
}


def _detect_mime_from_signature(file_path: str) -> str | None:
    """Detect MIME type from file signatures when libmagic is unavailable."""
    with open(file_path, "rb") as handle:
        header = handle.read(8)

    if header.startswith(b"%PDF-"):
        return "application/pdf"

    if zipfile.is_zipfile(file_path):
        with zipfile.ZipFile(file_path) as archive:
            names = archive.namelist()
            if any(name.startswith("word/") for name in names):
                return "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
            if "mimetype" in names:
                with archive.open("mimetype") as mime_file:
                    if b"epub" in mime_file.read().lower():
                        return "application/epub+zip"
            if any("META-INF/container.xml" in name for name in names):
                return "application/epub+zip"

    with open(file_path, "rb") as handle:
        sample = handle.read(8192)
    try:
        sample.decode("utf-8")
        return "text/plain"
    except UnicodeDecodeError:
        return None


def detect_mime_type(file_path: str) -> str:
    """Detect MIME type by inspecting file magic bytes.

    Args:
        file_path: Absolute path to the file.

    Returns:
        MIME type string (e.g., 'application/pdf').
    """
    try:
        import magic

        return magic.from_file(file_path, mime=True)
    except (ImportError, OSError) as exc:
        logger.warning("libmagic unavailable, using signature fallback", error=str(exc))
        mime_type = _detect_mime_from_signature(file_path)
        if mime_type is None:
            raise UnsupportedFormatError(extension=Path(file_path).suffix.lstrip(".")) from exc
        return mime_type


def compute_sha256(file_path: str) -> str:
    """Compute SHA-256 hash of a file.

    Args:
        file_path: Absolute path to the file.

    Returns:
        Hex string of the SHA-256 hash (64 characters).
    """
    sha256 = hashlib.sha256()
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            sha256.update(chunk)
    return sha256.hexdigest()


def get_file_extension_from_mime(mime_type: str) -> str | None:
    """Map a MIME type to our extension label.

    Args:
        mime_type: MIME type string.

    Returns:
        Extension label ('pdf', 'epub', 'docx', 'txt') or None if unsupported.
    """
    if mime_type == "application/zip":
        return None
    return _SUPPORTED_MIME_TYPES.get(mime_type)


def validate_upload(file_path: str, claimed_size_bytes: int) -> str:
    """Run all pre-processing validation checks on an uploaded file.

    Checks: existence, empty, size limit, MIME type (by magic bytes).

    Args:
        file_path: Absolute path to the uploaded file.
        claimed_size_bytes: File size in bytes (from client or OS stat).

    Returns:
        Detected extension label ('pdf', 'epub', 'docx', 'txt').

    Raises:
        EmptyFileError: If file has no content.
        FileTooLargeError: If file exceeds size limit.
        UnsupportedFormatError: If MIME type is not supported.
    """
    # Empty file check
    actual_size = os.path.getsize(file_path)
    if actual_size == 0:
        raise EmptyFileError()

    # Size limit check
    max_bytes = cfg.UPLOAD_MAX_FILE_SIZE_MB * 1024 * 1024
    if actual_size > max_bytes:
        size_mb = actual_size / (1024 * 1024)
        raise FileTooLargeError(size_mb=size_mb, max_mb=cfg.UPLOAD_MAX_FILE_SIZE_MB)

    # MIME type check by magic bytes
    mime_type = detect_mime_type(file_path)
    extension = get_file_extension_from_mime(mime_type)
    if extension is None:
        detected_ext = Path(file_path).suffix.lstrip(".")
        raise UnsupportedFormatError(extension=detected_ext or mime_type)

    logger.debug("File validation passed", path=file_path, mime=mime_type, size_mb=actual_size / 1024 / 1024)
    return extension


def ensure_directory(path: str) -> None:
    """Create a directory (and all parents) if it doesn't exist.

    Args:
        path: Directory path to create.
    """
    Path(path).mkdir(parents=True, exist_ok=True)


def get_author_upload_path(author_id: str) -> str:
    """Get the upload directory path for a specific author.

    Args:
        author_id: UUID of the author.

    Returns:
        Absolute directory path string.
    """
    base = cfg.UPLOAD_STORAGE_PATH
    path = os.path.join(base, "authors", author_id)
    ensure_directory(path)
    return path