decisions/review/build_reel.py
#!/usr/bin/env python3
"""Assemble docs/review/STORY_REEL.md from the spine sources.
Deterministic build: body prose extracted BYTE-IDENTICAL, transitions pulled
verbatim, one missing seam (Prologue->1) derived from carry-forward blocks and
flagged. Self-verifies: byte-identity of every body block, full seam coverage,
Prologue/Epilogue bounded. Never hand-edit the reel; rebuild instead.
"""
import csv, re, os, sys, glob
try:
sys.stdout.reconfigure(encoding="utf-8")
except Exception:
pass
_HERE = os.path.dirname(os.path.abspath(__file__))
REPO = os.path.abspath(os.path.join(_HERE, "..", ".."))
SPINE = os.path.join(REPO, "docs", "spine")
TRANS = os.path.join(SPINE, "transitions")
OUT_DIR = os.path.join(REPO, "docs", "review")
HEADER = os.path.join(_HERE, "reel_header.md")
# ---- node sequence -----------------------------------------------------------
NODE_FILES = {} # node-key -> filename
SEQ = ["PRO"] + list(range(1, 78)) + ["EPI"]
NODE_FILES["PRO"] = "CH_PROLOGUE.md"
NODE_FILES["EPI"] = "CH_EPILOGUE.md"
for n in range(1, 78):
NODE_FILES[n] = f"CH_{n:02d}.md"
CHAP_ID = {"PRO": "CH_PROLOGUE", "EPI": "CH_EPILOGUE"}
for n in range(1, 78):
CHAP_ID[n] = f"CH_{n:02d}"
def read(path):
with open(path, encoding="utf-8") as f:
return f.read()
# ---- body-prose extraction (byte-identical) ----------------------------------
def extract_body(text):
"""Return the exact prose text between '## Body prose' and the next '## '."""
lines = text.split("\n")
start = None
for i, ln in enumerate(lines):
if ln.strip() == "## Body prose":
start = i + 1
break
if start is None:
raise RuntimeError("no '## Body prose' header")
end = None
for j in range(start, len(lines)):
if re.match(r"^##\s", lines[j]):
end = j
break
if end is None:
raise RuntimeError("no closing '## ' after Body prose")
block = "\n".join(lines[start:end])
return block.strip("\n"), start, end
def h1_beat(text):
for ln in text.split("\n"):
if ln.startswith("# "):
beat = ln[2:].strip()
beat = re.sub(r"^(Chapter\s+\d+|Prologue|Epilogue)\s*[—:-]\s*", "", beat)
return beat
return ""
def care_tier(text):
m = re.search(r"cultural_depiction_tier\s+([A-Za-z]+)", text)
return m.group(1) if m else "standard"
def threads_active(text):
"""Thread numbers listed in the Threads-advancing/at-origin/resolving section."""
lines = text.split("\n")
start = None
for i, ln in enumerate(lines):
if re.match(r"^##\s+Threads\b", ln):
start = i + 1
break
if start is None:
return ""
nums = []
for j in range(start, len(lines)):
if re.match(r"^##\s", lines[j]):
break
m = re.match(r"^-\s*Thread\s+(\d+)\b", lines[j])
if m:
nums.append(m.group(1))
return ",".join(nums)
# ---- CSV labels --------------------------------------------------------------
csv_path = glob.glob(os.path.join(REPO, "registries", "T0_Chapter_Index*", "*.csv"))[0]
LBL = {}
with open(csv_path, encoding="utf-8-sig", newline="") as f:
for row in csv.reader(f):
if not row or row[0] == "chapter_id":
continue
LBL[row[0]] = row
JARGON = re.compile(r"Vimana|mechanic|controlled-time-travel|world-state|narrative design|Hard Line|referent|precedent|first-visit composite|per CVD", re.I)
def strip_build_jargon(s):
# drop parentheticals that contain build jargon (keep legit ones like dates)
s = re.sub(r"\([^()]*\)", lambda m: "" if JARGON.search(m.group(0)) else m.group(0), s)
# cut a trailing em-dash / double-dash scope note whose remainder is build jargon
for sep in (" — ", " -- "):
if sep in s:
head, tail = s.split(sep, 1)
if JARGON.search(tail):
s = head
return re.sub(r"\s+", " ", s).strip().rstrip(",").strip()
def clean_region(s):
"""Display region: drop internal build-QA scope-notes and coordinates."""
s = re.sub(r"\[[^\]]*\]", "", s) # [Scope note ...], [POPULATE...]
s = re.split(r"\s+--\s+|\s+—\s+PRIMARY|;|\bPRIMARY\b|\(~", s)[0]
s = strip_build_jargon(s)
return re.sub(r"\s+", " ", s).strip().rstrip(",").strip()
def clean_era(s):
s = re.sub(r"\[[^\]]*\]", "", s)
s = strip_build_jargon(s)
return re.sub(r"\s+", " ", s).strip()
def clean_type(s):
s = s.strip()
if "POPULATE" in s or not s:
return "—"
return s.replace("_derived", "")
def label_line(node, care):
cid = CHAP_ID[node]
r = LBL[cid]
title, reg1, reg2, era, typ, hard = r[2], r[4], r[5], r[6], r[8], r[38]
region = clean_region(reg1) + ((" / " + clean_region(reg2)) if reg2.strip() else "")
num = {"PRO": "Prologue", "EPI": "Epilogue"}.get(node, f"Ch {node}")
return f"{num} · {title} | {region} | era: {clean_era(era)} | type: {clean_type(typ)} | care: {care} | hard-line: {hard}"
def arc_row(node, threads):
cid = CHAP_ID[node]
r = LBL[cid]
num = {"PRO": "Pro", "EPI": "Epi"}.get(node, str(node))
beat = h1_beat(BODY_SRC[node])
boss = r[14].strip() or "—"
hard = "HL" if r[38].strip().upper() in ("TRUE", "YES", "1") else "—"
return f"{num} | {clean_region(r[4])} | {clean_era(r[6])} | {clean_type(r[8])} | {threads or '—'} | {hard} | {boss} | {beat}"
# ---- transitions -------------------------------------------------------------
def norm_to(tok):
tok = tok.strip()
if tok.lower().startswith("the epilogue") or tok.lower() == "epilogue":
return "EPI"
m = re.match(r"(\d+)", tok)
return int(m.group(1)) if m else None
SEAMS = {} # (from,to) -> verbatim text
SEAM_HDR = re.compile(r"^###\s+(?:Ch\s+(\d+)|(Prologue))\s+to\s+(?:Ch\s+(\d+)|the\s+Epilogue)\b", re.I)
def _seam_key(ln):
m = SEAM_HDR.match(ln)
if not m:
return None
frm = int(m.group(1)) if m.group(1) else "PRO"
to = int(m.group(3)) if m.group(3) else "EPI"
return (frm, to)
def parse_format_A(text):
lines = text.split("\n")
idxs = [i for i, ln in enumerate(lines) if SEAM_HDR.match(ln)]
for i in idxs:
key = _seam_key(lines[i])
# block from header to next '### ' or '## '
end = len(lines)
for j in range(i + 1, len(lines)):
if re.match(r"^###\s", lines[j]) or re.match(r"^##\s", lines[j]):
end = j
break
SEAMS[key] = "\n".join(lines[i:end]).strip("\n")
def parse_format_B(text):
for ln in text.split("\n"):
m = re.match(r"^-\s+Ch\s+(\d+)\s+→\s+(?:Ch\s+)?(\d+|the Epilogue|the epilogue)\b", ln)
if m:
frm = int(m.group(1))
to = norm_to(m.group(2))
SEAMS[(frm, to)] = ln.rstrip("\n")
for fn in sorted(os.listdir(TRANS)):
if not fn.endswith(".md"):
continue
t = read(os.path.join(TRANS, fn))
if any(SEAM_HDR.match(ln) for ln in t.split("\n")):
parse_format_A(t)
else:
parse_format_B(t)
# ---- load all bodies ---------------------------------------------------------
BODY_SRC = {}
BODY_BOUNDS = {}
CARE = {}
THREADS = {}
for node in SEQ:
text = read(os.path.join(SPINE, NODE_FILES[node]))
body, s, e = extract_body(text)
BODY_SRC[node] = text # store full text for h1 beat
BODY_BOUNDS[node] = (body, s, e)
CARE[node] = care_tier(text)
THREADS[node] = threads_active(text)
def body_of(node):
return BODY_BOUNDS[node][0]
# ---- derive the missing Prologue->1 seam -------------------------------------
def extract_section(text, header_pat):
lines = text.split("\n")
start = None
for i, ln in enumerate(lines):
if re.match(header_pat, ln):
start = i + 1
break
if start is None:
return ""
end = len(lines)
for j in range(start, len(lines)):
if re.match(r"^##\s", lines[j]):
end = j
break
return "\n".join(lines[start:end]).strip("\n")
pro_text = BODY_SRC["PRO"]
ch1_text = BODY_SRC[1]
pro_carry = extract_section(pro_text, r"^##\s+End-state carry-forward")
ch1_anchor = extract_section(ch1_text, r"^##\s+Continuity anchor")
derived = (
"- Ch Prologue → Ch 1 [carry-forward-derived] "
"(no authored transition exists for this seam; reconstructed from the Prologue's "
"End-state carry-forward and Ch 1's Continuity anchor)\n\n"
"Departure (Prologue End-state carry-forward):\n" + pro_carry + "\n\n"
"Arrival (Ch 1 Continuity anchor):\n" + ch1_anchor
)
DERIVED = set()
if ("PRO", 1) not in SEAMS:
SEAMS[("PRO", 1)] = derived
DERIVED.add(("PRO", 1))
# ---- verification ------------------------------------------------------------
errors = []
# 1. seam coverage
missing = []
for a, b in zip(SEQ, SEQ[1:]):
if (a, b) not in SEAMS:
missing.append((a, b))
if missing:
errors.append(f"MISSING SEAMS: {missing}")
# 2. body prose present + re-extract byte-identity
for node in SEQ:
text = read(os.path.join(SPINE, NODE_FILES[node]))
fresh, _, _ = extract_body(text)
if fresh.strip("\n") != body_of(node):
errors.append(f"BODY MISMATCH at {node}")
if not body_of(node).strip():
errors.append(f"EMPTY BODY at {node}")
# 3. word-count soft cross-check vs CSV col[36]
wc_notes = []
for node in SEQ:
csv_wc = LBL[CHAP_ID[node]][36].strip()
actual = len(body_of(node).split())
if csv_wc.isdigit():
d = abs(int(csv_wc) - actual)
if d > max(60, int(int(csv_wc) * 0.15)):
wc_notes.append(f"{CHAP_ID[node]}: csv={csv_wc} actual={actual} (Δ{d})")
# ---- assemble ----------------------------------------------------------------
arc_lines = [arc_row(node, THREADS[node]) for node in SEQ]
arc_table = "```\n" + "\n".join(arc_lines) + "\n```"
hdr = read(HEADER).replace("[ARC_MAP_TABLE]", arc_table)
parts = [hdr.rstrip("\n"), "\n\n" + "=" * 78 + "\n= THE STORY — player-experienced order (Prologue → Epilogue)\n" + "=" * 78 + "\n"]
def seam_block(a, b):
txt = SEAMS[(a, b)]
txt = re.sub(r"[ \t]*\[SRC:[^\]]*\]", "", txt) # story-only read: drop citation tags from seams
an = {"PRO": "Prologue", "EPI": "Epilogue"}.get(a, f"Ch {a}")
bn = {"PRO": "Prologue", "EPI": "Epilogue"}.get(b, f"Ch {b}")
derived_tag = " [carry-forward-derived]" if (a, b) in DERIVED else ""
return (f"\n── TRANSITION · {an} → {bn}{derived_tag} ──\n\n{txt}\n")
BAR = "━" * 78
for idx, node in enumerate(SEQ):
lbl = label_line(node, CARE[node])
parts.append(f"\n{BAR}\n▌ {lbl}\n{BAR}\n\n{body_of(node)}\n")
if idx < len(SEQ) - 1:
nxt = SEQ[idx + 1]
parts.append(seam_block(node, nxt))
reel = "\n".join(parts).rstrip("\n") + "\n"
os.makedirs(OUT_DIR, exist_ok=True)
out_path = os.path.join(OUT_DIR, "STORY_REEL.md")
with open(out_path, "w", encoding="utf-8", newline="\n") as f:
f.write(reel)
# ---- report ------------------------------------------------------------------
total_words = sum(len(body_of(n).split()) for n in SEQ)
print("=== REEL BUILD REPORT ===")
print(f"nodes: {len(SEQ)} (Prologue + 77 + Epilogue = 79)")
print(f"seams: {len(SEQ)-1} required; authored={len(SEQ)-2}; derived=1 (Prologue→1)")
print(f"body-prose total words: {total_words:,}")
print(f"reel bytes: {len(reel.encode('utf-8')):,}")
print(f"written: {out_path}")
if wc_notes:
print("\n-- word-count soft-check notes (source-vs-CSV; informational) --")
for n in wc_notes:
print(" " + n)
else:
print("\nword-count soft-check: all within tolerance")
print("\n=== VERIFICATION ===")
if errors:
print("FAIL:")
for e in errors:
print(" " + e)
sys.exit(1)
else:
print("PASS: full seam coverage (78/78), every body block byte-identical to source, no empty bodies.")