#!/usr/bin/env python3 """ Episode Metrics and Fix Generator Measures word count and exchanges in Fountain-format episodes, then generates targeted edit instructions if out of range. Usage: python3 episode_metrics.py python3 episode_metrics.py --fix python3 episode_metrics.py --json python3 episode_metrics.py --prompt Exit codes: 0 = Within all constraints 1 = Out of range (fix instructions provided if --fix) 2 = File error """ import re import sys import json import argparse from pathlib import Path from dataclasses import dataclass from typing import Tuple, List # ============================================================================= # CONSTANTS (imported from engine_constants.py - single source of truth) # ============================================================================= # Add engine tools to path for imports script_dir = Path(__file__).parent if str(script_dir) not in sys.path: sys.path.insert(0, str(script_dir)) from engine_constants import ( WORD_COUNT_MIN, WORD_COUNT_MAX, MAX_EXCHANGES, DIALOGUE_MAX_PERCENT, MAX_ACTION_BLOCK_LINES, count_words as _shared_count_words, parse_dialogue_blocks, count_dialogue_words as _shared_count_dialogue_words, count_exchanges as _shared_count_exchanges, calculate_dialogue_percent, is_character_cue as _shared_is_character_cue, is_parenthetical as _shared_is_parenthetical, ) # Derived target values (not in CONSTANTS.md - computed for edit guidance) WORD_COUNT_TARGET = (WORD_COUNT_MIN + WORD_COUNT_MAX) // 2 # 475 TARGET_EXCHANGES = MAX_EXCHANGES - 1 # Comfortable buffer # Rename for consistency with rest of script MAX_DIALOGUE_PERCENT = DIALOGUE_MAX_PERCENT ACTION_BLOCK_MAX_LINES = MAX_ACTION_BLOCK_LINES # Kill Box section patterns (mirroring validate_batch.py) KILL_BOX_SECTIONS = [ (r'#\s*\[00:00\s*-\s*00:05\].*THE HOOK', 'THE HOOK', '[00:00 - 00:05]'), (r'#\s*\[00:05\s*-\s*00:15\].*THE SETUP', 'THE SETUP', '[00:05 - 00:15]'), (r'#\s*\[00:15\s*-\s*00:40\].*THE ESCALATION', 'THE ESCALATION', '[00:15 - 00:40]'), (r'#\s*\[00:40\s*-\s*00:70\].*THE TURN', 'THE TURN', '[00:40 - 00:70]'), (r'#\s*\[00:70\s*-\s*00:90\].*THE CLIFFHANGER', 'THE CLIFFHANGER', '[00:70 - 00:90]'), ] # ============================================================================= # DATA STRUCTURES # ============================================================================= @dataclass class EpisodeMetrics: """Measured metrics for an episode.""" word_count: int exchange_count: int dialogue_word_count: int action_word_count: int dialogue_percent: float action_block_count: int longest_action_block: int character_names: List[str] kill_box_sections_found: List[str] kill_box_sections_missing: List[str] kill_box_order_valid: bool kill_box_issues: List[str] @property def kill_box_status(self) -> str: if not self.kill_box_sections_missing and self.kill_box_order_valid and not self.kill_box_issues: return "OK" if self.kill_box_sections_missing: return "MISSING" if not self.kill_box_order_valid: return "MISORDERED" return "MALFORMED" @property def word_count_status(self) -> str: if self.word_count < WORD_COUNT_MIN: return "LOW" elif self.word_count > WORD_COUNT_MAX: return "HIGH" return "OK" @property def exchange_status(self) -> str: if self.exchange_count > MAX_EXCHANGES: return "HIGH" return "OK" @property def dialogue_status(self) -> str: if self.dialogue_percent > MAX_DIALOGUE_PERCENT: return "HIGH" return "OK" @property def is_valid(self) -> bool: return (self.word_count_status == "OK" and self.exchange_status == "OK" and self.dialogue_status == "OK" and self.kill_box_status == "OK") @dataclass class FixInstruction: """A specific edit instruction to fix a constraint violation.""" issue: str severity: str # "critical", "warning" instruction: str target_change: str # e.g., "+35 words" or "-2 exchanges" # ============================================================================= # PARSING FUNCTIONS # ============================================================================= def extract_content_sections(text: str) -> Tuple[str, str]: """ Separate episode content from metadata header. Returns (metadata, content). """ lines = text.split('\n') content_start = 0 # Skip YAML-style header if present if lines and lines[0].strip() == '---': for i, line in enumerate(lines[1:], 1): if line.strip() == '---': content_start = i + 1 break # Also skip markdown metadata (lines starting with ** or #) while content_start < len(lines): line = lines[content_start].strip() if line.startswith('**') or line.startswith('#') or line == '': content_start += 1 else: break metadata = '\n'.join(lines[:content_start]) content = '\n'.join(lines[content_start:]) return metadata, content def is_character_cue(line: str) -> bool: """ Detect if a line is a character cue (speaker name) in Fountain format. Delegates to shared canonical implementation from engine_constants. """ return _shared_is_character_cue(line) def is_parenthetical(line: str) -> bool: """Detect parenthetical direction within dialogue. Delegates to shared canonical implementation from engine_constants.""" return _shared_is_parenthetical(line) def is_action_line(line: str, prev_was_dialogue: bool) -> bool: """Determine if a line is action description.""" line = line.strip() if not line: return False if is_character_cue(line): return False if is_parenthetical(line): return False # Scene headers if re.match(r'^(INT\.|EXT\.|INT/EXT\.)', line): return False # Transitions if line.endswith(':') and line.isupper(): return False return True def parse_episode(text: str) -> dict: """ Parse a Fountain-format episode into structured components. Returns dict with dialogue blocks, action blocks, and metadata. """ metadata, content = extract_content_sections(text) lines = content.split('\n') dialogue_blocks = [] # List of (character, dialogue_text) action_blocks = [] # List of action text blocks current_character = None current_dialogue = [] current_action = [] in_dialogue = False for line in lines: stripped = line.strip() if is_character_cue(stripped): # Save previous action block if current_action: action_blocks.append('\n'.join(current_action)) current_action = [] # Save previous dialogue block if current_character and current_dialogue: dialogue_blocks.append((current_character, '\n'.join(current_dialogue))) # Start new dialogue block current_character = re.sub(r'\s*\([^)]+\)\s*', '', stripped).strip() current_dialogue = [] in_dialogue = True elif in_dialogue: if is_parenthetical(stripped): # Skip parentheticals from word count continue if stripped: current_dialogue.append(stripped) else: # Empty line ends dialogue if current_character and current_dialogue: dialogue_blocks.append((current_character, '\n'.join(current_dialogue))) current_character = None current_dialogue = [] in_dialogue = False elif not in_dialogue or (in_dialogue and not stripped): # Action line or dialogue ended if in_dialogue and not stripped: if current_character and current_dialogue: dialogue_blocks.append((current_character, '\n'.join(current_dialogue))) current_character = None current_dialogue = [] in_dialogue = False if stripped and is_action_line(stripped, in_dialogue): current_action.append(stripped) elif not stripped and current_action: # Empty line ends action block action_blocks.append('\n'.join(current_action)) current_action = [] # Don't forget trailing blocks if current_character and current_dialogue: dialogue_blocks.append((current_character, '\n'.join(current_dialogue))) if current_action: action_blocks.append('\n'.join(current_action)) return { 'metadata': metadata, 'dialogue_blocks': dialogue_blocks, 'action_blocks': action_blocks, 'raw_content': content } # ============================================================================= # MEASUREMENT FUNCTIONS # ============================================================================= def count_words(text: str) -> int: """Count words in text. Delegates to shared canonical implementation. Note: No longer strips markdown — counts ALL words for consistency with engine_constants.count_words().""" return _shared_count_words(text) def count_exchanges(dialogue_blocks: List[Tuple[str, str]]) -> int: """ Count total dialogue blocks (matches validate_batch.py). Each character cue counts as one exchange. Delegates to shared canonical implementation. """ return _shared_count_exchanges(dialogue_blocks) def validate_kill_box_sections(text: str) -> dict: """ Validate Kill Box section structure: presence, order, and non-empty content. Returns dict with sections_found, sections_missing, order_valid, issues. """ sections_found = [] sections_missing = [] positions = [] issues = [] for pattern, name, timestamp in KILL_BOX_SECTIONS: match = re.search(pattern, text, re.IGNORECASE) if match: sections_found.append(name) positions.append(match.start()) else: sections_missing.append(name) # Check order: match positions must be ascending order_valid = all(positions[i] < positions[i+1] for i in range(len(positions) - 1)) if not order_valid and len(positions) > 1: issues.append("Kill Box sections are out of order") # Check non-empty content between sections for pattern, name, timestamp in KILL_BOX_SECTIONS: match = re.search(pattern, text, re.IGNORECASE) if not match: continue header_end = match.end() # Find next section header or end-of-file marker next_section = None for next_pattern, next_name, _ in KILL_BOX_SECTIONS: next_match = re.search(next_pattern, text[header_end:], re.IGNORECASE) if next_match: if next_section is None or next_match.start() < next_section: next_section = next_match.start() # Also check for --- separator or EOF separator = re.search(r'^---\s*$', text[header_end:], re.MULTILINE) if separator and (next_section is None or separator.start() < next_section): next_section = separator.start() if next_section is None: content_between = text[header_end:] else: content_between = text[header_end:header_end + next_section] # Strip whitespace and check if any real content exists stripped = content_between.strip() if not stripped: issues.append(f"{name} section ({timestamp}) has no content") return { 'sections_found': sections_found, 'sections_missing': sections_missing, 'order_valid': order_valid, 'issues': issues, } def measure_episode(filepath: Path) -> EpisodeMetrics: """ Measure all metrics for an episode file. Uses shared counting functions from engine_constants for consistency. """ text = filepath.read_text(encoding='utf-8') parsed = parse_episode(text) # Word counts — use shared canonical function total_words = _shared_count_words(text) # Dialogue analysis — use shared canonical functions for consistent counting dialogue_blocks = parse_dialogue_blocks(text) dialogue_words = _shared_count_dialogue_words(dialogue_blocks) exchanges = _shared_count_exchanges(dialogue_blocks) dialogue_pct = calculate_dialogue_percent(total_words, dialogue_words) # Action word count (still from local parse for action block analysis) action_words = sum(len(a.split()) for a in parsed['action_blocks']) # Action block analysis action_block_lines = [len(a.split('\n')) for a in parsed['action_blocks']] longest_action = max(action_block_lines) if action_block_lines else 0 # Character names — from shared dialogue blocks characters = list(set(c for c, _ in dialogue_blocks)) # Kill Box section validation kb = validate_kill_box_sections(text) return EpisodeMetrics( word_count=total_words, exchange_count=exchanges, dialogue_word_count=dialogue_words, action_word_count=action_words, dialogue_percent=round(dialogue_pct, 1), action_block_count=len(parsed['action_blocks']), longest_action_block=longest_action, character_names=characters, kill_box_sections_found=kb['sections_found'], kill_box_sections_missing=kb['sections_missing'], kill_box_order_valid=kb['order_valid'], kill_box_issues=kb['issues'], ) # ============================================================================= # FIX GENERATION # ============================================================================= def generate_fix_instructions(metrics: EpisodeMetrics) -> List[FixInstruction]: """ Generate specific edit instructions based on constraint violations. """ fixes = [] # Word count fixes if metrics.word_count < WORD_COUNT_MIN: diff = WORD_COUNT_MIN - metrics.word_count # Prefer expanding action over dialogue to avoid dialogue % issues fixes.append(FixInstruction( issue=f"Word count too low: {metrics.word_count} (min: {WORD_COUNT_MIN})", severity="critical", instruction=f"""Add approximately {diff} words by expanding action descriptions. SPECIFIC GUIDANCE: - Add sensory details to 2-3 existing action blocks (what characters see, hear, feel) - Expand the physical environment description in the opening beat - Add a character's physical reaction (body language, gesture) to a tense moment - DO NOT add new dialogue exchanges - DO NOT add new plot beats—expand existing ones TARGET: Reach {WORD_COUNT_TARGET} words total.""", target_change=f"+{diff} words" )) elif metrics.word_count > WORD_COUNT_MAX: diff = metrics.word_count - WORD_COUNT_MAX if metrics.dialogue_percent > MAX_DIALOGUE_PERCENT * 0.85: # Cut from dialogue when it's close to or above the cap fixes.append(FixInstruction( issue=f"Word count too high: {metrics.word_count} (max: {WORD_COUNT_MAX})", severity="critical", instruction=f"""Cut approximately {diff} words, prioritizing dialogue reduction. SPECIFIC GUIDANCE: - Tighten dialogue: remove filler words, combine redundant lines - Cut any dialogue that restates what action already shows - Replace 2-3 line dialogue exchanges with single impactful lines - Preserve emotional beats and character voice - Dialogue is at {metrics.dialogue_percent}%—aim to reduce below {MAX_DIALOGUE_PERCENT}% TARGET: Reach {WORD_COUNT_TARGET} words total.""", target_change=f"-{diff} words" )) else: # Cut from action fixes.append(FixInstruction( issue=f"Word count too high: {metrics.word_count} (max: {WORD_COUNT_MAX})", severity="critical", instruction=f"""Cut approximately {diff} words from action descriptions. SPECIFIC GUIDANCE: - Remove adjectives and adverbs that don't add visual information - Cut any line that tells rather than shows - Combine consecutive short action blocks into tighter prose - Remove any unfilmable content (character thoughts, backstory) - Preserve all dialogue unchanged TARGET: Reach {WORD_COUNT_TARGET} words total.""", target_change=f"-{diff} words" )) # Exchange fixes if metrics.exchange_count > MAX_EXCHANGES: excess = metrics.exchange_count - TARGET_EXCHANGES fixes.append(FixInstruction( issue=f"Too many exchanges: {metrics.exchange_count} (max: {MAX_EXCHANGES})", severity="critical", instruction=f"""Reduce from {metrics.exchange_count} to {TARGET_EXCHANGES} exchanges. SPECIFIC GUIDANCE: - Identify the 2-3 weakest exchanges (least emotional weight, most expository) - Option A: Convert dialogue to action (character does instead of says) - Option B: Combine adjacent same-speaker lines if interrupted - Option C: Replace back-and-forth with single impactful line + reaction - PRESERVE the emotionally critical exchanges (usually first, middle turn, and final) TARGET: {TARGET_EXCHANGES} exchanges maximum.""", target_change=f"-{excess} exchanges" )) # Dialogue percentage fixes if metrics.dialogue_percent > MAX_DIALOGUE_PERCENT: excess_pct = metrics.dialogue_percent - MAX_DIALOGUE_PERCENT # Calculate words to cut target_dialogue_words = int(metrics.word_count * (MAX_DIALOGUE_PERCENT / 100)) words_to_cut = metrics.dialogue_word_count - target_dialogue_words fixes.append(FixInstruction( issue=f"Dialogue percentage too high: {metrics.dialogue_percent}% (max: {MAX_DIALOGUE_PERCENT}%)", severity="critical", instruction=f"""Reduce dialogue from {metrics.dialogue_percent}% to under {MAX_DIALOGUE_PERCENT}%. SPECIFIC GUIDANCE: - Cut approximately {words_to_cut} words of dialogue - Target: expository dialogue, redundant exchanges, filler lines - Preserve: emotional peaks, character-voice moments, plot-critical info - Consider: Can any dialogue become action instead? CURRENT: {metrics.dialogue_word_count} dialogue words / {metrics.word_count} total TARGET: ~{target_dialogue_words} dialogue words maximum.""", target_change=f"-{words_to_cut} dialogue words" )) # Action block length warning if metrics.longest_action_block > ACTION_BLOCK_MAX_LINES: fixes.append(FixInstruction( issue=f"Action block too long: {metrics.longest_action_block} lines (max: {ACTION_BLOCK_MAX_LINES})", severity="warning", instruction=f"""Break up long action block(s) exceeding {ACTION_BLOCK_MAX_LINES} lines. SPECIFIC GUIDANCE: - Apply the Camera Test: each paragraph = one shot - If camera points at different subject → new paragraph - If camera changes scale (wide to close-up) → new paragraph - Cut any unfilmable content (thoughts, backstory, history) The longest block is {metrics.longest_action_block} lines—break into 2-3 line blocks.""", target_change=f"max {ACTION_BLOCK_MAX_LINES} lines per block" )) # Kill Box structure fixes if metrics.kill_box_status != "OK": issue_details = [] if metrics.kill_box_sections_missing: issue_details.append(f"Missing sections: {', '.join(metrics.kill_box_sections_missing)}") if not metrics.kill_box_order_valid: issue_details.append("Sections are out of order") for iss in metrics.kill_box_issues: issue_details.append(iss) fixes.append(FixInstruction( issue=f"Kill Box structure invalid: {'; '.join(issue_details)}", severity="critical", instruction=f"""Fix the Kill Box section structure. Every episode MUST have all 5 sections in order. REQUIRED FORMAT (exact headers): # [00:00 - 00:05] THE HOOK # [00:05 - 00:15] THE SETUP # [00:15 - 00:40] THE ESCALATION # [00:40 - 00:70] THE TURN # [00:70 - 00:90] THE CLIFFHANGER ISSUES FOUND: {chr(10).join('- ' + d for d in issue_details)} RULES: - Each section header must be a # heading with timestamp and section name - Sections must appear in the order above (HOOK → SETUP → ESCALATION → TURN → CLIFFHANGER) - Each section must contain content (no empty sections)""", target_change="fix Kill Box structure" )) return fixes def format_fix_prompt(metrics: EpisodeMetrics, fixes: List[FixInstruction], episode_content: str) -> str: """ Format a complete prompt for Claude to fix the episode. """ fix_sections = [] for i, fix in enumerate(fixes, 1): fix_sections.append(f""" ### Fix {i}: {fix.issue} **Severity:** {fix.severity.upper()} **Target Change:** {fix.target_change} {fix.instruction} """) return f"""## Episode Edit Required The following episode has constraint violations that need fixing. ### Current Metrics - **Word Count:** {metrics.word_count} (target: {WORD_COUNT_MIN}-{WORD_COUNT_MAX}) - **Exchanges:** {metrics.exchange_count} (max: {MAX_EXCHANGES}) - **Dialogue %:** {metrics.dialogue_percent}% (max: {MAX_DIALOGUE_PERCENT}%) - **Characters:** {', '.join(metrics.character_names)} - **Kill Box:** {metrics.kill_box_status} ({len(metrics.kill_box_sections_found)}/5 sections) ### Required Fixes {''.join(fix_sections)} ### Instructions Edit the episode below to address ALL fixes marked as CRITICAL. Maintain character voice, emotional beats, and story progression. Output the complete corrected episode. --- ## Episode to Edit {episode_content} --- ## Output Provide the corrected episode in full, maintaining Fountain format. """ # ============================================================================= # MAIN / CLI # ============================================================================= def main(): parser = argparse.ArgumentParser( description='Measure episode metrics and generate fix instructions' ) parser.add_argument('episode', type=Path, help='Path to episode file') parser.add_argument('--fix', action='store_true', help='Generate fix instructions if out of range') parser.add_argument('--json', action='store_true', help='Output metrics as JSON') parser.add_argument('--prompt', action='store_true', help='Output full edit prompt for Claude') parser.add_argument('--quiet', action='store_true', help='Only output if issues found') args = parser.parse_args() if not args.episode.exists(): print(f"ERROR: File not found: {args.episode}", file=sys.stderr) sys.exit(2) try: metrics = measure_episode(args.episode) except Exception as e: print(f"ERROR: Failed to parse episode: {e}", file=sys.stderr) sys.exit(2) # JSON output mode if args.json: output = { 'file': str(args.episode), 'word_count': metrics.word_count, 'word_count_status': metrics.word_count_status, 'exchange_count': metrics.exchange_count, 'exchange_status': metrics.exchange_status, 'dialogue_percent': metrics.dialogue_percent, 'dialogue_status': metrics.dialogue_status, 'action_block_count': metrics.action_block_count, 'longest_action_block': metrics.longest_action_block, 'characters': metrics.character_names, 'kill_box_status': metrics.kill_box_status, 'kill_box_sections_found': metrics.kill_box_sections_found, 'kill_box_sections_missing': metrics.kill_box_sections_missing, 'kill_box_order_valid': metrics.kill_box_order_valid, 'kill_box_issues': metrics.kill_box_issues, 'is_valid': metrics.is_valid } if args.fix and not metrics.is_valid: fixes = generate_fix_instructions(metrics) output['fixes'] = [ { 'issue': f.issue, 'severity': f.severity, 'instruction': f.instruction, 'target_change': f.target_change } for f in fixes ] print(json.dumps(output, indent=2)) sys.exit(0 if metrics.is_valid else 1) # Full prompt output mode if args.prompt and not metrics.is_valid: fixes = generate_fix_instructions(metrics) episode_content = args.episode.read_text(encoding='utf-8') print(format_fix_prompt(metrics, fixes, episode_content)) sys.exit(1) # Standard output mode if not args.quiet or not metrics.is_valid: status = "VALID" if metrics.is_valid else "INVALID" print(f"\n{'='*60}") print(f"Episode: {args.episode.name}") print(f"Status: {status}") print(f"{'='*60}") print(f"\nMETRICS:") print(f" Word Count: {metrics.word_count:4d} [{metrics.word_count_status}] (target: {WORD_COUNT_MIN}-{WORD_COUNT_MAX})") print(f" Exchanges: {metrics.exchange_count:4d} [{metrics.exchange_status}] (max: {MAX_EXCHANGES})") print(f" Dialogue %: {metrics.dialogue_percent:5.1f}% [{metrics.dialogue_status}] (max: {MAX_DIALOGUE_PERCENT}%)") print(f" Action Blocks: {metrics.action_block_count:4d}") print(f" Longest Block: {metrics.longest_action_block:4d} lines (max: {ACTION_BLOCK_MAX_LINES})") print(f" Characters: {', '.join(metrics.character_names)}") print(f" Kill Box: [{metrics.kill_box_status}] ({len(metrics.kill_box_sections_found)}/5 sections)") if metrics.kill_box_sections_missing: print(f" Missing: {', '.join(metrics.kill_box_sections_missing)}") for iss in metrics.kill_box_issues: print(f" Issue: {iss}") # Generate fix instructions if requested if args.fix and not metrics.is_valid: fixes = generate_fix_instructions(metrics) print(f"\nFIX INSTRUCTIONS:") for i, fix in enumerate(fixes, 1): print(f"\n--- Fix {i} ({fix.severity.upper()}) ---") print(f"Issue: {fix.issue}") print(f"Target: {fix.target_change}") print(f"\n{fix.instruction}") sys.exit(0 if metrics.is_valid else 1) if __name__ == '__main__': main()