"""Pictures at three levels: the provenance vocabulary, the one display cascade, the tier the featured algorithm reads, the one writer and the derivative store (Stream AW3.2). A picture lives on a brand, a product line or a product variant, each row carrying the same seven columns (`image_url`, `thumb_url`, `image_source`, `image_level`, `image_licence`, `image_attribution`, `image_set_at`; migration `aw3b1c2d3e4f`). Every reader shows the best picture the catalogue holds for the product variant, else its product line, else its brand, through `resolve()`, with the level exposed so a brand mark is styled as one and never enters a product's structured data as its image. Every writer goes through `set_image()`, which refuses to replace a supplied picture (`admin:*`) with a fetched one (`public:*`): a human value is never overwritten by a machine (agents.md). `store()` writes the derivatives a page serves; the original is kept and never served. No network here, and `resolve()` is pure. The vocabulary is controlled and tested: `admin:` for a picture the client or a brand supplied (`admin:william-grant-via-adam`), `public:openfoodfacts:barcode` and `public:openfoodfacts:name` for Open Food Facts (a barcode match is a fact, a name match a judgement, and the audit reports them apart), `public:wikimedia-commons` for a Commons logo. The two strings written before this vocabulary are moved by `backfill image_sources`. """ from __future__ import annotations import hashlib import io import re from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path from typing import Any #: What a picture depicts, worst to best in the cascade's fallback order. LEVELS = ("brand", "line", "variant") #: The levels a page may emit as a product's `image` in structured data and as its Open Graph #: picture: a bottle, or the line's bottle. A brand mark shown in a bottle's place is a #: fallback the page styles as one (`thumb--brand`), never the product's image to a crawler. STRUCTURED_DATA_LEVELS = ("variant", "line") #: The class every reader puts on a picture shown at brand level (Thumb.css). BRAND_MARK_CLASS = "thumb--brand" ADMIN_PREFIX = "admin:" PUBLIC_PREFIX = "public:" PUBLIC_OFF_BARCODE = "public:openfoodfacts:barcode" PUBLIC_OFF_NAME = "public:openfoodfacts:name" PUBLIC_COMMONS = "public:wikimedia-commons" PUBLIC_SOURCES = frozenset({PUBLIC_OFF_BARCODE, PUBLIC_OFF_NAME, PUBLIC_COMMONS}) _ADMIN_RE = re.compile(r"^admin:[a-z0-9]+(?:-[a-z0-9]+)*$") #: The strings `images fetch` wrote before the vocabulary, and what each became. A bare #: "Open Food Facts" (older still) is split by whether the row holds a barcode. LEGACY_SOURCES = { "Open Food Facts (barcode)": PUBLIC_OFF_BARCODE, "Open Food Facts (name)": PUBLIC_OFF_NAME, } LEGACY_BARE = "Open Food Facts" #: What `set_image` did. The third is counted by every writer and printed by the commands. APPLIED = "applied" UNCHANGED = "unchanged" IMAGE_ADMIN_KEPT = "image_admin_kept" #: The tier the featured selection reads (AW2): none, then public by level, then supplied by #: level. Supplied outranks fetched at every level because a brand's own mark is a fact and a #: community photo is a judgement. TIERS: dict[tuple[str, str], int] = { ("public", "brand"): 1, ("public", "line"): 2, ("public", "variant"): 3, ("admin", "brand"): 4, ("admin", "line"): 5, ("admin", "variant"): 6, } TIER_NONE = 0 #: The served derivatives: the long side of the page picture and of the card thumbnail, and #: the WebP quality. Alpha is kept (a cut-out bottle sits on any card background), EXIF is #: dropped, CMYK is converted. The content hash in the file name makes the address immutable. LONG_SIDE = 1200 THUMB_SIDE = 400 WEBP_QUALITY = 85 UPLOADS_ROOT = Path("/srv/uploads") URL_PREFIX = "/uploads/images" _EXT_OF = {"JPEG": ".jpg", "PNG": ".png", "WEBP": ".webp", "TIFF": ".tif", "GIF": ".gif", "BMP": ".bmp"} # --- the vocabulary --------------------------------------------------------------------------- def is_admin(source: str | None) -> bool: return bool(source) and source.startswith(ADMIN_PREFIX) def is_public(source: str | None) -> bool: return bool(source) and source.startswith(PUBLIC_PREFIX) def valid_source(source: str | None) -> bool: """A source the vocabulary admits: `admin:` or one of the public sources.""" if not source: return False return source in PUBLIC_SOURCES or bool(_ADMIN_RE.match(source)) def normalise_source(source: str | None, *, has_gtin: bool = False) -> str | None: """The vocabulary's word for a stored source: a legacy string becomes its successor, a valid one is returned as it is, anything else is returned untouched for the caller to count.""" if source in LEGACY_SOURCES: return LEGACY_SOURCES[source] if source == LEGACY_BARE: return PUBLIC_OFF_BARCODE if has_gtin else PUBLIC_OFF_NAME return source def credit(source: str | None) -> str: """The short credit a page prints beside a picture ("Photo: Open Food Facts"). A supplied picture credits the supplier named in its slug, up to the `-via-` that says who handed it over; a legacy string credits Open Food Facts; an unknown string is printed as it is.""" if not source: return "" if source.startswith("public:openfoodfacts") or source.startswith(LEGACY_BARE): return "Open Food Facts" if source == PUBLIC_COMMONS: return "Wikimedia Commons" if is_admin(source): slug = source[len(ADMIN_PREFIX):].split("-via-", 1)[0] return " ".join(w.capitalize() for w in slug.split("-") if w) return source # --- the cascade ------------------------------------------------------------------------------ @dataclass(frozen=True, slots=True) class Picture: url: str thumb_url: str | None #: What the picture depicts: brand, line or variant. A reader styles a brand mark as one #: and never emits it as a product's `image`. level: str source: str | None credit: str licence: str | None = None attribution: str | None = None @property def tier(self) -> int: kind = "admin" if is_admin(self.source) else "public" return TIERS.get((kind, self.level), TIER_NONE) def picture_of(row: Any, table_level: str) -> Picture | None: """The picture one row carries, or None. `image_level` is the row's own word for what it depicts; a row written before the level existed (a product variant's Open Food Facts photo from before `backfill image_sources`) is read at its table's level so the site keeps its pictures between the deploy and the backfill.""" if row is None or not getattr(row, "image_url", None): return None level = getattr(row, "image_level", None) or table_level source = getattr(row, "image_source", None) return Picture( url=row.image_url, thumb_url=getattr(row, "thumb_url", None) or row.image_url, level=level, source=source, credit=credit(source), licence=getattr(row, "image_licence", None), attribution=getattr(row, "image_attribution", None), ) def resolve(variant: Any = None, line: Any = None, brand: Any = None) -> Picture | None: """The one cascade: the product variant's own picture, else its product line's, else its brand's, else nothing. Pure: rows in, a Picture out, the level exposed.""" for row, level in ((variant, "variant"), (line, "line"), (brand, "brand")): picture = picture_of(row, level) if picture is not None: return picture return None def in_structured_data(level: str | None) -> bool: """Whether a picture at this level may be a product's `image` to a crawler.""" return level in STRUCTURED_DATA_LEVELS def image_tier(variant: Any = None, line: Any = None, brand: Any = None) -> int: """The tier of the picture `resolve` would show: 0 none, 1 public brand, 2 public line, 3 public variant, 4 admin brand, 5 admin line, 6 admin variant.""" picture = resolve(variant, line, brand) return picture.tier if picture is not None else TIER_NONE # --- the one writer --------------------------------------------------------------------------- def set_image( row: Any, *, url: str, thumb_url: str | None, source: str, level: str, licence: str | None = None, attribution: str | None = None, set_at: datetime | None = None, ) -> str: """Write a picture onto a brand, product line or product variant row. Returns APPLIED, UNCHANGED (the same picture from the same source at the same level is already there, so an import re-run is a no-op) or IMAGE_ADMIN_KEPT (the row holds a supplied picture and this one was fetched: refused, and the caller counts it). A source outside the vocabulary or a level outside the three is a programming error and raises.""" if level not in LEVELS: raise ValueError(f"image level must be one of {LEVELS}, not {level!r}") source = normalise_source(source, has_gtin=bool(getattr(row, "gtin", None))) if not valid_source(source): raise ValueError(f"image source outside the vocabulary: {source!r}") if not url: raise ValueError("an image needs a url") if row.image_url == url and row.image_source == source and row.image_level == level: return UNCHANGED if is_admin(row.image_source) and is_public(source) and row.image_url: return IMAGE_ADMIN_KEPT row.image_url = url row.thumb_url = thumb_url or url row.image_source = source row.image_level = level row.image_licence = licence row.image_attribution = attribution row.image_set_at = set_at or datetime.now(UTC) return APPLIED # --- the store -------------------------------------------------------------------------------- @dataclass(frozen=True, slots=True) class Stored: original_path: Path url: str thumb_url: str sha8: str def sha8_of(data: bytes) -> str: return hashlib.sha256(data).hexdigest()[:8] def derivative_paths(level: str, id: str | int, sha8: str, *, uploads: Path = UPLOADS_ROOT) -> tuple[Path, Path]: """Where the served derivatives of one picture live: the page picture and its thumbnail, under `/images//-[-thumb].webp`.""" if level not in LEVELS: raise ValueError(f"image level must be one of {LEVELS}, not {level!r}") folder = uploads / "images" / level return folder / f"{id}-{sha8}.webp", folder / f"{id}-{sha8}-thumb.webp" def derivative_urls(level: str, id: str | int, sha8: str) -> tuple[str, str]: return f"{URL_PREFIX}/{level}/{id}-{sha8}.webp", f"{URL_PREFIX}/{level}/{id}-{sha8}-thumb.webp" def store(data: bytes, level: str, id: str | int, *, uploads: Path = UPLOADS_ROOT) -> Stored: """Keep the original under `originals//` (never served) and write the two WebP derivatives under `images//`: orientation applied from EXIF, then EXIF dropped, CMYK converted, alpha kept, the long side at most 1200 and 400. `` is the target's token (a brand or product line slug, a product variant id) and `` the content hash, so the same bytes land at the same address and a changed picture gets a new one. Idempotent: existing derivatives are not re-encoded. Bytes Pillow cannot open raise ValueError.""" from PIL import Image, ImageOps, UnidentifiedImageError if level not in LEVELS: raise ValueError(f"image level must be one of {LEVELS}, not {level!r}") try: image = Image.open(io.BytesIO(data)) image.load() except (UnidentifiedImageError, OSError) as exc: raise ValueError(f"not an image Pillow can read: {type(exc).__name__}") from exc ext = _EXT_OF.get(image.format or "", ".bin") sha8 = sha8_of(data) original = uploads / "originals" / level / f"{id}-{sha8}{ext}" full_path, thumb_path = derivative_paths(level, id, sha8, uploads=uploads) url, thumb_url = derivative_urls(level, id, sha8) original.parent.mkdir(parents=True, exist_ok=True) full_path.parent.mkdir(parents=True, exist_ok=True) if not original.exists(): original.write_bytes(data) if full_path.exists() and thumb_path.exists(): return Stored(original, url, thumb_url, sha8) image = ImageOps.exif_transpose(image) or image has_alpha = "A" in image.getbands() or (image.mode == "P" and "transparency" in image.info) image = image.convert("RGBA" if has_alpha else "RGB") for path, side in ((full_path, LONG_SIDE), (thumb_path, THUMB_SIDE)): derived = image.copy() derived.thumbnail((side, side), Image.LANCZOS) derived.save(path, format="WEBP", quality=WEBP_QUALITY) return Stored(original, url, thumb_url, sha8)