"""Shared upload validation: size + format constraints applied before any heavy work (detection, embedding, storage).""" import io from PIL import Image, UnidentifiedImageError ALLOWED_FORMATS = {"JPEG", "PNG", "WEBP"} MAX_FILE_BYTES = 10 * 1024 * 1024 # 10 MB class InvalidUploadError(Exception): """Raised when an uploaded file fails size/format validation.""" def decode_image(raw: bytes) -> Image.Image: """Decode + validate raw upload bytes. Raises InvalidUploadError on any failure.""" if len(raw) > MAX_FILE_BYTES: mb = len(raw) / (1024 * 1024) raise InvalidUploadError( f"File is {mb:.1f} MB; max is {MAX_FILE_BYTES // (1024 * 1024)} MB." ) try: pil = Image.open(io.BytesIO(raw)) pil.load() except (UnidentifiedImageError, OSError) as exc: raise InvalidUploadError("File is not a valid image.") from exc if pil.format not in ALLOWED_FORMATS: raise InvalidUploadError( f"Format {pil.format!r} not allowed. Use JPEG, PNG, or WebP." ) return pil