`enrich_corporate_colors.py`

61 lines
#!/usr/bin/env python3
"""Local color-ID enrichment for the harvested vendor-corporate products ($0, Pillow — same
logic as the distributor-color pipeline). Reads staging/corporate/*.jsonl, computes dominant
hex + palette + family + hue for each product image, writes staging/corporate-colors.jsonl.
Resumable. NO paid API, NO tokens."""
import json, os, sys, glob, tempfile, colorsys, urllib.request
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__)
CORP = os.path.join(HERE, 'staging', 'corporate')
OUT = os.path.join(HERE, 'staging', 'corporate-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(vendor, p):
    img = p.get('image')
    if not img or not img.startswith('http'): return None
    try:
        data = urllib.request.urlopen(urllib.request.Request(img, 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': vendor, 'title': p.get('title'), 'url': p.get('url'), 'image': img,
                '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: done.add(json.loads(l)['url'])
            except Exception: pass
    jobs = []
    for f in glob.glob(os.path.join(CORP, '*.jsonl')):
        vendor = os.path.basename(f)[:-6]
        for l in open(f):
            l = l.strip()
            if not l: continue
            p = json.loads(l)
            if p.get('url') not in done and p.get('image'): jobs.append((vendor, p))
    print(f"color-enriching {len(jobs)} corporate products ({len(done)} done)", flush=True)
    n = 0
    with open(OUT, 'a') as out, ThreadPoolExecutor(max_workers=8) as ex:
        for fut in as_completed({ex.submit(process, v, p): 1 for v, p in jobs}):
            r = fut.result()
            if r: out.write(json.dumps(r, ensure_ascii=False) + '\n'); out.flush()
            n += 1
            if n % 300 == 0: print(f"  {n}/{len(jobs)}", flush=True)
    print("corporate color enrichment pass done", flush=True)

if __name__ == '__main__':
    main()