← back to Dw Color Image Audit
scripts/pr_raw930_vision_spotcheck.py
239 lines
#!/usr/bin/env python3
"""
PR raw-930 vision spot-check (CLOSE-OUT measurement for the disposition memo).
Selects the ~40 STRONGEST chromatic PR raw-HIGH candidates under the CORRECTED mapper
and runs a Gemini 2.5-flash-lite YES/NO vision check implementing the memo's exact
genuine-defect test:
genuine wrong-IMAGE defect <=> the image's dominant color is inconsistent with
BOTH (a) the DW declared color name AND (b) the upstream-source color word parsed
from the product image filename.
READ-ONLY: only GETs the public Shopify image + calls Gemini. Writes NOTHING to
dw_unified / Shopify. Privacy: only color WORDS (emerald/ocean/...) from the public
image filename are used; the upstream private-label company name is never referenced.
"""
import csv, json, os, sys, time, base64, urllib.request, urllib.error, re, subprocess
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__))))
import reextract2 as M # corrected mapper: name_family, fam, classify, is_chrom, hue_dist
# reextract2 loads its color lexicon from sys.argv[1] at import; on import that's empty,
# so hydrate it explicitly from the shipped lexicon file (else every name maps to None).
_LEX = os.path.join(os.path.dirname(os.path.abspath(__file__)), "name_family.json")
M.NAME_FAMILY = json.loads(open(_LEX).read())
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
OUT_JSONL = os.path.join(ROOT, "results", "pr_raw930_vision_spotcheck.jsonl")
OUT_CSV = os.path.join(ROOT, "results", "pr_raw930_vision_spotcheck.csv")
MODEL = "gemini-2.5-flash-lite"
COST_PER_CALL = 0.0006
N_TARGET = 40
PR_VENDORS = ("Phillipe Romano","Philippe Romano Fabrics","Phillipe Romano Wide Textures",
"Phillipe Romano Naturals","Phillipe Romano Metals","Phillipe Romano Vinyls",
"Phillipe Romano Wallpaper","Phillipe Romano Type 2 Vinyls") # excl Maya Romanoff
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 hex2rgb(h):
h=h.lstrip('#'); return tuple(int(h[i:i+2],16) for i in (0,2,4))
def filename_color_words(image_url):
"""Color tokens from the public image filename slug (the upstream-source color word)."""
base = image_url.split('/')[-1].split('?')[0]
base = re.sub(r'\.(jpg|jpeg|png|webp|gif)$','',base,flags=re.I)
# strip trailing shopify uuid / version hashes
base = re.sub(r'_[0-9a-f]{8}(-[0-9a-f]{4}){3}-[0-9a-f]{12}.*$','',base,flags=re.I)
toks = re.split(r'[-_]', base.lower())
words=[]
for t in toks:
if t in M.NAME_FAMILY and t not in M.MATERIAL_STOPWORDS:
words.append(t)
# dedupe, preserve order
seen=set(); out=[]
for w in words:
if w not in seen: seen.add(w); out.append(w)
return out
def pull_rows():
vlist = ",".join("'"+v.replace("'","''")+"'" for v in PR_VENDORS)
q = (f"SELECT shopify_product_id, vendor, declared, real_hex, image_url "
f"FROM product_colors_v2 WHERE verdict='HIGH' AND lower(status)='active' "
f"AND vendor IN ({vlist}) AND real_hex IS NOT NULL AND real_hex<>'';")
p = subprocess.run(["psql","-d","dw_color_staging","-F","\t","-A","-t","-c",q],
capture_output=True, text=True)
if p.returncode!=0: sys.exit("psql: "+p.stderr)
rows=[]
for line in p.stdout.splitlines():
if not line.strip(): continue
spid,vendor,declared,real_hex,image_url = line.split("\t")
rows.append(dict(spid=spid,vendor=vendor,declared=declared,
real_hex=real_hex,image_url=image_url))
return rows
def declared_color_word(declared):
seg = M._colorway_segment(declared)
return seg.strip()
def select_strongest(rows):
cand=[]
seen=set()
for r in rows:
if r["spid"] in seen: continue
try: rgb=hex2rgb(r["real_hex"])
except Exception: continue
imf,s,l = M.fam(*rgb)
chroma=(max(rgb)-min(rgb))/255.0
df = M.name_family(r["declared"])
if not df: continue
verdict,reason = M.classify(df,imf,chroma,l)
if verdict!="HIGH": continue
dashed = " - " in r["declared"]
dC, iC = M.is_chrom(df), M.is_chrom(imf)
# strength score: pure chromatic-vs-chromatic conflict is the strongest defect signal
if dC and iC:
strength = 1000 + M.hue_dist(df,imf)*10 + (5 if dashed else 0)
elif dC and imf in ("beige","brown") and df in ("blue","indigo","violet","cyan","teal","green"):
strength = 500 + int(chroma*100) + (5 if dashed else 0) # cool name vs warm-neutral
elif dC and imf in ("gray","black"):
strength = 300 + int(chroma*100) + (5 if dashed else 0) # chromatic name, dark-neutral
else:
strength = 100 + int(chroma*50)
seen.add(r["spid"])
cand.append(dict(**r, declared_family=df, image_family=imf,
chroma=round(chroma,3), classify_reason=reason,
dashed=dashed, strength=strength,
declared_word=declared_color_word(r["declared"]),
filename_words=filename_color_words(r["image_url"])))
cand.sort(key=lambda c:(-c["strength"], -c["chroma"]))
return cand[:N_TARGET]
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 photo of a wallcovering/fabric. Look ONLY at the actual "
"wallcovering/fabric surface and IGNORE any white studio background, border, or "
"room-scene props. The product is merchandised under the color name \"{declared}\".{alt} "
"Question: is the DOMINANT color of the wallcovering itself plausibly consistent with "
"the declared color name \"{declared_word}\"{alt_q}? A busy/multi-color pattern where that "
"color is plausibly one of the inks counts as CONSISTENT. Answer matches=false ONLY if the "
"wallcovering is unmistakably a different color than ANY of those names imply. "
"Respond with STRICT JSON only, no prose: "
'{{"dominant_color":"<plain english>","matches":true|false,'
'"which_name_matched":"<declared|alt|none>","confidence":<0.0-1.0>,"note":"<=12 words"}}'
)
def ask_gemini(img_bytes, ct, declared, declared_word, alt_words):
alts = [w for w in alt_words if w.lower()!=declared_word.lower()]
if alts:
alt = (" The same item also carries the alternate color term"
f" \"{', '.join(alts)}\" in its source filename.")
alt_q = f" OR the alternate color term \"{', '.join(alts)}\""
else:
alt=""; alt_q=""
prompt = PROMPT_TMPL.format(declared=declared, declared_word=declared_word, alt=alt, alt_q=alt_q)
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():
rows=pull_rows()
print(f"PR-family raw-HIGH live rows pulled: {len(rows)}", file=sys.stderr)
cand=select_strongest(rows)
print(f"strongest chromatic candidates selected: {len(cand)}", file=sys.stderr)
results=[]; calls=0
for i,c in enumerate(cand,1):
img,ct=fetch_image(c["image_url"])
rec=dict(c)
if img is None:
rec.update(gate="ERROR", error="fetch:"+str(ct))
else:
out,err=ask_gemini(img,ct,c["declared"],c["declared_word"],c["filename_words"])
calls+=1
if out is None:
rec.update(gate="ERROR", error=err)
else:
matches=bool(out.get("matches"))
rec.update(gate="DEFECT" if not matches else "OK_CONSISTENT",
vision_dominant=out.get("dominant_color"),
vision_matches=matches,
which_matched=out.get("which_name_matched"),
vision_confidence=out.get("confidence"),
vision_note=out.get("note"))
results.append(rec)
print(f"[{i}/{len(cand)}] {rec.get('gate'):13s} str={c['strength']:4d} "
f"{c['vendor'][:20]:20s} declared='{c['declared'][:34]:34s}' "
f"df={c['declared_family']:8s} imf={c['image_family']:7s} "
f"alt={','.join(w for w in c['filename_words'] if w!=c['declared_word'].lower()) or '-':10s} "
f"vision='{rec.get('vision_dominant','')}'", file=sys.stderr)
time.sleep(0.3)
os.makedirs(os.path.dirname(OUT_JSONL), exist_ok=True)
with open(OUT_JSONL,"w") as f:
for r in results: f.write(json.dumps(r)+"\n")
fields=["spid","vendor","declared","declared_word","declared_family","image_family",
"real_hex","chroma","dashed","strength","classify_reason","filename_words",
"gate","vision_dominant","vision_matches","which_matched","vision_confidence",
"vision_note","image_url"]
with open(OUT_CSV,"w",newline="") as f:
w=csv.DictWriter(f,fieldnames=fields,extrasaction="ignore")
w.writeheader()
for r in results:
rr=dict(r); rr["filename_words"]="|".join(r.get("filename_words",[]))
w.writerow(rr)
defects=[r for r in results if r.get("gate")=="DEFECT"]
okc=[r for r in results if r.get("gate")=="OK_CONSISTENT"]
errs=[r for r in results if r.get("gate")=="ERROR"]
print("\n==== PR RAW-930 VISION SPOT-CHECK SUMMARY ====", file=sys.stderr)
print(f"candidates={len(results)} OK_CONSISTENT={len(okc)} DEFECT={len(defects)} ERROR={len(errs)}", file=sys.stderr)
print(f"gemini model={MODEL} calls={calls} actual cost=${calls*COST_PER_CALL:.4f}", file=sys.stderr)
if defects:
print("\n--- GENUINE WRONG-IMAGE DEFECT CANDIDATES ---", file=sys.stderr)
for d in defects:
print(f" SKU/spid={d['spid']} declared='{d['declared']}' "
f"vision='{d.get('vision_dominant')}' conf={d.get('vision_confidence')} "
f"url={d['image_url']}", file=sys.stderr)
print(json.dumps(dict(candidates=len(results),ok_consistent=len(okc),
defects=len(defects),errors=len(errs),calls=calls,
actual_cost_usd=round(calls*COST_PER_CALL,4))))
if __name__=="__main__":
main()