import os import shutil import tempfile import uuid from pathlib import Path from datetime import datetime from zoneinfo import ZoneInfo from fastapi import BackgroundTasks, HTTPException, UploadFile, status from fastapi.responses import FileResponse from PIL import Image from app.lib.handwriting.config import FONT_NAME, OUT_FONT_DIR from app.lib.handwriting.pipeline import run_pipeline from app.utils.fs import cleanup_dir from app.lib.handwriting.errors import PipelineInputError from app.utils.logger import get_logger from app.services.storage.spaces_client import upload_buffer from app.utils.media_io import download_image_to_file, validate_jpeg_file from app.utils.messages import JPEG_ONLY_MESSAGE from app.settings import settings MAX_BYTES = 10 * 1024 * 1024 logger = get_logger(__name__) def _create_work_dir(prefix: str) -> str: random_id = uuid.uuid4().hex return tempfile.mkdtemp(prefix=f"{prefix}_{random_id}_") def _safe_token(value: str) -> str: cleaned = "".join(ch for ch in value if ch.isalnum() or ch in ("-", "_")) return cleaned[:64] if cleaned else "req" def _create_work_dir_with_ids( prefix: str, request_id: str | None, user_id: str | None ) -> str: random_id = uuid.uuid4().hex parts = [prefix] if user_id: parts.append(f"user-{_safe_token(user_id)}") if request_id: parts.append(f"req-{_safe_token(request_id)}") parts.append(random_id) return tempfile.mkdtemp(prefix="_".join(parts) + "_") def _current_app_date_str() -> str: return datetime.now(ZoneInfo(settings.APP_TIMEZONE)).strftime("%Y-%m-%d") def _maybe_preserve_work_dir(work_dir: str, status: str) -> bool: if not settings.HANDWRITING_PRESERVE_WORK_DIR: return False target_root = settings.HANDWRITING_PRESERVE_DIR or "/tmp" target_dir = os.path.join(target_root, status, _current_app_date_str()) os.makedirs(target_dir, exist_ok=True) target_path = os.path.join(target_dir, os.path.basename(work_dir)) try: shutil.move(work_dir, target_path) logger.info("Preserved work dir: %s", target_path) return True except Exception as exc: logger.error("Failed to preserve work dir: %s", exc) return False def _finalize_success_work_dir(work_dir: str) -> None: # For download responses, defer moving/cleanup until after response is sent. if not _maybe_preserve_work_dir(work_dir, "success"): cleanup_dir(work_dir) async def process_handwriting_upload( background_tasks: BackgroundTasks, file: UploadFile, ) -> FileResponse: if not file.filename: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Missing filename." ) if not file.content_type or file.content_type.lower() not in { "image/jpeg", "image/jpg", }: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=JPEG_ONLY_MESSAGE, ) if not file.filename.lower().endswith((".jpg", ".jpeg")): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Only .jpg/.jpeg files are supported.", ) work_dir = _create_work_dir("handwriting") safe_name = Path(file.filename).name input_path = os.path.join(work_dir, safe_name) try: total_bytes = 0 with open(input_path, "wb") as out_file: while chunk := await file.read(1024 * 1024): out_file.write(chunk) total_bytes += len(chunk) if total_bytes > MAX_BYTES: raise HTTPException( status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail="File too large.", ) if total_bytes == 0: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Uploaded file is empty.", ) try: Image.open(input_path).verify() except Exception: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid image file.", ) run_pipeline(input_path, work_dir) ttf_path = os.path.join(work_dir, OUT_FONT_DIR, f"{FONT_NAME}.ttf") if not os.path.exists(ttf_path): raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Font generation failed.", ) background_tasks.add_task(_finalize_success_work_dir, work_dir) return FileResponse( ttf_path, media_type="font/ttf", filename=f"{FONT_NAME}.ttf", headers={"Cache-Control": "no-store"}, ) except HTTPException: if not _maybe_preserve_work_dir(work_dir, "failed"): cleanup_dir(work_dir) raise except PipelineInputError as exc: if not _maybe_preserve_work_dir(work_dir, "failed"): cleanup_dir(work_dir) raise except Exception as exc: if not _maybe_preserve_work_dir(work_dir, "failed"): cleanup_dir(work_dir) raise finally: await file.close() def process_handwriting_url( url: str, request_id: str | None = None, user_id: str | None = None, ) -> dict: work_dir = _create_work_dir_with_ids("handwriting_url", request_id, user_id) input_path = os.path.join(work_dir, "input.jpg") success = False try: download_image_to_file(url, input_path, MAX_BYTES) validate_jpeg_file(input_path) run_pipeline(input_path, work_dir) ttf_path = os.path.join(work_dir, OUT_FONT_DIR, f"{FONT_NAME}.ttf") if not os.path.exists(ttf_path): raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Font generation failed.", ) with open(ttf_path, "rb") as f: data = f.read() key = f"fonts/{uuid.uuid4().hex}.ttf" upload_result = upload_buffer( key=key, data=data, content_type="font/ttf", acl="public-read", ) success = True return {"url": upload_result.get("url"), "key": upload_result.get("key")} except HTTPException: if not _maybe_preserve_work_dir(work_dir, "failed"): cleanup_dir(work_dir) raise except PipelineInputError as exc: if not _maybe_preserve_work_dir(work_dir, "failed"): cleanup_dir(work_dir) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc), ) except Exception as exc: if not _maybe_preserve_work_dir(work_dir, "failed"): cleanup_dir(work_dir) raise finally: if success: _finalize_success_work_dir(work_dir)