`scripts/reextract.py`

139 lines
#!/usr/bin/env python3
"""Re-extract TRUE dominant color from real product images and classify vs declared name.
Reads TSV lines on stdin: vendor, handle, dwsku, mfrsku, title, cached_hex, image_url
Writes JSON-lines on stdout. Local, free. Center-crops to avoid white borders."""
import sys, json, io, urllib.request, colorsys
from collections import Counter
try:
    from PIL import Image
except Exception as e:
    sys.stderr.write("PIL missing\n"); sys.exit(1)

CHROM = ["red","orange","yellow","chartreuse","green","teal","cyan","blue","indigo","violet","magenta","pink"]
WHEEL = CHROM
def fam(r,g,b):
    h,l,s = colorsys.rgb_to_hls(r/255,g/255,b/255)
    spread = (max(r,g,b)-min(r,g,b))/255
    if l>0.86 and spread<0.22: return "white", s, l
    if l<0.16 and spread<0.22: return "black", s, l
    if s<0.15: return "gray", s, l
    H = h*360
    if s<0.45 and 18<=H<=65: return ("beige" if l>0.55 else "brown"), s, l
    bands=[(345,361,"red"),(0,15,"red"),(15,40,"orange"),(40,67,"yellow"),(67,95,"chartreuse"),
           (95,150,"green"),(150,175,"teal"),(175,195,"cyan"),(195,240,"blue"),(240,265,"indigo"),
           (265,290,"violet"),(290,320,"magenta"),(320,345,"pink")]
    for a,b2,f in bands:
        if a<=H<b2: return f, s, l
    return "red", s, l

NAME_FAMILY = json.loads(open(sys.argv[1]).read()) if len(sys.argv)>1 else {}

# Substrate/finish words that name the material, never the colorway. As a trailing token
# "leather"->brown used to hijack the family (DW titles read "Pattern - Colorway Material").
# linen/silk/raffia/jute/grasscloth intentionally excluded — they double as beige colorways.
MATERIAL_STOPWORDS = {"vegan","leather","suede","vinyl","velvet","wallcovering","wallpaper",
                      "commercial","contract","fabric","paper","cloth","faux","textile","weave"}
# Multi-word colorways whose family != any single token's (sea-green is teal, per the lexicon's
# seafoam/seaglass->teal). Checked before single tokens.
MULTIWORD_FAMILY = {"sea green":"teal","sea foam":"teal","sea glass":"teal","sea blue":"teal"}
# Botanical/place/motif homonyms that ARE in the lexicon but, in a DW title, almost always name
# the PATTERN/theme/material, not the colorway. They only mislead on the NO-DASH / whole-title
# path; on the clean after-dash colorway segment the word genuinely IS the colorway, so demotion
# is gated to the no-dash path (see name_family). Demotion only bites when the homonym is the SOLE
# color signal -> fall through to UNKNOWN (DTD verdict A 3/3, 2026-06-21). Mirror of JS.
AMBIGUOUS_COLORWORDS = {"grass","ocean","marine","leaf","apple","forest"}

def _colorway_segment(name):
    s=(name or "").strip(); i=s.rfind(" - ")
    return s[i+3:] if i>=0 else s

def _family_from_phrase(phrase, demote_ambiguous=False):
    toks=''.join(c.lower() if c.isalpha() or c==' ' else ' ' for c in phrase).split()
    mw=None
    for i in range(len(toks)-1):
        bg=toks[i]+" "+toks[i+1]
        if bg in MULTIWORD_FAMILY: mw=MULTIWORD_FAMILY[bg]
    if mw: return mw
    mappable=[t for t in toks if t in NAME_FAMILY]
    non_mat=[t for t in mappable if t not in MATERIAL_STOPWORDS]
    pool=non_mat if non_mat else mappable
    if demote_ambiguous:
        non_amb=[t for t in pool if t not in AMBIGUOUS_COLORWORDS]
        if non_amb: pool=non_amb        # a real colorway survives -> use it
        else: return None               # only pattern/place/material homonyms -> UNKNOWN
    fam_=None
    for t in pool: fam_=NAME_FAMILY[t]
    return fam_

