"""The `images` command group: pictures at three levels (Stream AW3). `fetch` is the former bare `app.cli images` (Open Food Facts by barcode, then the strict name search). `manifest` classifies the client's folder and the brand-owner export into a reviewable JSON without reading a pixel or touching the network (`services/image_manifest.py`); it runs on the host against a copy of the database, because the folder is not in the container. `stage` copies each chosen entry's picture into the uploads home as content-hashed derivatives (`imagery.store`) and writes the served addresses back onto the manifest; it runs on the host too, and it is the one command here that downloads (a brand-owner export names addresses, one request per second, the bot user agent, refused on anything but an image). `import` is database only: it runs in the container after a deploy, verifies each derivative exists, resolves the target by slug, id or barcode and writes the picture through the one writer (`imagery.set_image`), so a supplied picture is never replaced by a fetched one and a re-run is a no-op. `logos` looks each brand without a picture up on Wikidata and Wikimedia Commons (`services/images_public.py`: an exact label, a kind a brand can be, a free licence, else refused with the reason), downloads the accepted logo, stores it and writes it as `public:wikimedia-commons`; each host's robots.txt is read before its first request and a matching Disallow exits the command refused (2); `lines` promotes a product line's representative variant's Open Food Facts photo to the line, and leaves a line whose pictures sit only on other sizes empty. `coverage` prints, per level, how many rows hold a picture (supplied and fetched apart), how many hold none, and what the cascade shows for those (`services/image_ask.coverage`, the same count the /images stat cards read); it is run before and after an import so the handoff carries both. """ from __future__ import annotations import argparse import hashlib import json import time from collections import Counter from datetime import UTC, datetime from pathlib import Path from typing import Any, Callable from sqlalchemy import select from sqlalchemy.orm import Session from app.db import SessionLocal from app.models import Brand, ProductLine, ProductVariant from app.services import image_manifest, imagery, images_public from app.services.collectors import robots from app.services.images import enrich_product_variants from app.services.images_public import download_image # noqa: F401 (the stage command's default fetch; tests name it here) def cmd_images_fetch(args: argparse.Namespace) -> int: with SessionLocal() as db: stats = enrich_product_variants( db, limit=args.limit, delay=args.delay, recheck=args.recheck, by_name=not args.barcode_only, ) print( f"checked={stats['checked']} images_found={stats['found']} " f"(barcode={stats['by_barcode']} name={stats['by_name']})" ) return 0 def _digests(root: Path) -> dict[str, str]: """sha256 per file, keyed by the path relative to the folder: byte duplicates are found by content, the one thing a name cannot say.""" out: dict[str, str] = {} for p in sorted(root.rglob("*")): if p.is_file(): out[p.relative_to(root).as_posix()] = hashlib.sha256(p.read_bytes()).hexdigest() return out def cmd_images_manifest(args: argparse.Namespace) -> int: from app.services import catalog_queries folder = Path(args.folder) if args.folder else None if folder and not folder.is_dir(): print(f"images manifest: not a folder: {folder}") return 2 if args.csv and not Path(args.csv).is_file(): print(f"images manifest: not a file: {args.csv}") return 2 classifieds: list[image_manifest.Classified] = [] digests: dict[str, str] = {} if folder: classifieds.extend(image_manifest.walk_folder(folder)) digests = _digests(folder) if args.csv: classifieds.extend(image_manifest.read_csv(args.csv)) with SessionLocal() as db: def representative_ml(line) -> int | None: detail = catalog_queries.get_product_line(db, line.slug) if detail is None or detail.representative_variant_id is None: return None return next((v.quantity_ml for v in detail.variants if v.id == detail.representative_variant_id), None) entries, summary = image_manifest.build(db, classifieds, digests=digests, representative_ml=representative_ml) document = image_manifest.manifest_document(entries, summary, folder=str(folder) if folder else None, csv_path=args.csv) out = Path(args.out) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(json.dumps(document, indent=1, ensure_ascii=False) + "\n", encoding="utf-8") for level in image_manifest.LEVELS: s = summary[level] reasons = "; ".join(f"{r['reason']} ({r['count']})" for r in s["top_reasons"]) print(f"{level:8} entries={s['entries']} with_target={s['with_target']} chosen={s['chosen']} " f"targets={s['targets']} unassigned={s['unassigned']}" + (f" top: {reasons}" if reasons else "")) print(f"wrote {out} ({len(entries)} entries)") return 0 # --- stage: files and addresses into the uploads home ------------------------------------------ #: How a manifest entry's target names the row: the level's table and what the token is. _TARGET_PREFIXES = ("brand", "line", "variant") def target_of(entry: dict[str, Any]) -> tuple[str, str] | None: """`(level, token)` from an entry's `target` (`brand:`, `line:`, `variant:`). A person may type a bare slug or id on an entry the matcher refused; its `level` then names the table. None when the entry has no target.""" raw = (entry.get("target") or "").strip() if not raw: return None prefix, sep, token = raw.partition(":") if sep and prefix in _TARGET_PREFIXES and token: return prefix, token level = entry.get("level") return (level, raw) if level in _TARGET_PREFIXES else None def derivative_path_of(url: str, uploads: Path) -> Path | None: """Where a served address lives on disk, or None for an address outside the served prefix.""" prefix = imagery.URL_PREFIX + "/" if not url or not url.startswith(prefix): return None return uploads / "images" / url[len(prefix):] def stage_manifest( document: dict[str, Any], *, folder: Path | None, uploads: Path, check: bool = False, fetch: Callable[[str], bytes] = download_image, delay: float = 1.0, sleep: Callable[[float], None] = time.sleep, ) -> Counter: """Every chosen entry's picture into `/originals//` and its two derivatives into `/images//`, and `url` and `thumb_url` written back onto the entry. Idempotent: an entry whose derivative already exists is skipped without a read or a request; the same file or address chosen at two levels is read or fetched once. `--check` reads and requests nothing and reports what a run would do. Counts, per outcome and level.""" counts: Counter = Counter() bytes_of: dict[str, bytes] = {} failed: dict[str, str] = {} last_request = 0.0 for entry in document.get("entries", []): if not entry.get("chosen"): continue level = entry.get("level") target = target_of(entry) if target is None or level not in imagery.LEVELS: counts[f"{level}:no_target"] += 1 continue existing = derivative_path_of(entry.get("url") or "", uploads) if existing is not None and existing.exists(): counts[f"{level}:existing"] += 1 continue source = entry.get("source_path") or entry.get("source_url") or "" entry.pop("stage_error", None) if source in failed: entry["stage_error"] = failed[source] counts[f"{level}:failed"] += 1 continue if check: counts[f"{level}:would_{'read' if entry.get('source_path') else 'download'}"] += 1 continue data = bytes_of.get(source) if data is None: try: if entry.get("source_path"): if folder is None: raise ValueError("no --folder for a file entry") data = (folder / entry["source_path"]).read_bytes() else: wait = delay - (time.monotonic() - last_request) if wait > 0: sleep(wait) last_request = time.monotonic() data = fetch(entry["source_url"]) counts["downloads"] += 1 except (OSError, ValueError) as exc: failed[source] = entry["stage_error"] = str(exc) if str(exc) else type(exc).__name__ counts[f"{level}:failed"] += 1 continue bytes_of[source] = data try: stored = imagery.store(data, level, target[1], uploads=uploads) except ValueError as exc: failed[source] = entry["stage_error"] = str(exc) counts[f"{level}:failed"] += 1 continue entry["url"], entry["thumb_url"] = stored.url, stored.thumb_url counts[f"{level}:stored"] += 1 if not check: document["staged_at"] = datetime.now(UTC).isoformat(timespec="seconds") return counts def _print_counts(counts: Counter) -> None: for level in imagery.LEVELS: bits = {k.split(":", 1)[1]: v for k, v in sorted(counts.items()) if k.startswith(level + ":")} if bits: print(f"{level:8} " + " ".join(f"{k}={v}" for k, v in bits.items())) rest = {k: v for k, v in counts.items() if ":" not in k} if rest: print(" ".join(f"{k}={v}" for k, v in sorted(rest.items()))) def cmd_images_stage(args: argparse.Namespace) -> int: path = Path(args.manifest) if not path.is_file(): print(f"images stage: not a file: {path}") return 2 folder = Path(args.folder) if args.folder else None if folder is not None and not folder.is_dir(): print(f"images stage: not a folder: {folder}") return 2 uploads = Path(args.uploads) if args.uploads else imagery.UPLOADS_ROOT document = json.loads(path.read_text(encoding="utf-8")) counts = stage_manifest(document, folder=folder, uploads=uploads, check=args.check, delay=args.delay) _print_counts(counts) if args.check: print("[check, nothing written]") return 0 path.write_text(json.dumps(document, indent=1, ensure_ascii=False) + "\n", encoding="utf-8") print(f"wrote {path}") return 0 # --- import: the manifest into the database ------------------------------------------------------- _CHAIN_LIMIT = 8 def resolve_target(db: Session, entry: dict[str, Any]) -> Any | None: """The row an entry names, or None. A brand or a line by its slug, an alias followed to the fold row; a variant by its id, a merged-away id followed to its survivor, else by the barcode the manifest matched it on (the id may have changed between manifest and import).""" target = target_of(entry) if target is None: return None level, token = target if level == "brand": row = db.scalar(select(Brand).where(Brand.slug == token)) hops = 0 while row is not None and row.alias_of_id is not None and hops < _CHAIN_LIMIT: row, hops = db.get(Brand, row.alias_of_id), hops + 1 return row if level == "line": row = db.scalar(select(ProductLine).where(ProductLine.slug == token)) hops = 0 while row is not None and row.alias_of_id is not None and hops < _CHAIN_LIMIT: row, hops = db.get(ProductLine, row.alias_of_id), hops + 1 return row row = db.get(ProductVariant, int(token)) if token.isdigit() else None hops = 0 while row is not None and row.merged_into_id is not None and hops < _CHAIN_LIMIT: row, hops = db.get(ProductVariant, row.merged_into_id), hops + 1 if row is None and entry.get("gtin"): row = db.scalar(select(ProductVariant).where(ProductVariant.gtin == entry["gtin"], ProductVariant.merged_into_id.is_(None))) return row def import_manifest(db: Session, document: dict[str, Any], *, uploads: Path, now: datetime | None = None) -> Counter: """Each chosen entry with a staged address onto its row through `imagery.set_image`. An entry whose derivative is not on disk, or whose target no row answers to, is counted and skipped; an entry already applied is a no-op; a fetched picture never replaces a supplied one (`image_admin_kept`). Nothing is committed here: the caller commits, or rolls back on `--check`.""" counts: Counter = Counter() now = now or datetime.now(UTC) for entry in document.get("entries", []): if not entry.get("chosen"): continue level = entry.get("level") if level not in imagery.LEVELS: counts[f"{level}:bad_level"] += 1 continue url = entry.get("url") if not url: counts[f"{level}:not_staged"] += 1 continue derivative = derivative_path_of(url, uploads) if derivative is None or not derivative.exists(): counts[f"{level}:missing_derivative"] += 1 continue row = resolve_target(db, entry) if row is None: counts[f"{level}:unresolved"] += 1 continue outcome = imagery.set_image( row, url=url, thumb_url=entry.get("thumb_url"), source=entry["image_source"], level=level, licence=entry.get("licence"), attribution=entry.get("attribution"), set_at=now, ) counts[f"{level}:{outcome}"] += 1 if outcome == imagery.IMAGE_ADMIN_KEPT: counts[imagery.IMAGE_ADMIN_KEPT] += 1 return counts def cmd_images_import(args: argparse.Namespace) -> int: path = Path(args.manifest) if not path.is_file(): print(f"images import: not a file: {path}") return 2 uploads = Path(args.uploads) if args.uploads else imagery.UPLOADS_ROOT document = json.loads(path.read_text(encoding="utf-8")) with SessionLocal() as db: counts = import_manifest(db, document, uploads=uploads) if args.check: db.flush() db.rollback() else: db.commit() _print_counts(counts) print(f"{imagery.IMAGE_ADMIN_KEPT}={counts[imagery.IMAGE_ADMIN_KEPT]}") if args.check: print("[check, nothing written]") return 0 # --- logos: a brand mark from Wikidata and Commons ------------------------------------------------- def logos_for_brands( db: Session, brands: list[Brand], *, uploads: Path, check: bool = False, fetch_json: Callable[[str], dict | None] = images_public.fetch_json, fetch_bytes: Callable[[str], bytes] = download_image, pace: images_public.Pace | None = None, read_robots: Callable[[str], robots.Robots] = robots.read, now: datetime | None = None, out: Callable[[str], None] = print, ) -> Counter: """Each brand looked up, one line printed per brand with the outcome, the accepted logo downloaded, stored and written unless `check`. Every request goes through `pace`, and every host's robots.txt is read (paced, once per host) before its first request: a Disallow that matches, a host refusing us at robots.txt, or a robots.txt we cannot read stops the run there, counted as `stopped:`, no further brand looked up. Counts per outcome and refusal reason; nothing committed here.""" pace = pace or images_public.Pace() guard = images_public.RobotsGuard(read=pace.paced(read_robots)) paced_json, paced_bytes = guard.guarded(pace.paced(fetch_json)), guard.guarded(pace.paced(fetch_bytes)) counts: Counter = Counter() for brand in brands: try: found = images_public.wikidata_brand_logo(brand.name, fetch_json=paced_json) outcome = _logo_outcome(brand, found, check=check, fetch_bytes=paced_bytes, uploads=uploads, now=now, out=out) except images_public.RobotsRefused as exc: counts[f"stopped:{exc.reason}"] += 1 out(f"{brand.slug}: stopped, {exc.reason} ({exc.detail}); no further request") break counts[outcome] += 1 if outcome == f"brand:{imagery.IMAGE_ADMIN_KEPT}": counts[imagery.IMAGE_ADMIN_KEPT] += 1 counts["requests"] = pace.requests return counts def _logo_outcome( brand: Brand, found: images_public.Logo | images_public.Refusal, *, check: bool, fetch_bytes: Callable[[str], bytes], uploads: Path, now: datetime | None, out: Callable[[str], None], ) -> str: """One brand's outcome key, its line printed.""" if isinstance(found, images_public.Refusal): out(f"{brand.slug}: refused {found.reason}" + (f" ({found.detail})" if found.detail else "")) return f"brand:refused:{found.reason}" if check: out(f"{brand.slug}: would apply {found.file} ({found.licence}; {found.entity}; {found.mime or 'mime unknown'})") return "brand:would_apply" outcome = images_public.apply_logo(brand, found, fetch_bytes=fetch_bytes, uploads=uploads, now=now) out(f"{brand.slug}: {outcome} {found.file} ({found.licence})") return f"brand:{outcome}" def cmd_images_logos(args: argparse.Namespace) -> int: uploads = Path(args.uploads) if args.uploads else imagery.UPLOADS_ROOT with SessionLocal() as db: brands = images_public.brands_without_a_picture(db, limit=args.limit, slug=args.brand) if args.brand and not brands: print(f"images logos: no canonical brand with slug {args.brand!r}") return 2 print(f"{len(brands)} brand(s) to look up, {args.delay:.1f} s between requests" + (" [check]" if args.check else "")) counts = logos_for_brands(db, brands, uploads=uploads, check=args.check, pace=images_public.Pace(args.delay)) stopped = next((k for k in counts if k.startswith("stopped:")), None) if args.check or stopped: db.rollback() else: db.commit() _print_counts(counts) print(f"{imagery.IMAGE_ADMIN_KEPT}={counts[imagery.IMAGE_ADMIN_KEPT]}") if stopped: print(f"images logos: REFUSED ({stopped.removeprefix('stopped:')}); a robots.txt said no, nothing written, no workaround") return 2 if args.check: print("[check, nothing written]") return 0 # --- lines: the representative's Open Food Facts photo promoted to the line ------------------------- def cmd_images_lines(args: argparse.Namespace) -> int: with SessionLocal() as db: counts = images_public.promote_lines(db, limit=args.limit) if args.check: db.flush() db.rollback() else: db.commit() _print_counts(counts) if args.check: print("[check, nothing written]") return 0 # --- coverage: the per-level count, before and after an import -------------------------------------- def cmd_images_coverage(args: argparse.Namespace) -> int: from app.services import image_ask with SessionLocal() as db: levels = image_ask.coverage(db) print_coverage(levels) return 0 def print_coverage(levels: dict[str, dict[str, int]]) -> None: """One line per level: `brand total=2418 admin=0 public=0 without=2418 none=2418`.""" for level in imagery.LEVELS: bits = levels.get(level, {}) print(f"{level:8} " + " ".join(f"{k}={v}" for k, v in bits.items())) def register(sub: argparse._SubParsersAction) -> None: images = sub.add_parser("images", help="pictures at three levels: fetch (Open Food Facts), manifest, stage, import, logos, lines, coverage") isub = images.add_subparsers(dest="images_command", required=True) fetch = isub.add_parser("fetch", help="attach openly licensed product imagery from Open Food Facts") fetch.add_argument("--limit", type=int, default=None) fetch.add_argument("--delay", type=float, default=0.8) fetch.add_argument("--recheck", action="store_true", help="re-check product variants already looked up") fetch.add_argument("--barcode-only", action="store_true", help="skip the name-search fallback") fetch.set_defaults(func=cmd_images_fetch) manifest = isub.add_parser("manifest", help="classify a picture folder and a brand-owner export into a reviewable manifest (no network)") manifest.add_argument("--folder", default=None, help="the folder of supplied pictures (read only)") manifest.add_argument("--csv", default=None, help="a brand-owner product export (the Pernod Ricard USA shape)") manifest.add_argument("--out", required=True, help="the manifest to write, e.g. import/images/-.json") manifest.set_defaults(func=cmd_images_manifest) stage = isub.add_parser("stage", help="copy each chosen manifest entry into the uploads home as derivatives and write its address back (files, and one download per address)") stage.add_argument("--manifest", required=True, help="the manifest `images manifest` wrote") stage.add_argument("--folder", default=None, help="the folder the manifest's source_path entries are relative to") stage.add_argument("--uploads", default=None, help=f"the uploads home (default {imagery.UPLOADS_ROOT}; on the host, the mounted folder)") stage.add_argument("--delay", type=float, default=1.0, help="seconds between downloads (default 1)") stage.add_argument("--check", action="store_true", help="report what a run would read, download and store; nothing written") stage.set_defaults(func=cmd_images_stage) imp = isub.add_parser("import", help="set each staged manifest entry's picture on its brand, product line or product variant (database only)") imp.add_argument("--manifest", required=True, help="the staged manifest (entries carry url and thumb_url)") imp.add_argument("--uploads", default=None, help=f"where the derivatives are verified (default {imagery.UPLOADS_ROOT})") imp.add_argument("--check", action="store_true", help="resolve and count, then roll back; nothing written") imp.set_defaults(func=cmd_images_import) logos = isub.add_parser("logos", help="a brand mark from Wikidata and Wikimedia Commons for each brand without a picture (exact label, a kind a brand can be, a free licence, else refused)") logos.add_argument("--limit", type=int, default=None, help="at most this many brands, the ones with the most product variants first") logos.add_argument("--brand", default=None, help="one brand by slug") logos.add_argument("--delay", type=float, default=1.0, help="seconds between requests, every host counted (default 1)") logos.add_argument("--uploads", default=None, help=f"the uploads home the derivatives are written to (default {imagery.UPLOADS_ROOT})") logos.add_argument("--check", action="store_true", help="look up and report; no download, nothing stored, nothing written") logos.set_defaults(func=cmd_images_logos) lines = isub.add_parser("lines", help="promote the representative variant's Open Food Facts photo to each product line without a picture (no network)") lines.add_argument("--limit", type=int, default=None, help="at most this many lines, lowest id first") lines.add_argument("--check", action="store_true", help="count what a run would set, then roll back") lines.set_defaults(func=cmd_images_lines) coverage = isub.add_parser("coverage", help="per level, how many brands, product lines and product variants hold a picture (supplied and fetched apart), how many none, and what the cascade shows (no network, nothing written)") coverage.set_defaults(func=cmd_images_coverage)