SLICE-7: images pipeline end-to-end — fetch/host/serve + INV-12 over images (SD-0002 §7.2) #30
@@ -0,0 +1,77 @@
|
|||||||
|
"""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):
|
||||||
|
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()
|
||||||
@@ -7,3 +7,4 @@ pytest>=8.0
|
|||||||
import-linter>=2.0
|
import-linter>=2.0
|
||||||
nh3>=0.2
|
nh3>=0.2
|
||||||
python-multipart>=0.0.9
|
python-multipart>=0.0.9
|
||||||
|
Pillow>=11.0
|
||||||
|
|||||||
@@ -0,0 +1,55 @@
|
|||||||
|
"""platform/images — pure decode + resolution bar + WebP renditions (SD-0002 §6.2)."""
|
||||||
|
import io
|
||||||
|
|
||||||
|
from PIL import Image
|
||||||
|
|
||||||
|
from app.platform import images
|
||||||
|
|
||||||
|
|
||||||
|
def _png(width: int, height: int, color=(120, 80, 200)) -> bytes:
|
||||||
|
buf = io.BytesIO()
|
||||||
|
Image.new("RGB", (width, height), color).save(buf, format="PNG")
|
||||||
|
return buf.getvalue()
|
||||||
|
|
||||||
|
|
||||||
|
def _jpeg(width: int, height: int) -> bytes:
|
||||||
|
buf = io.BytesIO()
|
||||||
|
Image.new("RGB", (width, height), (10, 200, 90)).save(buf, format="JPEG")
|
||||||
|
return buf.getvalue()
|
||||||
|
|
||||||
|
|
||||||
|
def test_good_image_produces_three_webp_renditions():
|
||||||
|
result = images.process(_png(1200, 900))
|
||||||
|
assert isinstance(result, images.Processed)
|
||||||
|
assert result.source_format == "PNG"
|
||||||
|
assert set(result.renditions) == {"thumb", "card", "detail"}
|
||||||
|
for name, blob in result.renditions.items():
|
||||||
|
with Image.open(io.BytesIO(blob)) as im:
|
||||||
|
assert im.format == "WEBP"
|
||||||
|
with Image.open(io.BytesIO(result.renditions["thumb"])) as im:
|
||||||
|
assert max(im.size) <= images.RENDITIONS["thumb"]
|
||||||
|
with Image.open(io.BytesIO(result.renditions["detail"])) as im:
|
||||||
|
assert max(im.size) <= images.RENDITIONS["detail"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_downscale_only_never_upscales_small_source():
|
||||||
|
result = images.process(_png(520, 520))
|
||||||
|
with Image.open(io.BytesIO(result.renditions["detail"])) as im:
|
||||||
|
assert im.size == (520, 520)
|
||||||
|
|
||||||
|
|
||||||
|
def test_below_resolution_bar_rejected_low_res():
|
||||||
|
result = images.process(_png(300, 1200)) # shorter side 300 < 500
|
||||||
|
assert isinstance(result, images.Rejected)
|
||||||
|
assert result.reason == "rejected_low_res"
|
||||||
|
|
||||||
|
|
||||||
|
def test_not_an_image_rejected_not_image():
|
||||||
|
result = images.process(b"this is not an image")
|
||||||
|
assert isinstance(result, images.Rejected)
|
||||||
|
assert result.reason == "rejected_not_image"
|
||||||
|
|
||||||
|
|
||||||
|
def test_jpeg_source_format_preserved_in_result():
|
||||||
|
result = images.process(_jpeg(800, 800))
|
||||||
|
assert result.source_format == "JPEG"
|
||||||
Reference in New Issue
Block a user