from __future__ import annotations import re from enum import Enum from typing import Any, Dict, List, Literal, Optional from pydantic import BaseModel, Field, field_validator, model_validator from app.core.config import settings _PAGE_SPEC_RE = re.compile(r"^\s*(\d+(-\d+)?)(\s*,\s*(\d+(-\d+)?))*\s*$") class ImageFormat(str, Enum): PNG = "PNG" JPEG = "JPEG" class JobStatus(str, Enum): PENDING = "pending" PROCESSING = "processing" COMPLETED = "completed" FAILED = "failed" class ConversionParams(BaseModel): format: ImageFormat = Field(ImageFormat.PNG) dpi: int = Field(settings.DEFAULT_DPI, ge=72, le=settings.MAX_DPI) quality: int = Field(85, ge=1, le=100) pages: Optional[str] = Field(None) grayscale: bool = Field(False) transparent_bg: bool = Field(False) split_page: Optional[bool] = Field( False, description=( "Controls output granularity. " "True: each page → separate image URL. " "False (default): all pages stitched into one tall image → single URL." ), ) @field_validator("format", mode="before") @classmethod def normalise_format(cls, v: str) -> str: if isinstance(v, str): upper = v.upper() if upper == "JPG": return "JPEG" return upper return v @field_validator("pages") @classmethod def validate_page_spec_syntax(cls, v: Optional[str]) -> Optional[str]: if v is None or v.strip() == "": return None if not _PAGE_SPEC_RE.match(v): raise ValueError( f"Invalid page spec '{v}'. Use comma-separated numbers or ranges, " "e.g. '1', '1-3', '1,3,5-7'." ) for token in v.split(","): token = token.strip() if "-" in token: parts = token.split("-", 1) start, end = int(parts[0]), int(parts[1]) if start < 1: raise ValueError(f"Page numbers are 1-indexed; got '{token}'") if start > end: raise ValueError(f"Invalid range '{token}': start must be <= end") else: if int(token) < 1: raise ValueError(f"Page numbers are 1-indexed; got '{token}'") return v @model_validator(mode="after") def validate_transparent_bg(self) -> ConversionParams: if self.transparent_bg and self.format != ImageFormat.PNG: raise ValueError("transparent_bg is only supported for PNG output") return self def parse_pages(self, total_pages: int) -> List[int]: if not self.pages: return list(range(total_pages)) indices: set[int] = set() for token in self.pages.split(","): token = token.strip() if "-" in token: parts = token.split("-", 1) start, end = int(parts[0]) - 1, int(parts[1]) - 1 indices.update(range(start, end + 1)) else: indices.add(int(token) - 1) valid = sorted(i for i in indices if 0 <= i < total_pages) if not valid: raise ValueError( f"Page spec '{self.pages}' contains no pages within " f"the document's {total_pages} page(s)" ) return valid class PageResult(BaseModel): page_number: int download_url: str width: int height: int size_bytes: int format: str file_id: Optional[str] = None file_url: Optional[str] = None upload_error: Optional[str] = None class UploadSummary(BaseModel): upload_mode: str total_uploaded: int failed_uploads: int files: List["UploadedFileInfo"] class ConversionResult(BaseModel): job_id: str status: JobStatus total_pages: int converted_pages: int pages: List[PageResult] duration_ms: float upload_summary: Optional[UploadSummary] = None class AsyncJobResponse(BaseModel): job_id: str status: JobStatus message: str status_url: str class JobStatusResponse(BaseModel): job_id: str status: JobStatus progress: float = Field(..., ge=0, le=100) result: Optional[ConversionResult] = None error: Optional[str] = None class HealthResponse(BaseModel): status: str version: str environment: str class ErrorResponse(BaseModel): error: str request_id: Optional[str] = None class UploadedFileInfo(BaseModel): file_id: str file_url: str filename: str size_bytes: int