`tools/pattern-id-titlefirst-final.py`

157 lines
#!/usr/bin/env python3
"""WS-1 FINAL: TITLE-FIRST + COLOR-INVARIANT IMAGE-CONFIRMED pattern_id.

Same title-first derivation as the dry-run, but the image-confirm step uses the
COLOR-INVARIANT signature pair from tools/ci-hashes.jsonl (dHash + tonal inverse),
so light-on-dark vs dark-on-light colorways of one motif stay merged while genuinely
different designs sharing a title split apart.

Two products are the same pattern when:
    min( ham(A.dh,B.dh), ham(A.inv,B.inv), ham(A.dh,B.inv), ham(A.inv,B.dh) ) <= THRESH

Default THRESH=12 (the pattern-image-verify.py value tuned for the CI pair).

Modes:
  (default) print before/after diff + worst-offender spotlight
  --emit    write tools/pattern-id-titlefirst-by-dwsku.json {dw_sku:{pid,sid}}
            for the gated re-backfill (sid carried over from the old map).

Input on stdin: dw_sku<TAB>vendor<TAB>title<TAB>mfr_sku<TAB>image_url
"""
import sys, json, re, os
from collections import defaultdict, Counter

THRESH = int(os.environ.get("HAM_THRESH", "30"))  # WS-1 chosen 2026-06-26: lowest TH that fully merges Yakatore(1)+La Corka(1); over-merge flat at 42, under-merge 5682->559
HERE = os.path.dirname(__file__)
CI_PATH  = os.path.join(HERE, "ci-hashes.jsonl")
OLD_MAP  = os.path.join(HERE, "pattern-id-by-dwsku.json")
EMIT = "--emit" in sys.argv
EMIT_PATH = os.path.join(HERE, "pattern-id-titlefirst-by-dwsku.json")

NOISE = ['wallcovering','wallcoverings','wallpaper','wall covering','fabric','fabrics',
    'durable vinyl','vinyl','grasscloth','print','commercial','residential','type ii',
    'type iii','mural','wall mural','peel and stick','peel-and-stick','self adhesive',
    'self-adhesive','by phillipe romano','phillipe romano']
CW = set('light dark deep pale soft warm cool muted bright antique metallic natural neutral multi multicolor multicolour black white grey gray silver gold beige cream ivory taupe brown tan red crimson burgundy pink rose blush coral orange rust terracotta yellow ochre mustard green olive sage emerald teal aqua turquoise blue navy indigo cobalt azure purple violet lilac lavender plum mauve charcoal slate sand stone smoke linen pearl champagne bronze copper platinum marine lawn fog snow azalea cloud concord cyan dandelion mushroom seaweed'.split())
DENY = set(['','various','assorted','sample','default','tbd','test','n a','na'])

def nt(t):
    s = t.split('|')[0].lower()
    for n in NOISE: s = s.replace(n, ' ')
    s = re.sub(r'[^a-z0-9 ]+', ' ', s); s = re.sub(r'\s+', ' ', s).strip()
    return ' '.join(w for w in s.split() if w not in CW).strip() or s
def vs(v):
    s = re.sub(r'[^a-z0-9]+', '-', (v or '').lower()).strip('-'); return s or 'no-vendor'
def mfr_root(raw):
    if not raw: return None
    s = str(raw).strip().upper()
    toks = [t for t in re.split(r'[-/_.\s]+', s) if t]
    if len(toks) >= 2 and re.fullmatch(r'\d{1,3}[A-Z]?', toks[-1]): return '-'.join(toks[:-1])
    return '-'.join(toks) or None
def title_key(ven, title, mfr):
    t2 = nt(title)
    if t2 and t2 not in DENY and len(t2) >= 3: return f"{vs(ven)}::t2::{t2}"
    r = mfr_root(mfr)
    return f"{vs(ven)}::mfr::{r}" if r else f"{vs(ven)}::t2::{t2 or 'ungrouped'}"

def ham(a, b): return bin(int(a,16) ^ int(b,16)).count("1")
def ci_dist(A, B):
    return min(ham(A[0],B[0]), ham(A[1],B[1]), ham(A[0],B[1]), ham(A[1],B[0]))

# load color-invariant hashes by url
ci = {}
if os.path.exists(CI_PATH):
    for line in open(CI_PATH):
        try:
            r = json.loads(line)
            if r.get("status") == "ok": ci[r["url"]] = (r["dhash"], r["inv"])
        except Exception: pass

old = {}
try:
    for k, v in json.load(open(OLD_MAP)).items():
        if v and v.get("pid"): old[k] = (v["pid"], v.get("sid"))
