← back to Designer Wallcoverings
collection hero pipeline: heuristic-shortlist + Gemini vision-confirm room-setting hero per collection; dry-run apply script
a2ad88093a980ccb1c2d03ec0d738616f474de80 · 2026-06-23 13:12:08 -0700 · Steve
Resolves correct room-setting hero per DW collection sourced from a product
inside that same collection (DTD verdict C hybrid). Fixed Gemini endpoint:
this key 404s on v1beta+gemini-2.0-flash; uses v1+gemini-2.5-flash.
apply script is DRY-RUN by default, never writes collection.image without --execute (Steve-gated).
Files touched
A shopify/apply-collection-heroes.pyA shopify/collection-hero-pipeline.py
Diff
commit a2ad88093a980ccb1c2d03ec0d738616f474de80
Author: Steve <steve@designerwallcoverings.com>
Date: Tue Jun 23 13:12:08 2026 -0700
collection hero pipeline: heuristic-shortlist + Gemini vision-confirm room-setting hero per collection; dry-run apply script
Resolves correct room-setting hero per DW collection sourced from a product
inside that same collection (DTD verdict C hybrid). Fixed Gemini endpoint:
this key 404s on v1beta+gemini-2.0-flash; uses v1+gemini-2.5-flash.
apply script is DRY-RUN by default, never writes collection.image without --execute (Steve-gated).
---
shopify/apply-collection-heroes.py | 110 ++++++++++
shopify/collection-hero-pipeline.py | 399 ++++++++++++++++++++++++++++++++++++
2 files changed, 509 insertions(+)
diff --git a/shopify/apply-collection-heroes.py b/shopify/apply-collection-heroes.py
new file mode 100644
index 00000000..d3d6e27c
--- /dev/null
+++ b/shopify/apply-collection-heroes.py
@@ -0,0 +1,110 @@
+#!/usr/bin/env python3
+"""
+APPLY collection hero images from the manifest (DRY-RUN BY DEFAULT).
+
+Reads audit/collection-hero-manifest.json and, for each RESOLVED row, would set
+collection.image to the chosen room-setting image (sourced from a product INSIDE
+that same collection, vision-confirmed).
+
+ - custom collections -> PUT /custom_collections/{id}.json
+ - smart collections -> PUT /smart_collections/{id}.json
+ image is set by src; Shopify fetches the CDN URL (same-CDN, so instant).
+
+SAFETY:
+ * DRY-RUN by default — prints exactly what it WOULD do, writes nothing.
+ * --execute is required to write. Even then it honors --limit and a per-write
+ gap so we never hammer the store; ≥90s batch gaps are unnecessary here
+ (single PUTs, not bulk variant creates) but we cap rate at the 80-bucket.
+ * Sets the image ALT to the collection title (clean, no vendor/private-label).
+ * Never touches flagged/already-good rows.
+ * This script is intentionally NOT run by the officer — Steve gates the live
+ write (customer-facing). The pending-approval memo carries the exact command.
+
+Usage:
+ python3 apply-collection-heroes.py # dry-run, all resolved
+ python3 apply-collection-heroes.py --limit 5 # dry-run first 5
+ python3 apply-collection-heroes.py --execute # LIVE (Steve only)
+"""
+import sys, os, json, time, argparse, urllib.request, urllib.error
+
+STORE = "designer-laboratory-sandbox.myshopify.com"
+API = "2024-10"
+HERE = os.path.dirname(os.path.abspath(__file__))
+SECRETS = os.path.expanduser("~/Projects/secrets-manager/.env")
+MANIFEST = os.path.join(HERE, "audit", "collection-hero-manifest.json")
+
+
+def load_token():
+ t = os.environ.get("SHOPIFY_ADMIN_TOKEN")
+ if t:
+ return t
+ if os.path.exists(SECRETS):
+ for line in open(SECRETS):
+ for k in ("SHOPIFY_ADMIN_TOKEN", "SHOPIFY_THEME_TOKEN"):
+ if line.startswith(k + "="):
+ return line.split("=", 1)[1].strip().strip('"').strip("'")
+ sys.exit("No Shopify token.")
+
+
+TOK = load_token()
+H = {"X-Shopify-Access-Token": TOK, "Content-Type": "application/json"}
+
+
+def put_image(kind, coll_id, src, alt):
+ resource = "custom_collections" if kind == "custom" else "smart_collections"
+ url = f"https://{STORE}/admin/api/{API}/{resource}/{coll_id}.json"
+ payload = {resource[:-1]: {"id": coll_id, "image": {"src": src, "alt": alt}}}
+ body = json.dumps(payload).encode()
+ req = urllib.request.Request(url, data=body, headers=H, method="PUT")
+ resp = urllib.request.urlopen(req, timeout=60)
+ out = json.load(resp)
+ used = resp.headers.get("x-shopify-shop-api-call-limit", "0/80")
+ return out, used
+
+
+def main():
+ ap = argparse.ArgumentParser()
+ ap.add_argument("--execute", action="store_true", help="LIVE write (default: dry-run)")
+ ap.add_argument("--limit", type=int, default=0)
+ args = ap.parse_args()
+
+ m = json.load(open(MANIFEST))
+ rows = [r for r in m.get("rows", []) if r.get("chosen_image_src")]
+ if args.limit:
+ rows = rows[:args.limit]
+
+ mode = "EXECUTE (LIVE WRITE)" if args.execute else "DRY-RUN (no writes)"
+ print(f"=== apply-collection-heroes [{mode}] {len(rows)} resolved rows ===\n")
+ if not args.execute:
+ print("(pass --execute to write — Steve-gated)\n")
+
+ wrote = 0
+ for i, r in enumerate(rows, 1):
+ alt = r["title"] # clean collection title, no private-label
+ line = (f"[{i}/{len(rows)}] {r['kind']:6s} /{r['handle']} "
+ f"<- {r['chosen_image_src'][:90]}")
+ if not args.execute:
+ print("WOULD PUT " + line)
+ continue
+ try:
+ _, used = put_image(r["kind"], r["collection_id"], r["chosen_image_src"], alt)
+ wrote += 1
+ print(f"OK ({used}) " + line)
+ # respect 80-bucket; single PUTs are light
+ try:
+ cur, cap = (int(x) for x in used.split("/"))
+ time.sleep(0.6 if cur > cap * 0.7 else 0.3)
+ except Exception:
+ time.sleep(0.3)
+ except urllib.error.HTTPError as e:
+ print(f"FAIL http {e.code}: {e.read().decode()[:160]} " + line)
+ time.sleep(2)
+ except Exception as e:
+ print(f"FAIL {e} " + line)
+ time.sleep(2)
+
+ print(f"\n=== {'wrote ' + str(wrote) if args.execute else 'dry-run only, wrote 0'} of {len(rows)} ===")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/shopify/collection-hero-pipeline.py b/shopify/collection-hero-pipeline.py
new file mode 100644
index 00000000..4981983c
--- /dev/null
+++ b/shopify/collection-hero-pipeline.py
@@ -0,0 +1,399 @@
+#!/usr/bin/env python3
+"""
+Collection HERO image resolver for Designer Wallcoverings (READ-ONLY by default).
+
+DTD verdict C HYBRID (2026-06-23, 3/3): heuristic shortlist -> vision-confirm
+top-3 -> flag zero-candidate collections.
+
+GOAL: every DW Shopify collection (custom + smart) must have a correct
+ROOM-SETTING / lifestyle hero image sourced FROM A PRODUCT INSIDE THAT SAME
+COLLECTION. Not a flat swatch. Never an image from outside the collection.
+
+Pipeline (per collection):
+ a. fetch member products (Admin API products.json?collection_id=) + their images
+ b. gather candidate images per product (Admin API product.images[].src/alt)
+ (dw_unified room_setting_images/all_images is a preferred source but the
+ local mirror is not reachable with the registry cred right now, so we
+ rely on Admin API images, which is sufficient — vision still gates.)
+ c. HEURISTIC rank: bonus if filename/alt has room|scene|interior|setting|
+ installed|lifestyle|roomset|insitu|mockup; landscape (w>h) > square;
+ penalty for swatch|memo|sample|flat|cutout|white|tub|paste|accessory.
+ Shortlist top 3.
+ d. VISION-CONFIRM the shortlist with Gemini 2.0 Flash: "installed/room-setting/
+ lifestyle interior shot, not a flat product swatch? yes/no + why".
+ Pick the first confirmed.
+ e. empty shortlist OR all top-3 fail vision -> FLAG the collection.
+
+For the 599 collections that ALREADY have collection.image: cheaply heuristic-
+check whether the existing image looks like a room-setting; only re-pick if it
+scores like a swatch/flat (below SWATCH_THRESHOLD) — and only then spend vision.
+
+NEVER writes collection.image. Emits a reviewable manifest + flagged list.
+
+Env:
+ SHOPIFY_ADMIN_TOKEN / SHOPIFY_THEME_TOKEN (secrets-manager/.env)
+ GEMINI_API_KEY (secrets-manager/.env)
+Flags:
+ --limit N process at most N collections (after ordering missing-first)
+ --no-vision skip Gemini (heuristic-only, $0) — for a fast dry pass
+ --only-missing only process the 66 with no collection.image
+ --resume merge into an existing manifest, skip already-resolved handles
+"""
+import sys, os, json, time, re, argparse, datetime, urllib.request, urllib.parse, urllib.error
+
+STORE = "designer-laboratory-sandbox.myshopify.com"
+API = "2024-10"
+HERE = os.path.dirname(os.path.abspath(__file__))
+SECRETS = os.path.expanduser("~/Projects/secrets-manager/.env")
+AUDIT_DIR = os.path.join(HERE, "audit")
+MANIFEST = os.path.join(AUDIT_DIR, "collection-hero-manifest.json")
+MANIFEST_MD = os.path.join(AUDIT_DIR, "collection-hero-manifest.md")
+
+# NOTE: this key 404s on v1beta + gemini-2.0-flash ("no longer available"); the
+# v1 endpoint + gemini-2.5-flash is confirmed-200 and vision-capable. (2026-06-23)
+GEMINI_MODEL = "gemini-2.5-flash"
+GEMINI_API_VER = "v1"
+GEMINI_COST_PER_IMG = 0.0006 # standing-rule ballpark for a product image vision call
+
+ROOM_POS = re.compile(r"room|scene|interior|setting|installed|lifestyle|roomset|insitu|in-situ|mockup|mock[-_]?up|styled|vignette|wall", re.I)
+SWATCH_NEG = re.compile(r"swatch|memo|sample|flat|cutout|cut-out|white-?ground|tearsheet|paste|tub|accessory|brush|tool|\bpaste\b", re.I)
+
+SWATCH_THRESHOLD = 1 # existing-image heuristic score below this => re-pick
+
+
+def load_secret(*keys):
+ env = {k: os.environ.get(k) for k in keys}
+ if os.path.exists(SECRETS):
+ for line in open(SECRETS):
+ line = line.strip()
+ for k in keys:
+ if line.startswith(k + "=") and not env.get(k):
+ env[k] = line.split("=", 1)[1].strip().strip('"').strip("'")
+ return env
+
+
+SEC = load_secret("SHOPIFY_THEME_TOKEN", "SHOPIFY_ADMIN_TOKEN", "GEMINI_API_KEY")
+TOK = SEC.get("SHOPIFY_ADMIN_TOKEN") or SEC.get("SHOPIFY_THEME_TOKEN")
+GKEY = SEC.get("GEMINI_API_KEY")
+if not TOK:
+ sys.exit("No Shopify token in env/secrets.")
+H = {"X-Shopify-Access-Token": TOK, "Content-Type": "application/json"}
+
+
+def shopify_get(url, tries=4):
+ last = None
+ for i in range(tries):
+ try:
+ req = urllib.request.Request(url, headers=H)
+ resp = urllib.request.urlopen(req, timeout=40)
+ body = json.load(resp)
+ link = resp.headers.get("Link", "")
+ nxt = None
+ for part in link.split(","):
+ if 'rel="next"' in part:
+ nxt = part[part.find("<") + 1:part.find(">")]
+ # be gentle with the 80-bucket
+ used = resp.headers.get("x-shopify-shop-api-call-limit", "0/80")
+ try:
+ cur, cap = (int(x) for x in used.split("/"))
+ if cur > cap * 0.8:
+ time.sleep(0.6)
+ except Exception:
+ pass
+ return body, nxt
+ except urllib.error.HTTPError as e:
+ last = e
+ if e.code in (429, 502, 503):
+ time.sleep(2 + i * 2)
+ continue
+ raise
+ except urllib.error.URLError as e:
+ last = e
+ time.sleep(1 + i)
+ raise last
+
+
+def fetch_collections():
+ out = []
+ for resource, kind in (("custom_collections", "custom"), ("smart_collections", "smart")):
+ url = f"https://{STORE}/admin/api/{API}/{resource}.json?limit=250&fields=id,handle,title,image"
+ while url:
+ body, url = shopify_get(url)
+ for c in body.get(resource, []):
+ out.append({
+ "id": c["id"], "handle": c.get("handle"), "title": c.get("title"),
+ "kind": kind, "image": c.get("image"),
+ })
+ return out
+
+
+def fetch_member_images(collection_id, cap_products=60):
+ """Return list of {product_id, product_handle, src, alt, w, h} across member products."""
+ imgs = []
+ url = (f"https://{STORE}/admin/api/{API}/products.json?collection_id={collection_id}"
+ f"&limit=250&fields=id,handle,title,images")
+ seen = 0
+ while url and seen < cap_products * 4:
+ body, url = shopify_get(url)
+ prods = body.get("products", [])
+ for p in prods:
+ seen += 1
+ for im in (p.get("images") or []):
+ imgs.append({
+ "product_id": p["id"], "product_handle": p.get("handle"),
+ "product_title": p.get("title"),
+ "src": im.get("src"), "alt": im.get("alt") or "",
+ "w": im.get("width") or 0, "h": im.get("height") or 0,
+ })
+ if seen >= cap_products * 4:
+ break
+ return imgs
+
+
+def heuristic_score(src, alt, w, h):
+ s = 0
+ blob = f"{src or ''} {alt or ''}"
+ fname = (src or "").split("?")[0].rsplit("/", 1)[-1]
+ text = f"{fname} {alt or ''}"
+ if ROOM_POS.search(text):
+ s += 3
+ if SWATCH_NEG.search(text):
+ s -= 3
+ if w and h:
+ if w > h:
+ s += 2 # landscape favored for a hero bg
+ elif w == h:
+ s += 0
+ else:
+ s -= 1 # portrait often a swatch/tearsheet
+ return s
+
+
+def shortlist(imgs, top=3):
+ scored = []
+ for im in imgs:
+ sc = heuristic_score(im["src"], im["alt"], im["w"], im["h"])
+ scored.append((sc, im))
+ # prefer higher score; tie-break: landscape first, then larger area
+ scored.sort(key=lambda t: (t[0], 1 if (t[1]["w"] > t[1]["h"]) else 0,
+ (t[1]["w"] * t[1]["h"])), reverse=True)
+ # dedupe by src
+ out, seen = [], set()
+ for sc, im in scored:
+ if im["src"] in seen:
+ continue
+ seen.add(im["src"])
+ out.append((sc, im))
+ if len(out) >= top:
+ break
+ return out
+
+
+# ---- Gemini vision ----
+_vision_calls = {"n": 0, "cost": 0.0}
+
+
+def fetch_image_b64(src, max_bytes=6_000_000):
+ req = urllib.request.Request(src, headers={"User-Agent": "Mozilla/5.0 dw-hero/1.0"})
+ resp = urllib.request.urlopen(req, timeout=40)
+ data = resp.read(max_bytes + 1)
+ if len(data) > max_bytes:
+ data = data[:max_bytes]
+ ct = resp.headers.get("Content-Type", "image/jpeg").split(";")[0].strip()
+ if ct not in ("image/jpeg", "image/png", "image/webp"):
+ ct = "image/jpeg"
+ import base64
+ return base64.b64encode(data).decode(), ct
+
+
+def vision_is_roomsetting(src):
+ """Return (yes:bool, reason:str, cost:float). Counts cost regardless of verdict."""
+ if not GKEY:
+ return (None, "no GEMINI_API_KEY", 0.0)
+ try:
+ b64, mime = fetch_image_b64(src)
+ except Exception as e:
+ return (None, f"img-fetch-fail: {e}", 0.0)
+ prompt = ("You are auditing a wallcovering e-commerce hero image. Is THIS image an "
+ "installed / room-setting / lifestyle interior shot (wallcovering shown on a "
+ "real wall in a styled room), NOT a flat product swatch, memo, cutout, white-ground "
+ "studio shot, or accessory/tool photo? Answer strictly as: 'YES - <=10 words why' "
+ "or 'NO - <=10 words why'.")
+ payload = {
+ "contents": [{"parts": [
+ {"text": prompt},
+ {"inline_data": {"mime_type": mime, "data": b64}},
+ ]}],
+ "generationConfig": {"temperature": 0, "maxOutputTokens": 40},
+ }
+ url = (f"https://generativelanguage.googleapis.com/{GEMINI_API_VER}/models/"
+ f"{GEMINI_MODEL}:generateContent?key={GKEY}")
+ body = json.dumps(payload).encode()
+ _vision_calls["n"] += 1
+ _vision_calls["cost"] += GEMINI_COST_PER_IMG
+ try:
+ req = urllib.request.Request(url, data=body,
+ headers={"Content-Type": "application/json"})
+ resp = urllib.request.urlopen(req, timeout=60)
+ out = json.load(resp)
+ txt = out["candidates"][0]["content"]["parts"][0]["text"].strip()
+ except urllib.error.HTTPError as e:
+ return (None, f"gemini-http-{e.code}", GEMINI_COST_PER_IMG)
+ except Exception as e:
+ return (None, f"gemini-err: {e}", GEMINI_COST_PER_IMG)
+ yes = txt.upper().lstrip().startswith("YES")
+ return (yes, txt, GEMINI_COST_PER_IMG)
+
+
+def main():
+ ap = argparse.ArgumentParser()
+ ap.add_argument("--limit", type=int, default=0)
+ ap.add_argument("--no-vision", action="store_true")
+ ap.add_argument("--only-missing", action="store_true")
+ ap.add_argument("--resume", action="store_true")
+ args = ap.parse_args()
+
+ os.makedirs(AUDIT_DIR, exist_ok=True)
+ colls = fetch_collections()
+ # order: missing-image first, then existing
+ colls.sort(key=lambda c: (1 if c.get("image") else 0, c["kind"], c["handle"] or ""))
+
+ done = {}
+ if args.resume and os.path.exists(MANIFEST):
+ prev = json.load(open(MANIFEST))
+ for r in prev.get("rows", []):
+ done[r["handle"]] = r
+ for r in prev.get("flagged", []):
+ done[r["handle"]] = {**r, "_flagged": True}
+ print(f"resume: {len(done)} handles already processed")
+
+ rows, flagged, already_good = [], [], []
+ processed = 0
+ t0 = time.time()
+
+ for c in colls:
+ h = c["handle"]
+ if h in done:
+ r = done[h]
+ (flagged if r.get("_flagged") else rows).append({k: v for k, v in r.items() if k != "_flagged"})
+ continue
+ if args.only_missing and c.get("image"):
+ continue
+ if args.limit and processed >= args.limit:
+ break
+ processed += 1
+
+ existing = c.get("image") or {}
+ existing_src = existing.get("src")
+ # For collections that ALREADY have an image, cheap heuristic gate first.
+ if existing_src:
+ esc = heuristic_score(existing_src, existing.get("alt", ""),
+ existing.get("width", 0), existing.get("height", 0))
+ if esc >= SWATCH_THRESHOLD:
+ already_good.append({
+ "handle": h, "title": c["title"], "kind": c["kind"],
+ "existing_image_src": existing_src, "heuristic_score": esc,
+ })
+ print(f"[ok-keep] {h} (existing scores {esc}) — kept")
+ continue
+ # else falls through to re-pick
+
+ imgs = fetch_member_images(c["id"])
+ sl = shortlist(imgs, top=3)
+ chosen = None
+ verdict = None
+ if not sl:
+ flagged.append({
+ "handle": h, "title": c["title"], "kind": c["kind"],
+ "reason": "no candidate images in any member product",
+ "member_image_count": len(imgs),
+ "had_existing": bool(existing_src),
+ "existing_image_src": existing_src,
+ })
+ print(f"[FLAG] {h} — 0 candidate images (members imgs={len(imgs)})")
+ continue
+
+ if args.no_vision:
+ sc, im = sl[0]
+ chosen, verdict = im, "heuristic-only (no vision)"
+ confidence = "heuristic"
+ else:
+ confidence = "vision"
+ for sc, im in sl:
+ yes, reason, cost = vision_is_roomsetting(im["src"])
+ if yes is True:
+ chosen, verdict = im, reason
+ break
+ # keep last reason for flag detail
+ verdict = reason
+ if chosen is None:
+ flagged.append({
+ "handle": h, "title": c["title"], "kind": c["kind"],
+ "reason": "no member image passed vision (all top-3 are swatch/flat)",
+ "shortlist": [{"src": im["src"], "score": sc} for sc, im in sl],
+ "last_vision": verdict,
+ "had_existing": bool(existing_src),
+ "existing_image_src": existing_src,
+ })
+ print(f"[FLAG] {h} — top-3 failed vision (${_vision_calls['cost']:.4f} batch)")
+ continue
+
+ rows.append({
+ "handle": h, "title": c["title"], "kind": c["kind"],
+ "collection_id": c["id"],
+ "chosen_product_id": chosen["product_id"],
+ "chosen_product_handle": chosen["product_handle"],
+ "chosen_image_src": chosen["src"],
+ "source": "shopify_admin_product_image",
+ "vision_verdict": verdict,
+ "confidence": confidence,
+ "reason": ("re-pick: existing image scored as swatch" if existing_src
+ else "no existing collection.image"),
+ "replacing_existing": existing_src,
+ })
+ tag = "RE-PICK" if existing_src else "RESOLVE"
+ print(f"[{tag}] {h} -> {chosen['src'][:80]} (${_vision_calls['cost']:.4f} batch)")
+
+ # periodic checkpoint
+ if processed % 15 == 0:
+ write_manifest(rows, flagged, already_good, partial=True)
+
+ write_manifest(rows, flagged, already_good, partial=False)
+ dt = time.time() - t0
+ print(f"\n=== DONE processed={processed} in {dt:.0f}s ===")
+ print(f"resolved (write candidates): {len(rows)}")
+ print(f"flagged (no room-setting): {len(flagged)}")
+ print(f"already-good (kept): {len(already_good)}")
+ print(f"vision calls: {_vision_calls['n']} | $ {_vision_calls['cost']:.4f} (Gemini {GEMINI_MODEL} @ ~${GEMINI_COST_PER_IMG}/img)")
+ print(f"manifest: {MANIFEST}")
+
+
+def write_manifest(rows, flagged, already_good, partial):
+ out = {
+ "generated_at": datetime.datetime.now().isoformat(),
+ "store": STORE, "api": API, "model": GEMINI_MODEL, "partial": partial,
+ "counts": {"resolved": len(rows), "flagged": len(flagged), "already_good": len(already_good)},
+ "vision": {"calls": _vision_calls["n"], "cost_usd": round(_vision_calls["cost"], 4)},
+ "rows": rows, "flagged": flagged, "already_good": already_good,
+ }
+ json.dump(out, open(MANIFEST, "w"), indent=2)
+ md = []
+ md.append(f"# Collection Hero Image Manifest")
+ md.append(f"\n_generated {out['generated_at']} | store {STORE} | model {GEMINI_MODEL}_\n")
+ md.append(f"**resolved (write candidates):** {len(rows)} ")
+ md.append(f"**flagged (need render / decision):** {len(flagged)} ")
+ md.append(f"**already-good (existing image kept):** {len(already_good)} ")
+ md.append(f"**vision spend:** ${_vision_calls['cost']:.4f} over {_vision_calls['n']} calls\n")
+ if flagged:
+ md.append("## Flagged — no room-setting image inside collection\n")
+ for f in flagged:
+ md.append(f"- `/{f['handle']}` — {f['title']} [{f['kind']}] — {f['reason']}")
+ md.append("\n## Resolved (sample of first 30)\n")
+ for r in rows[:30]:
+ md.append(f"- `/{r['handle']}` -> product `{r['chosen_product_handle']}` "
+ f"({r['confidence']}): {r['chosen_image_src']}")
+ open(MANIFEST_MD, "w").write("\n".join(md) + "\n")
+
+
+if __name__ == "__main__":
+ main()
← 2228022a Add Tres Tintas completeness audit (desc+spec+room, roll-vs-
·
back to Designer Wallcoverings
·
Fix room-setting pattern scale: drive tiling by real repeat, d4a51d6f →