`build_logo_fingerprints.py`

77 lines
#!/usr/bin/env python3
"""Auto-populate the LLM's per-vendor logo knowledge: run the local VLM (qwen2.5vl, $0) over
each harvested reference logo ONCE and store a canonical fingerprint — what the logo looks
like, its typesetting, and the brand text it reads as — into the vendor profile. At scan time
this grounds /api/identify (compare the swatch's logo to the known reference) and shows on
/learn. Idempotent: skips vendors already fingerprinted unless --force. Local only. $0."""
import base64, json, os, re, sys, urllib.request

ROOT = os.path.dirname(os.path.abspath(__file__))
PROF = os.path.join(ROOT, "data", "vendor_profiles.json")
OLLAMA = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434")
MODEL = os.environ.get("OLLAMA_VISION_MODEL", "qwen2.5vl:7b")
FORCE = "--force" in sys.argv

PROMPT = ('This image is a BRAND LOGO for a wallcovering/fabric maker. Reply ONLY as compact '
          'JSON: {"reads_as":"<the brand text in the logo>","typeface":"<serif wordmark / sans '
          'caps / script / geometric / etc>","logo":"<one-line description of the mark>",'
          '"colors":"<dominant colors>"}')

def vlm(b64):
    body = json.dumps({"model": MODEL, "prompt": PROMPT, "images": [b64],
                       "stream": False, "format": "json", "options": {"temperature": 0}}).encode()
    r = urllib.request.Request(OLLAMA + "/api/generate", data=body,
                               headers={"Content-Type": "application/json"})
    d = json.load(urllib.request.urlopen(r, timeout=90))
    try: return json.loads(d.get("response", "{}"))
    except Exception: return {}

prof = json.load(open(PROF))
profiles = prof.get("profiles", {})
todo = [(v, p) for v, p in profiles.items()
        if p.get("logo_file") and (FORCE or not p.get("logo_desc"))]
print(f"fingerprinting {len(todo)} logos via {MODEL} …")

done = 0
for v, p in todo:
    fp = os.path.join(ROOT, p["logo_file"])
    if not os.path.exists(fp) or fp.lower().endswith(".svg"):   # VLM needs raster
        continue
    try:
        b64 = base64.b64encode(open(fp, "rb").read()).decode()
        out = vlm(b64)
        if out:
            reads = out.get("reads_as") or ""
            # VALIDATE: the logo must read back as THIS vendor. If the VLM sees a different
            # brand (e.g. a mislabeled file), reject the logo so nothing downstream trusts it.
            def norm(s): return re.sub(r"[^a-z0-9]", "", (s or "").lower())
            nr, nv = norm(reads), norm(v)
            # whole-WORD comparison (substring "century" in "20thcenturyfox" was the false-accept).
            vwords = {w for w in re.findall(r"[a-z0-9]+", v.lower()) if len(w) >= 4}
            rwords = {w for w in re.findall(r"[a-z0-9]+", reads.lower()) if len(w) >= 3}
            shared = vwords & rwords
            valid = bool(nr) and len(nr) >= 4 and (              # ≥4 chars: a 3-char read like "Sch" must not pass via containment
                nr == nv or nr in nv or nv in nr or            # one name contains the other
                len(shared) >= 2 or                            # ≥2 shared significant words
                (len(vwords) <= 1 and bool(shared))            # single-word vendor, exact word hit
            )
            p["logo_desc"] = out.get("logo") or ""
            p["logo_typeface"] = out.get("typeface") or ""
            p["logo_reads"] = reads
            p["logo_colors"] = out.get("colors") or ""
            p["logo_valid"] = valid
            if not valid:
                # don't let a wrong logo surface on /learn or bias /api/identify
                p.pop("logo_url", None); p.pop("logo_file", None)
                print(f"  ✗ {v:28} REJECTED — logo reads as '{reads}' (not this vendor)")
            else:
                done += 1
                print(f"  ✓ {v:28} reads='{reads}' type='{p['logo_typeface']}'")
            json.dump(prof, open(PROF, "w"), indent=0)   # save incrementally (crash-safe)
    except Exception as e:
        print(f"  ! {v}: {e}")

prof.setdefault("_meta", {})["logos_fingerprinted"] = sum(1 for p in profiles.values() if p.get("logo_desc"))
json.dump(prof, open(PROF, "w"), indent=0)
print(f"\nfingerprinted {done} logos -> profiles updated")