`build_logo_fingerprints.py`

90 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.

TK-12090 Lane P: the VLM call now goes through the shared fleet vision lib
(~/Projects/_shared/lib/exo_vision.py) — exo ring primary ($0), Gemini fallback (~$0.0006/logo,
cost-ledgered by the lib). The old direct Ollama :11434 path was retired 2026-09-18. When the
lib reports not_measured (ring down + fallback off/failed) the logo is SKIPPED, never written
as an empty/false fingerprint. Set VISION_FALLBACK=none to stay strictly $0."""
import base64, json, os, re, sys

ROOT = os.path.dirname(os.path.abspath(__file__))
PROF = os.path.join(ROOT, "data", "vendor_profiles.json")
sys.path.insert(0, os.environ.get("EXO_VISION_LIB_DIR",
                                  os.path.join(ROOT, "..", "_shared", "lib")))
import exo_vision  # noqa: E402 — shared lib, path set above
MODEL = exo_vision.EXO_MODEL
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>"}')

COST = {"usd": 0.0, "calls": 0}

def vlm(b64, mime="image/png"):
    r = exo_vision.vision_chat(PROMPT, {"b64": b64, "mime": mime}, timeout=90, max_tokens=512)
    if not r.get("ok"):
        raise RuntimeError("vision not measured: %s" % r.get("error"))
    COST["usd"] += r.get("cost_usd") or 0; COST["calls"] += 1
    t = r.get("text") or ""
    m = re.search(r"```(?:json)?\s*([\s\S]*?)```", t)
    try: return json.loads((m.group(1) if m else t).strip() or "{}")
    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()
        ext = os.path.splitext(fp)[1].lower()
        out = vlm(b64, {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp"}.get(ext, "image/png"))
        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")
print(f"cost: ${COST['usd']:.4f} over {COST['calls']} vision calls (exo ring = $0; Gemini fallback ~$0.0006 each)")