Publish snapshot: v1->v2 character port + accumulated fixes
- Tag-builder: port ~1,477 v1 characters into the v2 chip format so the character picker shows 1,486 characters (was 9); 1,390 carry rich per-chip descriptions restructured from each character's own bio. Adds scripts/port_v1_characters_to_v2.py. - Ideogram: rewrite caption to the real Ideogram-4 schema + broaden refusal detection; sampler updates. - Pose-studio, regional-detailer, attention-couple: assorted fixes. - Add validate_templates / test_regions / test_crop_windows scripts.
This commit is contained in:
chrmc40
@@ -0,0 +1,272 @@
|
||||
#!/usr/bin/env python
|
||||
"""
|
||||
Port v1 characters into v2 tag-builder chip format.
|
||||
|
||||
Restructures each character's OWN base_tags + base_natlang paragraph into:
|
||||
- appearance_chip_tags (JSON array of appearance_items.item_tag that slot)
|
||||
- base_extras (tokens with no matching chip, ride alongside)
|
||||
- character_chip_overrides (per-chip rich natlang carved from the paragraph)
|
||||
- natlang_status = 'normalized' (so the v2 picker shows them)
|
||||
- base_tags recomposed from the above (the cached flat projection)
|
||||
|
||||
The LLM call is RESTRUCTURING provided text — it needs no knowledge of the
|
||||
character, so accuracy does not depend on how obscure the character is.
|
||||
|
||||
Runs on the Anthropic API (your usage credits), independent of any Claude Code
|
||||
session limit. Safe by default: --dry-run prints intended writes and changes
|
||||
nothing. Resumable: only touches characters still natlang_status='unprocessed'.
|
||||
|
||||
export ANTHROPIC_API_KEY=sk-ant-...
|
||||
python port_v1_to_v2.py --dry-run --limit 5 # preview
|
||||
python port_v1_to_v2.py --limit 50 # port 50 for real
|
||||
python port_v1_to_v2.py # port all remaining
|
||||
"""
|
||||
import argparse, json, os, re, sys, sqlite3, time
|
||||
|
||||
DB_PATH = r"C:/comfyui/comfyui/custom_nodes/ComfyUI-PromptChain/data/tag-builder/tag-builder.db"
|
||||
MODEL = "claude-sonnet-4-5"
|
||||
PALETTE_GROUPS = ["modifiers", "hair_color", "hair_length", "hair_style", "hair_accessories",
|
||||
"eye_color", "eyes", "skin", "body_type", "breast_size", "ass_size",
|
||||
"body_marks", "fantasy"]
|
||||
|
||||
# Near-synonym tokens that should map onto an existing palette chip instead of
|
||||
# falling to base_extras. Keeps common danbooru variants slottable.
|
||||
VOCAB_ALIASES = {
|
||||
"silver_hair": "white_hair",
|
||||
"platinum_blonde_hair": "blonde_hair",
|
||||
"dark_green_eyes": "green_eyes",
|
||||
"light_brown_hair": "brown_hair",
|
||||
}
|
||||
|
||||
PORT_TOOL = {
|
||||
"name": "emit_port",
|
||||
"description": "Return the v2 port for each character.",
|
||||
"input_schema": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"results": {
|
||||
"type": "array",
|
||||
"items": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"tag": {"type": "string"},
|
||||
"chip_tags": {"type": "array", "items": {"type": "string"},
|
||||
"description": "base_tags tokens that are verbatim palette keys"},
|
||||
"base_extras": {"type": "string", "description": "comma-joined non-palette tokens"},
|
||||
"overrides": {"type": "array", "items": {"type": "object", "properties": {
|
||||
"chip_tag": {"type": "string"}, "natlang": {"type": "string"}},
|
||||
"required": ["chip_tag", "natlang"]},
|
||||
"description": "per-chip rich natlang carved ONLY from the paragraph"},
|
||||
"terse_natlang": {"type": "string"},
|
||||
},
|
||||
"required": ["tag", "chip_tags", "base_extras", "overrides", "terse_natlang"],
|
||||
},
|
||||
}
|
||||
},
|
||||
"required": ["results"],
|
||||
},
|
||||
}
|
||||
|
||||
SYSTEM_RULES = (
|
||||
"You PORT v1 characters into the v2 tag-builder chip format. This is pure RESTRUCTURING "
|
||||
"of each character's OWN provided data. NEVER add an appearance fact not present in that "
|
||||
"character's base_tags or base_natlang. You need no outside knowledge of who they are.\n"
|
||||
"For each character:\n"
|
||||
"1. chip_tags: every base_tags token (weights stripped, identity lead and the character's own "
|
||||
"tag excluded) that is a VERBATIM key in the palette below.\n"
|
||||
"2. base_extras: comma-joined base_tags tokens that are NOT palette keys.\n"
|
||||
"3. overrides: for each chip the paragraph describes more specifically than its generic phrase, "
|
||||
"a short natlang using ONLY paragraph wording. Never invent; never restate the generic; omit "
|
||||
"chips the paragraph does not elaborate.\n"
|
||||
"4. terse_natlang: one line, display name + series only.\n"
|
||||
"Call emit_port with one result per character."
|
||||
)
|
||||
|
||||
|
||||
def strip_weight(tok):
|
||||
# Greedy inner capture so a character tag that itself contains parens
|
||||
# (e.g. "(artoria_pendragon_(fate):1.1)") strips to its bare form, not a
|
||||
# truncated fragment that would leak into base_extras.
|
||||
tok = tok.strip()
|
||||
m = re.match(r"^\((.+):[0-9.]+\)$", tok)
|
||||
if m:
|
||||
return m.group(1)
|
||||
return tok
|
||||
|
||||
|
||||
def load_palette(conn):
|
||||
palette = {}
|
||||
for g in PALETTE_GROUPS:
|
||||
palette[g] = {r["item_tag"]: (r["base_natlang"] or "")
|
||||
for r in conn.execute(
|
||||
"SELECT item_tag, base_natlang FROM appearance_items "
|
||||
"WHERE item_group=? ORDER BY sort_order, item_tag", (g,))}
|
||||
return palette
|
||||
|
||||
|
||||
def palette_keys(palette):
|
||||
keys = set()
|
||||
for g in palette.values():
|
||||
keys.update(g.keys())
|
||||
return keys
|
||||
|
||||
|
||||
def select_pending(conn, limit):
|
||||
q = ("SELECT tag, display, series, base_tags, base_natlang FROM characters "
|
||||
"WHERE natlang_status='unprocessed' AND base_tags IS NOT NULL AND base_tags!=''")
|
||||
if limit:
|
||||
q += f" ORDER BY post_count DESC LIMIT {int(limit)}"
|
||||
return conn.execute(q).fetchall()
|
||||
|
||||
|
||||
def compose_base_tags(tag, base_extras, chip_tags):
|
||||
parts = [f"({tag}:1.1)"]
|
||||
if base_extras.strip():
|
||||
parts.append(base_extras.strip())
|
||||
parts.extend(chip_tags)
|
||||
return ", ".join(p for p in parts if p)
|
||||
|
||||
|
||||
def reconcile(conn, char, result, valid_keys):
|
||||
"""Trust the source base_tags, not the model, for which tokens exist.
|
||||
Re-derive chips/extras mechanically from base_tags so a model slip cannot
|
||||
invent or drop a token; keep only the model's overrides (the real value-add)
|
||||
after confirming each targets a real chip."""
|
||||
tag = char["tag"]
|
||||
tokens = [strip_weight(t) for t in (char["base_tags"] or "").split(",") if t.strip()]
|
||||
chips, extras = [], []
|
||||
for tok in tokens:
|
||||
if tok == tag:
|
||||
continue
|
||||
key = VOCAB_ALIASES.get(tok, tok)
|
||||
if key in valid_keys and key not in chips:
|
||||
chips.append(key)
|
||||
elif key not in valid_keys:
|
||||
extras.append(tok)
|
||||
chip_set = set(chips)
|
||||
overrides = [o for o in result.get("overrides", [])
|
||||
if o.get("chip_tag") in chip_set and (o.get("natlang") or "").strip()]
|
||||
return {
|
||||
"tag": tag,
|
||||
"chip_tags": chips,
|
||||
"base_extras": ", ".join(extras),
|
||||
"overrides": overrides,
|
||||
"terse_natlang": (result.get("terse_natlang") or f"{char['display']}.").strip(),
|
||||
}
|
||||
|
||||
|
||||
def ensure_backup_column(conn):
|
||||
# base_natlang is reduced to a terse identity line on port; preserve the
|
||||
# original rich paragraph so the change is reversible and the dropped
|
||||
# details (skin/height/build) stay available for later enrichment.
|
||||
try:
|
||||
conn.execute("ALTER TABLE characters ADD COLUMN base_natlang_orig TEXT")
|
||||
conn.commit()
|
||||
except sqlite3.OperationalError:
|
||||
pass # already exists
|
||||
|
||||
|
||||
def write_port(conn, port):
|
||||
tag = port["tag"]
|
||||
conn.execute(
|
||||
"UPDATE characters SET base_natlang_orig = COALESCE(base_natlang_orig, base_natlang) WHERE tag=?",
|
||||
(tag,))
|
||||
conn.execute(
|
||||
"UPDATE characters SET appearance_chip_tags=?, base_extras=?, base_natlang=?, "
|
||||
"base_tags=?, natlang_status='normalized' WHERE tag=?",
|
||||
(json.dumps(port["chip_tags"]), port["base_extras"], port["terse_natlang"],
|
||||
compose_base_tags(tag, port["base_extras"], port["chip_tags"]), tag),
|
||||
)
|
||||
conn.execute("DELETE FROM character_chip_overrides WHERE character_tag=?", (tag,))
|
||||
for o in port["overrides"]:
|
||||
conn.execute(
|
||||
"INSERT INTO character_chip_overrides (character_tag, chip_tag, natlang, status) "
|
||||
"VALUES (?,?,?, 'normalized')", (tag, o["chip_tag"], o["natlang"]))
|
||||
|
||||
|
||||
def call_api(client, palette, batch):
|
||||
palette_text = json.dumps(palette, ensure_ascii=False)
|
||||
chars = [{"tag": r["tag"], "display": r["display"], "series": r["series"],
|
||||
"base_tags": r["base_tags"], "base_natlang": r["base_natlang"]} for r in batch]
|
||||
msg = client.messages.create(
|
||||
model=MODEL, max_tokens=4096,
|
||||
system=[
|
||||
{"type": "text", "text": SYSTEM_RULES},
|
||||
# cache_control keeps the (large, identical) palette cheap across calls
|
||||
{"type": "text", "text": "PALETTE (group -> {item_tag: generic_natlang}):\n" + palette_text,
|
||||
"cache_control": {"type": "ephemeral"}},
|
||||
],
|
||||
tools=[PORT_TOOL], tool_choice={"type": "tool", "name": "emit_port"},
|
||||
messages=[{"role": "user", "content":
|
||||
"Port these characters:\n" + json.dumps(chars, ensure_ascii=False)}],
|
||||
)
|
||||
for block in msg.content:
|
||||
if block.type == "tool_use":
|
||||
return block.input.get("results", [])
|
||||
return []
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--dry-run", action="store_true", help="print intended writes, change nothing")
|
||||
ap.add_argument("--limit", type=int, default=0, help="only the N most popular pending chars")
|
||||
ap.add_argument("--batch", type=int, default=8, help="characters per API call")
|
||||
ap.add_argument("--db", default=DB_PATH)
|
||||
args = ap.parse_args()
|
||||
|
||||
conn = sqlite3.connect(args.db)
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.text_factory = lambda x: x.decode("utf-8", "replace") # some rows have bad bytes
|
||||
if not args.dry_run:
|
||||
ensure_backup_column(conn)
|
||||
palette = load_palette(conn)
|
||||
valid_keys = palette_keys(palette)
|
||||
pending = select_pending(conn, args.limit)
|
||||
print(f"{len(pending)} characters to port (dry_run={args.dry_run}, batch={args.batch})")
|
||||
if not pending:
|
||||
return
|
||||
|
||||
client = None
|
||||
if not args.dry_run:
|
||||
from anthropic import Anthropic # pip install anthropic
|
||||
client = Anthropic() # reads ANTHROPIC_API_KEY
|
||||
|
||||
done = 0
|
||||
for i in range(0, len(pending), args.batch):
|
||||
batch = pending[i:i + args.batch]
|
||||
by_tag = {r["tag"]: r for r in batch}
|
||||
if args.dry_run:
|
||||
# No API spend: derive chips/extras mechanically, skip override text.
|
||||
results = [{"tag": r["tag"], "overrides": [], "terse_natlang": f"{r['display']}."} for r in batch]
|
||||
else:
|
||||
for attempt in range(3):
|
||||
try:
|
||||
results = call_api(client, palette, batch)
|
||||
break
|
||||
except Exception as e:
|
||||
print(f" API error ({e}); retry {attempt+1}/3", file=sys.stderr)
|
||||
time.sleep(2 * (attempt + 1))
|
||||
else:
|
||||
print(" batch failed after retries; skipping", file=sys.stderr)
|
||||
continue
|
||||
|
||||
for res in results:
|
||||
char = by_tag.get(res.get("tag"))
|
||||
if not char:
|
||||
continue
|
||||
port = reconcile(conn, char, res, valid_keys)
|
||||
if args.dry_run:
|
||||
print(f" {port['tag']:40} chips={len(port['chip_tags'])} "
|
||||
f"extras=[{port['base_extras']}] overrides={len(port['overrides'])}")
|
||||
else:
|
||||
write_port(conn, port)
|
||||
done += 1
|
||||
if not args.dry_run:
|
||||
conn.commit()
|
||||
print(f" committed {done}/{len(pending)}")
|
||||
|
||||
print("done." if not args.dry_run else "dry-run complete (no changes written).")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,94 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Unit tests for the crop-window remap (core/crop_windows.py).
|
||||
|
||||
Locks the invariants of the most-regressed geometry in the codebase: output
|
||||
dims, the nearest-vs-bilinear mask choice (the confirmed region-edge bleed fix),
|
||||
full-canvas identity, and that distinct windows select distinct source regions
|
||||
(the mask_win != ctrl_win moved-content mechanism). torch-only; run anywhere.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
|
||||
import torch
|
||||
|
||||
_HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
_spec = importlib.util.spec_from_file_location(
|
||||
"crop_windows", os.path.join(_HERE, "core", "crop_windows.py"))
|
||||
_cw = importlib.util.module_from_spec(_spec)
|
||||
_spec.loader.exec_module(_cw)
|
||||
crop = _cw.crop_hint_to_window
|
||||
|
||||
_failures = []
|
||||
|
||||
|
||||
def check(name, cond):
|
||||
print(f" {'PASS' if cond else 'FAIL'} {name}")
|
||||
if not cond:
|
||||
_failures.append(name)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
H, W = 120, 80 # non-square canvas (portrait), the case aspect bugs hide in
|
||||
|
||||
# 1. Output dims are always (sh, sw) regardless of input size or window.
|
||||
src = torch.rand(1, 1, 64, 48)
|
||||
out = crop(src, (10, 20, 70, 100), H, W, 32, 24)
|
||||
check("output dims == (sh, sw)", tuple(out.shape[-2:]) == (32, 24))
|
||||
|
||||
# 2. nearest-exact keeps a binary mask binary (no soft bleed band).
|
||||
mask = torch.zeros(1, 1, H, W)
|
||||
mask[..., 40:80, 20:60] = 1.0
|
||||
out_n = crop(mask, (0, 0, W, H), H, W, 60, 40, mode="nearest-exact")
|
||||
uniq = set(torch.unique(out_n).tolist())
|
||||
check("nearest mask stays binary {0,1}", uniq.issubset({0.0, 1.0}))
|
||||
|
||||
# 3. bilinear DOES introduce an intermediate edge value — proves the mode
|
||||
# actually changes behaviour (so the mask fix is meaningful, not a no-op).
|
||||
# Upscale a hard edge so it must ramp (an exact-2x downsample with the edge
|
||||
# on an even boundary would sample exactly 0/1 and hide the difference).
|
||||
small = torch.zeros(1, 1, 8, 8)
|
||||
small[..., 4:, :] = 1.0
|
||||
up_b = crop(small, (0, 0, 8, 8), 8, 8, 32, 32, mode="bilinear")
|
||||
up_n = crop(small, (0, 0, 8, 8), 8, 8, 32, 32, mode="nearest-exact")
|
||||
check("bilinear softens the edge (mode matters)",
|
||||
bool(((up_b > 0.01) & (up_b < 0.99)).any()))
|
||||
check("nearest stays hard on the same upscale",
|
||||
set(torch.unique(up_n).tolist()).issubset({0.0, 1.0}))
|
||||
|
||||
# 4. Full-canvas window is identity-ish: a uniform field stays uniform.
|
||||
flat = torch.full((1, 1, H, W), 0.5)
|
||||
out_f = crop(flat, (0, 0, W, H), H, W, H, W)
|
||||
check("full-canvas uniform field preserved", torch.allclose(out_f, flat, atol=1e-5))
|
||||
|
||||
# 5. A window over an all-ones sub-region returns ~all-ones; a disjoint
|
||||
# window returns ~all-zeros. This is what makes a tile sample the right
|
||||
# region — and the basis of mask_win != ctrl_win moved-content selection.
|
||||
field = torch.zeros(1, 1, H, W)
|
||||
field[..., 0:40, 0:40] = 1.0 # top-left block is the "content"
|
||||
inside = crop(field, (0, 0, 40, 40), H, W, 16, 16, mode="nearest-exact")
|
||||
outside = crop(field, (40, 60, 80, 120), H, W, 16, 16, mode="nearest-exact")
|
||||
check("window over content -> all ones", bool((inside > 0.5).all()))
|
||||
check("disjoint window -> all zeros", bool((outside < 0.5).all()))
|
||||
|
||||
# 6. Distinct windows select DISTINCT regions of the same source (the
|
||||
# mask_win vs ctrl_win divergence relies on this).
|
||||
grad = torch.linspace(0, 1, W).view(1, 1, 1, W).expand(1, 1, H, W).contiguous()
|
||||
left = crop(grad, (0, 0, 20, H), H, W, 8, 8)
|
||||
right = crop(grad, (60, 0, 80, H), H, W, 8, 8)
|
||||
check("distinct windows -> distinct content", float(left.mean()) < float(right.mean()))
|
||||
|
||||
# 7. Input already at canvas size skips the upscale but still crops+resizes.
|
||||
at_canvas = torch.rand(1, 1, H, W)
|
||||
out_ac = crop(at_canvas, (10, 10, 50, 50), H, W, 20, 20)
|
||||
check("pre-sized input crops to (sh,sw)", tuple(out_ac.shape[-2:]) == (20, 20))
|
||||
|
||||
print(f"\n{'ALL PASS' if not _failures else str(len(_failures)) + ' FAILURE(S)'}")
|
||||
return 1 if _failures else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,124 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Regression tests for regional prompt extraction + figure binding (core/compiler.py).
|
||||
|
||||
Locks the two confirmed correctness fixes:
|
||||
- $block bodies are extracted by brace DEPTH, so a wildcard ({a|b}) inside a
|
||||
region no longer truncates it or leaks the remainder into the global prompt.
|
||||
- a non-object pose JSON (scalar/array/null) degrades gracefully instead of
|
||||
crashing the single-source-of-truth entity resolver.
|
||||
Plus sanity coverage of the name -> id -> order binding ladder and orphan drop.
|
||||
torch-free; run anywhere.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
|
||||
_HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
_spec = importlib.util.spec_from_file_location(
|
||||
"pc_compiler", os.path.join(_HERE, "core", "compiler.py"))
|
||||
_c = importlib.util.module_from_spec(_spec)
|
||||
_spec.loader.exec_module(_c)
|
||||
|
||||
_failures = []
|
||||
|
||||
|
||||
def check(name, cond):
|
||||
print(f" {'PASS' if cond else 'FAIL'} {name}")
|
||||
if not cond:
|
||||
_failures.append(name)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
# ── brace-depth extraction (the HIGH fix) ────────────────────────────────
|
||||
spans = list(_c._iter_region_spans("$mannequin1 { {red|blue} dress }"))
|
||||
check("wildcard brace kept inside region body",
|
||||
len(spans) == 1 and "{red|blue}" in spans[0][1] and "dress" in spans[0][1])
|
||||
|
||||
# global must NOT receive the truncated remainder
|
||||
glob = _c._remove_region_spans("$m1{ {a|b} dress } and a tree",
|
||||
list(_c._iter_region_spans("$m1{ {a|b} dress } and a tree")))
|
||||
check("global free of region body leak", "dress" not in glob and "and a tree" in glob)
|
||||
|
||||
# nested parens + wildcard
|
||||
sp2 = list(_c._iter_region_spans("$hero { (a|{red|blue}) } tail"))
|
||||
check("nested braces fully captured",
|
||||
len(sp2) == 1 and sp2[0][1].strip() == "(a|{red|blue})")
|
||||
|
||||
# adjacent regions still split correctly
|
||||
sp3 = list(_c._iter_region_spans("$a{ x } $b{ y }"))
|
||||
check("adjacent regions both extracted",
|
||||
[s[0] for s in sp3] == ["a", "b"])
|
||||
|
||||
# strip_region_markers keeps nested body inline (flat/non-regional path)
|
||||
flat = _c.strip_region_markers("$m1{ hair {a|b} } scene")
|
||||
check("strip keeps nested body inline",
|
||||
"{a|b}" in flat and "$m1" not in flat and "scene" in flat)
|
||||
|
||||
# unbalanced brace degrades without hanging
|
||||
sp4 = list(_c._iter_region_spans("$x{ no close here"))
|
||||
check("unbalanced region handled", len(sp4) == 1 and "no close" in sp4[0][1])
|
||||
|
||||
# compile_regions end-to-end: region text non-empty, global clean
|
||||
obj = _c.compile_regions("a cat $m1{ red {hat|cap} } in a park")
|
||||
body_ok = obj["regions"] and obj["regions"][0]["text"]
|
||||
check("compile_regions: region body compiled non-empty", bool(body_ok))
|
||||
check("compile_regions: global has outside text, not region body",
|
||||
"cat" in obj["global"] and "park" in obj["global"] and "red" not in obj["global"])
|
||||
|
||||
# ── brace-internal newlines don't comma-pollute a wildcard ───────────────
|
||||
# A wildcard split across lines is layout, not tag separators.
|
||||
p, _, _ = _c.compile_prompt("a woman in a {\n red|blue\n } sundress")
|
||||
check("multi-line wildcard: no mid-phrase comma before the option",
|
||||
p in ("a woman in a red sundress", "a woman in a blue sundress"))
|
||||
# nested wildcard across lines stays clean
|
||||
p, _, _ = _c.compile_prompt("a {\n big | {tiny|small}\n} cat")
|
||||
check("nested multi-line wildcard resolves clean", "," not in p and "cat" in p)
|
||||
# but a top-level newline (lazy tag list) STILL comma-joins
|
||||
p, _, _ = _c.compile_prompt("masterpiece\n{detailed|intricate}\nforest")
|
||||
check("lazy multi-line tag list still comma-joins", p.count(",") == 2)
|
||||
# a multi-line wildcard OPTION keeps internal commas (newline→comma), but no
|
||||
# leading/trailing junk comma from the option edges
|
||||
picks = {_c.compile_prompt("{\na\nb\n|\nc\nd\n}")[0] for _ in range(60)}
|
||||
check("multi-line wildcard options -> 'a, b' / 'c, d'", picks == {"a, b", "c, d"})
|
||||
picks = {_c.compile_prompt("{\nred\n|\nblue\n}")[0] for _ in range(60)}
|
||||
check("split single-tag wildcard -> clean (no edge comma)", picks == {"red", "blue"})
|
||||
|
||||
# ── example $blocks inside // comments are NOT extracted as regions ───────
|
||||
scaffolded = ("// help: $background{ a forest } and $alice{ red dress }, $bob{ blue suit }\n"
|
||||
"$region1 {\na cat\n}\n\n$region2 {\na dog\n}\n\na grassy field")
|
||||
obj = _c.compile_regions(scaffolded)
|
||||
check("comment-scaffold $blocks ignored (only real regions)",
|
||||
[r["name"] for r in obj["regions"]] == ["region1", "region2"])
|
||||
check("comment-scaffold global is the real global",
|
||||
obj["global"] == "a grassy field")
|
||||
|
||||
# ── non-object pose JSON guard (the LOW fix) ─────────────────────────────
|
||||
for payload in ("42", "[]", "null", '"x"', "3.14"):
|
||||
try:
|
||||
names, nfig = _c._pose_entity_index(payload)
|
||||
ok = names == [] and nfig is None
|
||||
except Exception:
|
||||
ok = False
|
||||
check(f"non-object pose {payload!r} degrades to ([],None)", ok)
|
||||
|
||||
# ── binding ladder sanity ────────────────────────────────────────────────
|
||||
pose = '{"regionEntities":[{"kind":"figure","name":"alice"},{"kind":"figure","name":"bob"}]}'
|
||||
regions = [{"id": 1, "name": "bob", "text": "x"}, {"id": 2, "name": "alice", "text": "y"}]
|
||||
idx = _c.region_figure_indices(regions, pose)
|
||||
check("name-match binds across order (bob->1, alice->0)", idx == [1, 0])
|
||||
|
||||
# orphan: a $block whose figure was deleted (name unknown, id past figures)
|
||||
orphan_regions = [{"id": 1, "name": "alice", "text": "x"},
|
||||
{"id": 9, "name": "ghost", "text": "y"}]
|
||||
orphans = _c.region_orphans(orphan_regions, pose)
|
||||
check("deleted-figure block flagged orphan", orphans == [False, True])
|
||||
|
||||
print(f"\n{'ALL PASS' if not _failures else str(len(_failures)) + ' FAILURE(S)'}")
|
||||
return 1 if _failures else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,154 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Structural validator for PromptChain workflow templates.
|
||||
|
||||
Templates are the contract surface between every model architecture and the
|
||||
regional/depth/pose nodes — a dangling link or bad slot index silently breaks a
|
||||
template in the picker. This catches those before they ship; run it on /deploy.
|
||||
|
||||
It validates STRUCTURE, not render semantics (it can't tell a depth ControlNet
|
||||
is wired to the wrong stream — only a render does). Checks:
|
||||
- valid JSON, required top-level fields
|
||||
- scope.type is one of the known kinds, with the fields that kind needs
|
||||
- every node has a string `type`
|
||||
- every connection references existing node indices and non-negative slots
|
||||
- anchorConnections reference existing nodes/slots
|
||||
- template `id` is unique across the directory
|
||||
- duplicate (name + scope) pairs are reported (two indistinguishable entries)
|
||||
|
||||
Exit code 0 = clean, 1 = at least one error. Warnings never fail the build.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
|
||||
TEMPLATES_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||||
"data", "templates")
|
||||
|
||||
SCOPE_TYPES = {"architecture", "family", "version", "model"}
|
||||
SCOPE_REQUIRED = {
|
||||
"architecture": ["architecture"],
|
||||
"family": ["architecture", "family"],
|
||||
"version": ["model_hash"],
|
||||
"model": ["model_name"], # display_name is accepted as an alias below
|
||||
}
|
||||
|
||||
|
||||
def _validate(path: str, errors: list, warnings: list, ids: dict, signatures: dict) -> None:
|
||||
name = os.path.basename(path)
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
tpl = json.load(f)
|
||||
except (OSError, json.JSONDecodeError) as e:
|
||||
errors.append(f"{name}: not valid JSON — {e}")
|
||||
return
|
||||
if not isinstance(tpl, dict):
|
||||
errors.append(f"{name}: top level must be an object")
|
||||
return
|
||||
|
||||
if not tpl.get("name"):
|
||||
errors.append(f"{name}: missing 'name'")
|
||||
tid = tpl.get("id")
|
||||
if not tid:
|
||||
errors.append(f"{name}: missing 'id'")
|
||||
elif tid in ids:
|
||||
errors.append(f"{name}: duplicate id '{tid}' (also in {ids[tid]})")
|
||||
else:
|
||||
ids[tid] = name
|
||||
|
||||
scope = tpl.get("scope")
|
||||
if not isinstance(scope, dict):
|
||||
errors.append(f"{name}: missing/invalid 'scope'")
|
||||
else:
|
||||
st = scope.get("type", "architecture")
|
||||
if st not in SCOPE_TYPES:
|
||||
errors.append(f"{name}: scope.type '{st}' not one of {sorted(SCOPE_TYPES)}")
|
||||
else:
|
||||
req = SCOPE_REQUIRED.get(st, [])
|
||||
for key in req:
|
||||
if key == "model_name" and (scope.get("model_name") or scope.get("display_name")):
|
||||
continue
|
||||
if not scope.get(key):
|
||||
errors.append(f"{name}: scope.type '{st}' requires '{key}'")
|
||||
|
||||
nodes = tpl.get("nodes")
|
||||
if not isinstance(nodes, list) or not nodes:
|
||||
errors.append(f"{name}: 'nodes' must be a non-empty array")
|
||||
nodes = []
|
||||
for i, node in enumerate(nodes):
|
||||
if not isinstance(node, dict) or not isinstance(node.get("type"), str) or not node.get("type"):
|
||||
errors.append(f"{name}: node[{i}] missing a string 'type'")
|
||||
|
||||
n = len(nodes)
|
||||
|
||||
def _check_link(label: str, link: dict, has_from: bool) -> None:
|
||||
if not isinstance(link, dict):
|
||||
errors.append(f"{name}: {label} is not an object")
|
||||
return
|
||||
if has_from:
|
||||
fi = link.get("from_node_idx")
|
||||
if not isinstance(fi, int) or not (0 <= fi < n):
|
||||
errors.append(f"{name}: {label} from_node_idx {fi} out of range (0..{n - 1})")
|
||||
for slot_key in ("from_slot", "to_slot"):
|
||||
sv = link.get(slot_key)
|
||||
if not isinstance(sv, int) or sv < 0:
|
||||
errors.append(f"{name}: {label} {slot_key} {sv!r} must be a non-negative int")
|
||||
ti = link.get("to_node_idx")
|
||||
if not isinstance(ti, int) or not (0 <= ti < n):
|
||||
errors.append(f"{name}: {label} to_node_idx {ti} out of range (0..{n - 1})")
|
||||
|
||||
conns = tpl.get("connections", [])
|
||||
if not isinstance(conns, list):
|
||||
errors.append(f"{name}: 'connections' must be an array")
|
||||
else:
|
||||
for j, link in enumerate(conns):
|
||||
_check_link(f"connections[{j}]", link, has_from=True)
|
||||
|
||||
anchors = tpl.get("anchorConnections", [])
|
||||
if isinstance(anchors, list):
|
||||
for j, link in enumerate(anchors):
|
||||
_check_link(f"anchorConnections[{j}]", link, has_from=False)
|
||||
|
||||
# Two templates the matcher can't tell apart (same name + identical scope)
|
||||
# both surface for the same model — usually an authoring mistake.
|
||||
if isinstance(scope, dict) and tpl.get("name"):
|
||||
sig = (tpl.get("name"), json.dumps(scope, sort_keys=True))
|
||||
signatures[sig].append(name)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if not os.path.isdir(TEMPLATES_DIR):
|
||||
print(f"templates dir not found: {TEMPLATES_DIR}")
|
||||
return 1
|
||||
errors: list = []
|
||||
warnings: list = []
|
||||
ids: dict = {}
|
||||
signatures: dict = defaultdict(list)
|
||||
|
||||
files = sorted(f for f in os.listdir(TEMPLATES_DIR)
|
||||
if f.endswith(".json") and not f.startswith("_"))
|
||||
for f in files:
|
||||
_validate(os.path.join(TEMPLATES_DIR, f), errors, warnings, ids, signatures)
|
||||
|
||||
for sig, names in signatures.items():
|
||||
if len(names) > 1:
|
||||
warnings.append(f"same name+scope on {len(names)} templates: {', '.join(names)} "
|
||||
f"(name={sig[0]!r})")
|
||||
|
||||
print(f"validated {len(files)} templates")
|
||||
for w in warnings:
|
||||
print(f" WARN {w}")
|
||||
for e in errors:
|
||||
print(f" ERROR {e}")
|
||||
if errors:
|
||||
print(f"\n{len(errors)} error(s)")
|
||||
return 1
|
||||
print("OK — no structural errors" + (f" ({len(warnings)} warning(s))" if warnings else ""))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user