`build_vendor_logos.py`

125 lines
#!/usr/bin/env python3
"""Harvest vendor brand LOGOS from Shopify Files and attach them to the learned vendor
profiles, so the scanner can ground logo/typeface recognition against real reference marks
(instead of the VLM guessing a brand cold). The DW store's brand-logo app uploaded these as
Files named like 'maxwell-logo-black.png' — we match each to one of our known vendors.

Reads vendors from data/vendor_profiles.json, pages Shopify Files (query 'logo'), fuzzy-
matches filename -> vendor, downloads the match into data/logos/<slug>.<ext>, and writes
data/vendor_logos.json (vendor -> {url,file,w,h}). Also stamps profiles[vendor].logo_url so
/learn and /api/identify can use it. Read-only Shopify + local download. $0."""
import json, os, re, urllib.request
from datetime import datetime, timezone

SHOP = "designer-laboratory-sandbox.myshopify.com"; API = "2024-10"
ROOT = os.path.dirname(os.path.abspath(__file__))
LOGO_DIR = os.path.join(ROOT, "data", "logos"); os.makedirs(LOGO_DIR, exist_ok=True)
PROF = os.path.join(ROOT, "data", "vendor_profiles.json")
OUT = os.path.join(ROOT, "data", "vendor_logos.json")
# Files need read_files — only the DRAFT token has it (admin token does not).
T = ""
for l in open(os.path.expanduser("~/Projects/secrets-manager/.env")):
    if l.startswith("SHOPIFY_DRAFT_TOKEN="): T = l.strip().split("=", 1)[1]; break
assert T, "no SHOPIFY_DRAFT_TOKEN (needs read_files)"

def gql(q, v=None):
    r = urllib.request.Request(f"https://{SHOP}/admin/api/{API}/graphql.json",
        data=json.dumps({"query": q, "variables": v or {}}).encode(),
        headers={"X-Shopify-Access-Token": T, "Content-Type": "application/json"})
    return json.load(urllib.request.urlopen(r, timeout=60))

def slug(s): return re.sub(r"[^a-z0-9]", "", (s or "").lower())

# vendor -> set of match keys (full slug + distinctive leading words)
prof = json.load(open(PROF))
vendors = list(prof.get("profiles", {}).keys())
STOP = {"fabrics", "fabric", "wallcoverings", "wallcovering", "wallpaper", "home", "and",
        "the", "co", "inc", "design", "designs", "international", "studio", "textiles"}
vkeys = {}
for v in vendors:
    words = [w for w in re.findall(r"[a-z0-9]+", v.lower())]
    keys = {slug(v)}
    sig = [w for w in words if w not in STOP and len(w) >= 4]
    if sig: keys.add(sig[0])                       # distinctive first word (maxwell, brunschwig)
    if len(sig) >= 2: keys.add(sig[0] + sig[1])
    vkeys[v] = {k for k in keys if len(k) >= 4}

# ── page Shopify Files for logo images ────────────────────────────────────────
def clean_fileslug(url, alt):
    base = (url.split("/")[-1].split("?")[0])
    base = re.sub(r"\.(png|jpg|jpeg|svg|webp|gif)$", "", base, flags=re.I)
    base = re.sub(r"[-_]?(logo|nav|black|white|transparent|background|backgro|header|footer|trade|web|small|large|\d+)$", "", base, flags=re.I)
    return slug(base)

files = []
cursor = None
for _ in range(40):
    q = '''query($c:String){ files(first:50, after:$c, query:"logo"){
      pageInfo{ hasNextPage endCursor }
      edges{ node{ __typename ... on MediaImage{ alt image{ url width height } } } } } }'''
    d = gql(q, {"c": cursor}).get("data", {}).get("files", {})
    for e in d.get("edges", []):
        n = e.get("node") or {}
        img = n.get("image") or {}
        if img.get("url"): files.append({"url": img["url"], "alt": n.get("alt") or "",
                                         "w": img.get("width"), "h": img.get("height")})
    if not d.get("pageInfo", {}).get("hasNextPage"): break
    cursor = d["pageInfo"]["endCursor"]
print(f"scanned {len(files)} logo-ish files")

# ── match files -> vendors (TIGHTENED: must be a logo file + anchored key match) ─
def keymatch(k, fslug):
    if k == fslug: return 1.0                       # exact
    if fslug.startswith(k) and len(k) >= 5: return 0.9   # file leads with the key
    if k.startswith(fslug) and len(fslug) >= 5: return 0.85
    if len(k) >= 7 and k in fslug: return 0.7       # long distinctive token contained
    return 0.0
MIN_CONF = 0.7
matches, rejected = {}, []
for f in files:
    # HARD requirement: the file must actually be a brand logo (name/alt says so).
    if not ("logo" in f["url"].lower() or "logo" in (f["alt"] or "").lower()):
        continue
    fslug = clean_fileslug(f["url"], f["alt"])
    if not fslug or len(fslug) < 4: continue
    best = None   # (conf, vendor, key)
    for v, keys in vkeys.items():
        for k in keys:
            c = keymatch(k, fslug)
            if c and (best is None or c > best[0] or (c == best[0] and len(k) > len(best[2]))):
                best = (c, v, k)
    if not best or best[0] < MIN_CONF:
        if best: rejected.append((fslug, best[1], round(best[0], 2)))
        continue
    c, v, k = best
    if v not in matches or c > matches[v]["conf"]:
        matches[v] = {**f, "conf": round(c, 2), "key": k, "fslug": fslug}
if rejected:
    print(f"  ({len(rejected)} below-threshold pairs skipped, e.g. {rejected[:4]})")

print(f"matched {len(matches)} vendors to a logo file")

# ── download + write manifest + stamp profiles ────────────────────────────────
manifest = {}
for v, m in sorted(matches.items()):
    ext = (re.search(r"\.(png|jpg|jpeg|svg|webp|gif)", m["url"], re.I) or [".png"])[0].lower().lstrip(".")
    if ext == "jpeg": ext = "jpg"
    fn = slug(v) + "." + ext
    try:
        data = urllib.request.urlopen(m["url"], timeout=40).read()
        open(os.path.join(LOGO_DIR, fn), "wb").write(data)
        manifest[v] = {"file": "data/logos/" + fn, "url": m["url"], "w": m["w"], "h": m["h"],
                       "matched_on": m["key"], "conf": m["conf"], "bytes": len(data)}
        if v in prof["profiles"]: prof["profiles"][v]["logo_url"] = m["url"]; prof["profiles"][v]["logo_file"] = "data/logos/" + fn
    except Exception as e:
        print(f"  ! {v}: download failed {e}")

json.dump({"_meta": {"vendors_with_logo": len(manifest), "scanned_files": len(files),
                     "built_at": datetime.now(timezone.utc).isoformat()}, "logos": manifest},
          open(OUT, "w"), indent=0)
prof.setdefault("_meta", {})["logos_attached"] = len(manifest)
json.dump(prof, open(PROF, "w"), indent=0)
print(f"\ndownloaded {len(manifest)} vendor logos -> {LOGO_DIR}")
for v in list(manifest)[:14]:
    print(f"  {v:28} <- {manifest[v]['matched_on']:14} {manifest[v]['w']}x{manifest[v]['h']}")