"""Wider legitimate coverage (Stream AW3.4): a brand mark from Wikidata and Wikimedia Commons, and a product line's picture promoted from its representative variant's Open Food Facts photo. Both answer the same question as the rest of the pictures: what may we show without taking a retailer's photograph? A Commons logo is accepted only when the whole chain agrees, precision first: `wbsearchentities` on the brand's exact name, an entity whose label equals the brand after NFKD case folding, whose `P31` (instance of) is in a small allow set of things a brand can be, one such entity and no second; its `P154` (logo image), else `P18` (image); the file's `imageinfo` with `extmetadata`, and a licence that is free or public domain. Anything else is a refusal with its reason, and a refusal leaves the row empty: a wrong picture is worse than none. The bytes are downloaded and stored as derivatives (`imagery.store`), never hotlinked, and written through the one writer with `public:wikimedia-commons`, the licence short name and an attribution line naming the file, the artist and the description page. The promotion is narrower still: a product line with no picture takes its representative variant's Open Food Facts photo (same address, level `line`); a line whose pictures sit only on other sizes stays empty, because the representative is the bottle the line's card stands for and a photo of another size is a guess. Every fetcher takes an injectable `fetch_json` so the rules are tested without the network; the commands in `app/cli_images.py` pace every request at one per second with the bot user agent, and every host's robots.txt is read before its first request (`RobotsGuard`, the collectors' one matcher and policy): any matching Disallow stops the run with `robots_disallow` before a request is made. The first probe of this command asked `/w/api.php` on both Wikimedia hosts without reading either robots.txt; both publish `Disallow: /w/` under `User-agent: *`, and under the project's rule that is a no until rian rules otherwise (the running list holds the decision). """ from __future__ import annotations import html import json import logging import re import time import unicodedata import urllib.error import urllib.parse import urllib.request from collections import Counter from dataclasses import dataclass from datetime import datetime from typing import Any, Callable from sqlalchemy import and_, func, select from sqlalchemy.orm import Session from app.models import Brand, Listing, PriceObservation, ProductLine, ProductVariant, Shop, ShopPlace from app.services import imagery, places from app.services.collectors import robots from app.services.collectors.fetch import SourceBlocked from app.services.images import USER_AGENT logger = logging.getLogger(__name__) WIKIDATA_API = "https://www.wikidata.org/w/api.php" COMMONS_API = "https://commons.wikimedia.org/w/api.php" COMMONS_FILE_PREFIX = "File:" #: `P31` values a brand's entity may carry: the things a brand can be. Anything else (a person, a #: place, a film, a grape) is refused, however well the label matches. Data, extended as brands #: are met; a value here is a Wikidata item id, checked against the entity's own claims. BRAND_KINDS: dict[str, str] = { "Q431289": "brand", "Q167270": "trademark", "Q4830453": "business", "Q6881511": "enterprise", "Q783794": "company", "Q1207302": "distillery", "Q156362": "winery", "Q13235160": "manufacturer", } P_INSTANCE_OF = "P31" P_LOGO = "P154" P_IMAGE = "P18" #: The long side the Commons render is asked for: the same as the served derivative, so an #: SVG (most logos) arrives as a PNG of the size the page will show, never larger. RENDER_WIDTH = imagery.LONG_SIDE #: Refusal reasons, the vocabulary the command counts. NO_ENTITY = "no_entity" LABEL_MISMATCH = "label_mismatch" KIND_OUTSIDE_SET = "kind_outside_set" AMBIGUOUS = "ambiguous" NO_LOGO = "no_logo" NO_FILE_INFO = "no_file_info" LICENCE_NOT_FREE = "licence_not_free" NO_ANSWER = "no_answer" #: The two reasons that stop the whole run rather than one brand: a host's robots.txt refuses #: the address (or the host refuses us at robots.txt), and a robots.txt we could not read. ROBOTS_DISALLOW = "robots_disallow" ROBOTS_UNAVAILABLE = "robots_unavailable" @dataclass(frozen=True, slots=True) class Logo: """A Commons logo accepted for one brand, with what the row will record.""" entity: str file: str download_url: str description_url: str licence: str licence_url: str | None artist: str | None mime: str | None @property def attribution(self) -> str: by = f" by {self.artist}" if self.artist else "" return f"Wikimedia Commons: {self.file}{by}, {self.description_url}" @dataclass(frozen=True, slots=True) class Refusal: reason: str detail: str = "" # --- the fetchers ----------------------------------------------------------------------------- def fetch_json(url: str, *, timeout: int = 20) -> dict | None: """One JSON answer with the bot identity, or None on any failure (logged, never raised): a lookup that fails is a refusal, not an error, and the command carries on to the next brand.""" request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "application/json"}) try: with urllib.request.urlopen(request, timeout=timeout) as response: content_type = (response.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower() if content_type not in ("application/json", "text/json"): logger.debug("public_image_not_json content_type=%s", content_type) return None return json.load(response) except urllib.error.HTTPError as exc: logger.debug("public_image_http_error code=%s", exc.code) except Exception as exc: logger.debug("public_image_request_failed error=%s", type(exc).__name__) return None def download_image(url: str, *, timeout: int = 30) -> bytes: """One picture from an address, with the bot identity. Refused (ValueError) when the answer is not an image by its own content type: a login page or an error page saved as a picture would be staged as one, and the refusal is what `imagery.store` cannot see.""" request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "image/*"}) try: with urllib.request.urlopen(request, timeout=timeout) as response: content_type = (response.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower() if not content_type.startswith("image/"): raise ValueError(f"not an image: content type {content_type or 'missing'!r}") return response.read() except urllib.error.HTTPError as exc: raise ValueError(f"HTTP {exc.code}") from exc except urllib.error.URLError as exc: raise ValueError(f"unreachable: {exc.reason}") from exc class RobotsRefused(RuntimeError): """A host's robots.txt (or its absence) forbids the next request. Raised by `RobotsGuard` before the request is made; `logos_for_brands` stops the run on it, never skips a brand.""" def __init__(self, reason: str, detail: str) -> None: super().__init__(detail) self.reason, self.detail = reason, detail class RobotsGuard: """Reads each host's robots.txt before its first request and refuses any address a Disallow matches, through the collectors' one matcher and policy (`collectors/robots.py`: any matching Disallow is a no, a longer Allow does not out-lawyer it; 401/403 at robots.txt is the host refusing us; 5xx, 429 or a timeout means we cannot know and the run stops). `read` is injectable so a test never touches the network; the command paces it like any request.""" def __init__(self, *, read: Callable[[str], robots.Robots] = robots.read) -> None: self._read = read self.rules: dict[str, robots.Robots] = {} def check(self, url: str) -> None: parts = urllib.parse.urlsplit(url) origin = f"{parts.scheme}://{parts.netloc}" rules = self.rules.get(origin) if rules is None: try: rules = self._read(origin) except SourceBlocked as exc: raise RobotsRefused(ROBOTS_DISALLOW, str(exc)) from exc except robots.RobotsUnavailable as exc: raise RobotsRefused(ROBOTS_UNAVAILABLE, str(exc)) from exc self.rules[origin] = rules if not rules.allows(url): path = parts.path + (f"?{parts.query}" if parts.query else "") raise RobotsRefused(ROBOTS_DISALLOW, f"{parts.netloc} robots.txt disallows {path}") def guarded(self, fn: Callable[..., Any]) -> Callable[..., Any]: def call(url: str, *args, **kwargs): self.check(url) return fn(url, *args, **kwargs) return call class Pace: """One request per second across every host the command touches. `wait()` before each request; the clock is injectable so a test never sleeps.""" def __init__(self, delay: float = 1.0, *, sleep: Callable[[float], None] = time.sleep, clock: Callable[[], float] = time.monotonic): self.delay, self._sleep, self._clock = delay, sleep, clock self._last: float | None = None self.requests = 0 def wait(self) -> None: if self._last is not None: gap = self.delay - (self._clock() - self._last) if gap > 0: self._sleep(gap) self._last = self._clock() self.requests += 1 def paced(self, fn: Callable[..., Any]) -> Callable[..., Any]: def call(*args, **kwargs): self.wait() return fn(*args, **kwargs) return call # --- the rules -------------------------------------------------------------------------------- _SPACE_RE = re.compile(r"\s+") def fold(text: str | None) -> str: """The form two names are compared in: NFKD, combining marks dropped, case folded, one space between words. "Moët & Chandon" and "MOET & CHANDON" fold the same; "Glenfiddich" and "Glenfiddich Distillery" do not.""" decomposed = unicodedata.normalize("NFKD", text or "") stripped = "".join(ch for ch in decomposed if not unicodedata.combining(ch)) return _SPACE_RE.sub(" ", stripped.casefold()).strip() _TAG_RE = re.compile(r"<[^>]+>") def plain_text(value: str | None) -> str | None: """Commons' `Artist` and `Credit` arrive as HTML fragments; the row keeps words.""" if not value: return None text = html.unescape(_TAG_RE.sub("", value)) text = _SPACE_RE.sub(" ", text).strip() return text or None _FREE_RE = re.compile(r"^(public domain|pd(?:[- ].*)?|cc0(?:[ -].*)?|cc[ -]by(?:[ -]sa)?(?:[ -][0-9.]+.*)?)$") def free_licence(short_name: str | None) -> bool: """Whether a Commons `LicenseShortName` lets us show the file: public domain, CC0, CC BY and CC BY-SA at any version. Everything else is refused, including CC BY-NC, CC BY-ND, fair use and a copyrighted logo Commons holds under a trademark notice.""" name = fold(short_name) if not name or "-nc" in name or " nc" in name or "-nd" in name or " nd" in name: return False return bool(_FREE_RE.match(name)) def _claim_values(entity: dict, prop: str) -> list[Any]: """The values of one property on an entity, preferred rank first, deprecated dropped.""" out: list[tuple[int, Any]] = [] for claim in (entity.get("claims") or {}).get(prop) or []: rank = claim.get("rank", "normal") if rank == "deprecated": continue value = ((claim.get("mainsnak") or {}).get("datavalue") or {}).get("value") if value is None: continue out.append((0 if rank == "preferred" else 1, value)) return [v for _, v in sorted(out, key=lambda p: p[0])] def wikidata_brand_logo(brand_name: str, *, fetch_json: Callable[[str], dict | None] = fetch_json) -> Logo | Refusal: """A Commons logo for one brand name, or the refusal that says why not. Three requests at most: the entity search, the entities' claims, the file's imageinfo.""" query = urllib.parse.urlencode({ "action": "wbsearchentities", "search": brand_name, "language": "en", "uselang": "en", "type": "item", "limit": 7, "format": "json", }) search = fetch_json(f"{WIKIDATA_API}?{query}") if search is None: return Refusal(NO_ANSWER, "wbsearchentities") hits = search.get("search") or [] if not hits: return Refusal(NO_ENTITY) wanted = fold(brand_name) same_label = [h["id"] for h in hits if h.get("id") and fold(h.get("label")) == wanted] if not same_label: return Refusal(LABEL_MISMATCH, ", ".join(f"{h.get('id')} {h.get('label')!r}" for h in hits[:3])) query = urllib.parse.urlencode({ "action": "wbgetentities", "ids": "|".join(same_label[:5]), "props": "claims|labels", "languages": "en", "format": "json", }) answer = fetch_json(f"{WIKIDATA_API}?{query}") if answer is None: return Refusal(NO_ANSWER, "wbgetentities") entities = answer.get("entities") or {} accepted: list[str] = [] kinds_seen: list[str] = [] for qid in same_label[:5]: entity = entities.get(qid) or {} if fold(((entity.get("labels") or {}).get("en") or {}).get("value")) != wanted: continue kinds = [v.get("id") for v in _claim_values(entity, P_INSTANCE_OF) if isinstance(v, dict)] kinds_seen.extend(k for k in kinds if k) if any(k in BRAND_KINDS for k in kinds): accepted.append(qid) if not accepted: return Refusal(KIND_OUTSIDE_SET, ", ".join(sorted(set(kinds_seen))) or "no P31") if len(accepted) > 1: return Refusal(AMBIGUOUS, ", ".join(accepted)) entity_id = accepted[0] files = [v for v in _claim_values(entities[entity_id], P_LOGO) if isinstance(v, str)] files = files or [v for v in _claim_values(entities[entity_id], P_IMAGE) if isinstance(v, str)] if not files: return Refusal(NO_LOGO, entity_id) file_name = files[0] query = urllib.parse.urlencode({ "action": "query", "titles": COMMONS_FILE_PREFIX + file_name, "prop": "imageinfo", "iiprop": "url|mime|extmetadata", "iiurlwidth": RENDER_WIDTH, "format": "json", }) answer = fetch_json(f"{COMMONS_API}?{query}") if answer is None: return Refusal(NO_ANSWER, "imageinfo") pages = ((answer.get("query") or {}).get("pages") or {}) info = next((p.get("imageinfo") or [None] for p in pages.values()), [None])[0] if not info or not (info.get("thumburl") or info.get("url")): return Refusal(NO_FILE_INFO, file_name) meta = info.get("extmetadata") or {} def field(name: str) -> str | None: value = (meta.get(name) or {}).get("value") return str(value) if value not in (None, "") else None licence = plain_text(field("LicenseShortName")) if not free_licence(licence): return Refusal(LICENCE_NOT_FREE, licence or "no licence") return Logo( entity=entity_id, file=COMMONS_FILE_PREFIX + file_name, download_url=info.get("thumburl") or info["url"], description_url=info.get("descriptionurl") or f"https://commons.wikimedia.org/wiki/{urllib.parse.quote(COMMONS_FILE_PREFIX + file_name)}", licence=licence or "", licence_url=field("LicenseUrl"), artist=plain_text(field("Artist")), mime=info.get("mime"), ) # --- the commands' work ----------------------------------------------------------------------- def brands_without_a_picture(db: Session, *, limit: int | None = None, slug: str | None = None) -> list[Brand]: """The brands `images logos` visits: canonical, not hidden, with no picture at brand level, the ones with the most live product variants first (a logo there stands in on the most cards). `slug` names one brand whatever its state; the writer decides what happens to it.""" from app.services import publish if slug: row = db.scalar(select(Brand).where(Brand.slug == slug)) return [row] if row is not None and row.alias_of_id is None else [] live = func.count(ProductVariant.id) stmt = ( select(Brand) .outerjoin(ProductVariant, and_(ProductVariant.brand_id == Brand.id, ProductVariant.merged_into_id.is_(None))) .where(Brand.alias_of_id.is_(None), Brand.image_url.is_(None)) .group_by(Brand.id) .order_by(live.desc(), Brand.id) ) rows = [b for b in db.scalars(stmt) if not publish.is_hidden(b)] return rows[:limit] if limit else rows def apply_logo( brand: Brand, logo: Logo, *, fetch_bytes: Callable[[str], bytes], uploads=imagery.UPLOADS_ROOT, now: datetime | None = None, ) -> str: """Download, store and write one accepted logo onto its brand; the writer's outcome, or `download_failed`. The row records the licence short name and the attribution line. A brand holding a supplied picture is refused before the download: the writer would refuse it after.""" if imagery.is_admin(brand.image_source) and brand.image_url: return imagery.IMAGE_ADMIN_KEPT try: data = fetch_bytes(logo.download_url) stored = imagery.store(data, "brand", brand.slug, uploads=uploads) except ValueError as exc: logger.info("logo_download_failed brand=%s error=%s", brand.slug, exc) return "download_failed" return imagery.set_image( brand, url=stored.url, thumb_url=stored.thumb_url, source=imagery.PUBLIC_COMMONS, level="brand", licence=logo.licence[:200], attribution=logo.attribution[:400], set_at=now, ) def representatives_of(db: Session, lines: list[ProductLine]) -> dict[int, ProductVariant | None]: """The representative variant per line, by the line page's rule (`catalog_queries. representative_variant`): among the line's live variants with a publishable, live listing that has a price, the most comparison units (airport shops, never a catalogue), then the lowest id. Two queries for every line at once.""" from app.services import catalog_queries as cq ids = [ln.id for ln in lines] if not ids: return {} members = list(db.scalars( select(ProductVariant).where(ProductVariant.product_line_id.in_(ids), ProductVariant.merged_into_id.is_(None)) )) member_ids = [v.id for v in members] units_of: dict[int, int] = {} if member_ids: latest = cq._latest_observation_subquery(member_ids) rows = db.execute( select(Listing.variant_id, places.unit_count().filter(Shop.is_catalogue_only.is_(False))) .join(PriceObservation, PriceObservation.listing_id == Listing.id) .join(Shop, Shop.id == Listing.shop_id).outerjoin(ShopPlace, places.primary_on()) .where(Listing.variant_id.in_(member_ids), PriceObservation.id.in_(select(latest.c.obs_id)), cq.publishable(db), cq.live_listings()) .group_by(Listing.variant_id) ).all() units_of = {vid: int(units) for vid, units in rows} by_line: dict[int, list[ProductVariant]] = {} for v in members: if v.id in units_of: by_line.setdefault(v.product_line_id, []).append(v) out: dict[int, ProductVariant | None] = {} for ln in lines: priced = by_line.get(ln.id, []) out[ln.id] = min(priced, key=lambda v: (-units_of[v.id], v.id)) if priced else None return out OFF_SOURCES = frozenset({imagery.PUBLIC_OFF_BARCODE, imagery.PUBLIC_OFF_NAME}) #: The promotion's outcomes, counted per line. NO_REPRESENTATIVE = "no_representative" REPRESENTATIVE_WITHOUT_PICTURE = "representative_without_picture" OTHER_SIZE_ONLY = "other_size_only" REPRESENTATIVE_NOT_OFF = "representative_not_off" def lines_without_a_picture(db: Session, *, limit: int | None = None) -> list[ProductLine]: stmt = select(ProductLine).where(ProductLine.alias_of_id.is_(None), ProductLine.image_url.is_(None)).order_by(ProductLine.id) if limit: stmt = stmt.limit(limit) return list(db.scalars(stmt)) def promote_lines(db: Session, *, limit: int | None = None, now: datetime | None = None, batch: int = 500) -> Counter: """Each product line with no picture takes its representative variant's Open Food Facts photo, same address, level `line`, through the one writer. A line with no priced variant, one whose representative has no picture, one whose pictures sit only on other sizes, and one whose representative's picture is not from Open Food Facts each stay empty and are counted. Nothing is committed here: the caller commits, or rolls back on `--check`.""" counts: Counter = Counter() lines = lines_without_a_picture(db, limit=limit) for start in range(0, len(lines), batch): chunk = lines[start:start + batch] representatives = representatives_of(db, chunk) for line in chunk: rep = representatives.get(line.id) if rep is None: counts[f"line:{NO_REPRESENTATIVE}"] += 1 continue if not rep.image_url: siblings = db.scalars( select(ProductVariant.id).where(ProductVariant.product_line_id == line.id, ProductVariant.merged_into_id.is_(None), ProductVariant.image_url.isnot(None)).limit(1) ).first() counts[f"line:{OTHER_SIZE_ONLY if siblings is not None else REPRESENTATIVE_WITHOUT_PICTURE}"] += 1 continue source = imagery.normalise_source(rep.image_source, has_gtin=bool(rep.gtin)) if source not in OFF_SOURCES: counts[f"line:{REPRESENTATIVE_NOT_OFF}"] += 1 continue outcome = imagery.set_image( line, url=rep.image_url, thumb_url=rep.thumb_url, source=source, level="line", licence=rep.image_licence, attribution=rep.image_attribution, set_at=now, ) counts[f"line:{outcome}"] += 1 db.flush() return counts