`vision_classify.py`

164 lines
#!/usr/bin/env python3
"""
Wave-2 VISION reclassify — Gemini 2.0 Flash classifies the material of each
Textured/Woven residual product from its image, then writes a material-group tag.

Vision sees APPEARANCE, not substrate, so the prompt forces a conservative default:
a flat/printed pattern (not a genuine 3D textured material) -> "Non-Woven".

  python3 vision_classify.py --canary 5     # classify 5, NO writes, show cost
  python3 vision_classify.py --apply        # classify + tag all residual, running cost
  python3 vision_classify.py --apply --limit 500   # bounded batch

Reads GEMINI_API_KEY + SHOPIFY_ADMIN_TOKEN from ~/Projects/secrets-manager/.env.
Every run appends a timestamped record to data/vision-runs.jsonl.
"""
import os, re, sys, json, time, base64, subprocess, urllib.request, urllib.error
from datetime import datetime, timezone
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import classify as C
import apply_tags as A   # reuse slug() + api() + token()

MODEL = "qwen2.5vl:7b"   # LOCAL Ollama vision (Gemini generateContent was deprecated;
                         # local also matches Steve's local-first rule + is $0)
OLLAMA = "http://localhost:11434/api/generate"
RATE  = 0.0             # $0 — local inference
ENV   = os.path.expanduser("~/Projects/secrets-manager/.env")
HERE  = os.path.dirname(os.path.abspath(__file__))
RUNLOG = os.path.join(HERE, "data", "vision-runs.jsonl")

LABELS = ["Grasscloth","Paperweave","Natural Fiber","Cork","Silk","Linen","Leather",
          "Glass Bead","Flock / Velvet","Metallic / Foil","Wood Veneer",
          "Vinyl / Type II","Non-Woven","Textured"]
PROMPT = (
  "You are classifying the physical MATERIAL of a wallcovering from its product image.\n"
  "Choose EXACTLY ONE label from this list:\n" + ", ".join(LABELS) + ".\n"
  "Rules: pick a specific natural material (Grasscloth, Silk, Cork, Linen, Natural Fiber, "
  "Wood Veneer, Glass Bead, Leather, Metallic/Foil, Flock) ONLY if the image clearly shows "
  "that real 3D texture/weave. If it looks like a FLAT PRINTED design that merely depicts a "
  "texture, or you are unsure, answer 'Non-Woven'. Answer with the label text only, nothing else."
)

def env(name):
    for line in open(ENV):
        if line.startswith(name+"="):
            return line.split("=",1)[1].strip().strip('"').strip("'")
    return ""

def residual():
    """rows classify() marks needs_vision, with a numeric Shopify id."""
    out=[]
    for r in C.load_rows(scope_all=True):
        tier,bucket,ev = C.classify(r)
        if bucket is None and tier=="needs_vision":
            gid=r.get("shopify_id") or ""; nid=gid.rsplit("/",1)[-1]
            if nid.isdigit():
                out.append((nid, r["title"]))
    out.sort(key=lambda x: x[0])   # deterministic order for reproducible canary/batches
    return out

def live_product(nid, tok):
    """(image_src, already_material_tagged) from the Admin API — avoids stale mirror
    URLs AND lets us skip anything the free-tag pass already classified (safe to run
    concurrently with the free job; no double-tagging)."""
    p=A.api("GET", f"products/{nid}.json?fields=id,images,tags", tok)["product"]
    imgs=p.get("images") or []
    src=imgs[0]["src"] if imgs else ""
    tags=[t.strip() for t in (p.get("tags") or "").split(",") if t.strip()]
    already=any(t.startswith("material-group-") for t in tags)
    return src, already, tags

def write_tag(nid, tag, tags, tok):
    """Single PUT using the tags we already fetched (no re-GET/verify). 2 REST
    calls/product total keeps us under Shopify's 2 req/s cap."""
    A.api("PUT", f"products/{nid}.json", tok, {"product":{"id":int(nid),"tags":", ".join(tags+[tag])}})

