Files
wiggleverse-ecomm/backend/app/platform/images.py
T
ben.stull c929282e07
ci / check (push) Has been cancelled
ci / check (pull_request) Has been cancelled
fix(products): bomb-safe image processing + per-image error isolation; honest claim/resumability docs (SLICE-7 review)
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-06-12 00:58:10 -07:00

78 lines
2.6 KiB
Python

"""platform/images — pure image processing (SD-0002 §6.2).
Model-free, deterministic, no I/O of its own: decode bytes, enforce the
resolution bar (INV-18 / Q-3), and emit downscale-only WebP renditions. The
caller (the image-fetch phase) owns fetching, storage, and status.
"""
from __future__ import annotations
import io
from dataclasses import dataclass
from PIL import Image, UnidentifiedImageError
# Q-3 (SD-0002 §13): the minimum shorter-side dimension; below it an image is
# rejected_low_res. Rejects icons/thumbnails, accepts normal product photos.
MIN_IMAGE_SHORT_SIDE = 500
# Rendition longest-side caps (px); downscale-only — a smaller source is kept
# at its own size, never upscaled. Encoded WebP.
RENDITIONS = {"thumb": 160, "card": 480, "detail": 1600}
# Source formats we accept (Pillow format names).
_ACCEPTED = {"JPEG", "PNG", "WEBP"}
_WEBP_QUALITY = 82
@dataclass(frozen=True)
class Processed:
source_format: str
width: int
height: int
renditions: dict[str, bytes] # name -> WebP bytes
@dataclass(frozen=True)
class Rejected:
reason: str # "rejected_low_res" | "rejected_not_image"
def process(data: bytes) -> Processed | Rejected:
"""Decode + bar-check + renditions, or a typed rejection. Never raises on
bad input — undecodable/unsupported bytes are Rejected('rejected_not_image')."""
try:
with Image.open(io.BytesIO(data)) as im:
im.load()
source_format = im.format or ""
if source_format not in _ACCEPTED:
return Rejected(reason="rejected_not_image")
rgb = im.convert("RGB")
width, height = rgb.size
if min(width, height) < MIN_IMAGE_SHORT_SIDE:
return Rejected(reason="rejected_low_res")
renditions = {
name: _rendition(rgb, cap) for name, cap in RENDITIONS.items()
}
return Processed(
source_format=source_format,
width=width,
height=height,
renditions=renditions,
)
except (UnidentifiedImageError, OSError, ValueError, Image.DecompressionBombError):
return Rejected(reason="rejected_not_image")
def _rendition(rgb: Image.Image, cap: int) -> bytes:
longest = max(rgb.size)
if longest > cap:
scale = cap / longest
size = (max(1, round(rgb.width * scale)), max(1, round(rgb.height * scale)))
resized = rgb.resize(size, Image.LANCZOS)
else:
resized = rgb
buf = io.BytesIO()
resized.save(buf, format="WEBP", quality=_WEBP_QUALITY, method=6)
return buf.getvalue()