← back to Dw Yolo Loop

artmura-site/extract-colors-cdn.py

73 lines

#!/usr/bin/env python3
"""Dominant color (hex+hue+sat+val+bucket) per SKU for a build-line line whose images are
Shopify CDN urls. Reads lines/<slug>.json, fetches images[0] at ?width=120 (tiny), writes
lines/<slug>-colors.json keyed by mfr_sku. Same logic as scripts/artmura-onboard/extract_colors.py.

    python3 extract-colors-cdn.py thibaut
"""
import json, os, sys, colorsys, io, urllib.request
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
from PIL import Image

HERE = os.path.dirname(os.path.abspath(__file__))
slug = sys.argv[1] if len(sys.argv) > 1 else "thibaut"
PKG = json.load(open(os.path.join(HERE, "lines", f"{slug}.json")))

def fetch(url):
    sep = '&' if '?' in url else '?'
    req = urllib.request.Request(url + sep + "width=120", headers={"User-Agent": "dw-color/1.0"})
    with urllib.request.urlopen(req, timeout=20) as r:
        return Image.open(io.BytesIO(r.read())).convert("RGB").resize((64, 64))

def dominant(im):
    q = im.quantize(colors=16).convert("RGB")
    ranked = Counter(q.getdata()).most_common()
    def score(item):
        (r, g, b), n = item
        h, s, v = colorsys.rgb_to_hsv(r/255, g/255, b/255)
        return n * (0.5 + s)
    (r, g, b), _ = max(ranked[:6], key=score)
    return r, g, b

def bucket(h, s, v):
    if v < 0.18: return "black"
    if s < 0.12 and v > 0.85: return "white"
    if s < 0.14: return "grey"
    deg = h * 360
    if deg < 16 or deg >= 345: return "red"
    if deg < 45: return "orange" if v > 0.5 else "brown"
    if deg < 70: return "gold"
    if deg < 160: return "green"
    if deg < 200: return "teal"
    if deg < 255: return "blue"
    if deg < 290: return "purple"
    return "pink"

def one(p):
    sku, imgs = p.get("mfr_sku"), p.get("images") or []
    if not sku or not imgs:
        return None
    try:
        r, g, b = dominant(fetch(imgs[0]))
        h, s, v = colorsys.rgb_to_hsv(r/255, g/255, b/255)
        return sku, {"hex": f"#{r:02x}{g:02x}{b:02x}", "hue": round(h*360, 1),
                     "sat": round(s, 3), "val": round(v, 3), "bucket": bucket(h, s, v)}
    except Exception:
        return None

out, done, total = {}, 0, len(PKG["products"])
with ThreadPoolExecutor(max_workers=16) as ex:
    futs = [ex.submit(one, p) for p in PKG["products"]]
    for f in as_completed(futs):
        done += 1
        r = f.result()
        if r:
            out[r[0]] = r[1]
        if done % 200 == 0:
            print(f"  …{done}/{total}  ({len(out)} extracted)")

open(os.path.join(HERE, "lines", f"{slug}-colors.json"), "w").write(json.dumps(out, indent=1))
print(f"extracted {len(out)}/{total} colors → lines/{slug}-colors.json")
print("buckets:", dict(Counter(v["bucket"] for v in out.values()).most_common()))