`tools/ci-hash-build.py`

86 lines
#!/usr/bin/env python3
"""Color-invariant image hasher for WS-1 image confirmation (read-only, $0 local).

For each (dw_sku, image_url), download (disk-cached), compute a COLOR-INVARIANT
signature = dHash of grayscale AND dHash of the tonal inverse. Two products are the
"same pattern, different color" when min(ham(dA,dB), ham(invA,invB)) is small — so a
light-on-dark vs dark-on-light colorway of one motif still matches.

Writes JSONL (resumable, append-only) to tools/ci-hashes.jsonl:
  {"url": "...", "dhash": "<16hex>", "inv": "<16hex>", "status": "ok|err"}

Resumable: on restart, URLs already present in the JSONL are skipped. No DB writes,
no schema change. Input on stdin: "dw_sku<TAB>url" lines.

  cut -f1,5 multi-images.tsv | python3 tools/ci-hash-build.py
"""
import sys, os, io, json, hashlib, urllib.request, ssl
from concurrent.futures import ThreadPoolExecutor, as_completed
from PIL import Image, ImageOps

OUT   = os.path.join(os.path.dirname(__file__), "ci-hashes.jsonl")
CACHE = os.path.expanduser("~/.cache/ci-hash")
WORKERS = int(os.environ.get("CI_WORKERS", "16"))
HASH = 8
os.makedirs(CACHE, exist_ok=True)
_ctx = ssl.create_default_context(); _ctx.check_hostname = False; _ctx.verify_mode = ssl.CERT_NONE

def dhash(img, n=HASH):
    g = img.convert("L").resize((n + 1, n), Image.LANCZOS)
    px = list(g.getdata()); v = 0
    for r in range(n):
        b = r * (n + 1)
        for c in range(n):
            v = (v << 1) | (1 if px[b + c] > px[b + c + 1] else 0)
    return f"{v:016x}"

def inv_dhash(img, n=HASH):
    return dhash(ImageOps.invert(img.convert("L")), n)

def fetch(url):
    key = os.path.join(CACHE, hashlib.md5(url.encode()).hexdigest() + ".bin")
    if os.path.exists(key) and os.path.getsize(key) > 0:
        return open(key, "rb").read()
    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 dw-ci-hash"})
    with urllib.request.urlopen(req, timeout=20, context=_ctx) as r:
        data = r.read()
    if data: open(key, "wb").write(data)
    return data

def work(url):
    try:
        data = fetch(url)
        img = Image.open(io.BytesIO(data))
        img.load()
        return {"url": url, "dhash": dhash(img), "inv": inv_dhash(img), "status": "ok"}
    except Exception as e:
        return {"url": url, "status": "err", "error": str(e)[:80]}

# resume: skip URLs already hashed
done = set()
if os.path.exists(OUT):
    for line in open(OUT):
        try: done.add(json.loads(line)["url"])
        except Exception: pass

urls, seen = [], set()
for line in sys.stdin:
    p = line.rstrip("\n").split("\t")
    if len(p) < 2: continue
    u = p[1].strip()
    if u and u not in seen and u not in done:
        seen.add(u); urls.append(u)

total = len(urls)
print(f"[ci-hash] {total} new URLs to hash ({len(done)} already done), {WORKERS} workers", flush=True)
n = ok = err = 0
with open(OUT, "a") as out, ThreadPoolExecutor(max_workers=WORKERS) as ex:
    futs = {ex.submit(work, u): u for u in urls}
    for f in as_completed(futs):
        rec = f.result()
        out.write(json.dumps(rec) + "\n"); out.flush()
        n += 1; ok += rec["status"] == "ok"; err += rec["status"] == "err"
        if n % 500 == 0:
            print(f"[ci-hash] {n}/{total}  ok={ok} err={err}", flush=True)
print(f"[ci-hash] DONE {n}/{total}  ok={ok} err={err}  -> {OUT}", flush=True)