← back to Dw Color Image Audit

scripts/reextract2.py

175 lines

#!/usr/bin/env python3
"""Full-catalog TRUE dominant-color re-extraction (v2), local + free (PIL only).
Reads TSV on stdin: vendor, handle(=shopify_product_id), dwsku(=status), mfrsku, title, cached_hex, image_url
Writes JSON-lines on stdout.

Two extractions per (single) downloaded image:
  real_hex_v1 : original center-crop + saturation-weighted dominant (scripts/reextract.py method)
  real_hex    : v2 = same, but near-white/near-black PAPER/BACKGROUND pixels are dropped
                before choosing the swatch color (falls back to v1 only when the image is
                essentially all-neutral). Directly attacks the '#F2F2F2 white-bg capture' defect.
Classification (declared color name vs real image family) uses the v2 hex.
"""
import sys, json, io, urllib.request, colorsys
from collections import Counter
try:
    from PIL import Image
except Exception:
    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"
    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 _weighted_dominant(buckets):
    """Pick the saturation-boosted most-frequent bucket among the top 6."""
    top = buckets.most_common(6)
    best = top[0][0]; best_score = -1
    for (col, cnt) in top:
        s = (max(col)-min(col))/255.0
        score = cnt*(0.4+s)
        if score > best_score: best_score = score; best = col
    return best

def _is_paper(col):
    """Near-white paper/background or near-black: not a real swatch color."""
    r,g,b = col
    spread = (max(col)-min(col))/255.0
    mx = max(col)/255.0; mn = min(col)/255.0
    if mx > 0.90 and spread < 0.10: return True   # paper white
    if mn < 0.08 and spread < 0.10: return True   # ink black
    return False

def extract(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())
    q = Counter(tuple(c//20*20 for c in p) for p in px)
    # v1: original method (background-naive)
    dom_v1 = _weighted_dominant(q)
    # v2: drop paper/ink background buckets, then pick swatch dominant
    q2 = Counter({c:n for c,n in q.items() if not _is_paper(c)})
    total = sum(q.values()); kept = sum(q2.values())
    if kept >= max(1, int(total*0.12)):   # enough non-background content to trust
        dom_v2 = _weighted_dominant(q2)
    else:
        dom_v2 = dom_v1                    # essentially all white/black -> keep v1
    return dom_v1, dom_v2

def _main():
    for line in sys.stdin:
        line = line.rstrip("\n")
        if not line.strip(): continue
        parts = line.split("\t")
        if len(parts) < 7: continue
        vendor,spid,status,mfrsku,title,cached,url = parts[:7]
        declared = title.split("|")[0].strip()
        df = name_family(declared)
        rec = {"shopify_product_id":spid, "vendor":vendor, "status":status, "title":title,
               "declared":declared, "cached_hex":cached, "image_url":url, "declaredFamily":df}
        try:
            v1, v2 = extract(url)
            imf, s, l = fam(*v2)
            rec["real_hex_v1"] = "#%02X%02X%02X" % v1
            rec["real_hex"]    = "#%02X%02X%02X" % v2
            rec["imageFamily"] = imf
            verdict, reason = classify(df, imf, s, l)
            rec["verdict"] = verdict; 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")

if __name__ == "__main__":
    _main()