#!/usr/bin/env python3 """ prompt_compiler.py — Stateless Resolver-Compiler for T2I/Video Prompts Replaces prompt_engine.py with a single compile() function that resolves character, wardrobe, and location data from breakdown.json, then assembles 10 layers into a model-aware prompt string. Three entry points use this same function: 1. generate_storyboard_keyframes.py (generation) 2. serve.py /api/preview-prompt (editorial preview) 3. CLI (this file's __main__) Usage: # Preview all 3 frame prompts for a shot python3 prompt_compiler.py leviathan/ preview --episode 1 --shot 4 # Preview specific frame python3 prompt_compiler.py leviathan/ preview --episode 1 --shot 4 --frame hero # Validate verb strength across all shots python3 prompt_compiler.py leviathan/ validate --episode 1 # Add override python3 prompt_compiler.py leviathan/ override add \\ --scope character --key "char.jinx.wardrobe_props" \\ --operation append --value "wrist strap with gunmetal housing" \\ --reason "Standardize debt counter" # List overrides python3 prompt_compiler.py leviathan/ override list Dependencies: None (stdlib only). Imported by generate_storyboard_keyframes.py. """ import argparse import hashlib import json import logging import re import sys from dataclasses import dataclass, field, asdict from datetime import datetime, timezone from pathlib import Path from typing import List, Optional from recoil.core.config_loader import DEFAULT_PROJECT_CONFIG as _DEFAULT_PROJECT_CONFIG from recoil.core.config_loader import load_project_config as _config_loader_load from recoil.core.exceptions import PromptCompilerOverridesCorruptError log = logging.getLogger(__name__) # ── Layer Names (order matters) ────────────────────────────────────────── LAYER_ORDER = [ "lora_triggers", "subject", "action_pose", "wardrobe_props", "environment", "lighting", "color_objects", "camera_lens", "film_style", "quality_guard", ] # ── Frame Types → Verb Classes ─────────────────────────────────────────── VERB_CLASSES = { "hero": "ACTIVE", "first": "PREPARATION", "last": "RESULT", } # ── Model Capabilities ────────────────────────────────────────────────── MODEL_CAPABILITIES = { "z_image": {"negative_prompt": False, "guidance": False, "steps": 8}, "z_image_base": {"negative_prompt": True, "guidance": True, "steps": 28}, "flux2": {"negative_prompt": False, "guidance": True, "steps": 28}, "wan_2.2": {"negative_prompt": True, "guidance": True, "steps": 40}, } # ── Default Quality Guard ─────────────────────────────────────────────── DEFAULT_QUALITY_GUARD = ( "Correct human anatomy, anatomically correct proportions, " "five fingers per hand, sharp focus, clean detailed image, " "natural skin texture with pores" ) DEFAULT_NEGATIVE_PROMPT = ( "deformed hands, extra fingers, mutated hands, poorly drawn hands, " "extra limbs, fused fingers, too many fingers, long neck, " "blurry, low quality, illustration, cartoon, painting, drawing, " "3d render, anime, cgi, digital art, smooth skin" ) # ── Default Project Config ───────────────────────────────────────────── # Imported from config_loader.py — single source of truth for all tools. # ── VFX Language Blacklist (T2I only) ───────────────────────────────── # These phrases cause literal spirals, geometric noise, and artifacts on faces # when passed to T2I models. Safe for video/motion prompts only. # See visual_production_findings.md: "VFX language poisons T2I output" _VFX_BLACKLIST = [ r"targeting reticle[s]?", r"data stream[s]?", r"holographic (?:HUD|display|interface|overlay|projection)", r"heads?-up display", r"digital (?:interface|readout|display|overlay|scan|data)", r"scanning beam[s]?", r"laser grid[s]?", r"energy (?:field|beam|pulse|wave|shield|barrier)", r"force field[s]?", r"particle effect[s]?", r"glitch effect[s]?", r"pixel(?:ated|ation|s)?", r"matrix (?:code|rain|overlay)", r"wireframe", r"augmented reality", r"HUD (?:element|overlay|display)", r"targeting (?:system|overlay|lock|display)", r"scan(?:ning)? (?:line|grid|overlay)", r"cyber(?:netic)? (?:implant|overlay|interface)", r"neural (?:link|interface|network) (?:glow|overlay|visual)", ] _VFX_BLACKLIST_COMPILED = [re.compile(p, re.IGNORECASE) for p in _VFX_BLACKLIST] def _strip_vfx_language(prompt: str) -> tuple: """Strip VFX vocabulary from T2I prompts that causes face artifacts. Returns: (cleaned_prompt, list_of_stripped_phrases) """ stripped = [] cleaned = prompt for pattern in _VFX_BLACKLIST_COMPILED: match = pattern.search(cleaned) if match: stripped.append(match.group()) cleaned = pattern.sub("", cleaned) # Clean up double spaces/periods left by removal cleaned = re.sub(r"\s{2,}", " ", cleaned) cleaned = re.sub(r"\.\s*\.", ".", cleaned) cleaned = re.sub(r",\s*,", ",", cleaned) cleaned = cleaned.strip(" ,.") return cleaned, stripped # ── LoRA key map per model ────────────────────────────────────────────── _LORA_KEY_MAP = { "flux2": "t2i_path", "z_image": "z_image_t2i_path", "z_image_base": "z_image_base_t2i_path", "wan_2.2": "video_lora_high", } # ── Shot Type → Framing Language ────────────────────────────────────────── SHOT_TYPE_FRAMING = { "ECU": "Macro photograph, extreme close-up filling the entire frame, only this single detail visible, nothing else in frame, camera inches away, shallow depth of field, background completely out of focus", "CU": "Close-up portrait, tight crop from chin to top of head, face and shoulders filling the frame, shallow depth of field, soft bokeh background, minimal environment visible", "MCU": "Medium close-up, chest up framing, subject prominent in frame, shallow depth of field", "MS": "Medium shot, waist up framing", "LS": "Long shot, full body visible with environment context, deep focus", "WIDE": "Wide establishing shot, full environment visible, figures small in frame, deep focus, expansive composition, architectural scale", } # Layers to SKIP per shot type (these layers contain irrelevant info for the framing) _SHOT_TYPE_SKIP_LAYERS = { "ECU": {"lora_triggers", "subject", "wardrobe_props", "environment", "color_objects", "quality_guard"}, # ECU = macro detail, no character/body "CU": {"wardrobe_props"}, # Minimal wardrobe visible in CU "WIDE": set(), # WIDE uses all layers } # ── Z-Image Compile-to-Budget ──────────────────────────────────────────── # Z-Image Turbo attention fades at ~75 tokens (~55 words). # Instead of compiling a 200-word prompt and chopping it, we compile # *within budget* from the start: compact framing, per-layer word caps, # and low-value layers skipped entirely. _Z_IMAGE_COMPACT_FRAMING = { "ECU": "Extreme close-up, tight crop, shallow depth of field", "CU": "Close-up portrait, face and shoulders, shallow depth of field", "MCU": "Medium close-up, chest up, shallow depth of field", "MS": "Medium shot, waist up", "LS": "Full body, environment visible, deep focus", "WIDE": "Wide shot, expansive environment, figures small in frame", } _Z_IMAGE_BUDGET = 60 # hard word cap # Per-layer word budgets for Z-Image. Layers not listed are always skipped. # Priority order: framing > triggers > subject > action > quality > wardrobe > environment _Z_IMAGE_LAYER_BUDGETS = { "lora_triggers": 5, "subject": 15, "action_pose": 15, "wardrobe_props": 8, "environment": 8, "quality_guard": 8, } # Layers always skipped for Z-Image — low impact, waste attention window _Z_IMAGE_ALWAYS_SKIP = {"lighting", "color_objects", "camera_lens", "film_style"} def _cap_layer_words(text: str, max_words: int) -> str: """Truncate layer text to max_words, ending at a sentence or comma boundary.""" words = text.split() if len(words) <= max_words: return text truncated = " ".join(words[:max_words]) # Try to end at a sentence boundary last_period = truncated.rfind(".") if last_period > len(truncated) * 0.5: return truncated[:last_period + 1] # Try to end at a comma last_comma = truncated.rfind(",") if last_comma > len(truncated) * 0.5: return truncated[:last_comma] return truncated.rstrip(" ,.") CAMERA_MOVEMENT_LANGUAGE = { "dolly": "smooth dolly movement", "track": "tracking shot", "handheld": "handheld camera energy", "crane": "crane movement", "push in": "slow push in", "pull back": "slow pull back", "pan": "panning movement", "tilt": "tilting movement", "steadicam": "steadicam floating movement", } # ── Data Classes ──────────────────────────────────────────────────────── @dataclass class PromptLayers: """10-layer prompt structure.""" lora_triggers: str = "" subject: str = "" action_pose: str = "" wardrobe_props: str = "" environment: str = "" lighting: str = "" color_objects: str = "" camera_lens: str = "" film_style: str = "" quality_guard: str = "" @dataclass class PreviousShotContext: """Carries shot metadata + layers for transition-aware prompts. Unlike PromptLayers (just text), this includes the cinematic metadata needed to compute transition language between shots. """ layers: Optional[PromptLayers] = None shot_type: str = "" # ECU, CU, MCU, MS, LS, WIDE camera_angle: str = "" # eye, low, high, dutch, bird camera_side: str = "" # A, B (180° rule) camera_movement: str = "" # static, dolly, track, handheld, etc. screen_direction: str = "" # left-to-right, right-to-left shot_id: int = 0 def build_prev_context( shot: dict, breakdown: dict, episode: int, storyboard: dict = None, model: str = "z_image", lora_registry: dict = None, project_config: dict = None, frame_type: str = "last", ) -> "PreviousShotContext": """Build PreviousShotContext from a shot dict. Reusable helper for CLI preview, serve.py, previz, and any other consumer that needs transition-aware prompts. """ return PreviousShotContext( layers=build_layers(shot, breakdown, episode, storyboard, model=model, lora_registry=lora_registry, project_config=project_config, frame_type=frame_type), shot_type=shot.get("shot_type", ""), camera_angle=shot.get("camera_angle", "eye"), camera_side=shot.get("spatial", {}).get("camera_side", ""), camera_movement=shot.get("camera_movement", "static"), screen_direction=shot.get("spatial", {}).get("screen_direction", ""), shot_id=shot.get("id", 0), ) # Shot scale order for transition language (tighter = lower number) SCALE_ORDER = {"ECU": 0, "CU": 1, "MCU": 2, "MS": 3, "LS": 4, "WIDE": 5} @dataclass class OverrideEntry: """Single override entry.""" scope: str # global, character, episode, shot key: str # e.g. "char.jinx.wardrobe_props" operation: str # replace, prepend, append, remove_phrase value: str reason: str = "" created_at: str = "" @dataclass class NoteEntry: """Production note / feedback entry.""" episode: int shot_id: int frame_type: str = "hero" issue: str = "" fix: str = "" tags: List[str] = field(default_factory=list) override_key: Optional[str] = None created_at: str = "" # ── Override Store ─────────────────────────────────────────────────────── class OverrideStore: """Manages prompt_overrides.json — 4-scope override system.""" def __init__(self, project_dir: Path): self.path = project_dir / "visual" / "prompt_overrides.json" self.overrides: List[dict] = [] self._load() def _load(self): if self.path.exists(): try: with open(self.path) as f: data = json.load(f) self.overrides = data.get("overrides", []) except FileNotFoundError: self.overrides = [] # legitimate missing file except (json.JSONDecodeError, OSError) as e: log.warning( "prompt_compiler: corrupt overrides at %s — %s", self.path, e, ) raise PromptCompilerOverridesCorruptError( str(self.path), message=str(e) ) from e def save(self): self.path.parent.mkdir(parents=True, exist_ok=True) data = { "version": 1, "updated_at": datetime.now(timezone.utc).isoformat(), "overrides": self.overrides, } with open(self.path, "w") as f: json.dump(data, f, indent=2) def add(self, entry: OverrideEntry): entry.created_at = datetime.now(timezone.utc).isoformat() self.overrides.append(asdict(entry)) self.save() def remove(self, index: int): if 0 <= index < len(self.overrides): self.overrides.pop(index) self.save() def list_all(self) -> List[dict]: return self.overrides def get_applicable(self, layer: str, character: str = "", episode: int = 0, shot_id: int = 0, frame_type: str = "") -> List[dict]: """Get overrides applicable to a specific context, broadest first.""" applicable = [] for ov in self.overrides: key = ov.get("key", "") scope = ov.get("scope", "") if scope == "global" and key == f"global.{layer}": applicable.append(ov) elif scope == "character" and key == f"char.{character}.{layer}": applicable.append(ov) elif scope == "episode" and key == f"ep.{episode}.{layer}": applicable.append(ov) elif scope == "shot": shot_key = f"shot.{episode}.{shot_id}.{frame_type}.{layer}" shot_key_any = f"shot.{episode}.{shot_id}.*.{layer}" if key in (shot_key, shot_key_any): applicable.append(ov) return applicable # ── Note Store ─────────────────────────────────────────────────────────── class NoteStore: """Manages prompt_notes.json — production feedback.""" def __init__(self, project_dir: Path): self.path = project_dir / "visual" / "prompt_notes.json" self.notes: List[dict] = [] self._load() def _load(self): if self.path.exists(): try: with open(self.path) as f: data = json.load(f) self.notes = data.get("notes", []) except FileNotFoundError: self.notes = [] # legitimate missing file except (json.JSONDecodeError, OSError) as e: log.warning( "prompt_compiler: corrupt notes at %s — %s", self.path, e, ) raise PromptCompilerOverridesCorruptError( str(self.path), message=str(e) ) from e def save(self): self.path.parent.mkdir(parents=True, exist_ok=True) data = { "version": 1, "updated_at": datetime.now(timezone.utc).isoformat(), "notes": self.notes, } with open(self.path, "w") as f: json.dump(data, f, indent=2) def add(self, entry: NoteEntry): entry.created_at = datetime.now(timezone.utc).isoformat() self.notes.append(asdict(entry)) self.save() def list_all(self) -> List[dict]: return self.notes def get_by_tag(self, tag: str) -> List[dict]: return [n for n in self.notes if tag in n.get("tags", [])] # ── Resolution Functions ──────────────────────────────────────────────── def _resolve_character(name: str, episode: int, breakdown: dict) -> dict: """Resolve character visual data from breakdown for a given episode. Returns: { "visual": str, # visual_description from breakdown "wardrobe_desc": str, # wardrobe phase description "wardrobe_phase": str, # wardrobe phase name "hair_makeup": str, # combined hair + skin for episode "trigger": str, # LoRA trigger (from breakdown prompts) "reference_prompt": str, # reference prompt variant for phase } """ characters = breakdown.get("characters", {}) # Try uppercase key first, then original char_data = characters.get(name.upper(), characters.get(name, {})) if not char_data: return {} result = { "visual": char_data.get("visual_description", ""), "wardrobe_desc": "", "wardrobe_phase": "", "hair_makeup": "", "trigger": "", "reference_prompt": "", } # Resolve wardrobe phase for this episode wardrobe = char_data.get("wardrobe", {}) for phase_name, phase_data in wardrobe.items(): ep_range = phase_data.get("episodes", []) if len(ep_range) == 2 and ep_range[0] <= episode <= ep_range[1]: result["wardrobe_desc"] = phase_data.get("description", "") result["wardrobe_phase"] = phase_name break # Resolve hair/makeup state for this episode hair_makeup = char_data.get("hair_makeup", {}) for state_name, state_data in hair_makeup.items(): ep_range = state_data.get("episodes", []) if len(ep_range) == 2 and ep_range[0] <= episode <= ep_range[1]: parts = [] hair = state_data.get("hair", "") skin = state_data.get("skin", "") if hair: parts.append(f"Hair: {hair}") if skin: parts.append(f"Skin: {skin}") result["hair_makeup"] = ". ".join(parts) break # Reference prompt variant prompts = char_data.get("prompts", {}) ref_prompts = prompts.get("reference", {}) variants = ref_prompts.get("variants", {}) if result["wardrobe_phase"] and result["wardrobe_phase"] in variants: result["reference_prompt"] = variants[result["wardrobe_phase"]] return result def _resolve_location(shot: dict, breakdown: dict) -> dict: """Resolve location data from breakdown matching shot atmosphere/location. Returns: { "description_samples": list, "habitat_zone": str, "lighting_notes": list, } """ locations = breakdown.get("locations", {}) if not locations: return {} # Try to match by shot's location field or atmosphere atmosphere = shot.get("atmosphere", "") location_name = shot.get("location", "") beat = shot.get("beat", "") # Direct match by location name if location_name: for loc_key, loc_data in locations.items(): if location_name.lower() in loc_key.lower(): return { "description_samples": loc_data.get("description_samples", []), "habitat_zone": loc_data.get("habitat_zone", ""), "lighting_notes": loc_data.get("lighting_notes", []), } # Fuzzy match: try matching key words from atmosphere or beat search_terms = (atmosphere + " " + beat).lower().split() best_match = None best_score = 0 for loc_key, loc_data in locations.items(): loc_lower = loc_key.lower() score = sum(1 for term in search_terms if term in loc_lower and len(term) > 3) if score > best_score: best_score = score best_match = loc_data if best_match and best_score > 0: return { "description_samples": best_match.get("description_samples", []), "habitat_zone": best_match.get("habitat_zone", ""), "lighting_notes": best_match.get("lighting_notes", []), } return {} def _resolve_wardrobe(name: str, episode: int, breakdown: dict) -> str: """Resolve wardrobe phase description for a character at an episode. Subset of _resolve_character — just the wardrobe + hair/makeup combined. """ char = _resolve_character(name, episode, breakdown) parts = [] if char.get("wardrobe_desc"): parts.append(char["wardrobe_desc"]) if char.get("hair_makeup"): parts.append(char["hair_makeup"]) return ". ".join(parts) # ── Character Detection ───────────────────────────────────────────────── def _detect_characters(shot: dict, storyboard: dict) -> List[str]: """Detect characters in shot from explicit field or text scan.""" chars = shot.get("characters_in_shot", []) if chars: return [c.lower() for c in chars] search = " ".join([ shot.get("first_frame", ""), shot.get("last_frame", ""), shot.get("subject", ""), shot.get("action", ""), shot.get("name", ""), shot.get("script_excerpt", ""), ]).lower() storyboard_chars = storyboard.get("characters", {}) return [c.lower() for c in storyboard_chars if c.lower() in search] # ── Legacy Detection ──────────────────────────────────────────────────── def _is_legacy_prompt(shot: dict) -> bool: """Detect pre-baked prose prompts (legacy mode).""" if shot.get("prompt_mode") == "structured": return False if shot.get("hero_action"): return False first = shot.get("first_frame", "") word_count = len(first.split()) has_camera = "Shot on" in first or "Kodak" in first or "Arri" in first return word_count > 80 and has_camera def _is_rich_hero_frame(shot: dict, frame_type: str) -> bool: """Detect when hero_frame provides rich E-style prose that already contains camera, film stock, and wardrobe information. When true, the compiler skips wardrobe_props, camera_lens, and film_style layers to avoid redundancy (hero_frame already includes them). """ if frame_type != "hero": return False hero = shot.get("hero_frame", "") if not hero: return False word_count = len(hero.split()) # E-style prose is 150+ words and includes camera/film info has_camera = "Shot on" in hero or "Arri" in hero or "Kodak" in hero return word_count > 100 and has_camera # ── Layer Builders ────────────────────────────────────────────────────── def _build_lora_triggers(shot: dict, storyboard: dict, lora_registry: dict, model: str) -> str: """Layer 1: LoRA trigger words.""" chars = _detect_characters(shot, storyboard) lora_key = _LORA_KEY_MAP.get(model, "z_image_t2i_path") triggers = [] for c in chars: reg = lora_registry.get(c, {}) if reg.get("trigger") and reg.get(lora_key): triggers.append(reg["trigger"]) return ", ".join(triggers) def _build_subject(shot: dict, frame_type: str, episode: int, storyboard: dict, breakdown: dict) -> str: """Layer 2: Character visual DNA + emotion. Resolution priority: 1. breakdown → _resolve_character() visual description 2. storyboard characters.visual (fallback) """ chars = _detect_characters(shot, storyboard) emotion = shot.get("emotion", "") parts = [] for c in chars: # Try breakdown resolution first resolved = _resolve_character(c, episode, breakdown) visual = resolved.get("visual", "") # Fall back to storyboard character visual if not visual: char_data = storyboard.get("characters", {}).get(c, {}) visual = char_data.get("visual", "") if visual: parts.append(visual) if emotion: parts.append(f"expression: {emotion.lower()}") return ", ".join(parts) def _build_action_pose(shot: dict, frame_type: str) -> str: """Layer 3: Frame-specific verb state. Priority: structured verb-state fields > storyboard first/last frame prose > action. The storyboard's first_frame/last_frame fields describe different moments in the shot, so using them produces genuinely different prompts for first vs last keyframes. """ # Structured verb-state fields (populated by /storyboard when available) if frame_type == "hero" and shot.get("hero_action"): return shot["hero_action"] elif frame_type == "first" and shot.get("anticipation_action"): return shot["anticipation_action"] elif frame_type == "last" and shot.get("aftermath_action"): return shot["aftermath_action"] # Fall back to storyboard first_frame/last_frame prose descriptions. # These describe genuinely different moments (start vs end of shot action). if frame_type == "first" and shot.get("first_frame"): return shot["first_frame"] if frame_type == "last" and shot.get("last_frame"): return shot["last_frame"] if frame_type == "hero" and shot.get("hero_frame"): return shot["hero_frame"] # Default: action field (same for all frame types — the pre-fix behavior) action = shot.get("action", "") if action: return action return shot.get("subject", "") def _build_wardrobe_props(shot: dict, episode: int, storyboard: dict, breakdown: dict) -> str: """Layer 4: Wardrobe + props resolved from breakdown. Now resolves from breakdown.json wardrobe phases instead of returning "". """ chars = _detect_characters(shot, storyboard) parts = [] for c in chars: wardrobe_desc = _resolve_wardrobe(c, episode, breakdown) if wardrobe_desc: parts.append(wardrobe_desc) return ". ".join(parts) def _build_environment(shot: dict, breakdown: dict) -> str: """Layer 5: Habitat zone materials + shot atmosphere. Combines shot atmosphere with breakdown location data. """ atmosphere = shot.get("atmosphere", "") # Enrich with breakdown location data loc = _resolve_location(shot, breakdown) parts = [] if atmosphere: parts.append(atmosphere) # Add first description sample for environmental texture samples = loc.get("description_samples", []) if samples and not atmosphere: # Only use sample if no atmosphere was specified parts.append(samples[0][:150]) return ". ".join(parts) if parts else "" def _build_lighting(shot: dict, breakdown: dict) -> str: """Layer 6: Shot lighting + habitat zone lighting.""" lighting = shot.get("lighting", "") if not lighting: gen_meta = shot.get("generation_metadata", {}) light_data = gen_meta.get("lighting", {}) if light_data: parts = [] if light_data.get("type"): parts.append(light_data["type"]) if light_data.get("source"): parts.append(light_data["source"]) if light_data.get("color_temp"): parts.append(light_data["color_temp"]) lighting = ", ".join(parts) # Fallback: breakdown location lighting notes if not lighting: loc = _resolve_location(shot, breakdown) notes = loc.get("lighting_notes", []) if notes: lighting = ", ".join(notes) return lighting def _build_color_objects(shot: dict, storyboard: dict, project_config: dict) -> str: """Layer 7: HEX codes linked to specific in-frame objects.""" palette = shot.get("color_palette", []) if not palette: gen_meta = shot.get("generation_metadata", {}) palette = gen_meta.get("color_palette", []) if not palette: palette = storyboard.get("color_palette", []) if not palette: return "" hex_map = project_config.get("hex_object_map", _DEFAULT_PROJECT_CONFIG["hex_object_map"]) linked = [] for hex_code in palette: obj = hex_map.get(hex_code.upper(), hex_map.get(hex_code, "")) if obj: linked.append(f"{obj} {hex_code}") return ", ".join(linked) def _build_camera_lens(shot: dict, storyboard: dict, project_config: dict) -> str: """Layer 8: Camera body + lens + angle + movement (framing is now Layer 0).""" gen_meta = shot.get("generation_metadata", {}) camera = gen_meta.get("camera", {}) parts = [] camera_body = project_config.get("camera_body", _DEFAULT_PROJECT_CONFIG["camera_body"]) parts.append(camera_body) focal = shot.get("focal_length", "") aperture = shot.get("aperture", "") if focal and aperture: parts.append(f"{focal} {aperture}") elif camera.get("lens"): parts.append(camera["lens"]) dof = camera.get("depth_of_field", "") if dof: parts.append(f"{dof} depth of field") # Camera angle promoted to _build_spatial_directive() (Layer 0) so it's # always present — including ECU shots where this layer gets skipped. # Camera movement (skip "static") movement = shot.get("camera_movement", "static").lower() if movement and movement != "static": # Handle compound movements like "dolly (push in)" base_movement = movement.split("(")[0].strip() move_lang = CAMERA_MOVEMENT_LANGUAGE.get(base_movement, "") if not move_lang and movement != "static": move_lang = f"{movement} camera movement" if move_lang: parts.append(move_lang) return ", ".join(parts) def _build_film_style(shot: dict, storyboard: dict, project_config: dict) -> str: """Layer 9: Film stock + positive quality language.""" gen_meta = shot.get("generation_metadata", {}) film = gen_meta.get("film_stock", "") if not film: lens_pkg = storyboard.get("lens_package", {}) film = lens_pkg.get("film_stock", project_config.get("film_stock", _DEFAULT_PROJECT_CONFIG["film_stock"])) suffix = project_config.get("film_style_suffix", _DEFAULT_PROJECT_CONFIG["film_style_suffix"]) return f"{film}, {suffix}" def _build_quality_guard(project_config: dict) -> str: """Layer 10: Anti-deformity positive embedding (always on).""" return project_config.get("quality_guard", _DEFAULT_PROJECT_CONFIG["quality_guard"]) # ── Transition Language ──────────────────────────────────────────────── def _transition_language(shot: dict, prev: "PreviousShotContext") -> str: """Build transition description from scale/angle change between shots.""" curr_type = shot.get("shot_type", "").upper() prev_type = prev.shot_type.upper() if prev.shot_type else "" curr_order = SCALE_ORDER.get(curr_type, -1) prev_order = SCALE_ORDER.get(prev_type, -1) parts = [] # Scale transition if curr_order >= 0 and prev_order >= 0 and curr_order != prev_order: if curr_order < prev_order: # Tighter (punch in) parts.append(f"Punching in from {prev_type.lower()} shot") else: # Wider (pull out) parts.append(f"Pulling wide from {prev_type.lower()}") # Angle change curr_angle = shot.get("camera_angle", "eye").lower() prev_angle = prev.camera_angle.lower() if prev.camera_angle else "eye" if curr_angle != prev_angle and curr_angle != "eye": parts.append(f"{curr_angle}-angle view") # Screen direction reinforcement curr_screen_dir = shot.get("spatial", {}).get("screen_direction", "") prev_screen_dir = prev.screen_direction if prev.screen_direction else "" if curr_screen_dir and prev_screen_dir and curr_screen_dir == prev_screen_dir: parts.append(f"maintaining {curr_screen_dir} screen direction") return ", ".join(parts) # ── Spatial Directive Builder ────────────────────────────────────────── def _build_spatial_directive(shot: dict, model: str = "", prev_context: "PreviousShotContext" = None) -> str: """Build a transition-aware spatial directive from shot data. Injected at Layer 0 (highest attention weight) to control character facing direction, screen position, and inter-shot transition cues. Combines three signal sources: 1. edge_continuity.spatial_note — editorial transition note (always, not fallback) 2. Transition language — computed from prev_context scale/angle change 3. Blocking directive — from spatial.blocking data Camera angle is promoted here (was in Layer 8 _build_camera_lens, which gets skipped for ECU shots — the tightest, most composition-critical shots). For Z-Image (compile-to-budget): ~8-12 words max. For Flux 2 (full prompt): ~15-20 words. """ spatial = shot.get("spatial") is_z_image = model.startswith("z_image") if model else False # 1. spatial_note — editorial metadata for cinematographer review only. # NOT injected into T2I prompts (models can't interpret inter-shot # transition notes like "Angle shift eye→low"). The transition language # computed in step 2 handles what the model actually needs. spatial_note = "" # 2. Transition language from previous shot context transition = "" if prev_context: transition = _transition_language(shot, prev_context) # 3. Blocking directive from spatial data blocking_directive = "" if spatial: blocking = spatial.get("blocking", {}) screen_dir = spatial.get("screen_direction", "") if blocking: char_names = list(blocking.keys()) if len(char_names) == 1: # Single character char = blocking[char_names[0]] pos = char.get("position", "") facing = char.get("facing", "") # Skip "toward-camera" — that's the default composition and forces # uncinematic direct-to-camera frontal framing. Only inject facing # directives for meaningful directions (profile, away, left, right). if facing == "toward-camera": facing = "" if is_z_image: parts = [] if facing: parts.append(f"Subject faces camera-{facing}") if pos: parts.append(f"positioned {pos}") blocking_directive = ". ".join(parts) if parts else "" else: parts = [] if facing: parts.append(f"Subject faces camera-{facing}") if pos: parts.append(f"positioned {pos} in frame") if screen_dir: parts.append(f"action moves {screen_dir}") blocking_directive = ", ".join(parts) if parts else "" else: # Multi-character descs = [] for name in char_names: char = blocking[name] pos = char.get("position", "") facing = char.get("facing", "") descs.append(f"{pos} facing {facing}" if pos and facing else "") descs = [d for d in descs if d] if is_z_image: blocking_directive = f"Two figures: {', '.join(descs)}" if descs else "" else: blocking_directive = f"Two figures: {', '.join(descs)}" if descs else "" if screen_dir and blocking_directive: blocking_directive += f". Action moves {screen_dir}" elif screen_dir: blocking_directive = f"Action moves {screen_dir}" # 4. Camera angle promotion (moved from _build_camera_lens Layer 8, # which gets skipped for ECU shots) angle = shot.get("camera_angle", "eye").lower() angle_part = "" if angle and angle != "eye": angle_part = f"{angle}-angle" # Combine: spatial_note + transition + blocking + angle all_parts = [] if spatial_note: all_parts.append(spatial_note) if transition: all_parts.append(transition) if blocking_directive: all_parts.append(blocking_directive) if angle_part: all_parts.append(angle_part) return ". ".join(all_parts) + "." if all_parts else "" # ── Compile Function ──────────────────────────────────────────────────── def compile_layers(layers: PromptLayers, shot_type: str = "", model: str = "", spatial_directive: str = "") -> str: """Join layers into single prompt string. Rules: - Shot framing goes FIRST (before LoRA triggers) for composition control - LoRA triggers comma-prefixed after framing - Layers joined with ". " (period-space) - Empty layers skipped - ECU/CU skip irrelevant layers (wardrobe, environment) - Z-Image: compile-to-budget (compact framing, per-layer word caps, skip low-value layers) - Quality guard always last - Warns if > 250 words (other models) """ shot_type_upper = shot_type.upper() if shot_type else "" is_z_image = model.startswith("z_image") if model else False if is_z_image: return _compile_layers_z_image(layers, shot_type_upper, spatial_directive) # ── Standard compilation (Flux 2, WAN 2.2, etc.) ── framing = SHOT_TYPE_FRAMING.get(shot_type_upper, "") skip_layers = _SHOT_TYPE_SKIP_LAYERS.get(shot_type_upper, set()) parts = [] # Spatial directive at Layer 0 (highest attention weight) if spatial_directive: parts.append(spatial_directive.strip().rstrip(".")) # Framing goes FIRST (or after spatial) — before LoRA triggers if framing: parts.append(framing) if layers.lora_triggers and "lora_triggers" not in skip_layers: parts.append(layers.lora_triggers) for layer_name in LAYER_ORDER[1:-1]: # Skip layers that are irrelevant for this shot type if layer_name in skip_layers: continue val = getattr(layers, layer_name, "") if val and val.strip(): parts.append(val.strip().rstrip(".")) if layers.quality_guard and "quality_guard" not in skip_layers: parts.append(layers.quality_guard.strip().rstrip(".")) # Join: framing first, then LoRA triggers comma-separated, then rest period-separated has_triggers = layers.lora_triggers and "lora_triggers" not in skip_layers if framing and has_triggers and len(parts) >= 3: # parts[0] = framing, parts[1] = triggers, parts[2:] = rest compiled = f"{parts[0]}. {parts[1]}, {'. '.join(parts[2:])}" elif framing and len(parts) >= 2: compiled = f"{parts[0]}. {'. '.join(parts[1:])}" elif has_triggers and len(parts) >= 2: triggers = parts[0] rest = ". ".join(parts[1:]) compiled = f"{triggers}, {rest}" else: compiled = ". ".join(parts) word_count = len(compiled.split()) if word_count > 250: print(f" WARNING: Prompt is {word_count} words (budget: 250). " "Later layers may lose attention weight.", file=sys.stderr) return compiled def _compile_layers_z_image(layers: PromptLayers, shot_type_upper: str, spatial_directive: str = "") -> str: """Z-Image compile-to-budget: build within 60 words from the start. Instead of building a 200-word prompt and chopping it, each layer is capped at a word budget and low-value layers are skipped entirely. Compact framing replaces verbose framing strings. Layer priority (all within 60-word budget): spatial → framing → triggers → subject → action → quality → wardrobe → environment Skipped: lighting, color_objects, camera_lens, film_style """ budget = _Z_IMAGE_BUDGET shot_skip = _SHOT_TYPE_SKIP_LAYERS.get(shot_type_upper, set()) # Merge: Z-Image always-skip + shot-type skip skip_layers = _Z_IMAGE_ALWAYS_SKIP | shot_skip parts = [] layer_report = [] words_used = 0 # Spatial directive at Layer 0 (highest attention weight, ~8-12 words) if spatial_directive: spatial_words = len(spatial_directive.split()) # Cap spatial to 12 words for Z-Image budget capped_spatial = _cap_layer_words(spatial_directive.strip().rstrip("."), 12) spatial_words = len(capped_spatial.split()) words_used += spatial_words parts.append(capped_spatial) layer_report.append(f"spatial={spatial_words}w") # Compact framing (8-12 words vs 20-30 in standard) framing = _Z_IMAGE_COMPACT_FRAMING.get(shot_type_upper, "") framing_words = len(framing.split()) if framing else 0 words_used += framing_words if framing: parts.append(framing) layer_report.append(f"framing={framing_words}w") # Triggers (identity lock — must fit) if layers.lora_triggers and "lora_triggers" not in skip_layers: trigger_words = len(layers.lora_triggers.split()) words_used += trigger_words parts.append(layers.lora_triggers) layer_report.append(f"triggers={trigger_words}w") # Remaining layers: add in priority order, each capped at its budget # Only add if budget allows priority_layers = ["subject", "action_pose", "quality_guard", "wardrobe_props", "environment"] for layer_name in priority_layers: if layer_name in skip_layers: continue val = getattr(layers, layer_name, "") if not val or not val.strip(): continue layer_budget = _Z_IMAGE_LAYER_BUDGETS.get(layer_name, 10) remaining = budget - words_used if remaining <= 3: # No room left — stop adding layers layer_report.append(f"[BUDGET FULL at {words_used}w — skipped {layer_name}+]") break # Cap at the smaller of layer budget or remaining budget effective_budget = min(layer_budget, remaining) capped = _cap_layer_words(val.strip().rstrip("."), effective_budget) capped_words = len(capped.split()) words_used += capped_words parts.append(capped) layer_report.append(f"{layer_name}={capped_words}w") # Join: framing first, then triggers comma-separated, then rest period-separated has_triggers = layers.lora_triggers and "lora_triggers" not in skip_layers if framing and has_triggers and len(parts) >= 3: compiled = f"{parts[0]}. {parts[1]}, {'. '.join(parts[2:])}" elif framing and len(parts) >= 2: compiled = f"{parts[0]}. {'. '.join(parts[1:])}" elif has_triggers and len(parts) >= 2: compiled = f"{parts[0]}, {'. '.join(parts[1:])}" else: compiled = ". ".join(parts) print(f" Z-IMAGE COMPILE: {words_used}w/{budget}w budget " f"[{', '.join(layer_report)}]", file=sys.stderr) return compiled return compiled def _prompt_hash(prompt: str, negative_prompt: str) -> str: """SHA-256 first 12 chars of prompt + negative_prompt.""" content = prompt + "||" + negative_prompt return hashlib.sha256(content.encode()).hexdigest()[:12] # ── Main Compile API ──────────────────────────────────────────────────── def compile( shot: dict, breakdown: dict, episode: int, storyboard: dict = None, model: str = "z_image", lora_registry: dict = None, overrides: list = None, notes: list = None, project_config: dict = None, previous_shot_context: "Optional[PreviousShotContext]" = None, frame_type: str = "hero", ) -> dict: """Compile a resolved prompt for a shot. Args: shot: Shot dict from storyboard JSON. breakdown: Full breakdown.json contents. episode: Episode number (for wardrobe/hair phase resolution). storyboard: Full storyboard dict (for character fallback, color palette). model: Target model ("z_image" | "flux2" | "wan_2.2" | "z_image_base"). lora_registry: Flattened LoRA registry {name: {trigger, paths...}}. overrides: Override list (from prompt_overrides.json). notes: Notes list (from prompt_notes.json). project_config: Project visual config (camera, film stock, hex map, etc.). previous_shot_context: Previous shot's context (layers + metadata) for edge continuity. frame_type: "hero", "first", or "last". Returns: { "prompt": str, "negative_prompt": str, "prompt_hash": str, "layers": dict, "warnings": list, "model": str, } """ # If shot has a prompt_override (set via approved PromptRewriteProposal), # return it directly — bypasses all layer assembly. override = shot.get("prompt_override") if override: capabilities = MODEL_CAPABILITIES.get(model, MODEL_CAPABILITIES["z_image"]) if capabilities["negative_prompt"]: negative = (project_config or {}).get( "negative_prompt", _DEFAULT_PROJECT_CONFIG["negative_prompt"] ) else: negative = "" return { "prompt": override, "negative_prompt": negative, "prompt_hash": _prompt_hash(override, negative), "layers": {name: "" for name in LAYER_ORDER}, "warnings": ["prompt_override active — layer assembly bypassed"], "model": model, "override_applied": True, } if storyboard is None: storyboard = {} if lora_registry is None: lora_registry = {} if project_config is None: project_config = {} warnings = [] # Check for legacy mode if _is_legacy_prompt(shot): layers = _build_legacy_layers(shot, frame_type, storyboard, lora_registry, model, project_config) else: layers = PromptLayers( lora_triggers=_build_lora_triggers(shot, storyboard, lora_registry, model), subject=_build_subject(shot, frame_type, episode, storyboard, breakdown), action_pose=_build_action_pose(shot, frame_type), wardrobe_props=_build_wardrobe_props(shot, episode, storyboard, breakdown), environment=_build_environment(shot, breakdown), lighting=_build_lighting(shot, breakdown), color_objects=_build_color_objects(shot, storyboard, project_config), camera_lens=_build_camera_lens(shot, storyboard, project_config), film_style=_build_film_style(shot, storyboard, project_config), quality_guard=_build_quality_guard(project_config), ) # Hero-frame redundancy skip: when hero_frame is rich E-style prose # (150+ words with camera/film info), it already contains wardrobe, # camera body, lens, and film stock. Clear those layers to avoid # duplication that bloats prompts past Flux 2's ~250-word attention limit. if _is_rich_hero_frame(shot, frame_type): layers.wardrobe_props = "" layers.camera_lens = "" layers.film_style = "" # Edge continuity: inherit specified layers from previous shot edge = shot.get("edge_continuity") if edge and previous_shot_context and previous_shot_context.layers: inherit_layers = edge.get("inherit_layers", []) for layer_name in inherit_layers: if hasattr(previous_shot_context.layers, layer_name): prev_val = getattr(previous_shot_context.layers, layer_name) if prev_val: setattr(layers, layer_name, prev_val) # Apply overrides (broadest to narrowest) if overrides: chars = _detect_characters(shot, storyboard) main_char = chars[0] if chars else "" shot_id = shot.get("id", 0) override_store_overrides = overrides for layer_name in LAYER_ORDER: for ov in override_store_overrides: key = ov.get("key", "") scope = ov.get("scope", "") applicable = False if scope == "global" and key == f"global.{layer_name}": applicable = True elif scope == "character" and key == f"char.{main_char}.{layer_name}": applicable = True elif scope == "episode" and key == f"ep.{episode}.{layer_name}": applicable = True elif scope == "shot": shot_key = f"shot.{episode}.{shot_id}.{frame_type}.{layer_name}" shot_key_any = f"shot.{episode}.{shot_id}.*.{layer_name}" if key in (shot_key, shot_key_any): applicable = True if applicable: current = getattr(layers, layer_name, "") op = ov.get("operation", "replace") val = ov.get("value", "") if op == "replace": setattr(layers, layer_name, val) elif op == "prepend": setattr(layers, layer_name, f"{val}, {current}" if current else val) elif op == "append": setattr(layers, layer_name, f"{current}, {val}" if current else val) elif op == "remove_phrase": setattr(layers, layer_name, current.replace(val, "").strip(" ,")) # Build spatial directive from shot's spatial data (Layer 0) # Suppress transition language on scene breaks (new location = fresh composition) prev_ctx = previous_shot_context if not shot.get("scene_break_before") else None spatial_directive = _build_spatial_directive(shot, model=model, prev_context=prev_ctx) if spatial_directive: print(f" SPATIAL: Shot #{shot.get('id', '?')}: {spatial_directive}", file=sys.stderr) # Compile to string (shot_type drives framing position + layer filtering) # Model passed for budget enforcement (Z-Image Turbo: 60 words max) shot_type = shot.get("shot_type", "") prompt = compile_layers(layers, shot_type=shot_type, model=model, spatial_directive=spatial_directive) # Strip VFX language from T2I prompts (causes face artifacts: spirals, noise) # Safe for video prompts (wan_2.2) — VFX language is fine in motion context if model != "wan_2.2": prompt, vfx_stripped = _strip_vfx_language(prompt) if vfx_stripped: warnings.append(f"VFX language stripped (causes T2I artifacts): {', '.join(vfx_stripped)}") print(f" VFX STRIP: Removed {len(vfx_stripped)} phrases: {', '.join(vfx_stripped)}", file=sys.stderr) # Negative prompt (model-aware) capabilities = MODEL_CAPABILITIES.get(model, MODEL_CAPABILITIES["z_image"]) if capabilities["negative_prompt"]: negative = project_config.get("negative_prompt", _DEFAULT_PROJECT_CONFIG["negative_prompt"]) else: negative = "" # Word budget warning word_count = len(prompt.split()) if word_count > 250: warnings.append(f"Prompt is {word_count} words (budget: 250)") # Build layers dict for preview/debug layers_dict = {name: getattr(layers, name, "") for name in LAYER_ORDER} return { "prompt": prompt, "negative_prompt": negative, "prompt_hash": _prompt_hash(prompt, negative), "layers": layers_dict, "warnings": warnings, "model": model, } def _build_legacy_layers(shot: dict, frame_type: str, storyboard: dict, lora_registry: dict, model: str, project_config: dict) -> PromptLayers: """Legacy mode: whole prose goes in subject, guard still appended.""" if frame_type == "first": prose = shot.get("first_frame", "") elif frame_type == "last": prose = shot.get("last_frame", "") else: prose = shot.get("hero_frame") or shot.get("first_frame", "") return PromptLayers( lora_triggers=_build_lora_triggers(shot, storyboard, lora_registry, model), subject=prose, quality_guard=_build_quality_guard(project_config), ) # ── Convenience: Build Layers Without Compiling ───────────────────────── def build_layers( shot: dict, breakdown: dict, episode: int, storyboard: dict = None, model: str = "z_image", lora_registry: dict = None, project_config: dict = None, frame_type: str = "hero", ) -> PromptLayers: """Build layers without compiling — for edge continuity tracking.""" if storyboard is None: storyboard = {} if lora_registry is None: lora_registry = {} if project_config is None: project_config = {} if _is_legacy_prompt(shot): return _build_legacy_layers(shot, frame_type, storyboard, lora_registry, model, project_config) layers = PromptLayers( lora_triggers=_build_lora_triggers(shot, storyboard, lora_registry, model), subject=_build_subject(shot, frame_type, episode, storyboard, breakdown), action_pose=_build_action_pose(shot, frame_type), wardrobe_props=_build_wardrobe_props(shot, episode, storyboard, breakdown), environment=_build_environment(shot, breakdown), lighting=_build_lighting(shot, breakdown), color_objects=_build_color_objects(shot, storyboard, project_config), camera_lens=_build_camera_lens(shot, storyboard, project_config), film_style=_build_film_style(shot, storyboard, project_config), quality_guard=_build_quality_guard(project_config), ) # Hero-frame redundancy skip (same logic as compile()) if _is_rich_hero_frame(shot, frame_type): layers.wardrobe_props = "" layers.camera_lens = "" layers.film_style = "" return layers # ── File Loaders ──────────────────────────────────────────────────────── def _load_storyboard(project_dir: Path, episode: int) -> dict: """Load storyboard JSON for an episode.""" ep_str = f"{episode:03d}" path = project_dir / "storyboards" / f"storyboard_ep_{ep_str}.json" if not path.exists(): print(f"ERROR: Storyboard not found: {path}", file=sys.stderr) sys.exit(1) with open(path) as f: return json.load(f) def _load_breakdown(project_dir: Path) -> dict: """Load breakdown.json for a project.""" path = project_dir / "visual" / "breakdown.json" if not path.exists(): return {} with open(path) as f: return json.load(f) def _load_project_config(project_dir: Path) -> dict: """Load project_config.json, falling back to defaults via config_loader.""" return _config_loader_load(project_dir) def _load_lora_registry(project_dir: Path) -> dict: """Load LoRA registry and flatten to inference config.""" reg_path = project_dir / "visual" / "lora_registry.json" if not reg_path.exists(): return {} with open(reg_path) as f: data = json.load(f) raw = data.get("characters", {}) flat = {} for name, char_data in raw.items(): t2i = char_data.get("t2i", {}) z_image = char_data.get("z_image_t2i", {}) z_image_base = char_data.get("z_image_base_t2i", {}) video = char_data.get("video", {}) inference = char_data.get("inference", {}) flat[name] = { "trigger": char_data.get("trigger"), "t2i_path": t2i.get("path"), "z_image_t2i_path": z_image.get("path"), "z_image_base_t2i_path": z_image_base.get("path"), "video_lora_high": video.get("high_noise_path"), "video_lora_low": video.get("low_noise_path"), "scale_solo": inference.get("scale_solo", 1.0), "scale_dual": inference.get("scale_dual", 0.8), } return flat def _get_shot(storyboard: dict, shot_id: int) -> Optional[dict]: """Get a shot by ID from storyboard.""" for s in storyboard.get("shots", []): if s.get("id") == shot_id: return s return None # ── CLI ────────────────────────────────────────────────────────────────── def cmd_preview(args): """Preview compiled prompts for a shot.""" project_dir = Path(args.project_dir).resolve() storyboard = _load_storyboard(project_dir, args.episode) breakdown = _load_breakdown(project_dir) registry = _load_lora_registry(project_dir) config = _load_project_config(project_dir) override_store = OverrideStore(project_dir) shot = _get_shot(storyboard, args.shot) if not shot: print(f"ERROR: Shot {args.shot} not found", file=sys.stderr) sys.exit(1) # Build previous shot context for transition-aware prompts prev_shot = _get_shot(storyboard, args.shot - 1) if args.shot > 1 else None prev_context = None if prev_shot and not shot.get("scene_break_before"): prev_context = build_prev_context(prev_shot, breakdown, args.episode, storyboard, args.model, registry, config) frame_types = [args.frame] if args.frame else ["hero", "first", "last"] print(f"\n{'=' * 70}") print(f" PROMPT PREVIEW — Episode {args.episode}, Shot {args.shot}") print(f" Model: {args.model} | Shot: {shot.get('name', '')}") print(f"{'=' * 70}") for ft in frame_types: result = compile( shot=shot, breakdown=breakdown, episode=args.episode, storyboard=storyboard, model=args.model, lora_registry=registry, overrides=override_store.list_all(), project_config=config, previous_shot_context=prev_context, frame_type=ft, ) word_count = len(result["prompt"].split()) print(f"\n -- {ft.upper()} ({VERB_CLASSES.get(ft, '?')}) --") print() for layer_name in LAYER_ORDER: val = result["layers"].get(layer_name, "") if val: label = f" {layer_name:20s}" display = val[:100] + "..." if len(val) > 100 else val print(f"{label} | {display}") print() print(f" COMPILED ({word_count} words):") prompt = result["prompt"] for i in range(0, len(prompt), 80): print(f" {prompt[i:i+80]}") if result["negative_prompt"]: print("\n NEGATIVE PROMPT:") neg = result["negative_prompt"] for i in range(0, len(neg), 80): print(f" {neg[i:i+80]}") print(f"\n Hash: {result['prompt_hash']}") print(f" Model: {result['model']}") if result["warnings"]: for w in result["warnings"]: print(f" WARNING: {w}") # Verb strength check (import validator) try: from prompt_validators import validate_verb_strength vw = validate_verb_strength(result, shot, ft) if vw: print(f"\n WEAK VERB: {vw['current'][:80]}") print(f" Details found: {vw['detail_count']}") print(f" {vw['suggestion']}") except ImportError: pass print(f"\n{'=' * 70}") def cmd_validate(args): """Validate verb strength across all shots.""" project_dir = Path(args.project_dir).resolve() storyboard = _load_storyboard(project_dir, args.episode) breakdown = _load_breakdown(project_dir) registry = _load_lora_registry(project_dir) config = _load_project_config(project_dir) shots = storyboard.get("shots", []) print(f"\n{'=' * 70}") print(f" VERB STRENGTH VALIDATION — Episode {args.episode}") print(f" Shots: {len(shots)} | Model: {args.model}") print(f"{'=' * 70}") try: from prompt_validators import validate_verb_strength except ImportError: print(" ERROR: prompt_validators.py not found", file=sys.stderr) sys.exit(1) warnings = [] for shot in shots: gen_approach = shot.get("generation_approach", "standard_flf") if gen_approach in ("held_frame_static", "held_frame_push"): frame_types = ["first"] elif shot.get("hero_action"): frame_types = ["hero", "first", "last"] else: frame_types = ["hero"] for ft in frame_types: result = compile( shot=shot, breakdown=breakdown, episode=args.episode, storyboard=storyboard, model=args.model, lora_registry=registry, project_config=config, frame_type=ft, ) vw = validate_verb_strength(result, shot, ft) if vw: warnings.append(vw) if not warnings: print("\n PASS -- all shots have emotional payload") else: for w in warnings: status = "HALT" if args.strict else "WARNING" print(f"\n {status} [S{w['shot_id']:02d} {w['frame_type']}] " f"Layer 3 lacks emotional payload") print(f" Current: \"{w['current']}\"") print(f" Details found: {w['detail_count']}") print(f" {w['suggestion']}") print(f"\n Result: {len(warnings)} warnings out of {len(shots)} shots") print(f"{'=' * 70}") if args.strict and warnings: sys.exit(1) def cmd_override_add(args): """Add an override.""" project_dir = Path(args.project_dir).resolve() store = OverrideStore(project_dir) entry = OverrideEntry( scope=args.scope, key=args.key, operation=args.operation, value=args.value, reason=args.reason or "", ) store.add(entry) print(f" Override added: {args.key} ({args.operation})") print(f" Stored in: {store.path}") def cmd_override_list(args): """List all overrides.""" project_dir = Path(args.project_dir).resolve() store = OverrideStore(project_dir) overrides = store.list_all() if not overrides: print(" No overrides configured.") return print(f"\n{'=' * 70}") print(f" PROMPT OVERRIDES -- {project_dir.name}") print(f"{'=' * 70}") for i, ov in enumerate(overrides): print(f"\n [{i}] {ov.get('scope', '?')} | {ov.get('key', '?')}") print(f" Op: {ov.get('operation', '?')}") print(f" Value: {ov.get('value', '')[:80]}") if ov.get("reason"): print(f" Reason: {ov['reason']}") print(f"\n{'=' * 70}") def cmd_override_remove(args): """Remove an override by index.""" project_dir = Path(args.project_dir).resolve() store = OverrideStore(project_dir) store.remove(args.index) print(f" Override {args.index} removed.") def cmd_note(args): """Record a production note.""" project_dir = Path(args.project_dir).resolve() store = NoteStore(project_dir) tags = args.tags.split(",") if args.tags else [] entry = NoteEntry( episode=args.episode, shot_id=args.shot, frame_type=args.frame or "hero", issue=args.issue or "", fix=args.fix or "", tags=tags, ) store.add(entry) print(f" Note recorded for ep {args.episode} shot {args.shot}") print(f" Stored in: {store.path}") def main(): parser = argparse.ArgumentParser( description="Stateless prompt compiler for T2I/video keyframes", ) parser.add_argument("project_dir", help="Project directory (e.g., leviathan/)") parser.add_argument("--model", choices=list(MODEL_CAPABILITIES.keys()), default="z_image", help="Target model (affects LoRA selection + negative prompt)") subparsers = parser.add_subparsers(dest="command", required=True) # preview p_preview = subparsers.add_parser("preview", help="Preview compiled prompts") p_preview.add_argument("--episode", "-e", type=int, required=True) p_preview.add_argument("--shot", "-s", type=int, required=True) p_preview.add_argument("--frame", "-f", choices=["hero", "first", "last"]) # validate p_validate = subparsers.add_parser("validate", help="Validate verb strength") p_validate.add_argument("--episode", "-e", type=int, required=True) p_validate.add_argument("--strict", action="store_true", help="Halt on weak verbs (exit code 1)") # override p_override = subparsers.add_parser("override", help="Manage overrides") ov_sub = p_override.add_subparsers(dest="override_cmd", required=True) p_ov_add = ov_sub.add_parser("add", help="Add override") p_ov_add.add_argument("--scope", required=True, choices=["global", "character", "episode", "shot"]) p_ov_add.add_argument("--key", required=True, help="Override key") p_ov_add.add_argument("--operation", required=True, choices=["replace", "prepend", "append", "remove_phrase"]) p_ov_add.add_argument("--value", required=True, help="Override value") p_ov_add.add_argument("--reason", help="Why this override exists") ov_sub.add_parser("list", help="List overrides") p_ov_remove = ov_sub.add_parser("remove", help="Remove override by index") p_ov_remove.add_argument("--index", type=int, required=True) # note p_note = subparsers.add_parser("note", help="Record production note") p_note.add_argument("--episode", "-e", type=int, required=True) p_note.add_argument("--shot", "-s", type=int, required=True, dest="shot") p_note.add_argument("--frame", "-f", choices=["hero", "first", "last"]) p_note.add_argument("--issue", help="What went wrong") p_note.add_argument("--fix", help="How it was fixed") p_note.add_argument("--tags", help="Comma-separated tags (e.g., deformity,skin)") args = parser.parse_args() if args.command == "preview": cmd_preview(args) elif args.command == "validate": cmd_validate(args) elif args.command == "override": if args.override_cmd == "add": cmd_override_add(args) elif args.override_cmd == "list": cmd_override_list(args) elif args.override_cmd == "remove": cmd_override_remove(args) elif args.command == "note": cmd_note(args) if __name__ == "__main__": main() # ══════════════════════════════════════════════════════════════════════ # MIGRATED FROM tools/prompt_engine.py (CP-3 Phase 5, 2026-04-26) # Source: recoil/tools/prompt_engine.py @ pre-prompt-tools-stub tag # Audit: recoil/docs/prompt-engine-audit.md (Phase 4 + Phase 5 logs) # # Phase 4 verification proved that `class PromptEngine` lived ONLY at # recoil/tools/prompt_engine.py:342 — it had no canonical home elsewhere. # Sibling layered classes (PromptLayers, OverrideStore, NoteStore, # OverrideEntry, NoteEntry, compile_layers) ARE already canonical here. # This block migrates the PromptEngine class plus its private helpers # (verb-strength patterns, VerbWarning, _suggest_fix_legacy) verbatim # from the legacy tools file. The CLI (main / cmd_preview / cmd_validate / # cmd_override_*/ cmd_note / argparse setup) is NOT migrated — it's a # tools/-side artifact and Phase 8 deletes it with the stub file. # Reconstitute via `git show pre-prompt-tools-stub:tools/prompt_engine.py` # if a CLI revival is wanted later. # ══════════════════════════════════════════════════════════════════════ # ── Verb Strength Validator (legacy PromptEngine helpers) ────────────── # These patterns mirror the historical inline patterns of the legacy # PromptEngine.validate_verb_strength method. The richer canonical # patterns now live in lib/prompt_validators.py and are loaded from # verb_patterns.json — that newer code is what compile() uses post-hoc. # These are kept verbatim only to preserve PromptEngine's behavior. # Physical micro-detail patterns (body part + state, object + interaction) # Patterns allow up to 5 intervening words between subject and verb # to match prose like "frost on eyelashes shattering" or "fingers trailing through" _PE_GAP = r"(?:\s+\S+){0,5}\s+" # up to 5 words between subject and verb _PE_MICRO_DETAIL_PATTERNS = [ # Body micro-details (subject + gap + verb) rf"jaw{_PE_GAP}(tight|clench|set|grind|lock|press|work)", rf"fingers?{_PE_GAP}(curl|grip|white|clench|spread|dig|find|trace|press|wrap|drum|tap|trail|wedge|wrench|reach|extend|yank)", rf"knuckle[s]?{_PE_GAP}(white|whiten|tight|grip|crack|press|span|encrust)", rf"teeth{_PE_GAP}(bare|grit|clench|show|grind|set)", rf"eyes?{_PE_GAP}(wid|narrow|lock|track|blaz|squint|dart|scan|dilat|snap|shift|reflect|open|focus)", rf"iris(es)?{_PE_GAP}(blaz|glow|dilat|contract|shift|burn|flare)", rf"pupil[s]?{_PE_GAP}(dilat|contract|track|wid)", rf"gaze{_PE_GAP}(lock|shift|cast|fix|track|assess|sweep|narrow)", rf"brow{_PE_GAP}(furrow|raise|knit|crease|heavy)", rf"lips?{_PE_GAP}(press|part|crack|thin|curl|tighten|split)", rf"mouth{_PE_GAP}(open|press|set|work|crack|tighten)", rf"shoulder[s]?{_PE_GAP}(square|brace|hunch|drop|tense|set|pull)", rf"chest{_PE_GAP}(ris|heav|expand|compress|tight)", rf"breath{_PE_GAP}(held|catch|ragg|quick|shallow|visible|fog)", rf"breathing{_PE_GAP}(hard|heavy|ragg|shallow|fast)", rf"muscle[s]?{_PE_GAP}(tens|strain|flex|bulg|taut|lock|fire)", rf"tendon[s]?{_PE_GAP}(strain|vis|pull|tight|taut)", rf"vein[s]?{_PE_GAP}(bulg|vis|pulse|throb|stand)", rf"sweat{_PE_GAP}(bead|drip|glisten|sheen|slick)", rf"fist[s]?{_PE_GAP}(clench|tight|white|ball|curl)", rf"arm[s]?{_PE_GAP}(trem|strain|lock|brace|extend|reach|shake|lunge)", rf"leg[s]?{_PE_GAP}(brace|plant|spread|drive|strain|bent|kick|swing|dangle)", rf"boot[s]?{_PE_GAP}(plant|slam|dig|press|flat|slide|grip|swing|hit|clear|dangle|hover|leave|aim|land)", rf"wrist{_PE_GAP}(glow|amber|compress|strain|turn|twist|press|catch|buckle)", rf"skin{_PE_GAP}(flush|pale|tight|pore|goose|sweat|glow|catch|show|bloodless|marble|tighten)", rf"hair{_PE_GAP}(whip|catch|slick|fall|mat|cling|crop|frame|close)", rf"hand[s]?{_PE_GAP}(trem|shake|grip|clench|reach|press|steady|close|twitch|extend|lift)", rf"palm{_PE_GAP}(press|flat|slick|open|close|slam|push)", rf"chin{_PE_GAP}(tilt|lift|drop|set|jut|lower|raise)", rf"neck{_PE_GAP}(strain|tend|tilt|crane|twist)", rf"body{_PE_GAP}(torque|brace|compress|shift|lean|angle|coil|hang|suspend|lift)", rf"weight{_PE_GAP}(shift|plant|forward|back|lean|drop|throw|commit)", rf"spine{_PE_GAP}(straight|curve|stiffen|arch|rigid)", rf"face{_PE_GAP}(flush|pale|set|shift|fill|catch|lit|illuminat|carv|strain)", rf"expression{_PE_GAP}(shift|set|war|harden|soften|neutral|stun|falter)", rf"posture{_PE_GAP}(emerg|shift|set|square|sag|stiffen|straighten)", rf"grin{_PE_GAP}(split|spread|commit|falter|flash|widen|set)", rf"eyelash(es)?{_PE_GAP}(shatter|frost|crust|catch|cling|frozen|break)", rf"toe[s]?{_PE_GAP}(catch|point|clear|hover|curl|grip|dig)", rf"sole[s]?{_PE_GAP}(separ|hover|clear|worn|flat|press|land)", rf"knee[s]?{_PE_GAP}(compress|bend|buckle|lock|drive|bend|flex)", # Compound adjective forms r"frost[- ]whiten", r"sweat[- ]blacken", r"white[- ]knuckle", r"cryo[- ]pallor", r"scar(red)?\s+(fist|knuckle|skin|hand)", # Object interactions (subject + gap + verb) rf"rust{_PE_GAP}(cascad|flak|crumbl|dust|peel|scat|fall|drift|curl|land|streak|catch|melt)", rf"spark[s]?{_PE_GAP}(erupt|fly|shower|burst|scatter|arc|die|catch|ignit|trail)", rf"frost{_PE_GAP}(crack|shatter|melt|cryst|form|cling|break|radi|residu|encrust|catch|refract|fragment)", rf"metal{_PE_GAP}(groan|scream|creak|screech|bend|buckle|yield|gleam|catch)", rf"grating{_PE_GAP}(vibrat|rattle|clang|ring|shake|corrode)", rf"counter{_PE_GAP}(tick|glow|pulse|climb|flash|blaze|cast|bleed|radiat)", rf"gas{_PE_GAP}(erupt|vent|billow|blast|hiss|cloud|rush|clear|dissipat|burn)", rf"condensation{_PE_GAP}(bead|drip|form|fog|cloud|mist|weep)", rf"light{_PE_GAP}(catch|refract|scatter|pool|wash|cast|die|flick|carv|illuminat|plung|punch|intensif)", rf"shadow[s]?{_PE_GAP}(carv|pool|deep|shift|fall|stretch)", rf"debris{_PE_GAP}(suspend|float|scatter|fall|drift|hang|tumbl)", rf"particl[e]?{_PE_GAP}(hang|drift|float|suspend|swirl|ignit|glow|catch)", rf"vapor{_PE_GAP}(explod|billow|dissipat|clear|hang|drift|burn)", rf"digit[s]?{_PE_GAP}(tick|cast|climb|glow|pulse|flash)", rf"beam{_PE_GAP}(plung|punch|whip|carv|cut|catch|sweep|angle|point|die|swing)", rf"glow{_PE_GAP}(cast|wash|climb|pulse|radiat|bleed|intensif|warm|amber|steady)", # Environmental micro-details (no body part needed — for establishing shots) r"flakes?\s+drift", r"drip(ping|s)?\s+from", r"hum(ming|s)?\s+", r"vibrat(ing|e|es)\s+through", r"catching\s+(amber|light|glow|fire)", r"hanging\s+motionless", r"pulsing\s+(steady|with|blue|amber|warm|faint)", r"refracting\s+", r"gleam(ing)?\s+impossible", r"weeping\s+condensation", r"corroded\s+(wall|ladder|grating|metal|surface|catwalk|steel)", r"amber\s+(emergency|glow|light|LED|counter|readout)", r"blue-white\s+(glow|light|pulse|machinery|systems|pod|internal)", r"dissipating\s+(cryogenic|gas|vapor|mist|fog|smoke)", r"outline\s+of\s+something\s+(massive|enormous|huge|vast)", r"shrouded\s+in\s+(dissipating|cryogenic|gas|vapor|mist|fog)", ] # Bare verb + noun patterns (weak — no micro-detail) _PE_BARE_PATTERNS = [ r"^(standing|sitting|looking|holding|walking|waiting|watching|staring|leaning) (in|at|on|by|near|toward|against|into) ", r"^(a|the) (woman|man|figure|person|character|salvager) (stands|sits|looks|holds|walks|waits|watches|stares|leans) ", ] @dataclass class VerbWarning: """Warning from verb strength validator (legacy PromptEngine return type).""" shot_id: int frame_type: str current: str detail_count: int suggestion: str def _suggest_fix_legacy(action: str, frame_type: str) -> str: """Generate a suggestion for strengthening weak verb prompts. Renamed from `_suggest_fix` during CP-3 Phase 5 migration to avoid name collision with the canonical `_suggest_fix` in lib/prompt_validators.py (different return formatting). Behavior here is preserved verbatim from the legacy tools/prompt_engine.py. """ verb_class = VERB_CLASSES.get(frame_type, "ACTIVE") if verb_class == "ACTIVE": return ("Add 2+ physical micro-details to hero action. " "E.g.: body language (jaw tightening, knuckles whitening), " "object interaction (rust cascading, sparks erupting), " "or environmental response (grating vibrating, condensation dripping)") elif verb_class == "PREPARATION": return ("Add anticipation details: fingers finding grip, " "weight shifting, eyes tracking, breath held") else: return ("Add aftermath details: breathing hard, " "body settling, debris still falling, grip loosening") # ── Prompt Engine ──────────────────────────────────────────────────────── class PromptEngine: """Builds layered prompts from storyboard data + overrides. Migrated verbatim from tools/prompt_engine.py during CP-3 Phase 5. Sibling layered classes (PromptLayers, OverrideStore) and helpers (LAYER_ORDER, compile_layers, DEFAULT_QUALITY_GUARD) live in this same file already, so the migration is a straight class-body copy with its private regex/data helpers (_PE_GAP, _PE_MICRO_DETAIL_PATTERNS, _PE_BARE_PATTERNS, VerbWarning, _suggest_fix_legacy). Usage: engine = PromptEngine(project_dir, storyboard, lora_registry) prompt = engine.compile(shot, frame_type="hero") layers = engine.build_layers(shot, frame_type="hero") """ def __init__(self, project_dir: Path, storyboard: dict, lora_registry: dict, model_key: str = "z_image", previous_shot_layers: Optional[PromptLayers] = None): self.project_dir = project_dir self.storyboard = storyboard self.lora_registry = lora_registry self.model_key = model_key self.overrides = OverrideStore(project_dir) self.previous_shot_layers = previous_shot_layers # Model-specific LoRA key self._lora_key_map = { "flux2": "t2i_path", "z_image": "z_image_t2i_path", } def _get_lora_path(self, char_name: str) -> Optional[str]: """Get LoRA path for active model.""" reg = self.lora_registry.get(char_name.lower(), {}) lora_key = self._lora_key_map.get(self.model_key, "z_image_t2i_path") return reg.get(lora_key) def _detect_characters(self, shot: dict) -> List[str]: """Detect characters in shot.""" chars = shot.get("characters_in_shot", []) if chars: return [c.lower() for c in chars] search = " ".join([ shot.get("first_frame", ""), shot.get("last_frame", ""), shot.get("subject", ""), shot.get("action", ""), ]).lower() storyboard_chars = self.storyboard.get("characters", {}) return [c.lower() for c in storyboard_chars if c.lower() in search] def _is_legacy_prompt(self, shot: dict) -> bool: """Detect pre-baked prose prompts (legacy mode).""" if shot.get("prompt_mode") == "structured": return False # Check if hero_action exists — structured mode if shot.get("hero_action"): return False # Legacy: long prose in first_frame with camera/film baked in first = shot.get("first_frame", "") word_count = len(first.split()) has_camera = "Shot on" in first or "Kodak" in first or "Arri" in first return word_count > 80 and has_camera # ── Layer Builders ───────────────────────────────────────────────── def _build_lora_triggers(self, shot: dict) -> str: """Layer 1: LoRA trigger words.""" chars = self._detect_characters(shot) triggers = [] for c in chars: reg = self.lora_registry.get(c, {}) if reg.get("trigger") and self._get_lora_path(c): triggers.append(reg["trigger"]) return ", ".join(triggers) def _build_subject(self, shot: dict, frame_type: str) -> str: """Layer 2: Character visual DNA + emotion.""" chars = self._detect_characters(shot) emotion = shot.get("emotion", "") parts = [] for c in chars: char_data = self.storyboard.get("characters", {}).get(c, {}) visual = char_data.get("visual", "") if visual: parts.append(visual) if emotion: parts.append(f"expression: {emotion.lower()}") return ", ".join(parts) def _build_action_pose(self, shot: dict, frame_type: str) -> str: """Layer 3: Frame-specific verb state. Uses hero_action/anticipation_action/aftermath_action when available. Falls back to shot action/subject for legacy. """ if frame_type == "hero" and shot.get("hero_action"): return shot["hero_action"] elif frame_type == "first" and shot.get("anticipation_action"): return shot["anticipation_action"] elif frame_type == "last" and shot.get("aftermath_action"): return shot["aftermath_action"] # Fallback: use action field action = shot.get("action", "") if action: return action return shot.get("subject", "") def _build_wardrobe_props(self, shot: dict) -> str: """Layer 4: Wardrobe + props with HEX linked to objects.""" chars = self._detect_characters(shot) for c in chars: char_data = self.storyboard.get("characters", {}).get(c, {}) # Pull wardrobe from character data in storyboard wardrobe = char_data.get("wardrobe", "") if wardrobe and wardrobe != char_data.get("visual", ""): # Don't duplicate if wardrobe is same as visual pass # wardrobe phase name, not description # Props from subject/action # This is intentionally light — wardrobe is already in the subject layer # via character visual DNA. This layer is for EXTRA props and HEX-linked items. return "" def _build_environment(self, shot: dict) -> str: """Layer 5: Habitat zone materials + shot atmosphere.""" # Shot-specific atmosphere atmosphere = shot.get("atmosphere", "") # Episode location as fallback if not atmosphere: atmosphere = self.storyboard.get("atmosphere", "") # Trim to essentials — keep material textures and sensory details return atmosphere def _build_lighting(self, shot: dict) -> str: """Layer 6: Shot lighting + habitat zone lighting.""" lighting = shot.get("lighting", "") if not lighting: gen_meta = shot.get("generation_metadata", {}) light_data = gen_meta.get("lighting", {}) if light_data: parts = [] if light_data.get("type"): parts.append(light_data["type"]) if light_data.get("source"): parts.append(light_data["source"]) if light_data.get("color_temp"): parts.append(light_data["color_temp"]) lighting = ", ".join(parts) return lighting def _build_color_objects(self, shot: dict) -> str: """Layer 7: HEX codes linked to specific in-frame objects. BAD: "Palette: #B54A1A, #2C2C2C" GOOD: "Rust streaks #B54A1A on corridor walls, steel panels #2C2C2C" """ palette = shot.get("color_palette", []) if not palette: gen_meta = shot.get("generation_metadata", {}) palette = gen_meta.get("color_palette", []) if not palette: palette = self.storyboard.get("color_palette", []) if not palette: return "" # Map known HEX codes to objects for Leviathan hex_object_map = { "#B54A1A": "rust streaks on corridor walls", "#D4380D": "emergency warning light strips", "#1A3A5C": "cryo-pod glow and deep shadows", "#E8960C": "amber debt counter readout", "#2C2C2C": "gunmetal steel panels and grating", "#8B4513": "oxidized copper pipe bundles", "#C0C0C0": "chrome cryo-pod housing", } linked = [] for hex_code in palette: obj = hex_object_map.get(hex_code.upper(), hex_object_map.get(hex_code, "")) if obj: linked.append(f"{obj} {hex_code}") # Skip bare HEX codes without object links return ", ".join(linked) def _build_camera_lens(self, shot: dict) -> str: """Layer 8: Camera + lens from generation_metadata.""" gen_meta = shot.get("generation_metadata", {}) camera = gen_meta.get("camera", {}) parts = ["Arri Alexa Mini LF"] # Focal length + aperture from shot focal = shot.get("focal_length", "") aperture = shot.get("aperture", "") if focal and aperture: parts.append(f"{focal} {aperture}") elif camera.get("lens"): parts.append(camera["lens"]) # Depth of field dof = camera.get("depth_of_field", "") if dof: parts.append(f"{dof} depth of field") # Camera angle angle = shot.get("camera_angle", "eye") if angle and angle != "eye": parts.append(f"{angle}-angle") return ", ".join(parts) def _build_film_style(self, shot: dict) -> str: """Layer 9: Film stock + positive quality language.""" gen_meta = shot.get("generation_metadata", {}) film = gen_meta.get("film_stock", "") if not film: lens_pkg = self.storyboard.get("lens_package", {}) film = lens_pkg.get("film_stock", "Kodak Vision3 500T") return f"{film}, visible grain, photorealistic, ultra-detailed" def _build_quality_guard(self, shot: dict) -> str: """Layer 10: Anti-deformity positive embedding (always on).""" return DEFAULT_QUALITY_GUARD # ── Main Build / Compile ───────────────────────────────────────────── def build_layers(self, shot: dict, frame_type: str = "hero", inherit_from: Optional[PromptLayers] = None) -> PromptLayers: """Build all 10 layers for a shot + frame type. Args: shot: Shot data from storyboard. frame_type: "hero", "first", or "last". inherit_from: Previous shot's layers for edge continuity. Returns: PromptLayers with all 10 layers populated. """ # Check for legacy mode if self._is_legacy_prompt(shot): return self._build_legacy_layers(shot, frame_type) layers = PromptLayers( lora_triggers=self._build_lora_triggers(shot), subject=self._build_subject(shot, frame_type), action_pose=self._build_action_pose(shot, frame_type), wardrobe_props=self._build_wardrobe_props(shot), environment=self._build_environment(shot), lighting=self._build_lighting(shot), color_objects=self._build_color_objects(shot), camera_lens=self._build_camera_lens(shot), film_style=self._build_film_style(shot), quality_guard=self._build_quality_guard(shot), ) # Edge continuity: inherit specified layers from previous shot edge = shot.get("edge_continuity") if edge and inherit_from: inherit_layers = edge.get("inherit_layers", []) for layer_name in inherit_layers: if hasattr(inherit_from, layer_name): prev_val = getattr(inherit_from, layer_name) if prev_val: setattr(layers, layer_name, prev_val) # Apply overrides (broadest to narrowest) chars = self._detect_characters(shot) main_char = chars[0] if chars else "" ep = self.storyboard.get("episode", 0) shot_id = shot.get("id", 0) for layer_name in LAYER_ORDER: applicable = self.overrides.get_applicable( layer_name, main_char, ep, shot_id, frame_type ) for ov in applicable: current = getattr(layers, layer_name, "") op = ov.get("operation", "replace") val = ov.get("value", "") if op == "replace": setattr(layers, layer_name, val) elif op == "prepend": setattr(layers, layer_name, f"{val}, {current}" if current else val) elif op == "append": setattr(layers, layer_name, f"{current}, {val}" if current else val) elif op == "remove_phrase": setattr(layers, layer_name, current.replace(val, "").strip(" ,")) return layers def _build_legacy_layers(self, shot: dict, frame_type: str) -> PromptLayers: """Legacy mode: whole prose goes in subject, guard still appended.""" if frame_type == "first": prose = shot.get("first_frame", "") elif frame_type == "last": prose = shot.get("last_frame", "") else: # hero: use hero_frame if available, else first_frame prose = shot.get("hero_frame") or shot.get("first_frame", "") return PromptLayers( lora_triggers=self._build_lora_triggers(shot), subject=prose, quality_guard=self._build_quality_guard(shot), ) def compile(self, shot: dict, frame_type: str = "hero", inherit_from: Optional[PromptLayers] = None) -> str: """Build layers and compile to single prompt string. Returns: Single string for fal.ai `prompt` field. """ layers = self.build_layers(shot, frame_type, inherit_from) return compile_layers(layers) def compile_from_layers(self, layers: PromptLayers) -> str: """Compile pre-built layers to single prompt string.""" return compile_layers(layers) # ── Verb Strength Validator ────────────────────────────────────────── def validate_verb_strength(self, shot: dict, frame_type: str = "hero") -> Optional[VerbWarning]: """Check if Layer 3 (action_pose) has emotional payload. Returns: VerbWarning if weak, None if strong. """ action = self._build_action_pose(shot, frame_type) if not action: return VerbWarning( shot_id=shot.get("id", 0), frame_type=frame_type, current="(empty)", detail_count=0, suggestion="Add action description with 2+ physical micro-details", ) action_lower = action.lower() # Count physical micro-details detail_count = 0 for pattern in _PE_MICRO_DETAIL_PATTERNS: if re.search(pattern, action_lower): detail_count += 1 # Check for bare verb + noun (weak) is_bare = False for pattern in _PE_BARE_PATTERNS: if re.match(pattern, action_lower): is_bare = True break # Need at least 2 micro-details for hero, 1 for first/last min_details = 2 if frame_type == "hero" else 1 if detail_count < min_details or (is_bare and detail_count < 2): return VerbWarning( shot_id=shot.get("id", 0), frame_type=frame_type, current=action[:120], detail_count=detail_count, suggestion=_suggest_fix_legacy(action, frame_type), ) return None def validate_episode(self, shots: List[dict], strict: bool = False) -> List[VerbWarning]: """Validate all shots in an episode. Returns list of warnings.""" warnings = [] for shot in shots: # Determine which frame types to check based on generation approach gen_approach = shot.get("generation_approach", "standard_flf") if gen_approach in ("held_frame_static", "held_frame_push"): frame_types = ["first"] elif shot.get("hero_action"): frame_types = ["hero", "first", "last"] else: frame_types = ["hero"] for ft in frame_types: w = self.validate_verb_strength(shot, ft) if w: warnings.append(w) return warnings