#!/usr/bin/env python3 """ Puzzle Box Format Validator Validates episodes against Puzzle Box format constraints. 30-second mood/mystery microserial format. Interface: validate_episode(episode_path, constants=None) -> dict validate_batch(episode_paths, constants=None) -> dict """ import re from pathlib import Path from typing import Optional # ============================================================================= # CONSTANTS LOADING # ============================================================================= def _parse_constants_from_md(filepath: Path) -> dict: """Parse CONSTANTS.md and extract key values.""" content = filepath.read_text() constants = {} table_pattern = r'\|\s*`([A-Z_]+)`\s*\|\s*([^|]+)\s*\|' for match in re.finditer(table_pattern, content): name = match.group(1).strip() value_str = match.group(2).strip() # Parse value if value_str.endswith('%'): try: constants[name] = float(value_str.rstrip('%')) except ValueError: constants[name] = value_str elif value_str.lower() in ('true', 'false'): constants[name] = value_str.lower() == 'true' else: try: constants[name] = int(value_str) except ValueError: try: constants[name] = float(value_str) except ValueError: constants[name] = value_str return constants def _load_default_constants() -> dict: """Load constants from this format's CONSTANTS.md.""" constants_path = Path(__file__).parent / 'CONSTANTS.md' if constants_path.exists(): return _parse_constants_from_md(constants_path) # Hardcoded fallback return { 'NARRATIVE_MIN_WORDS': 40, 'NARRATIVE_MAX_WORDS': 80, 'LINKAGE_REQUIRED_FROM': 5, } # ============================================================================= # VALID VALUES # ============================================================================= VALID_ENDING_TYPES = {'RHYME', 'WITHHOLD', 'DISSONANCE', 'OBJECT', 'ABSENCE'} VALID_RHYTHM_TAGS = {'SUSPENDED', 'LAYERED', 'KINETIC', 'DRIFT'} REQUIRED_BEATS = {'ENTRY IMAGE', 'VOICE', 'LINGER'} FRACTURE_BEATS = {'BREAK', 'AFTERMATH'} # ============================================================================= # PARSING HELPERS # ============================================================================= def _count_words(text: str) -> int: """Count words in text.""" return len(text.split()) def _extract_section(content: str, header_pattern: str) -> Optional[str]: """Extract content under a markdown section header.""" match = re.search(header_pattern, content, re.IGNORECASE | re.MULTILINE) if not match: return None start = match.end() # Find next same-level or higher-level header next_header = re.search(r'^#{1,2}\s', content[start:], re.MULTILINE) if next_header: return content[start:start + next_header.start()].strip() return content[start:].strip() def _extract_layer_text(content: str, layer_marker: str) -> str: """ Extract text belonging to a specific layer (NARRATIVE SCRIPT or PIPELINE DIRECTION). Returns the text content of that layer section. """ # Look for the layer header as a ## heading pattern = rf'^##\s+{re.escape(layer_marker)}.*$' match = re.search(pattern, content, re.IGNORECASE | re.MULTILINE) if not match: return '' start = match.end() # Find next ## heading or --- separator that isn't inside the section next_section = re.search(r'^(?:##\s(?!#)|---\s*$)', content[start:], re.MULTILINE) if next_section: return content[start:start + next_section.start()].strip() return content[start:].strip() def _detect_episode_number(filepath: Path) -> Optional[int]: """Try to detect episode number from filename or content.""" # From filename: ep_003.md, ep03.md, etc. match = re.search(r'ep[_]?(\d+)', filepath.stem, re.IGNORECASE) if match: return int(match.group(1)) # From content: # EP03 or # EP 3 try: content = filepath.read_text(encoding='utf-8') match = re.search(r'#\s*EP\s*(\d+)', content, re.IGNORECASE) if match: return int(match.group(1)) except Exception: pass return None # ============================================================================= # PUBLIC API # ============================================================================= def validate_episode(episode_path: str, constants: dict = None) -> dict: """ Validate a single Puzzle Box episode. Args: episode_path: Path to the episode file. constants: Optional dict of constants. If None, loads from CONSTANTS.md. Returns: {valid: bool, errors: [], warnings: [], metrics: {}} """ filepath = Path(episode_path) errors = [] warnings = [] metrics = {} if constants is None: constants = _load_default_constants() narrative_min = constants.get('NARRATIVE_MIN_WORDS', 40) narrative_max = constants.get('NARRATIVE_MAX_WORDS', 80) linkage_from = int(constants.get('LINKAGE_REQUIRED_FROM', 5)) if not filepath.exists(): return { 'valid': False, 'errors': [f"File not found: {filepath}"], 'warnings': [], 'metrics': {}, } content = filepath.read_text(encoding='utf-8') # ----------------------------------------------------------------- # Narrative script layer # ----------------------------------------------------------------- narrative_text = _extract_layer_text(content, 'NARRATIVE SCRIPT') has_narrative = bool(narrative_text) metrics['has_narrative_layer'] = has_narrative if not has_narrative: errors.append("Missing NARRATIVE SCRIPT layer") # ----------------------------------------------------------------- # Word counts (narrative only) # ----------------------------------------------------------------- narrative_words = _count_words(narrative_text) if has_narrative else 0 metrics['narrative_word_count'] = narrative_words # Narrative word count: warn outside 40-80 range if narrative_words > narrative_max and has_narrative: warnings.append(f"Narrative word count above target: {narrative_words} (target: {narrative_min}-{narrative_max})") elif narrative_words < narrative_min and has_narrative: warnings.append(f"Narrative word count below target: {narrative_words} (target: {narrative_min}-{narrative_max})") # ----------------------------------------------------------------- # Beat sections: ENTRY IMAGE, VOICE, LINGER (or BREAK, AFTERMATH for FRACTURE) # Disruption episodes declare non-standard beat structure — skip beat check # ----------------------------------------------------------------- # A FRACTURE episode has "(FRACTURE)" in its title line is_fracture = bool(re.search(r'^#\s+EP\d+\s.*\(FRACTURE\)', content[:200], re.IGNORECASE | re.MULTILINE)) has_disruption = bool(re.search(r'Disruption\s*:', content, re.IGNORECASE)) metrics['is_fracture'] = is_fracture metrics['has_disruption'] = has_disruption if has_disruption: # Disruption episodes intentionally break standard beat structure metrics['beats_found'] = [] else: required = FRACTURE_BEATS if is_fracture else REQUIRED_BEATS beats_found = set() for beat in required: # Match ### [timestamp] BEAT_NAME or ## BEAT_NAME pattern = rf'#{{2,3}}\s*(?:\[[\d:s\-\s]+\]\s*)?{re.escape(beat)}' if re.search(pattern, content, re.IGNORECASE): beats_found.add(beat) metrics['beats_found'] = sorted(beats_found) missing_beats = required - beats_found if missing_beats: errors.append(f"Missing required beats: {', '.join(sorted(missing_beats))}") # ----------------------------------------------------------------- # WORLD VOTE or ORACLE section # ----------------------------------------------------------------- has_world_vote = bool(re.search(r'##\s*WORLD\s+VOTE', content, re.IGNORECASE)) has_oracle = bool(re.search(r'##\s*ORACLE', content, re.IGNORECASE)) metrics['has_world_vote'] = has_world_vote metrics['has_oracle'] = has_oracle if not has_world_vote and not has_oracle: errors.append("Missing WORLD VOTE or ORACLE section") # ----------------------------------------------------------------- # Ending type annotation # ----------------------------------------------------------------- ending_match = re.search( r'Ending\s*Type\s*:\s*(RHYME|WITHHOLD|DISSONANCE|OBJECT|ABSENCE)', content, re.IGNORECASE ) if ending_match: ending_type = ending_match.group(1).upper() metrics['ending_type'] = ending_type if ending_type not in VALID_ENDING_TYPES: errors.append(f"Invalid ending type: {ending_type} (must be one of {', '.join(sorted(VALID_ENDING_TYPES))})") else: errors.append(f"Missing ending type annotation (must be one of: {', '.join(sorted(VALID_ENDING_TYPES))})") # ----------------------------------------------------------------- # Fragment linkage (required from episode 5 onward) # ----------------------------------------------------------------- ep_num = _detect_episode_number(filepath) metrics['episode_number'] = ep_num has_fragment = bool(re.search(r'##\s*FRAGMENT', content, re.IGNORECASE)) metrics['has_fragment'] = has_fragment if ep_num is not None and ep_num >= linkage_from: if not has_fragment: errors.append(f"Missing FRAGMENT section (required from episode {linkage_from} onward)") # Cross-episode verification is a stub warnings.append("Cross-episode fragment linkage validation not yet implemented") # ----------------------------------------------------------------- # Rhythm tag # ----------------------------------------------------------------- rhythm_match = re.search(r'Rhythm\s*:\s*(\w+)', content, re.IGNORECASE) if rhythm_match: rhythm_tag = rhythm_match.group(1).upper() metrics['rhythm_tag'] = rhythm_tag if rhythm_tag not in VALID_RHYTHM_TAGS: errors.append(f"Invalid rhythm tag: {rhythm_tag} (must be one of {', '.join(sorted(VALID_RHYTHM_TAGS))})") else: errors.append(f"Missing rhythm tag (must be one of: {', '.join(sorted(VALID_RHYTHM_TAGS))})") # ----------------------------------------------------------------- # THE MOMENT annotation # ----------------------------------------------------------------- has_moment = bool(re.search(r'##\s*THE\s+MOMENT', content, re.IGNORECASE)) metrics['has_the_moment'] = has_moment if not has_moment: warnings.append("Missing THE MOMENT annotation") # ----------------------------------------------------------------- # Metadata section # ----------------------------------------------------------------- has_metadata = bool(re.search(r'##\s*Metadata', content, re.IGNORECASE)) metrics['has_metadata'] = has_metadata if has_metadata: metadata_text = _extract_section(content, r'^##\s*Metadata') required_fields = ['Exposure', 'Sequence', 'Rhythm', 'Ending Type'] if metadata_text: for field in required_fields: if not re.search(rf'{re.escape(field)}\s*:', metadata_text, re.IGNORECASE): warnings.append(f"Metadata missing field: {field}") else: errors.append("Missing Metadata section") return { 'valid': len(errors) == 0, 'errors': errors, 'warnings': warnings, 'metrics': metrics, } def validate_batch(episode_paths: list, constants: dict = None) -> dict: """ Validate a batch of Puzzle Box episodes. Args: episode_paths: List of file paths to episode files. constants: Optional dict of constants. If None, loads from CONSTANTS.md. Returns: {valid: bool, episode_results: [...]} """ if constants is None: constants = _load_default_constants() results = [] all_valid = True for ep_path in episode_paths: result = validate_episode(str(ep_path), constants=constants) result['file'] = str(ep_path) results.append(result) if not result['valid']: all_valid = False return { 'valid': all_valid, 'episode_results': results, }