"""Attachment storage on the local filesystem (a named Docker volume in production). Files live at ATTACHMENTS_DIR//; metadata is kept in the attachments table.""" import re import uuid from pathlib import Path from fastapi import UploadFile from app.config import get_settings ALLOWED_EXTENSIONS = { # images (camera capture on tablets produces jpg/png/heic) ".jpg", ".jpeg", ".png", ".gif", ".webp", ".heic", ".heif", ".bmp", ".tiff", ".tif", # documents ".pdf", ".doc", ".docx", ".xls", ".xlsx", ".csv", ".txt", ".msg", ".eml", } IMAGE_EXTENSIONS = { ".jpg", ".jpeg", ".png", ".gif", ".webp", ".heic", ".heif", ".bmp", ".tiff", ".tif", } CHUNK_SIZE = 1024 * 1024 class UploadValidationError(Exception): pass def _safe_filename(name: str) -> str: name = Path(name or "upload").name return re.sub(r"[^\w.\- ()]", "_", name)[:255] or "upload" async def save_attachment(upload: UploadFile, ncr_id: int) -> dict: """Validate and persist an uploaded file. Returns metadata for the Attachment row. Raises UploadValidationError on type/size violations.""" settings = get_settings() original = _safe_filename(upload.filename or "upload") ext = Path(original).suffix.lower() if ext not in ALLOWED_EXTENSIONS: raise UploadValidationError( f"File type '{ext or 'unknown'}' is not allowed. " f"Allowed: {', '.join(sorted(ALLOWED_EXTENSIONS))}" ) stored_rel = f"{ncr_id}/{uuid.uuid4().hex}{ext}" dest = Path(settings.attachments_dir) / stored_rel dest.parent.mkdir(parents=True, exist_ok=True) size = 0 max_bytes = settings.max_upload_bytes try: with dest.open("wb") as out: while chunk := await upload.read(CHUNK_SIZE): size += len(chunk) if size > max_bytes: raise UploadValidationError( f"File exceeds the {settings.max_upload_mb} MB limit." ) out.write(chunk) except UploadValidationError: dest.unlink(missing_ok=True) raise except Exception: dest.unlink(missing_ok=True) raise if size == 0: dest.unlink(missing_ok=True) raise UploadValidationError("Uploaded file is empty.") return { "original_filename": original, "stored_path": stored_rel, "content_type": upload.content_type or "application/octet-stream", "size_bytes": size, "is_image": ext in IMAGE_EXTENSIONS, } def attachment_abs_path(stored_path: str) -> Path: settings = get_settings() base = Path(settings.attachments_dir).resolve() p = (base / stored_path).resolve() if not str(p).startswith(str(base)): raise UploadValidationError("Invalid attachment path.") return p