except Exception: pass

rows = []
for line in sys.stdin:
    p = line.rstrip("\n").split("\t")
    if len(p) < 5: continue
    dw, ven, title, mfr, url = p[0], p[1], p[2], p[3], p[4]
    if not dw: continue
    rows.append((dw, ven, title, mfr, url, ci.get(url)))

# stage 1: title key
tkey = {r[0]: title_key(r[1], r[2], r[3]) for r in rows}
bykey = defaultdict(list)
for r in rows: bykey[tkey[r[0]]].append(r)

# stage 2: color-invariant confirm within each title group
final = {}; img_splits = 0; conf = 0; unconf = 0
for key, members in bykey.items():
    hashed = [m for m in members if m[5]]
    nohash = [m for m in members if not m[5]]
    conf += len(hashed); unconf += len(nohash)
    if len(members) == 1 or len(hashed) <= 1:
        for m in members: final[m[0]] = key
        continue
    clusters = []  # (rep_sig, [skus])
    for m in hashed:
        placed = False
        for cl in clusters:
            if ci_dist(m[5], cl[0]) <= THRESH: cl[1].append(m[0]); placed = True; break
        if not placed: clusters.append((m[5], [m[0]]))
    if len(clusters) == 1:
        for m in members: final[m[0]] = key
    else:
        img_splits += 1
        clusters.sort(key=lambda c: -len(c[1]))
        for ci_i, (rep, sk) in enumerate(clusters):
            sub = key if ci_i == 0 else f"{key}#v{ci_i+1}"
            for s in sk: final[s] = sub
        for m in nohash: final[m[0]] = key  # unhashed -> largest cluster

def stats(getkey):
    k2t = defaultdict(set); vt2k = defaultdict(set)
    for (dw, ven, title, mfr, url, sig) in rows:
        k = getkey(dw)
        if not k: continue
        n = nt(title)
        if n: k2t[k].add(n); vt2k[(ven.lower(), n)].add(k)
    return (len(k2t), sum(1 for ts in k2t.values() if len(ts) > 1),
            sum(1 for ks in vt2k.values() if len(ks) > 1))

b = stats(lambda dw: old.get(dw, (None,))[0])
a = stats(lambda dw: final.get(dw))
print("="*70)
print("TITLE-FIRST + COLOR-INVARIANT IMAGE-CONFIRMED  —  FINAL (no writes)")
print(f"rows {len(rows)} · THRESH {THRESH} · CI-hashed {conf} / title-only {unconf} "
      f"({conf*100//max(1,len(rows))}%) · image splits {img_splits}")
print("="*70)
print(f"{'metric':<42}{'BEFORE':>14}{'AFTER':>12}")
print(f"{'  distinct pattern tiles':<42}{b[0]:>14}{a[0]:>12}")
print(f"{'  OVER-merge keys (diff designs lumped)':<42}{b[1]:>14}{a[1]:>12}")
print(f"{'  UNDER-merge (1 design fragmented)':<42}{b[2]:>14}{a[2]:>12}")
print("="*70)

def show(label, match):
    bc, ac = Counter(), Counter()
    for (dw, ven, title, mfr, url, sig) in rows:
        if not match(ven, title): continue
        bc[old.get(dw,('(none)',))[0]] += 1; ac[final.get(dw,'(none)')] += 1
    print(f"\n— {label} —")
    print(f"   BEFORE {len(bc)} key(s): " + ", ".join(f"{k.split('::')[-1]}×{n}" for k,n in bc.most_common(4)))
    print(f"   AFTER  {len(ac)} key(s): " + ", ".join(f"{k.split('::')[-1]}×{n}" for k,n in ac.most_common(4)))

show("Yakatore (real colorways — MUST stay together)", lambda v,t:'phillipe' in v.lower() and 'yakatore' in t.lower())
show("La Corka (real 11-colorway — MUST stay together)", lambda v,t:'phillipe' in v.lower() and 'corka' in t.lower())
show("Ultrasuede 5538 (93 designs — MUST split)", lambda v,t:'ultrasuede' in v.lower())
show("Phillipe Romano CORK (16 designs — MUST split)", lambda v,t:'phillipe' in v.lower() and 'cork' in t.lower() and 'corka' not in t.lower())

if EMIT:
    out = {}
    for dw, k in final.items():
        out[dw] = {"pid": k, "sid": old.get(dw, (None, None))[1]}
    json.dump(out, open(EMIT_PATH, "w"))
    print(f"\n[emit] wrote {len(out)} keys -> {EMIT_PATH}")