#!/usr/bin/env python3 """ script_breakdown.py — Script Breakdown Extraction Pipeline Scans episodes and characters.md to extract every visual production asset. Writes structured JSON conforming to breakdown_schema.json. Usage: python3 script_breakdown.py /leviathan/ # Full (all episodes) python3 script_breakdown.py /leviathan/ --episodes 1-10 # Range python3 script_breakdown.py /leviathan/ --episodes 5 # Single python3 script_breakdown.py /leviathan/ --report # Human-readable summary python3 script_breakdown.py /leviathan/ --shots # Shot estimates only python3 script_breakdown.py /leviathan/ --status # Lock progress summary Exit codes: 0 = success, 1 = partial (warnings), 2 = error """ # ╔════════════════════════════════════════════════════════════════════╗ # ║ DEPRECATED — Superseded by Starsend equivalents (Feb 2026). ║ # ║ Kept alive for Recoil agent protocols + referencing scripts. ║ # ║ Do NOT delete until agents/breakdown_agent.md, storyboard_agent, ║ # ║ engine_checks/structural.py, and batch_threepass.py are updated. ║ # ╚════════════════════════════════════════════════════════════════════╝ import argparse import json import re import sys from datetime import datetime, timezone from pathlib import Path from dataclasses import dataclass, field from typing import Dict, List, Optional, Set # ── Constants ────────────────────────────────────────────────────────────── BEAT_NAMES = ["THE HOOK", "THE SETUP", "THE ESCALATION", "THE TURN", "THE CLIFFHANGER"] # Promptable SFX keywords (generated directly in frame) # NOTE: Uses word-boundary matching (\b) to prevent substring false positives. # Removed overly general terms: # "glow", "flicker" → lighting effects (captured in lighting_notes) # "rust", "corrosion" → environmental texture (captured in location descriptions) # "flash" → too ambiguous (camera flash, flashback, etc.) SFX_KEYWORDS = [ "fire", "flame", "smoke", "sparks", "spark", "dust", "fog", "mist", "debris", "explosion", "rain", "steam", "haze", "snow", "ice", "water", "drip", "leak", "burst", "scatter", "crumble", "shatter", ] # Exclusion patterns for metaphorical/non-visual uses of SFX keywords. # If the line matches an exclusion for the detected keyword, skip it. SFX_EXCLUSIONS = { "fire": [r'\bline of fire\b', r'\bopen\s*fire\b', r'\bunder fire\b', r'\bcease.?fire\b', r'\bcross.?fire\b', r'\bgun.?fire\b', r'\bfireflies?\b', r'\bfire\s*sale\b'], "ice": [r'\bon ice\b', r'\bthin ice\b', r'\bice.?cold\b'], "dust": [r'\bbite the dust\b', r'\bdust\s*off\b'], "rain": [r'\brain\s*check\b'], "water": [r'\bwater\s*down\b', r'\bwaterfall', r'\bwater\s*tight\b', r'\bwater\s*mark\b'], } # Post-composite VFX keywords (added in post-production) # NOTE: Removed "display" (too general — tactical display, display case, etc.) # and "glitch" (commonly used as nickname/dialogue, not VFX). VFX_KEYWORDS = [ "ar overlay", "hud", "hologram", "holographic", "data stream", "targeting reticle", "reticle", "readout", "interface", "overlay", "text overlay", "countdown", "timer display", "scan", "scanning", "digital", "static", ] # Prop detection stoplist (verbs, sound effects, and non-prop words that appear in ALL CAPS) VERB_STOPLIST = { # Verbs "pulls", "pushes", "grips", "grabs", "reaches", "touches", "finds", "opens", "closes", "drops", "lifts", "holds", "takes", "gives", "throws", "catches", "breaks", "cuts", "slams", "strikes", "hits", "swings", "turns", "moves", "steps", "runs", "walks", "falls", "climbs", "jumps", "slides", "snaps", "crashes", "lands", "fires", "pulls", "presses", "wraps", "yanks", "twists", "rips", "tears", "locks", "flips", "digs", "shoves", "drags", "kicks", "pins", "seize", "convulses", "die", "dies", "hums", "buzzes", # Sound effects / emphasis words (often ALL CAPS in scripts) "pulses", "groans", "hisses", "screams", "whirr", "clicks", "cracks", "thuds", "bangs", "booms", "roars", "whistles", "rattles", "creaks", "snaps", "pops", "shatters", "rumbles", "echoes", "beeps", "chirps", "whines", "drones", "buzzes", "clangs", "squeals", "grinds", "flicker", "flickers", "sparks", "explode", "explodes", "hum", "hums", "humming", # Singular sound effect forms "whine", "thud", "bang", "clang", "groan", "howl", "crack", "click", "shriek", "crash", "sound", # Non-prop nouns that appear capitalized "oxygen", "credit", "collection", "squad", "mins", "minutes", "sector", "level", "grid", "mark", "index", "alpha", "beta", "night", "day", "continuous", "later", "dawn", "dusk", "man", "woman", "boy", "girl", "figure", "body", "voice", # NOTE: Do NOT hardcode project-specific character names here. # Character names are loaded dynamically from characters.md via known_characters set. } # Beat duration in seconds (for shot estimation) BEAT_DURATIONS = { "THE HOOK": 5, "THE SETUP": 10, "THE ESCALATION": 25, "THE TURN": 30, "THE CLIFFHANGER": 20, } # Shots per second estimate (for shot counts) SHOTS_PER_EPISODE_ESTIMATE = 20 # ── Data Structures ──────────────────────────────────────────────────────── @dataclass class CharacterData: display_name: str = "" episodes: List[int] = field(default_factory=list) episode_count: int = 0 first_appearance: int = 0 dialogue_count: int = 0 visual_description: str = "" color_palette: List[str] = field(default_factory=list) story_days_present: List[int] = field(default_factory=list) wardrobe: Dict = field(default_factory=dict) hair_makeup: Dict = field(default_factory=dict) state_changes: List[Dict] = field(default_factory=list) signature_props: List[str] = field(default_factory=list) reference_status: str = "not_started" reference_images: Dict = field(default_factory=lambda: { "front": None, "profile": None, "three_quarter": None, "full_body": None, "back": None, "_flux2_slot_map": "Slots 1-5: character angles (front, profile, 3/4, full body, back). Slot 6: signature props. Slots 7-8: environment. Slot 9: lighting. Slot 10: pose/structure." }) prompts: Dict = field(default_factory=lambda: { "reference": None, "flux2": None }) @dataclass class LocationData: type: str = "INT" episodes: List[int] = field(default_factory=list) episode_count: int = 0 description_samples: List[str] = field(default_factory=list) lighting_notes: List[str] = field(default_factory=list) color_palette: List[str] = field(default_factory=list) reference_status: str = "not_started" reference_images: Dict = field(default_factory=lambda: { "wide_establishing": None, "detail_texture": None }) prompts: Dict = field(default_factory=lambda: { "reference": None, "flux2": None }) @dataclass class PropData: display_name: str = "" owner: Optional[str] = None episodes: List[int] = field(default_factory=list) episode_count: int = 0 description_samples: List[str] = field(default_factory=list) states: List[str] = field(default_factory=lambda: ["default"]) confidence: str = "medium" reference_status: str = "not_started" reference_images: Dict = field(default_factory=lambda: { "default": None, "active_state": None }) prompts: Dict = field(default_factory=lambda: { "reference": None, "flux2": None }) @dataclass class SFXElement: display_name: str = "" type: str = "promptable" episodes: List[int] = field(default_factory=list) episode_count: int = 0 description_samples: List[str] = field(default_factory=list) production_method: str = "prompt_directly" @dataclass class VFXElement: display_name: str = "" type: str = "post_composite" episodes: List[int] = field(default_factory=list) episode_count: int = 0 description_samples: List[str] = field(default_factory=list) colors: List[str] = field(default_factory=list) production_method: str = "post_composite" # ── Path Resolution ──────────────────────────────────────────────────────── def resolve_project_path(project_arg: str) -> Path: """Resolve project path from argument. Handles relative and absolute paths.""" # Find the Recoil root script_dir = Path(__file__).resolve().parent # tools/ -> go up two levels to Recoil root if script_dir.name == "tools" and script_dir.parent.name == "recoil": root = script_dir.parent.parent else: root = Path.cwd() # Strip leading/trailing slashes for folder name matching project_name = project_arg.strip("/").strip("\\") # Try as subfolder of root candidate = root / project_name if candidate.is_dir(): return candidate # Try as absolute path abs_path = Path(project_arg) if abs_path.is_dir(): return abs_path # Try from cwd cwd_path = Path.cwd() / project_name if cwd_path.is_dir(): return cwd_path print(f"ERROR: Project directory not found: {project_arg}", file=sys.stderr) print(f" Tried: {candidate}", file=sys.stderr) sys.exit(2) def find_episodes(project_path: Path, ep_range: Optional[str] = None) -> List[Path]: """Find episode files matching the requested range.""" episodes_dir = project_path / "episodes" if not episodes_dir.is_dir(): print(f"ERROR: Episodes directory not found: {episodes_dir}", file=sys.stderr) sys.exit(2) # Parse range start, end = 1, 60 if ep_range: if "-" in ep_range: parts = ep_range.split("-") start, end = int(parts[0]), int(parts[1]) else: start = end = int(ep_range) files = [] for ep_num in range(start, end + 1): ep_file = episodes_dir / f"ep_{ep_num:03d}.md" if ep_file.exists(): files.append(ep_file) return files # ── Pre-Scan ────────────────────────────────────────────────────────────── def pre_scan_characters(episode_files: List[Path]) -> Set[str]: """Quick pre-scan to collect ALL character names from dialogue cues across all episodes. This runs BEFORE full episode parsing so that prop detection in every episode has the complete character set (prevents character names like "VAREK SORN" from being detected as props). """ characters: Set[str] = set() for filepath in episode_files: text = filepath.read_text(encoding="utf-8") lines = text.split("\n") in_metadata = True for line in lines: stripped = line.strip() if stripped == "---": in_metadata = not in_metadata continue if in_metadata: continue # Character dialogue cue: ALL CAPS name on its own line char_match = re.match(r'^([A-Z][A-Z\s]+?)(?:\s*\(.*\))?\s*$', stripped) if char_match and stripped not in BEAT_NAMES and not stripped.startswith("AR "): name = char_match.group(1).strip() if (len(name.split()) <= 3 and not name.startswith(("EPISODE", "WORD", "CLIFFHANGER", "HOOK", "NEXT", "INT", "EXT")) and len(name) > 1): characters.add(name) # Also add first word for multi-word names (e.g., "VAREK" from "VAREK SORN") first_word = name.split()[0] if len(first_word) > 1: characters.add(first_word) return characters # ── Character Parsing ────────────────────────────────────────────────────── def parse_characters_md(project_path: Path) -> Dict[str, CharacterData]: """Parse characters.md to extract character data for breakdown.""" chars_file = project_path / "bible" / "characters.md" if not chars_file.exists(): print(f"WARNING: characters.md not found at {chars_file}", file=sys.stderr) return {} text = chars_file.read_text(encoding="utf-8") characters: Dict[str, CharacterData] = {} # Split by ## CHARACTER NAME sections # Pattern: ## NAME — TITLE or ## NAME char_sections = re.split(r'\n## ([A-Z][A-Z\s]+?)(?:\s*—\s*|\s*$)', text) # char_sections[0] is preamble, then alternating name/content for i in range(1, len(char_sections) - 1, 2): raw_name = char_sections[i].strip() content = char_sections[i + 1] if i + 1 < len(char_sections) else "" # Skip non-character sections if raw_name in ("PURPOSE", "VOICE CONSISTENCY RULES", "CHARACTER ENFORCEMENT RULES", "VALIDATION CHECKLIST"): continue # Extract the character key (ALL CAPS first word or full name) key = raw_name.split("—")[0].strip().split()[0].upper() char = CharacterData() # Display name: Title case of the first word char.display_name = key.capitalize() # Visual description vis_match = re.search(r'\*\*Visual Design:\*\*\n((?:- .+\n)+)', content) if vis_match: lines = [l.strip("- ").strip() for l in vis_match.group(1).strip().split("\n") if l.strip()] char.visual_description = ". ".join(lines) # Signature props from Visual Design bullet points (generic extraction) # Matches noun phrases before spatial/attachment indicators if vis_match: body_words = {'eyes', 'hair', 'skin', 'fingers', 'hands', 'face', 'scar', 'build', 'frame', 'bearing', 'moves', 'lean', 'tall', 'wiry', 'head', 'arm', 'arms', 'leg', 'legs'} for line in vis_match.group(1).strip().split("\n"): line_text = line.strip("- ").strip() pm = re.search( r'^([\w][\w\s-]{2,30}?)\s+' r'(?:on|in|around|across|at|welded|embedded|attached|strapped|' r'holstered|clipped|hanging|slung)\s', line_text, re.IGNORECASE ) if pm: candidate = pm.group(1).strip().lower() if not any(w in body_words for w in candidate.split()): prop_key = candidate.replace(" ", "_").replace("-", "_") if prop_key and prop_key not in char.signature_props: char.signature_props.append(prop_key) # Wardrobe from Transformation Beats table wardrobe_match = re.search( r'\*\*Transformation Beats:\*\*\s*\n\s*\|[^|]*\|[^|]*\|[^|]*\|\s*\n\s*\|[-|]+\|\s*\n((?:\|[^\n]+\n)+)', content ) if wardrobe_match: rows = wardrobe_match.group(1).strip().split("\n") for idx, row in enumerate(rows): cells = [c.strip() for c in row.split("|") if c.strip()] if len(cells) >= 3: ep_range_str = cells[0].strip() state = cells[1].strip() # Strip markdown bold/italic markers from state text state_clean = re.sub(r'\*+', '', state).strip() ep_match = re.match(r'(\d+)-(\d+)', ep_range_str) if ep_match: ep_start = int(ep_match.group(1)) ep_end = int(ep_match.group(2)) # Build a readable phase key: "phase_N_ep_START_END" phase_key = f"phase_{idx + 1}_ep_{ep_start}_{ep_end}" char.wardrobe[phase_key] = { "episodes": [ep_start, ep_end], # NOTE: This is the arc/emotional state from Transformation Beats, # NOT visual wardrobe. Claude enrichment should replace this with # actual clothing/appearance descriptions. "description": f"[ENRICHMENT NEEDED] Arc state: {state_clean}", "reference_images": { "front": None, "profile": None, "three_quarter": None, "full_body": None, "back": None, }, } # State changes from Mask Crack Schedule or transformation beats crack_match = re.search( r'\*\*Mask Crack Schedule:\*\*\s*\n\s*\|[^|]*\|[^|]*\|[^|]*\|\s*\n\s*\|[-|]+\|\s*\n((?:\|[^\n]+\n)+)', content ) if crack_match: rows = crack_match.group(1).strip().split("\n") for row in rows: cells = [c.strip() for c in row.split("|") if c.strip()] if len(cells) >= 3: try: ep_num = int(cells[0]) change_desc = cells[1] + " — " + cells[2] char.state_changes.append({ "episode": ep_num, "change": change_desc }) except ValueError: continue characters[key] = char return characters # ── Episode Parsing ──────────────────────────────────────────────────────── @dataclass class EpisodeExtraction: """Data extracted from a single episode file.""" episode_number: int = 0 title: str = "" characters_present: Set[str] = field(default_factory=set) dialogue_cues: Dict[str, int] = field(default_factory=dict) # char -> count locations: List[str] = field(default_factory=list) location_types: Dict[str, str] = field(default_factory=dict) # normalized -> INT/EXT action_text: str = "" dialogue_text: str = "" beats: List[Dict] = field(default_factory=list) sfx_found: List[Dict] = field(default_factory=list) vfx_found: List[Dict] = field(default_factory=list) props_found: List[Dict] = field(default_factory=list) description_samples: Dict[str, List[str]] = field(default_factory=dict) # location -> samples lighting_samples: List[str] = field(default_factory=list) audio_flags: List[Dict] = field(default_factory=list) specialty_shots: List[Dict] = field(default_factory=list) def parse_episode(filepath: Path, known_characters: Set[str]) -> EpisodeExtraction: """Parse a single episode file and extract all visual production data.""" text = filepath.read_text(encoding="utf-8") lines = text.split("\n") ext = EpisodeExtraction() # Episode number and title ep_match = re.search(r'Episode\s+(\d+):\s*(.+)', text, re.IGNORECASE) if ep_match: ext.episode_number = int(ep_match.group(1)) ext.title = ep_match.group(2).strip() else: # Try from filename fn_match = re.search(r'ep_(\d+)', filepath.stem) if fn_match: ext.episode_number = int(fn_match.group(1)) # Track current state for context current_beat = None in_metadata = True action_lines = [] current_location = None for i, line in enumerate(lines): stripped = line.strip() # Skip metadata header if stripped == "---": in_metadata = not in_metadata continue if in_metadata and (stripped.startswith("**Word Count") or stripped.startswith("**Dialogue")): continue # Beat detection beat_match = re.match(r'^#\s*\[(\d{2}:\d{2})\s*-\s*(\d{2}:\d{2})\]\s*(THE\s+\w+)', stripped) if beat_match: current_beat = beat_match.group(3).upper().strip() ext.beats.append({ "name": current_beat, "time_start": beat_match.group(1), "time_end": beat_match.group(2), }) continue # Scene headings (INT./EXT.) loc_match = re.match(r'^(INT\.|EXT\.|INT\./EXT\.)\s*(.+?)(?:\s*-\s*(DAY|NIGHT|CONTINUOUS|LATER|DAWN|DUSK|MORNING|EVENING))?$', stripped, re.IGNORECASE) if loc_match: loc_type = loc_match.group(1).upper().rstrip(".") if loc_type == "INT./EXT": loc_type = "INT/EXT" elif loc_type == "INT": loc_type = "INT" elif loc_type == "EXT": loc_type = "EXT" raw_location = loc_match.group(2).strip().rstrip(" -") # Normalize: strip time-of-day suffix normalized = re.sub(r'\s*-\s*(DAY|NIGHT|CONTINUOUS|LATER|DAWN|DUSK|MORNING|EVENING)\s*$', '', raw_location, flags=re.IGNORECASE).strip() full_heading = f"{loc_type}. {normalized}" if full_heading not in ext.locations: ext.locations.append(full_heading) ext.location_types[full_heading] = loc_type current_location = full_heading continue # Character dialogue cues (ALL CAPS name on its own line) char_match = re.match(r'^([A-Z][A-Z\s]+?)(?:\s*\(.*\))?\s*$', stripped) if char_match and stripped not in BEAT_NAMES and not stripped.startswith("AR "): char_name = char_match.group(1).strip() # Must be a known character or plausible new one (not a heading) if char_name in known_characters or ( len(char_name.split()) <= 3 and not char_name.startswith("EPISODE") and not char_name.startswith("WORD") and not char_name.startswith("CLIFFHANGER") and not char_name.startswith("HOOK") and not char_name.startswith("NEXT") and not char_name.startswith("INT") and not char_name.startswith("EXT") and len(char_name) > 1 ): ext.characters_present.add(char_name) ext.dialogue_cues[char_name] = ext.dialogue_cues.get(char_name, 0) + 1 # Next non-empty line is dialogue for j in range(i + 1, min(i + 5, len(lines))): dl = lines[j].strip() if dl and not re.match(r'^[A-Z][A-Z\s]+$', dl): ext.dialogue_text += dl + " " break continue # Skip footer metadata if stripped.startswith("**CLIFFHANGER TYPE") or stripped.startswith("**HOOK TYPE") or stripped.startswith("**NEXT"): continue # Action blocks (everything else that's content) if stripped and not stripped.startswith("#") and not stripped.startswith("[["): action_lines.append(stripped) # SFX detection (word-boundary matching to prevent substring false positives) lower = stripped.lower() for kw in SFX_KEYWORDS: if re.search(r'\b' + re.escape(kw) + r'\b', lower): # Check exclusion patterns for metaphorical uses exclusions = SFX_EXCLUSIONS.get(kw, []) if any(re.search(exc, lower) for exc in exclusions): continue ext.sfx_found.append({ "keyword": kw, "context": stripped[:120], "beat": current_beat, }) break # VFX detection for kw in VFX_KEYWORDS: if kw in lower: ext.vfx_found.append({ "keyword": kw, "context": stripped[:120], "beat": current_beat, }) break # Prop detection: CAPITALIZED objects (not character names, not verbs) # Look for CAPS words that are preceded by articles or are multi-word noun phrases cap_matches = re.findall(r'\b([A-Z]{2,}(?:[\s-]+[A-Z]{2,})*)\b', stripped) for cap in cap_matches: cap_clean = cap.strip() cap_lower = cap_clean.lower() # Check each word against stoplist cap_words = cap_clean.split() if any(w.lower() in VERB_STOPLIST for w in cap_words): continue # Check if ANY word in the phrase is a known character name # (catches multi-word names like "VAREK SORN" where "VAREK" is known) if any(w in known_characters for w in cap_words): continue if (cap_clean not in known_characters and cap_clean not in ext.characters_present and cap_clean not in BEAT_NAMES and cap_clean not in {"INT", "EXT", "AR", "OVERLAY", "RED", "POV", "VO"} and len(cap_clean) > 2 and not re.match(r'^(EPISODE|WORD|CLIFFHANGER|HOOK|NEXT|THE)\b', cap_clean) and # Must be a noun-like word: check context for article/possessive before it # or be a known compound (CRYO-POD, DEBT COUNTER) (re.search(r'(?:a|an|the|her|his|its|their|this|that)\s+' + re.escape(cap_clean), stripped, re.IGNORECASE) or len(cap_words) >= 2 or # Multi-word CAPS phrases are likely props cap_clean in {"POD", "TORCH", "MASK", "HOOK", "CABLE", "COUNTER", "BLADE", "GUN", "WEAPON"}) # Known prop words ): ext.props_found.append({ "name": cap_clean, "context": stripped[:120], "beat": current_beat, }) # Location description samples if current_location and len(stripped) > 20: if current_location not in ext.description_samples: ext.description_samples[current_location] = [] if len(ext.description_samples[current_location]) < 3: ext.description_samples[current_location].append(stripped[:150]) # Lighting keywords lighting_patterns = [ r'(amber|orange|red|blue|green|white|golden|dim|bright|dark|shadow|glow|flicker|neon)\s+(light|lighting|glow|lamp|strip|emergency)', r'(emergency lighting|industrial shadows|overhead light|single.*light)', r'(silhouette|backlit|spotlight|volumetric)', ] for pat in lighting_patterns: lm = re.search(pat, lower) if lm: ext.lighting_samples.append(lm.group(0)) ext.action_text = "\n".join(action_lines) # Audio flags if ext.dialogue_cues: ext.audio_flags.append({ "episode": ext.episode_number, "type": "vo", "note": f"Dialogue from: {', '.join(sorted(ext.dialogue_cues.keys()))}" }) # Ambient/music detection lower_action = ext.action_text.lower() ambient_keywords = ["groaning", "hiss", "hum", "rumble", "clang", "screech", "drip", "creak", "wind", "alarm", "siren", "beep"] ambient_found = [kw for kw in ambient_keywords if kw in lower_action] if ambient_found: ext.audio_flags.append({ "episode": ext.episode_number, "type": "ambient", "note": f"Ambient cues: {', '.join(ambient_found[:5])}" }) # Specialty shot detection (extended motion sequences) # Build a map of action lines to their beats for accurate beat assignment action_line_beats = [] beat_for_action = None ss_in_metadata = True for line in lines: stripped_line = line.strip() # Track metadata boundaries if stripped_line == "---": ss_in_metadata = not ss_in_metadata continue if ss_in_metadata: continue beat_match_ss = re.match(r'^#\s*\[(\d{2}:\d{2})\s*-\s*(\d{2}:\d{2})\]\s*(THE\s+\w+)', stripped_line) if beat_match_ss: beat_for_action = beat_match_ss.group(3).upper().strip() elif (stripped_line and not stripped_line.startswith("#") and not stripped_line.startswith("[[") and not stripped_line.startswith("**") and not re.match(r'^(INT\.|EXT\.)', stripped_line) and # Skip scene headings not re.match(r'^[A-Z][A-Z\s]+(?:\s*\(.*\))?\s*$', stripped_line)): # Skip dialogue cues action_line_beats.append((stripped_line, beat_for_action)) # Detect extended motion sequences that need sandwich workflow or complex compositing. # Criteria: 6+ consecutive action lines within a single beat, with 2+ distinct motion verbs. # Blocks break at beat boundaries (beat changes) and empty lines. action_block_lengths = [] current_block = [] current_block_beat = None for line_text, line_beat in action_line_beats: # Break block on beat change if line_beat != current_block_beat and current_block: if len(current_block) >= 6: action_block_lengths.append((current_block[:], current_block_beat)) current_block = [] current_block_beat = line_beat if line_text.strip(): current_block.append(line_text) else: if len(current_block) >= 6: action_block_lengths.append((current_block[:], current_block_beat)) current_block = [] if len(current_block) >= 6: action_block_lengths.append((current_block[:], current_block_beat)) for block, block_beat in action_block_lengths: block_text = " ".join(block) block_lower = block_text.lower() motion_words = ["swings", "falls", "drops", "crashes", "runs", "chases", "leaps", "jumps", "slides", "rolls", "spins", "tumbles", "convulses", "explode", "explodes", "collapses"] # Require 2+ distinct motion verbs to qualify as extended motion sequence motion_found = [w for w in motion_words if re.search(r'\b' + w + r'\b', block_lower)] if len(motion_found) >= 2: ext.specialty_shots.append({ "episode": ext.episode_number, "beat": block_beat or "unknown", "description": block_text[:200], "technique": "sandwich_workflow", "estimated_duration_seconds": 8, }) return ext # ── Aggregation ──────────────────────────────────────────────────────────── def aggregate_extractions( extractions: List[EpisodeExtraction], char_data: Dict[str, CharacterData], ) -> dict: """Aggregate per-episode extractions into the full breakdown JSON.""" ep_numbers = sorted([e.episode_number for e in extractions]) ep_range = [min(ep_numbers), max(ep_numbers)] if ep_numbers else [0, 0] # ── Characters ── characters = {} for key, cd in char_data.items(): characters[key] = cd # Merge episode data into character records for ext in extractions: for char_name in ext.characters_present: if char_name not in characters: characters[char_name] = CharacterData(display_name=char_name.capitalize()) c = characters[char_name] if ext.episode_number not in c.episodes: c.episodes.append(ext.episode_number) c.dialogue_count += ext.dialogue_cues.get(char_name, 0) for key, c in characters.items(): c.episodes = sorted(set(c.episodes)) c.episode_count = len(c.episodes) c.first_appearance = c.episodes[0] if c.episodes else 0 # ── Locations ── locations: Dict[str, LocationData] = {} for ext in extractions: for loc in ext.locations: if loc not in locations: locations[loc] = LocationData( type=ext.location_types.get(loc, "INT"), ) ld = locations[loc] if ext.episode_number not in ld.episodes: ld.episodes.append(ext.episode_number) # Add description samples for sample in ext.description_samples.get(loc, []): if sample not in ld.description_samples and len(ld.description_samples) < 5: ld.description_samples.append(sample) # Add lighting notes for note in ext.lighting_samples: if note not in ld.lighting_notes and len(ld.lighting_notes) < 5: ld.lighting_notes.append(note) for key, ld in locations.items(): ld.episodes = sorted(set(ld.episodes)) ld.episode_count = len(ld.episodes) # ── Props ── props: Dict[str, PropData] = {} # First: add signature props from characters for char_key, cd in characters.items(): for prop_key in cd.signature_props: if prop_key not in props: props[prop_key] = PropData( display_name=prop_key.replace("_", " ").title(), owner=char_key, confidence="high", episodes=list(cd.episodes), episode_count=cd.episode_count, ) # Then: add props found in episodes for ext in extractions: for pf in ext.props_found: prop_key = pf["name"].lower().replace(" ", "_") if prop_key not in props: props[prop_key] = PropData( display_name=pf["name"].replace("_", " ").title(), confidence="medium", ) pd = props[prop_key] if ext.episode_number not in pd.episodes: pd.episodes.append(ext.episode_number) if pf["context"] not in pd.description_samples and len(pd.description_samples) < 5: pd.description_samples.append(pf["context"]) for key, pd in props.items(): pd.episodes = sorted(set(pd.episodes)) pd.episode_count = len(pd.episodes) # ── Dedup: Props vs VFX ── # VFX keywords can also match as props (e.g., "AR OVERLAY" detected as both). # After building both, remove props whose key matches a VFX key. # This is done after VFX aggregation below, see "Dedup pass" section. # ── SFX Elements ── sfx_elements: Dict[str, SFXElement] = {} for ext in extractions: for sf in ext.sfx_found: sfx_key = sf["keyword"].replace(" ", "_") if sfx_key not in sfx_elements: sfx_elements[sfx_key] = SFXElement( display_name=sf["keyword"].replace("_", " ").title(), type="promptable", production_method="prompt_directly", ) se = sfx_elements[sfx_key] if ext.episode_number not in se.episodes: se.episodes.append(ext.episode_number) if sf["context"] not in se.description_samples and len(se.description_samples) < 5: se.description_samples.append(sf["context"]) for key, se in sfx_elements.items(): se.episodes = sorted(set(se.episodes)) se.episode_count = len(se.episodes) # ── VFX Elements ── vfx_elements: Dict[str, VFXElement] = {} for ext in extractions: for vf in ext.vfx_found: vfx_key = vf["keyword"].replace(" ", "_") if vfx_key not in vfx_elements: vfx_elements[vfx_key] = VFXElement( display_name=vf["keyword"].replace("_", " ").title(), type="post_composite", production_method="post_composite", ) ve = vfx_elements[vfx_key] if ext.episode_number not in ve.episodes: ve.episodes.append(ext.episode_number) if vf["context"] not in ve.description_samples and len(ve.description_samples) < 5: ve.description_samples.append(vf["context"]) # Extract colors from AR OVERLAY lines color_match = re.search(r'\(([A-Z]+)\)', vf["context"]) if color_match: color_name = color_match.group(1) color_map = { "RED": "#FF0000", "GREEN": "#00FF00", "BLUE": "#0000FF", "AMBER": "#FFBF00", "ORANGE": "#FF8C00", "WHITE": "#FFFFFF", "YELLOW": "#FFFF00", } hex_color = color_map.get(color_name) if hex_color and hex_color not in ve.colors: ve.colors.append(hex_color) for key, ve in vfx_elements.items(): ve.episodes = sorted(set(ve.episodes)) ve.episode_count = len(ve.episodes) # ── Dedup pass: remove props that overlap with VFX keys ── vfx_keys = set(vfx_elements.keys()) props_to_remove = [k for k in props if k in vfx_keys] for k in props_to_remove: del props[k] # ── Dedup pass: remove props that overlap with SFX keys ── sfx_keys = set(sfx_elements.keys()) props = {k: v for k, v in props.items() if k not in sfx_keys} # ── Filter: remove props matching system readout patterns ── system_suffixes = ( "_detected", "_active", "_expired", "_complete", "_failure", "_status", "_override", "_conflict", "_denied", ) props = {k: v for k, v in props.items() if not k.endswith(system_suffixes)} # ── Filter: remove medium-confidence props with 2+ underscored words ── # Multi-word ALL CAPS in scripts are almost always readout text, not physical props. # Real physical props are captured as high-confidence signature props from characters.md. props = {k: v for k, v in props.items() if not (v.confidence == "medium" and len(k.split("_")) >= 2)} # ── Specialty Shots ── specialty_shots = [] for ext in extractions: specialty_shots.extend(ext.specialty_shots) # ── Audio Flags ── audio_flags = [] for ext in extractions: audio_flags.extend(ext.audio_flags) # ── Shot Estimates ── total_episodes = len(extractions) total_shots = total_episodes * SHOTS_PER_EPISODE_ESTIMATE # Estimate character shots based on episode appearances character_shots = {} total_char_appearances = sum(c.episode_count for c in characters.values()) if total_char_appearances > 0: for key, c in characters.items(): ratio = c.episode_count / total_char_appearances character_shots[key] = round(total_shots * ratio) # ── Lock Status ── chars_locked = sum(1 for c in characters.values() if c.reference_status == "locked") locs_locked = sum(1 for l in locations.values() if l.reference_status == "locked") props_locked = sum(1 for p in props.values() if p.reference_status == "locked") vfx_locked = sum(1 for v in vfx_elements.values() if getattr(v, 'reference_status', 'not_started') == "locked") # ── Build Output ── def serialize_dataclass(obj): """Convert dataclass to dict, handling sets.""" if hasattr(obj, '__dataclass_fields__'): d = {} for f in obj.__dataclass_fields__: val = getattr(obj, f) if isinstance(val, set): val = sorted(val) d[f] = val return d return obj output = { "version": 1, "project": extractions[0].title.lower() if extractions else "unknown", "generated": datetime.now(timezone.utc).isoformat(), "episodes_processed": total_episodes, "episode_range": ep_range, "story_timeline": { "total_story_days": 0, # Default 1:1 mapping — enriched by breakdown agent with actual story days from treatment.md "episodes_to_days": {str(ep): ep for ep in ep_numbers}, }, "characters": {k: serialize_dataclass(v) for k, v in characters.items()}, "style_references": { "color_grade": None, "film_stock": "Kodak Vision3 500T", "look_dev_images": {"slot_9": None}, "notes": "Project-level style lock. Slot 9: lighting reference. Slot 10: pose/structure from storyboard.", }, "locations": {k: serialize_dataclass(v) for k, v in locations.items()}, "props": {k: serialize_dataclass(v) for k, v in props.items()}, "sfx_elements": {k: serialize_dataclass(v) for k, v in sfx_elements.items()}, "vfx_elements": {k: serialize_dataclass(v) for k, v in vfx_elements.items()}, "specialty_shots": specialty_shots, "audio_flags": audio_flags, "shot_estimates": { "total_episodes": total_episodes, "shots_per_episode": SHOTS_PER_EPISODE_ESTIMATE, "total_shots": total_shots, "character_shots": character_shots, "specialty_shots_count": len(specialty_shots), }, "asset_lock_status": { "characters_locked": chars_locked, "characters_total": len(characters), "locations_locked": locs_locked, "locations_total": len(locations), "props_locked": props_locked, "props_total": len(props), "vfx_locked": vfx_locked, "vfx_total": len(vfx_elements), }, } return output # ── Report Generation ────────────────────────────────────────────────────── def print_report(breakdown: dict): """Print a human-readable summary of the breakdown.""" print("=" * 60) print("SCRIPT BREAKDOWN REPORT") print("=" * 60) print(f"Project: {breakdown['project']}") print(f"Episodes: {breakdown['episodes_processed']} ({breakdown['episode_range'][0]}-{breakdown['episode_range'][1]})") print(f"Generated: {breakdown['generated']}") print() # Characters chars = breakdown["characters"] print(f"CHARACTERS ({len(chars)})") print("-" * 40) for key, c in sorted(chars.items(), key=lambda x: -x[1].get("episode_count", 0)): lock_icon = "[LOCKED]" if c.get("reference_status") == "locked" else "[------]" print(f" {lock_icon} {c['display_name']:15s} {c.get('episode_count', 0):3d} eps {c.get('dialogue_count', 0):3d} cues") print() # Locations locs = breakdown["locations"] print(f"LOCATIONS ({len(locs)})") print("-" * 40) for key, l in sorted(locs.items(), key=lambda x: -x[1].get("episode_count", 0)): lock_icon = "[LOCKED]" if l.get("reference_status") == "locked" else "[------]" print(f" {lock_icon} {key[:35]:35s} {l.get('episode_count', 0):3d} eps") print() # Props props = breakdown["props"] print(f"PROPS ({len(props)})") print("-" * 40) for key, p in sorted(props.items(), key=lambda x: -x[1].get("episode_count", 0)): conf = p.get("confidence", "?")[0].upper() owner = p.get("owner", "") owner_str = f" ({owner})" if owner else "" print(f" [{conf}] {p['display_name']:25s} {p.get('episode_count', 0):3d} eps{owner_str}") print() # VFX vfx = breakdown["vfx_elements"] sfx = breakdown["sfx_elements"] print(f"VFX ELEMENTS ({len(vfx)} post-composite, {len(sfx)} promptable SFX)") print("-" * 40) for key, v in sorted(vfx.items(), key=lambda x: -x[1].get("episode_count", 0)): print(f" [POST] {v['display_name']:25s} {v.get('episode_count', 0):3d} eps") for key, s in sorted(sfx.items(), key=lambda x: -x[1].get("episode_count", 0)): print(f" [SFX] {s['display_name']:25s} {s.get('episode_count', 0):3d} eps") print() # Shot Estimates se = breakdown["shot_estimates"] print("SHOT ESTIMATES") print("-" * 40) print(f" Total episodes: {se['total_episodes']}") print(f" Shots/episode: {se['shots_per_episode']}") print(f" Total shots: {se['total_shots']}") print(f" Specialty shots: {se['specialty_shots_count']}") if se.get("character_shots"): print(" Per character:") for char, count in sorted(se["character_shots"].items(), key=lambda x: -x[1]): print(f" {char:15s} ~{count} shots") print() # Lock Status ls = breakdown["asset_lock_status"] total_lockable = ls["characters_total"] + ls["locations_total"] + ls["props_total"] + ls["vfx_total"] total_locked = ls["characters_locked"] + ls["locations_locked"] + ls["props_locked"] + ls["vfx_locked"] pct = (total_locked / total_lockable * 100) if total_lockable > 0 else 0 print("LOCK STATUS") print("-" * 40) print(f" Characters: {ls['characters_locked']}/{ls['characters_total']}") print(f" Locations: {ls['locations_locked']}/{ls['locations_total']}") print(f" Props: {ls['props_locked']}/{ls['props_total']}") print(f" VFX: {ls['vfx_locked']}/{ls['vfx_total']}") print(f" TOTAL: {total_locked}/{total_lockable} ({pct:.0f}%)") print() def print_shots(breakdown: dict): """Print shot estimates only.""" se = breakdown["shot_estimates"] print(f"Shot Estimates for {breakdown['episodes_processed']} episodes:") print(f" {se['total_shots']} total shots ({se['shots_per_episode']}/ep)") print(f" {se['specialty_shots_count']} specialty shots") if se.get("character_shots"): for char, count in sorted(se["character_shots"].items(), key=lambda x: -x[1]): print(f" {char}: ~{count}") def print_status(breakdown: dict): """Print lock progress summary.""" ls = breakdown["asset_lock_status"] total_lockable = ls["characters_total"] + ls["locations_total"] + ls["props_total"] + ls["vfx_total"] total_locked = ls["characters_locked"] + ls["locations_locked"] + ls["props_locked"] + ls["vfx_locked"] pct = (total_locked / total_lockable * 100) if total_lockable > 0 else 0 print(f"Lock Progress: {total_locked}/{total_lockable} ({pct:.0f}%)") print(f" Characters: {ls['characters_locked']}/{ls['characters_total']}") print(f" Locations: {ls['locations_locked']}/{ls['locations_total']}") print(f" Props: {ls['props_locked']}/{ls['props_total']}") print(f" VFX: {ls['vfx_locked']}/{ls['vfx_total']}") if pct == 100: print("\nAll assets locked. Ready for storyboard generation.") elif pct >= 50: print(f"\n{total_lockable - total_locked} assets remaining.") else: print(f"\n{total_lockable - total_locked} assets need reference images and lock.") # ── Main ─────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Script Breakdown — Extract visual production assets from episodes" ) parser.add_argument("project", help="Project path (e.g., /leviathan/ or leviathan)") parser.add_argument("--episodes", "-e", help="Episode range (e.g., 1-10 or 5)") parser.add_argument("--report", "-r", action="store_true", help="Print human-readable summary") parser.add_argument("--shots", "-s", action="store_true", help="Shot estimates only") parser.add_argument("--status", action="store_true", help="Lock progress summary") parser.add_argument("--json", "-j", action="store_true", help="Output raw JSON to stdout") parser.add_argument("--output", "-o", help="Custom output path for breakdown.json") args = parser.parse_args() # Resolve paths project_path = resolve_project_path(args.project) project_name = project_path.name # If --status, try to load existing breakdown.json if args.status: existing = project_path / "visual" / "breakdown.json" if existing.exists(): try: data = json.loads(existing.read_text(encoding="utf-8")) except json.JSONDecodeError as e: print(f"ERROR: Invalid JSON in {existing}: {e}", file=sys.stderr) sys.exit(2) print_status(data) sys.exit(0) else: print(f"No breakdown.json found at {existing}", file=sys.stderr) print("Run script_breakdown.py first to generate.", file=sys.stderr) sys.exit(2) # Find episode files episode_files = find_episodes(project_path, args.episodes) if not episode_files: print(f"ERROR: No episodes found in {project_path / 'episodes'}", file=sys.stderr) sys.exit(2) print(f"Scanning {len(episode_files)} episodes in {project_name}...", file=sys.stderr) # Pre-scan: collect ALL character names from ALL episodes first # This prevents character names from being detected as props pre_scanned = pre_scan_characters(episode_files) print(f" Pre-scan found {len(pre_scanned)} character names across episodes", file=sys.stderr) # Parse characters.md char_data = parse_characters_md(project_path) known_characters = set(char_data.keys()) | pre_scanned print(f" {len(known_characters)} known characters (characters.md + pre-scan)", file=sys.stderr) # Parse each episode with full character set extractions = [] for ep_file in episode_files: ext = parse_episode(ep_file, known_characters) extractions.append(ext) # Still expand — catches edge cases where a character appears in action but not dialogue known_characters.update(ext.characters_present) print(f" Extracted data from {len(extractions)} episodes", file=sys.stderr) # Aggregate breakdown = aggregate_extractions(extractions, char_data) breakdown["project"] = project_name # Merge with existing breakdown.json to preserve locks and reference images output_path = Path(args.output) if args.output else project_path / "visual" / "breakdown.json" if output_path.exists(): try: existing = json.loads(output_path.read_text(encoding="utf-8")) # Preserve lock statuses and reference images from existing for cat in ["characters", "locations", "props"]: if cat in existing and cat in breakdown: for key, val in existing[cat].items(): if key in breakdown[cat]: # Preserve locked status if val.get("reference_status") == "locked": breakdown[cat][key]["reference_status"] = "locked" # Preserve reference images if val.get("reference_images"): for slot, img in val["reference_images"].items(): if img and not slot.startswith("_"): if "reference_images" not in breakdown[cat][key]: breakdown[cat][key]["reference_images"] = {} breakdown[cat][key]["reference_images"][slot] = img # Preserve prompts if val.get("prompts"): for ptype, prompt in val["prompts"].items(): if prompt: if "prompts" not in breakdown[cat][key]: breakdown[cat][key]["prompts"] = {} breakdown[cat][key]["prompts"][ptype] = prompt # Preserve style_references if "style_references" in existing: breakdown["style_references"] = existing["style_references"] print(f" Merged with existing breakdown (preserved locks & refs)", file=sys.stderr) except (json.JSONDecodeError, KeyError) as e: print(f" WARNING: Could not merge with existing breakdown: {e}", file=sys.stderr) # Recompute lock status after merge for cat, total_key, locked_key in [ ("characters", "characters_total", "characters_locked"), ("locations", "locations_total", "locations_locked"), ("props", "props_total", "props_locked"), ]: breakdown["asset_lock_status"][total_key] = len(breakdown[cat]) breakdown["asset_lock_status"][locked_key] = sum( 1 for v in breakdown[cat].values() if v.get("reference_status") == "locked" ) # Output if args.json: print(json.dumps(breakdown, indent=2)) elif args.shots: print_shots(breakdown) elif args.report: print_report(breakdown) else: # Write to file output_path.parent.mkdir(parents=True, exist_ok=True) output_path.write_text(json.dumps(breakdown, indent=2), encoding="utf-8") print(f"\nBreakdown written to: {output_path}", file=sys.stderr) print(f" Characters: {len(breakdown['characters'])}", file=sys.stderr) print(f" Locations: {len(breakdown['locations'])}", file=sys.stderr) print(f" Props: {len(breakdown['props'])}", file=sys.stderr) print(f" SFX: {len(breakdown['sfx_elements'])}", file=sys.stderr) print(f" VFX: {len(breakdown['vfx_elements'])}", file=sys.stderr) print(f" Specialty: {len(breakdown['specialty_shots'])}", file=sys.stderr) print(f" Total shots: ~{breakdown['shot_estimates']['total_shots']}", file=sys.stderr) # Also print report print() print_report(breakdown) sys.exit(0) if __name__ == "__main__": main()