`tools/pattern-image-verify.py`

108 lines
#!/usr/bin/env python3
"""WS-1 image confirmation (READ-ONLY, local, $0).

Reads "dw_sku<TAB>title<TAB>image_url" lines on stdin (one candidate group),
fetches each image, computes a COLOR-INVARIANT structural signature, and clusters
members by visual pattern. Same pattern in different colorways -> one cluster;
genuinely different patterns sharing a key (e.g. a templated-title vendor) ->
multiple clusters. This is the "are these really the same pattern, just different
colors?" confirmation on top of the SKU/title key.

Color invariance: image -> grayscale -> we compute dHash on BOTH the image and its
tonal inverse and take the smaller Hamming distance, so a light-on-dark vs
dark-on-light colorway of the same motif still matches.

  psql ... -tAc "<query yielding sku|title|url>" | tr '|' '\\t' | \
     python3 tools/pattern-image-verify.py "Coordonné" 30
"""
import sys, io, urllib.request, hashlib, os
from PIL import Image, ImageOps

LABEL = sys.argv[1] if len(sys.argv) > 1 else "group"
LIMIT = int(sys.argv[2]) if len(sys.argv) > 2 else 40
THRESH = int(os.environ.get("HAM_THRESH", "12"))   # dHash Hamming <= THRESH => same pattern
HASH = 8
CACHE = "/tmp/pattern-verify-cache"
os.makedirs(CACHE, exist_ok=True)

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):
        base = r * (n + 1)
        for c in range(n):
            v = (v << 1) | (1 if px[base + c] > px[base + c + 1] else 0)
    return v

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

def ham(a, b):
    return bin(a ^ b).count("1")

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

items = []
for line in sys.stdin:
    parts = line.rstrip("\n").split("\t")
    if len(parts) < 3:
        continue
    sku, title, url = parts[0], parts[1], parts[2]
    if not url.startswith("http"):
        continue
    items.append((sku, title, url))
    if len(items) >= LIMIT:
        break

print(f"=== {LABEL}: {len(items)} candidate members (thresh dHash<={THRESH}) ===")
hashed = []
for sku, title, url in items:
    try:
        im = Image.open(io.BytesIO(fetch(url)))
        hashed.append((sku, title, dhash(im), inv_dhash(im)))
    except Exception as e:
        print(f"  [skip] {sku}: {e}")

# single-linkage clustering: a member joins a cluster if it's within THRESH of ANY member
clusters = []
for sku, title, h, hi in hashed:
    placed = False
    for cl in clusters:
        for (_, _, ch, _chi) in cl:
            if min(ham(h, ch), ham(hi, ch)) <= THRESH:
                cl.append((sku, title, h, hi))
                placed = True
                break
        if placed:
            break
    if not placed:
        clusters.append([(sku, title, h, hi)])

clusters.sort(key=len, reverse=True)
print(f"  -> {len(hashed)} images formed {len(clusters)} distinct visual pattern(s)")
for i, cl in enumerate(clusters[:8], 1):
    egs = "; ".join(t for (_, t, _, _) in cl[:3])
    print(f"   cluster {i}: {len(cl):>3} member(s)  e.g. {egs[:90]}")
if len(clusters) > 8:
    print(f"   … +{len(clusters)-8} more clusters")

# verdict for this candidate group
if hashed:
    biggest = len(clusters[0]) / len(hashed)
    if len(clusters) == 1:
        print("  VERDICT: CONFIRMED single pattern — safe to group as colorways.")
    elif biggest >= 0.6:
        print(f"  VERDICT: MOSTLY one pattern ({len(clusters[0])}/{len(hashed)}) + {len(clusters)-1} outlier(s) to split off.")
    else:
        print(f"  VERDICT: OVER-GROUPED — {len(clusters)} real patterns share this key. Do NOT group by key alone.")