`scripts/_regen_v3.py`

110 lines
#!/usr/bin/env python3
"""Targeted _v3 regen for exactly two off-lane briefs (deco-01, texture-01).
Reuses gen_briefs.py machinery (same Replicate SDXL model, poll loop, cost accounting)
but with NEW tight per-brief prompts + negatives, fresh seeds, _v3 suffix.
NEVER overwrites v1/v2. Curator-mode, is_published untouched (no publish here)."""
import os, re, json, time, urllib.request, urllib.error

ROOT = "/Users/macstudio3/Projects/pattern-vault"
OUT = f"{ROOT}/assets/generated"
os.makedirs(OUT, exist_ok=True)

tok = None
with open(os.path.expanduser("~/Projects/secrets-manager/.env")) as f:
    m = re.search(r"REPLICATE_API_TOKEN=([A-Za-z0-9_-]+)", f.read())
    if m: tok = m.group(1)
assert tok, "no replicate token"

# NEW tight prompts + per-brief negatives (verbatim from task).
JOBS = {
    "BRIEF-2026-deco-01": {
        "prompt": ("Art-Deco geometric wallpaper, repeating fan and sunburst arcs, stepped chevrons and "
                   "concentric semicircles, crisp symmetrical linework in metallic gold on a CHARTREUSE "
                   "(yellow-green) ground, 1920s Gatsby geometric, flat vector, seamless tile, NO plants, "
                   "NO leaves, NO florals, NO figures. "
                   "seamless repeating wallpaper pattern, tileable, flat lay, even lighting, high-end wallcovering."),
        "negative": ("palm, fronds, leaves, botanical, flowers, birds, animals, photorealistic, "
                     "text, watermark, signature, logo, blurry, low quality, jpeg artifacts, seams, "
                     "mismatched edges, frame, border, vignette"),
        "w": 1024, "h": 1024,
    },
    "BRIEF-2026-texture-01": {
        "prompt": ("Photoreal grasscloth wallcovering texture, coarse woven natural fiber, pronounced "
                   "horizontal slubs and irregular reed strands, visible weave, warm greige, seamless tile, "
                   "high detail. "
                   "seamless repeating wallpaper pattern, tileable, flat lay, even lighting, high-end wallcovering."),
        "negative": ("smooth, flat, blurry, pattern, motif, floral, "
                     "text, watermark, signature, logo, low quality, jpeg artifacts, seams, "
                     "mismatched edges, frame, border, vignette"),
        "w": 1024, "h": 1024,
    },
}

UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
def post(url, payload):
    hdr = {"Authorization": f"Bearer {tok}", "Content-Type": "application/json", "User-Agent": UA}
    req = urllib.request.Request(url, data=json.dumps(payload).encode(), headers=hdr, method="POST")
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.load(r)

def get(url):
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {tok}", "User-Agent": UA})
    with urllib.request.urlopen(req, timeout=60) as r:
        return json.load(r)

MODEL = "stability-ai/sdxl"
VERSION = get(f"https://api.replicate.com/v1/models/{MODEL}")["latest_version"]["id"]
print(f"using {MODEL}@{VERSION[:12]}", flush=True)

results = []
total_secs = 0.0

# Generate exactly 2 variants per brief, both tagged _v3 (a/b) so we never touch v1/v2.
# Fresh seeds 303/404 so v3 differs from v1(101)/v2(202).
VARIANTS = [("v3a", 303), ("v3b", 404)]
for bid, spec in JOBS.items():
    for vlabel, seed in VARIANTS:
        tag = f"{bid}_{vlabel}"
        payload = {"version": VERSION, "input": {
            "prompt": spec["prompt"], "negative_prompt": spec["negative"],
            "width": spec["w"], "height": spec["h"], "num_inference_steps": 30,
            "guidance_scale": 7.5, "seed": seed, "refine": "no_refiner"
        }}
        ok = False
        for attempt in range(4):
            try:
                pred = post("https://api.replicate.com/v1/predictions", payload); ok = True; break
            except urllib.error.HTTPError as e:
                body = e.read().decode()[:150]
                if e.code == 429:
                    print(f"[{tag}] 429 throttled, backoff {6*(attempt+1)}s", flush=True)
                    time.sleep(6*(attempt+1)); continue
                print(f"[{tag}] HTTP {e.code}: {body}", flush=True); break
        if not ok:
            results.append({"tag": tag, "status": "error"}); continue
        for _ in range(60):
            if pred.get("status") in ("succeeded", "failed", "canceled"): break
            time.sleep(2); pred = get(pred["urls"]["get"])
        st = pred.get("status")
        secs = (pred.get("metrics") or {}).get("predict_time", 0) or 0
        total_secs += secs
        if st == "succeeded":
            out = pred["output"]
            imgurl = out[0] if isinstance(out, list) else out
            dest = f"{OUT}/{tag}.png"
            urllib.request.urlretrieve(imgurl, dest)
            sz = os.path.getsize(dest)
            print(f"[{tag}] OK {secs:.1f}s -> {dest} ({sz//1024}KB)", flush=True)
            results.append({"tag": tag, "brief": bid, "path": dest, "status": "ok",
                            "predict_time": secs, "prompt": spec["prompt"], "seed": seed})
        else:
            print(f"[{tag}] {st}: {pred.get('error')}", flush=True)
            results.append({"tag": tag, "brief": bid, "status": st})
        time.sleep(1.5)

RATE = 0.000725
cost = total_secs * RATE
print(f"\n=== TOTAL predict_time {total_secs:.1f}s  est cost ${cost:.4f} (rate ${RATE}/s A40) ===", flush=True)
json.dump({"results": results, "total_predict_secs": total_secs, "est_cost_usd": round(cost, 4)},
          open(f"{OUT}/_regen_v3_manifest.json", "w"), indent=2)