recordstoreguy/tools/make_flow_bank.py

#!/usr/bin/env python3
"""Compile production/flow/*_SHOTS.md into Flow Auto-Pilot prompt banks.

The flowext extension (v2 + attach patch) attaches each task's saved character
cards itself (task "characters" field drives the + / Characters / Add-to-Prompt
dance), then types the prompt. Each prompt ALSO embeds the characters' visual
descriptions so a silently-failed attach degrades consistency instead of losing
the character entirely. Model + output-count are still one-time composer state,
so there are two banks: 01 fast (set Veo Fast, 1x) and 02 quality (Veo Quality,
1x). One take per shot on pass one — review, then requeue what needs more takes.

Usage:
  make_flow_bank.py               # write banks to ~/Documents/flowext/banks/rsg/
  make_flow_bank.py --enqueue N   # POST bank N's tasks to localhost:8017/enqueue
"""
import json
import re
import sys
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
SHEETS = ["TRACK1", "TRACK2", "TRACK3", "TRACK4", "TRACK5", "TRACK6", "TRACK7",
          "SHIFTS", "EXTRAS"]
BANKS = Path.home() / "Documents/flowext/banks/rsg"
SKIP = {"Logging", "t1_whitelabel", "attract_loop", "intro_qual"}  # done or cut
TAKES = 1  # pass one; requeue rejects for more takes
SERVER = "http://localhost:8017"

# Compact visual locks from production/flow/CHARACTERS.md — embedded per prompt
# as insurance for the card attach. Keys = names as saved in the Flow project.
CHAR_LOCK = {
    "Dez": "Dez is a lanky 20-something record store clerk: tall messy black "
           "quiff, olive skin, thick expressive eyebrows, open red flannel shirt "
           "over a faded black band t-shirt, rolled dark jeans, scuffed white "
           "hi-top sneakers, oversized silver headphones around his neck, "
           "rubber-limbed expressive acting.",
    "The Dead Wax": "The Dead Wax is a towering villain made of glossy black "
           "molten liquid vinyl dripping upward, limbs of dripping turntable "
           "tonearms with chrome needle tips, its face a slowly spinning 12-inch "
           "record with a hypnotic red label eye, groove lines rippling across "
           "its wet surface, room-filling and looming.",
    "Mildred": "Mildred is a fat gray-striped tabby store cat with half-closed "
           "judgmental eyes and a white chest, utterly indifferent.",
    "The Boring Guy": "The Boring Guy is a beige man in his 50s: tucked-in taupe "
           "polo, pleated khakis, gray combover, lanyard with an unreadable "
           "badge, reusable coffee cup, entirely unremarkable face, oblivious to "
           "any catastrophe.",
    "The Name Dropper": "The Name Dropper is a smug man in an expensive vintage "
           "leather jacket over a deadstock-clean band t-shirt, designer "
           "glasses, precisely groomed stubble, phone always in one hand.",
    "The Postman": "The Postman is a postal worker in an immaculate uniform: "
           "pressed shorts, high socks, mirrored aviator sunglasses, an enormous "
           "parcel-laden satchel, granite jaw, terrifying procedural calm.",
    "The Accountant": "The Accountant is a gaunt figure in a slightly-too-tight "
           "gray suit, green banker's visor, sleeve garters, round glasses "
           "catching the light so his eyes are never visible, leather ledger "
           "under one arm, red pen behind his ear.",
}
ALIAS = {"The Name-Dropper": "The Name Dropper"}  # sheet spelling -> Flow card name


def parse():
    shots = []  # (id, chars tuple, quality bool, prompt)
    for name in SHEETS:
        text = (ROOT / f"production/flow/{name}_SHOTS.md").read_text()
        for m in re.finditer(r"^## (.+?)\n(.*?)(?=^## |\Z)", text, re.M | re.S):
            header, body = m.group(1), m.group(2)
            sid = header.split()[0]
            if sid in SKIP:
                continue
            quality = "QUALITY" in header
            am = re.search(r"\[attach: ([^\]]+)\]", header) or \
                 re.search(r"\[([^\]]+)\]", header)
            chars = () if not am or am.group(1) == "no characters" else \
                tuple(ALIAS.get(c.strip(), c.strip()) for c in am.group(1).split(","))
            for c in chars:
                assert c in CHAR_LOCK, f"{sid}: unknown character {c!r}"
            lines = [l.strip() for l in body.strip().splitlines() if l.strip()]
            if lines and lines[0].startswith("(") and lines[0].endswith(")"):
                lines = lines[1:]  # production note, not paste text
            prompt = re.sub(r"\s+", " ", " ".join(lines)).strip()
            if len(prompt) < 80:  # header wrap fragments / non-prompt sections
                print(f"  !! skipping {sid}: body too short ({len(prompt)} ch)")
                continue
            lock = " ".join(CHAR_LOCK[c] for c in chars)
            shots.append((sid, chars, quality, (lock + " " + prompt).strip()))
    return shots


def main():
    shots = parse()
    BANKS.mkdir(parents=True, exist_ok=True)
    for old in BANKS.glob("*.jsonl"):
        old.unlink()
    banks = [("01__fast.jsonl", "Veo 3.1 - Fast", [s for s in shots if not s[2]]),
             ("02__quality.jsonl", "Veo 3.1 - Quality", [s for s in shots if s[2]])]
    for fname, model, rows in banks:
        with open(BANKS / fname, "w") as f:
            for sid, chars, _, prompt in rows:
                for take in range(1, TAKES + 1):
                    f.write(json.dumps({
                        "id": f"rsg_{sid}__t{take}", "kind": "video", "model": model,
                        "title": f"{sid} take {take}", "prompt": prompt,
                        "characters": list(chars), "category": "rsg"}) + "\n")
        gens = len(rows) * TAKES
        cr = gens * (100 if "Quality" in model else 10)
        print(f"{fname}: {len(rows)} shots x{TAKES} = {gens} gens ~{cr} cr")
    if len(sys.argv) > 2 and sys.argv[1] == "--enqueue":
        bank = sorted(BANKS.glob(f"{int(sys.argv[2]):02d}__*.jsonl"))[0]
        for line in bank.read_text().splitlines():
            req = urllib.request.Request(SERVER + "/enqueue", line.encode(),
                                         {"Content-Type": "application/json"})
            urllib.request.urlopen(req, timeout=5).read()
        print(f"enqueued {bank.name}")


if __name__ == "__main__":
    main()