image repair on existing products

scripts/repair_brand_images.py

@@ -76,6 +76,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 from app.infrastructure.settings import DB_HOST, DB_NAME  # noqa: E402
 from app.services.brand_registry import resolve_parent_brand  # noqa: E402
 from app.services.generic_products import OWN_PRODUCTS_BRAND  # noqa: E402
+from app.services import produce_reference  # noqa: E402
 from app.services.vector_store import (  # noqa: E402
     _connect,
     _list_brand_table_suffixes,
@@ -440,18 +441,56 @@ def _is_own_products(brand: str) -> bool:
     return _sanitize_name(brand or "") == _sanitize_name(OWN_PRODUCTS_BRAND)
 
 
-def _search_replacement(product_name: str, brand: str, max_results: int) -> List[str]:
-    """Validated image URLs for one product, best first.
+def _search_replacement(product_name: str, brand: str, max_results: int,
+                        timeout: int = 8) -> List[str]:
+    """Image URLs for one product, best first.
 
-    find_all_image_urls validates every candidate internally, so what comes back
-    is live by construction. Ranking then reuses the same _select_best_images
-    the ingestion pipeline uses, so a repaired row is ranked by the rules
+    For a loose commodity in the Own Products bucket the answer is already known
+    and already reviewed - see the curated branch below. Everything else is
+    searched, then ranked with the same _select_best_images the ingestion
+    pipeline uses, so a repaired row is ranked by the rules
     tests/test_image_selection.py already pins.
+
+    On the searched path `find_all_image_urls(validate=True)` probes each
+    candidate, but it falls back to returning UNVALIDATED candidates when every
+    probe fails (image_search.py, the "validation killed everything" branch). So
+    a URL from here is usually live, not live by construction.
     """
     key = _search_key(product_name, brand)
     if key in _SEARCH_CACHE:
         return _SEARCH_CACHE[key]
 
+    # THE REVIEWED IMAGE WINS, AND IT HAS TO SHORT-CIRCUIT TO DO SO.
+    #
+    # These photos were picked by hand and signed off on a contact sheet.
+    # Sending one through the rest of this function destroys it twice over. The
+    # `_names_product` corroboration gate at the end keeps only URLs containing
+    # a distinctive word of the product title, and the reviewed Commons photo
+    # for "Apple" is `Honeycrisp.jpg` - a Commons filename names the cultivar,
+    # not the catalogue word, so the CORRECT image is precisely the one that
+    # gets rejected. Even past that, ranking sorts any searched URL containing
+    # "apple" above it, and index 0 is what becomes `image_url`.
+    #
+    # Gating on the bucket is not a shortcut either: `produce_reference.lookup`
+    # walks shorter and shorter leading prefixes, so "Apple Cider Vinegar" misses
+    # twice and then hits "apple". This branch must never see another brand's
+    # rows.
+    if _is_own_products(brand):
+        curated = produce_reference.image_url(product_name)
+        if curated:
+            # A failed probe WARNS and keeps the URL, rather than falling back
+            # to a search. upload.wikimedia.org throttles a fast sweep and
+            # validate_image_url_live treats 429 exactly like 404, so a
+            # transient rate-limit would otherwise swap a reviewed photo for a
+            # searched one - silently, and at scale. That is the exact failure
+            # this table exists to prevent. A genuinely dead URL is a fix in
+            # produce_reference.json, and this warning is what asks for it.
+            if not _url_is_live(curated, timeout):
+                logger.warning("    curated image did not verify, using anyway  "
+                               "%s  ->  %s", product_name[:38], curated[:70])
+            _SEARCH_CACHE[key] = [curated]
+            return [curated]
+
     from app.services.image_search import find_all_image_urls
 
     # `stage_6_images` already blanks the brand for this bucket
@@ -581,7 +620,7 @@ def repair(cur, conn, brands: List[str], apply: bool, timeout: int, workers: int
                 totals["unrepairable"] += 1
                 continue
 
-            found = _search_replacement(name, suffix, max_results)
+            found = _search_replacement(name, suffix, max_results, timeout)
             if not found:
                 logger.info("    no replacement found  %s", name[:58])
                 unrepairable.append(f"{suffix}/{row['image_id']} ({name})")