`scripts/settlement-vision.py`

100 lines
#!/usr/bin/env python3
"""Settlement post-gen vision gate for Groundworks Hunt Slonem images.
Runs Gemini 2.0 Flash on each image, extracting the 5 settlement detector booleans,
then applies the settlement-verdict combinator: BLOCK only when full Part A
(A1 directional foliage AND A2 open space AND A3 multiple ink colors) is satisfied
AND a Part B element (banana/grape/BIRD/BUTTERFLY) is present, absent an acceptable carve-out.
Rabbits/bunnies are NOT Part B prohibited elements.
"""
import os, sys, json, base64, urllib.request, time, subprocess

KEY = os.environ.get("GEMINI_API_KEY")
if not KEY:
    print("ERR: no GEMINI_API_KEY"); sys.exit(2)
MODEL = "gemini-2.5-flash"
URL = f"https://generativelanguage.googleapis.com/v1beta/models/{MODEL}:generateContent?key={KEY}"

PROMPT = """You are a strict visual detector for a wallcovering settlement gate. Look ONLY at what is actually visible in this repeating pattern image. Answer as JSON with these exact boolean/string fields:
{
 "a1_directional_foliage": <true if there are leaves/fronds/foliage shown in MORE THAN ONE orientation across the repeat, else false>,
 "a2_open_space": <true if there is visible negative/open space BETWEEN foliage motifs (not edge-to-edge coverage), else false>,
 "a3_multiple_ink_colors": <true if the foliage/leaf layer uses more than one ink color (excluding background), else false>,
 "partB_bananas": <true if bananas or banana pods visible>,
 "partB_grapes": <true if grapes visible>,
 "partB_birds": <true if birds visible>,
 "partB_butterflies": <true if butterflies or moths visible>,
 "acceptable_trunk_branch": <true if clearly represented tree trunks or branches visible>,
 "primary_motif": "<one short phrase describing the main visible motif>"
}
Return ONLY the JSON, no prose."""

def gemini(image_bytes, mime="image/jpeg"):
    # downscale big print-master JPEGs so the inline request stays small
    try:
        from io import BytesIO
        from PIL import Image
        im = Image.open(BytesIO(image_bytes)).convert("RGB")
        im.thumbnail((1024,1024))
        buf = BytesIO(); im.save(buf, format="JPEG", quality=85)
        image_bytes = buf.getvalue()
    except Exception:
        pass
    b64 = base64.b64encode(image_bytes).decode()
    body = {"contents":[{"parts":[{"text":PROMPT},{"inline_data":{"mime_type":mime,"data":b64}}]}],
            "generationConfig":{"temperature":0,"maxOutputTokens":2048,"responseMimeType":"application/json"}}
    req = urllib.request.Request(URL, data=json.dumps(body).encode(), headers={"Content-Type":"application/json"})
    try:
        with urllib.request.urlopen(req, timeout=90) as r:
            resp = json.load(r)
    except urllib.error.HTTPError as he:
        raise RuntimeError(f"gemini HTTP {he.code}: {he.read().decode()[:200]}")
    cand = resp["candidates"][0]
    parts = cand.get("content",{}).get("parts",[])
    txt = "".join(p.get("text","") for p in parts).strip()
    if not txt:
        raise RuntimeError(f"empty gemini response finish={cand.get('finishReason')}")
    if txt.startswith("```"):
        txt = txt.split("```")[1]
        if txt.startswith("json"): txt = txt[4:]
    # grab the first {...} block
    s,e = txt.find("{"), txt.rfind("}")
    if s>=0 and e>s: txt = txt[s:e+1]
    return json.loads(txt.strip())

def verdict(d):
    partA = d["a1_directional_foliage"] and d["a2_open_space"] and d["a3_multiple_ink_colors"]
    partB = d["partB_bananas"] or d["partB_grapes"] or d["partB_birds"] or d["partB_butterflies"]
    if partA and partB and not d.get("acceptable_trunk_branch"):
        return "BLOCK"
    if partB and not partA:
        # prohibited element present but not full-tropical composition -> defendant-favorable but flag
        return "NEEDS REVIEW"
    if partA and partB and d.get("acceptable_trunk_branch"):
        return "NEEDS REVIEW"
    return "OK"

items = json.load(open(sys.argv[1]))
out = []
for it in items:
    sku, url = it["mfr_sku"], it["image_url"]
    try:
        p = subprocess.run(["curl","-sL","-A","Mozilla/5.0","-e","https://www.kravet.com/",url],
                           capture_output=True, timeout=90)
        img = p.stdout
        if len(img) < 500:
            raise RuntimeError(f"image too small ({len(img)} bytes) http-fetch failed")
        d = gemini(img)
        v = verdict(d)
    except Exception as e:
        d = {"error": str(e)}; v = "BLOCK"  # unknown => BLOCK per hard rule
    rec = {"mfr_sku": sku, "verdict": v, "detectors": d, "name": it.get("name")}
    out.append(rec)
    print(f"{v:14s} {sku:16s} {it.get('name','')} :: partB_birds={d.get('partB_birds')} partB_butterflies={d.get('partB_butterflies')} motif={d.get('primary_motif','')}")
    time.sleep(0.4)

json.dump(out, open(sys.argv[2],"w"), indent=2)
n_block = sum(1 for r in out if r["verdict"]=="BLOCK")
n_rev = sum(1 for r in out if r["verdict"]=="NEEDS REVIEW")
n_ok = sum(1 for r in out if r["verdict"]=="OK")
print(f"\nSUMMARY: OK={n_ok} NEEDS_REVIEW={n_rev} BLOCK={n_block}  (audit -> {sys.argv[2]})")