88 lines
2.8 KiB
Python
88 lines
2.8 KiB
Python
|
"""Attachment storage on the local filesystem (a named Docker volume in
|
||
|
|
production). Files live at ATTACHMENTS_DIR/<ncr_id>/<uuid><ext>; 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
|