`compute_distributor_colors.py`

54 lines
#!/usr/bin/env python3
"""Compute the 'color ID' (dominant hex + palette + color-family + hue) for every distributor
catalog product, LOCALLY ($0, Pillow median-cut — same logic as our sangetsu enrichment).
This is the match key: wallcoverings keep their color across renamed catalogs, so color is the
strongest cross-catalog visual signal. Output: staging/distributor-colors.jsonl. Resumable."""
import json, os, sys, urllib.request, tempfile, colorsys
from concurrent.futures import ThreadPoolExecutor, as_completed
sys.path.insert(0, os.path.dirname(__file__))
from enrich import hex_palette, dominant_family

HERE = os.path.dirname(__file__)
CAT = json.load(open(os.path.join(HERE, 'staging', 'distributor-catalogs.json')))
OUT = os.path.join(HERE, 'staging', 'distributor-colors.jsonl')
UA = 'Mozilla/5.0 (Macintosh)'

def hue_of(hx):
    h = hx.lstrip('#'); r, g, b = [int(h[i:i+2], 16)/255 for i in (0, 2, 4)]
    return round(colorsys.rgb_to_hsv(r, g, b)[0]*360)

def process(p):
    try:
        data = urllib.request.urlopen(urllib.request.Request(p['image_url'], headers={'User-Agent': UA}), timeout=20).read()
        if len(data) < 200: return None
        f = tempfile.NamedTemporaryFile(suffix='.jpg', delete=False); f.write(data); f.close()
        pal = hex_palette(f.name); os.unlink(f.name)
        if not pal: return None
        return {'vendor': p['vendor_code'], 'sku': p['mfr_sku'], 'name': p.get('pattern_name'),
                'color_name': p.get('color_name'), 'image_url': p['image_url'], 'product_url': p.get('product_url'),
                'hex': pal[0]['hex'], 'palette': [x['hex'] for x in pal[:4]],
                'family': dominant_family(pal), 'hue': hue_of(pal[0]['hex'])}
    except Exception:
        return None

def main():
    done = set()
    if os.path.exists(OUT):
        for l in open(OUT):
            try:
                r = json.loads(l); done.add(r['sku'] + '|' + r['vendor'])
            except Exception: pass
    todo = [p for v in CAT.values() for p in v if (p['mfr_sku'] + '|' + p['vendor_code']) not in done]
    print(f"computing color for {len(todo)} distributor products ({len(done)} already done)", flush=True)
    n = 0
    with open(OUT, 'a') as out, ThreadPoolExecutor(max_workers=8) as ex:
        for f in as_completed({ex.submit(process, p): p for p in todo}):
            r = f.result()
            if r: out.write(json.dumps(r, ensure_ascii=False) + '\n'); out.flush()
            n += 1
            if n % 250 == 0: print(f"  {n}/{len(todo)}", flush=True)
    print(f"done — {sum(1 for _ in open(OUT))} products have a color ID", flush=True)

if __name__ == '__main__':
    main()