def small(url):
    # ask Shopify CDN for a smaller render to keep the call light
    if "cdn.shopify.com" in url and "width=" not in url:
        return url + (("&" if "?" in url else "?") + "width=384")
    return url

def vision_label(img_url, key=None):
    raw = urllib.request.urlopen(small(img_url), timeout=30).read()
    b64 = base64.b64encode(raw).decode()
    body = {"model":MODEL,"prompt":PROMPT,"images":[b64],"stream":False,
            "options":{"temperature":0,"num_predict":20}}
    req = urllib.request.Request(OLLAMA, method="POST",
        headers={"Content-Type":"application/json"}, data=json.dumps(body).encode())
    resp = json.load(urllib.request.urlopen(req, timeout=120))
    txt = (resp.get("response") or "").strip()
    for L in LABELS:                     # snap to a known label
        if L.lower() in txt.lower() or txt.lower() in L.lower():
            return L
    return "Non-Woven"                   # unparseable -> conservative default

def logrun(rec):
    os.makedirs(os.path.dirname(RUNLOG), exist_ok=True)
    with open(RUNLOG,"a") as f: f.write(json.dumps(rec)+"\n")

def main():
    canary = int(sys.argv[sys.argv.index("--canary")+1]) if "--canary" in sys.argv else None
    limit  = int(sys.argv[sys.argv.index("--limit")+1]) if "--limit" in sys.argv else None
    do_write = "--apply" in sys.argv
    rows = residual()
    print(f"Vision residual: {len(rows)} products")
    n = canary if canary is not None else (limit or len(rows))
    target = rows[:n]
    est = len(target)*RATE
    started = datetime.now(timezone.utc).isoformat()
    print(f"Run start (UTC): {started}")
    print(f"Batch: {len(target)}   cost: $0 (LOCAL {MODEL})   mode: {'WRITE' if do_write else 'DRY (canary)'}")
    print("-"*70)

    tok = A.token()   # always needed now (live image fetch is an Admin GET)
    from collections import Counter
    split=Counter(); done=0; errors=0; skipped=0; ghosts=0
    for nid,title in target:
        try:
            img, already, tags = live_product(nid, tok)
        except Exception as e:
            ghosts+=1                       # 404 = ghost/staged/deleted -> skip, $0
            if canary is not None: print(f"  GHOST {str(e)[:30]:30s} {title[:34]}")
            continue
        if already:                          # free-pass already tagged it -> don't double-tag
            skipped+=1
            if canary is not None: print(f"  skip(already-tagged)  {title[:40]}")
            continue
        try:
            if not img: raise RuntimeError("no image")
            label = vision_label(img)
        except Exception as e:
            errors+=1
            if canary is not None: print(f"  ERR {str(e)[:40]:40s} {title[:34]}")
            continue
        split[label]+=1; done+=1
        if do_write:
            try: write_tag(nid, A.slug(label), tags, tok)
            except Exception: errors+=1
        if canary is not None:
            print(f"  {label:18s} {title[:44]}")
        if done % 250 == 0:
            print(f"  ...{done} done  ($0 local)")
    ended = datetime.now(timezone.utc).isoformat()
    actual = done*RATE
    print("-"*70)
    print("split:", dict(split))
    print(f"classified {done}, skipped(already-tagged) {skipped}, ghosts(404) {ghosts}, errors {errors}")
    print(f"ACTUAL cost: $0 (local {MODEL}); {done} images classified")
    print(f"Run start {started}  ->  end {ended}")
    if do_write:
        logrun({"wave":"wave2-vision","model":MODEL,"started":started,"ended":ended,
                "classified":done,"skipped_already":skipped,"ghosts":ghosts,"errors":errors,
                "cost_usd":round(actual,4),"split":dict(split)})
        # log to the shared cost ledger too
        try:
            subprocess.run(["node", os.path.expanduser("~/.claude/skills/cost-tracker/log.js"),
                "--provider","gemini","--op","wave2-material-vision",
                "--qty",str(done),"--cost",f"{actual:.4f}"], capture_output=True, timeout=10)
        except Exception: pass
        print(f"logged run -> {RUNLOG}")

if __name__=="__main__":
    main()