def name_family(name):
    seg=_colorway_segment(name)
    has_dash = seg != (name or "").strip()
    # After-dash colorway is the declared colorway verbatim -> no demotion ("X - Ocean" IS teal).
    # No-dash whole title mixes pattern+colorway -> demote ambiguous homonyms.
    fam_=_family_from_phrase(seg, demote_ambiguous=not has_dash)
    if fam_ is None and has_dash:
        fam_=_family_from_phrase(name, demote_ambiguous=True)   # fallback over whole title
    return fam_

def is_chrom(f): return f in CHROM
def hue_dist(a,b):
    ia,ib=WHEEL.index(a),WHEEL.index(b); d=abs(ia-ib); return min(d,12-d)

def classify(df, imf, chroma, l):
    if not df: return "UNKNOWN","name-unmapped"
    if df==imf: return "OK","same-family"
    dC, iC = is_chrom(df), is_chrom(imf)
    if dC and iC:
        d=hue_dist(df,imf)
        if d>=2: return "HIGH", f"non-adjacent hue (dist {d})"
        return "REVIEW", f"adjacent hue (dist {d})"
    if dC != iC:
        if dC and imf=="white": return "REVIEW","chromatic name vs pale image (light-ground)"
        if dC and imf in ("gray","black"): return "HIGH","chromatic name, dark-neutral image"
        if dC and imf in ("beige","brown"):
            if df in ("blue","indigo","violet","cyan","teal","green"): return "HIGH","cool color vs warm-neutral image"
            return "REVIEW","warm color vs warm-neutral image"
        if not dC and iC and chroma>0.45: return "HIGH","neutral name, strongly chromatic image"
        return "REVIEW","chromatic/neutral boundary"
    # both neutral
    NG={"white":"light","gray":"mid","black":"dark","beige":"warm","brown":"warm"}
    if NG.get(df)==NG.get(imf): return "OK","same neutral group"
    if {df,imf}=={"white","black"}: return "HIGH","opposite-value neutrals"
    return "REVIEW",f"neutral shift {df}->{imf}"

def dominant(url):
    req=urllib.request.Request(url,headers={"User-Agent":"Mozilla/5.0"})
    data=urllib.request.urlopen(req,timeout=25).read()
    im=Image.open(io.BytesIO(data)).convert("RGB")
    w,h=im.size; cw,ch=int(w*0.3),int(h*0.3)
    im2=im.crop((cw,ch,w-cw,h-ch)).resize((60,60))
    px=list(im2.getdata())
    # ignore near-white border bleed: drop pixels that are near-white when a chromatic majority exists
    q=Counter(tuple(c//20*20 for c in p) for p in px)
    # weighted: prefer the most-saturated frequent bucket among top 5
    top=q.most_common(6)
    best=top[0][0]; best_score=-1
    for (col,cnt) in top:
        r,g,b=col; s=(max(col)-min(col))/255.0
        score=cnt*(0.4+s)   # weight frequency but boost chromatic
        if score>best_score: best_score=score; best=col
    return best

for line in sys.stdin:
    line=line.rstrip("\n")
    if not line.strip(): continue
    parts=line.split("\t")
    if len(parts)<7: continue
    vendor,handle,dwsku,mfrsku,title,cached,url=parts[:7]
    declared=title.split("|")[0].strip()
    df=name_family(declared)
    rec={"vendor":vendor,"handle":handle,"dwsku":dwsku,"mfrsku":mfrsku,"title":title,"declared":declared,"cached_hex":cached,"image_url":url,"declaredFamily":df}
    try:
        dom=dominant(url); imf,s,l=fam(*dom)
        rec["real_hex"]="#%02X%02X%02X"%dom; rec["imageFamily"]=imf
        v,reason=classify(df,imf,s,l)
        rec["verdict"]=v; rec["reason"]=reason
    except Exception as e:
        rec["verdict"]="ERROR"; rec["reason"]=f"{type(e).__name__}:{str(e)[:50]}"
    sys.stdout.write(json.dumps(rec)+"\n")