`scripts/extract_dominant_hex.py`

97 lines
#!/usr/bin/env python3
"""
Per-product dominant-color extraction for Sister Parish catalog.
Standing rule (memory: feedback_sort_skill_canonical): real-pixel hex beats
tag-derived hex. Run sips-style "shrink to 1x1, read pixel" via Pillow.

Strategy per image:
- Download primary_image to /tmp/sp_imgs/
- Resize to 200x200 (preserve detail), then bucket to dominant color
  using k=5 KMeans-lite via Pillow.quantize (mode P, 5 colors)
- Persist {dominant_hex, palette[]} to dw_unified.sisterparish_catalog.colors JSONB
"""
import os, json, subprocess, time, urllib.request, ssl
from pathlib import Path
from PIL import Image
from collections import Counter

ROOT   = Path.home() / 'Projects' / 'sister-parish-onboarding'
IMGDIR = Path('/tmp/sp_imgs'); IMGDIR.mkdir(exist_ok=True)
CTX    = ssl.create_default_context()

def fetch(url, dest):
    if dest.exists() and dest.stat().st_size > 1000:
        return True
    try:
        req = urllib.request.Request(url, headers={'User-Agent':'Mozilla/5.0 (DW)'})
        with urllib.request.urlopen(req, timeout=20, context=CTX) as r, dest.open('wb') as f:
            f.write(r.read())
        return True
    except Exception as e:
        print(f"  fetch fail: {e}")
        return False

def dominant_palette(path, k=5):
    img = Image.open(path).convert('RGB')
    img.thumbnail((200,200))
    pal = img.quantize(colors=k, method=Image.Quantize.MEDIANCUT).convert('RGB')
    pixels = list(pal.getdata())
    cnt = Counter(pixels)
    total = sum(cnt.values())
    top = cnt.most_common(k)
    palette = [{'hex': '#{:02x}{:02x}{:02x}'.format(*rgb), 'pct': round(n/total*100, 2)} for rgb,n in top]
    return palette

def list_products():
    sql = "SELECT json_agg(t) FROM (SELECT sp_product_id, title, primary_image FROM sisterparish_catalog WHERE primary_image IS NOT NULL ORDER BY title) t;"
    out = subprocess.run(['psql','dw_unified','-At','-q','-c',sql], capture_output=True, text=True, check=True).stdout.strip()
    return json.loads(out)

def ensure_column():
    subprocess.run(['psql','dw_unified','-q','-c',
        "ALTER TABLE sisterparish_catalog ADD COLUMN IF NOT EXISTS colors JSONB; ALTER TABLE sisterparish_catalog ADD COLUMN IF NOT EXISTS dominant_hex TEXT;"], check=True)

def save_to_pg(sp_id, dominant_hex, palette):
    sql = f"UPDATE sisterparish_catalog SET dominant_hex='{dominant_hex}', colors='{json.dumps(palette).replace(chr(39), chr(39)+chr(39))}'::jsonb WHERE sp_product_id={sp_id};"
    subprocess.run(['psql','dw_unified','-q','-c',sql], check=True)

def main():
    ensure_column()
    products = list_products()
    print(f"{len(products)} products with primary_image")
    sql_lines = []
    failed = 0
    for i, p in enumerate(products, 1):
        url = p['primary_image']
        ext = url.split('?')[0].rsplit('.', 1)[-1].lower()
        if ext not in ('jpg','jpeg','png','webp'):
            ext = 'jpg'
        dest = IMGDIR / f"sp_{p['sp_product_id']}.{ext}"
        ok = fetch(url, dest)
        if not ok:
            failed += 1
            print(f"[{i:03}/{len(products)}] ✗ {p['title'][:50]}")
            continue
        try:
            palette = dominant_palette(dest)
            dominant = palette[0]['hex']
            sql_lines.append((p['sp_product_id'], dominant, palette))
            print(f"[{i:03}/{len(products)}] {dominant}  {p['title'][:50]}")
        except Exception as e:
            failed += 1
            print(f"[{i:03}/{len(products)}] palette err: {e}  {p['title'][:50]}")
    # Bulk update
    fp = Path('/tmp/sp_colors.sql')
    with fp.open('w') as f:
        f.write("BEGIN;\n")
        for sp_id, dh, pal in sql_lines:
            pal_str = json.dumps(pal).replace("'", "''")
            f.write(f"UPDATE sisterparish_catalog SET dominant_hex='{dh}', colors='{pal_str}'::jsonb WHERE sp_product_id={sp_id};\n")
        f.write("COMMIT;\n")
    subprocess.run(['psql','dw_unified','-q','-f', str(fp)], check=True)
    fp.unlink()
    print(f"\nUpdated {len(sql_lines)} rows · failed {failed}")

if __name__ == '__main__':
    main()