"""Image pipeline. Validates, re-encodes (drops EXIF), thumbnails, saves to media. Re-encoding through Pillow strips metadata and neutralizes polyglot/malicious payloads. Filenames are randomized. Returns a ListingImage (uncommitted). """ import io import os import secrets from PIL import Image from flask import current_app from app.models.listing import ListingImage MAX_BYTES = 8 * 1024 * 1024 # 8 MB per upload MAX_DIM = 1600 # longest edge for the full image THUMB_DIM = 400 ALLOWED = {"JPEG", "PNG", "WEBP"} class ImageError(ValueError): pass def _media_root(): root = current_app.config.get("MEDIA_ROOT") or os.path.join( current_app.instance_path, "media") os.makedirs(root, exist_ok=True) return root def process_upload(file_storage, listing_id, sort_order=0): data = file_storage.read() if not data: raise ImageError("empty file") if len(data) > MAX_BYTES: raise ImageError("file too large") try: img = Image.open(io.BytesIO(data)) img.verify() # detect truncated/corrupt img = Image.open(io.BytesIO(data)) # re-open after verify except Exception: raise ImageError("not a valid image") if img.format not in ALLOWED: raise ImageError("unsupported format") img = img.convert("RGB") # normalize, drop alpha/EXIF img.thumbnail((MAX_DIM, MAX_DIM)) root = _media_root() sub = os.path.join(root, str(listing_id)) os.makedirs(sub, exist_ok=True) name = secrets.token_hex(16) full_rel = os.path.join(str(listing_id), f"{name}.jpg") thumb_rel = os.path.join(str(listing_id), f"{name}_t.jpg") img.save(os.path.join(root, full_rel), "JPEG", quality=85, optimize=True) thumb = img.copy() thumb.thumbnail((THUMB_DIM, THUMB_DIM)) thumb.save(os.path.join(root, thumb_rel), "JPEG", quality=80, optimize=True) return ListingImage( listing_id=listing_id, path=full_rel, thumb_path=thumb_rel, sort_order=sort_order, width=img.width, height=img.height, ) def delete_image_files(image): root = _media_root() for rel in (image.path, image.thumb_path): if rel: try: os.remove(os.path.join(root, rel)) except OSError: pass