← back to Dw Color Image Audit
vision-gate non-PR Tier-1 color-image set: 105->29 survivors->8 recrawl/18 hold/3 drop (DTD Q1=b)
07d2c0465f15d6365f4bf0f99c4bb2f437d08587 · 2026-06-20 11:47:07 -0700 · Steve
Files touched
A scripts/classify_survivors.pyA scripts/vision_gate.py
Diff
commit 07d2c0465f15d6365f4bf0f99c4bb2f437d08587
Author: Steve <steve@designerwallcoverings.com>
Date: Sat Jun 20 11:47:07 2026 -0700
vision-gate non-PR Tier-1 color-image set: 105->29 survivors->8 recrawl/18 hold/3 drop (DTD Q1=b)
---
scripts/classify_survivors.py | 108 +++++++++++++++++++++++++
scripts/vision_gate.py | 181 ++++++++++++++++++++++++++++++++++++++++++
2 files changed, 289 insertions(+)
diff --git a/scripts/classify_survivors.py b/scripts/classify_survivors.py
new file mode 100644
index 0000000..d4d74dc
--- /dev/null
+++ b/scripts/classify_survivors.py
@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+"""
+Classify the 29 vision-gate SURVIVORS into 3 dispositions per DTD Q2=a:
+ DROP_NAME_LEAK : the vision-detected color is literally present in the declared
+ name -> audit-side parser leak, NOT a real defect (false positive).
+ HOLD_BUSY : busy multi-color / botanical / commercial pattern where the
+ declared color plausibly names one ink -> HUMAN REVIEW, never
+ auto-remap (the memo's hard rule).
+ RECRAWL : solid / textured / neutral-ground product whose image is plainly
+ a different color than the name -> real wrong-image suspect ->
+ re-crawl source full-page to find the correct colorway->image pairing.
+Pure-local logic, $0. Writes out/survivor_dispositions.{jsonl,csv}.
+"""
+import json, os, re
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+IN = os.path.join(ROOT, "out", "vision_gate_results.jsonl")
+
+BASIC = ["red", "orange", "yellow", "green", "blue", "teal", "purple", "violet",
+ "pink", "brown", "beige", "tan", "gold", "magenta", "navy", "burgundy",
+ "white", "black", "gray", "grey", "cream", "lavender", "turquoise",
+ "crimson", "coral", "terracotta", "olive", "khaki", "chestnut"]
+
+# busy / multi-color pattern signals: doubled motif words + botanical/commercial lines
+BUSY_RE = re.compile(
+ r"\b(flowers?\s+flowers?|trees?\s+trees?|leaves?\s+leaves?|grasses?[- ]leaves?|"
+ r"landscapes?\s+landscapes?|waterscapes?\s+waterscapes?|floral|botanical|"
+ r"flower|tree|leaf|leaves|grass|vine|garden|meadow|reef|safari|woodland|"
+ r"damask|toile|paisley|medley|pinwheel|splash|sunset|coleus|petrouchka)\b", re.I)
+
+# solid / texture lines that SHOULD be one near-solid color (strong wrong-image signal)
+SOLID_RE = re.compile(
+ r"\b(paintable|faux suede|suede|faux linen|linen|grasscloth|grass cloth|"
+ r"raffia|vinyl|texture|weave|woven|silk|solid|plain)\b", re.I)
+
+
+def tokens(s):
+ return set(re.findall(r"[a-z]+", (s or "").lower()))
+
+
+def main():
+ surv = [json.loads(l) for l in open(IN)
+ if json.loads(l).get("gate") == "SURVIVOR"]
+ out = []
+ for r in surv:
+ declared = (r["declared_color"] or "")
+ dl = declared.lower()
+ vis = (r.get("vision_dominant") or "").lower()
+ vis_tokens = [t for t in re.findall(r"[a-z]+", vis) if t in BASIC]
+
+ name_leak = any(t in dl for t in vis_tokens)
+ is_busy = bool(BUSY_RE.search(declared))
+ is_solid = bool(SOLID_RE.search(declared))
+ # vision reporting >1 color => inherently multi-ink => can't be a wrong-solid
+ # defect; ambiguous busy pattern -> HOLD (a real wrong-image shows ONE color)
+ multi_vis = bool(re.search(r"\band\b|multi|gradient|/", vis))
+ conf = r.get("vision_confidence") or 0
+
+ if name_leak:
+ disp = "DROP_NAME_LEAK"
+ why = f"declared name already contains '{[t for t in vis_tokens if t in dl][0]}'"
+ elif multi_vis:
+ disp = "HOLD_BUSY"
+ why = "vision sees multiple colors -> multi-ink/busy -> human review"
+ elif is_solid and not is_busy:
+ disp = "RECRAWL"
+ why = "solid/textured line expected near-solid; image is clearly off-color"
+ elif is_busy:
+ disp = "HOLD_BUSY"
+ why = "busy/botanical pattern; declared may name one ink -> human review"
+ elif conf >= 0.95:
+ disp = "RECRAWL"
+ why = "non-busy ground, high-confidence color contradiction"
+ else:
+ disp = "HOLD_BUSY"
+ why = "ambiguous; conservative hold"
+ r2 = dict(r)
+ r2["disposition"] = disp
+ r2["disposition_reason"] = why
+ out.append(r2)
+
+ out.sort(key=lambda r: (r["disposition"], r["vendor"], r["dw_sku"]))
+ with open(os.path.join(ROOT, "out", "survivor_dispositions.jsonl"), "w") as f:
+ for r in out:
+ f.write(json.dumps(r) + "\n")
+ import csv
+ fields = ["disposition", "vendor", "dw_sku", "mfr_sku", "declared_color",
+ "vision_dominant", "vision_confidence", "disposition_reason",
+ "image_url", "handle"]
+ with open(os.path.join(ROOT, "out", "survivor_dispositions.csv"), "w", newline="") as f:
+ w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
+ w.writeheader()
+ for r in out:
+ w.writerow(r)
+
+ from collections import Counter
+ c = Counter(r["disposition"] for r in out)
+ print("DISPOSITIONS:", dict(c))
+ for disp in ["RECRAWL", "HOLD_BUSY", "DROP_NAME_LEAK"]:
+ print(f"\n=== {disp} ===")
+ for r in out:
+ if r["disposition"] == disp:
+ print(f" {r['vendor'][:16]:16} {r['dw_sku']:14} {r['declared_color'][:40]:40} "
+ f"vis={r.get('vision_dominant'):20} ({r['disposition_reason']})")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/vision_gate.py b/scripts/vision_gate.py
new file mode 100644
index 0000000..a24933d
--- /dev/null
+++ b/scripts/vision_gate.py
@@ -0,0 +1,181 @@
+#!/usr/bin/env python3
+"""
+Gemini-vision YES/NO gate for the non-PR Tier-1 color-image defect set (DTD Q1=b).
+For each row: fetch image_url, ask Gemini 2.0 Flash whether the wallcovering's
+DOMINANT color is consistent with the declared colorway name, or clearly different.
+Drops family-parser / busy-pattern FALSE POSITIVES; keeps real-defect candidates.
+
+Read-only against catalog (only GETs the public Shopify image + calls Gemini).
+Writes results to out/vision_gate_results.jsonl + out/vision_gate_summary.csv.
+"""
+import csv, json, os, sys, time, base64, urllib.request, urllib.error, re
+
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+CSV_IN = os.path.join(ROOT, "out", "tier1_confirmed_defects.csv")
+JSONL_OUT = os.path.join(ROOT, "out", "vision_gate_results.jsonl")
+CSV_OUT = os.path.join(ROOT, "out", "vision_gate_summary.csv")
+V2_JSONL = os.path.join(ROOT, "full", "reextracted_full.jsonl")
+
+MODEL = "gemini-2.5-flash-lite"
+COST_PER_CALL = 0.0006 # ~$ per product image (always-show-costs rule)
+
+def get_key():
+ env = os.path.expanduser("~/Projects/secrets-manager/.env")
+ with open(env) as f:
+ for line in f:
+ if line.startswith("GEMINI_API_KEY="):
+ return line.split("=", 1)[1].strip().strip('"').strip("'")
+ sys.exit("no GEMINI_API_KEY")
+
+API_KEY = get_key()
+URL = f"https://generativelanguage.googleapis.com/v1beta/models/{MODEL}:generateContent?key={API_KEY}"
+
+def load_v2_verdicts():
+ """Map image_url -> v2 verdict so we can tag survivors-of-v2 vs cleared."""
+ m = {}
+ if not os.path.exists(V2_JSONL):
+ return m
+ with open(V2_JSONL) as f:
+ for line in f:
+ try:
+ r = json.loads(line)
+ m[r.get("image_url")] = r.get("verdict")
+ except Exception:
+ pass
+ return m
+
+def fetch_image(url, tries=3):
+ for i in range(tries):
+ try:
+ req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
+ with urllib.request.urlopen(req, timeout=30) as r:
+ data = r.read()
+ ct = r.headers.get("Content-Type", "image/jpeg")
+ if "image" not in ct:
+ ct = "image/jpeg"
+ return data, ct
+ except Exception as e:
+ if i == tries - 1:
+ return None, str(e)
+ time.sleep(2)
+ return None, "exhausted"
+
+PROMPT_TMPL = (
+ "This is a product image for a wallcovering/fabric that is merchandised online "
+ "under the color name \"{declared}\". Look ONLY at the actual wallcovering/fabric "
+ "surface (ignore any white studio background or border). "
+ "Is the DOMINANT color of the wallcovering consistent with the name \"{declared}\", "
+ "or is the image CLEARLY a different color than that name implies? "
+ "A multi-color or busy pattern where \"{declared}\" is plausibly one of the inks "
+ "counts as CONSISTENT (matches_declared=true). Only answer false when the image is "
+ "unmistakably the wrong color for that name. "
+ "Respond with STRICT JSON only, no prose: "
+ '{{"dominant_color":"<plain english>","matches_declared":true|false,'
+ '"confidence":<0.0-1.0>,"note":"<=12 words"}}'
+)
+
+def ask_gemini(img_bytes, ct, declared):
+ prompt = PROMPT_TMPL.format(declared=declared)
+ body = {
+ "contents": [{
+ "parts": [
+ {"text": prompt},
+ {"inline_data": {"mime_type": ct, "data": base64.b64encode(img_bytes).decode()}},
+ ]
+ }],
+ "generationConfig": {"temperature": 0, "maxOutputTokens": 200},
+ }
+ req = urllib.request.Request(
+ URL, data=json.dumps(body).encode(),
+ headers={"Content-Type": "application/json"})
+ for i in range(3):
+ try:
+ with urllib.request.urlopen(req, timeout=60) as r:
+ resp = json.loads(r.read())
+ txt = resp["candidates"][0]["content"]["parts"][0]["text"]
+ mt = re.search(r"\{.*\}", txt, re.S)
+ if mt:
+ return json.loads(mt.group(0)), None
+ return None, "no-json:" + txt[:120]
+ except urllib.error.HTTPError as e:
+ err = e.read().decode()[:200]
+ if e.code == 429:
+ time.sleep(15)
+ continue
+ if i == 2:
+ return None, f"http{e.code}:{err}"
+ time.sleep(3)
+ except Exception as e:
+ if i == 2:
+ return None, str(e)[:160]
+ time.sleep(3)
+ return None, "exhausted"
+
+def main():
+ v2 = load_v2_verdicts()
+ rows = []
+ with open(CSV_IN) as f:
+ for r in csv.DictReader(f):
+ if r["vendor"] != "Phillipe Romano":
+ rows.append(r)
+ print(f"non-PR Tier-1 rows: {len(rows)}", file=sys.stderr)
+
+ results = []
+ calls = 0
+ for i, r in enumerate(rows, 1):
+ declared = r["declared_color"]
+ img_bytes, ct = fetch_image(r["image_url"])
+ rec = {
+ "vendor": r["vendor"], "dw_sku": r["dw_sku"], "mfr_sku": r["mfr_sku"],
+ "declared_color": declared, "declared_family": r["declared_family"],
+ "image_family": r["real_image_family"], "image_hex": r["real_image_hex"],
+ "image_url": r["image_url"], "handle": r["handle"],
+ "v2_verdict": v2.get(r["image_url"]),
+ }
+ if img_bytes is None:
+ rec.update({"gate": "ERROR", "error": "fetch:" + str(ct)})
+ results.append(rec)
+ print(f"[{i}/{len(rows)}] FETCH-ERR {r['vendor']} {r['dw_sku']}", file=sys.stderr)
+ continue
+ out, err = ask_gemini(img_bytes, ct, declared)
+ calls += 1
+ if out is None:
+ rec.update({"gate": "ERROR", "error": err})
+ else:
+ matches = bool(out.get("matches_declared"))
+ rec.update({
+ "gate": "FALSE_POSITIVE" if matches else "SURVIVOR",
+ "vision_dominant": out.get("dominant_color"),
+ "vision_matches": matches,
+ "vision_confidence": out.get("confidence"),
+ "vision_note": out.get("note"),
+ })
+ results.append(rec)
+ g = rec.get("gate")
+ print(f"[{i}/{len(rows)}] {g:14s} {r['vendor'][:18]:18s} {r['dw_sku']:14s} "
+ f"declared='{declared[:30]}' vision='{rec.get('vision_dominant','')}'", file=sys.stderr)
+ time.sleep(0.3)
+
+ with open(JSONL_OUT, "w") as f:
+ for r in results:
+ f.write(json.dumps(r) + "\n")
+ fields = ["vendor", "dw_sku", "mfr_sku", "declared_color", "declared_family",
+ "image_family", "v2_verdict", "gate", "vision_dominant",
+ "vision_matches", "vision_confidence", "vision_note", "image_url", "handle"]
+ with open(CSV_OUT, "w", newline="") as f:
+ w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
+ w.writeheader()
+ for r in results:
+ w.writerow(r)
+
+ surv = [r for r in results if r.get("gate") == "SURVIVOR"]
+ fp = [r for r in results if r.get("gate") == "FALSE_POSITIVE"]
+ errs = [r for r in results if r.get("gate") == "ERROR"]
+ print("\n==== VISION GATE SUMMARY ====", file=sys.stderr)
+ print(f"total={len(results)} SURVIVOR={len(surv)} FALSE_POSITIVE={len(fp)} ERROR={len(errs)}", file=sys.stderr)
+ print(f"gemini calls={calls} est cost=${calls*COST_PER_CALL:.4f}", file=sys.stderr)
+ from collections import Counter
+ print("survivors by vendor:", dict(Counter(r["vendor"] for r in surv)), file=sys.stderr)
+
+if __name__ == "__main__":
+ main()
← 8c15981 v2 background-suppressed full-catalog re-extract + staging l
·
back to Dw Color Image Audit
·
track vision-gate result artifacts (summary + dispositions) db0ed7e →