""" prompt_engine.py — Cinematic prompt builder for frontier model generation. Plan-first: when a shot has structured plan data (prompt_skeleton, lighting sources, kinetic_action), uses those directly. Falls back to regex parsing for legacy Recoil storyboard shots. Implements Gemini-validated prompt architecture: - Kinetic descriptors over semantic emphasis - Positive constraints over negative language - Wide-shot prompt branching (strip facial demands for WIDE/LS) - Lighting vector locking (explicit directional coords) - ENV sanitization (strip human-presence language) - Visual Anchors block (constant traits for grid consistency) - Structured grid panel assignments """ import hashlib as _hashlib import logging import pathlib as _pathlib import re from dataclasses import dataclass, field from enum import Enum from typing import Any from typing import Optional logger = logging.getLogger(__name__) @dataclass class BoundPrompt: """Deterministically bound prompt text plus payload reference metadata.""" text: str payload_refs: dict[str, Any] = field(default_factory=dict) manifest: dict[str, Any] = field(default_factory=dict) class BindAssertionError(AssertionError): """Raised when deterministic prose binding violates a post-bind invariant.""" # Phase 7 (Bug O): schema-version hash of this module's own bytes. Computed # once at import time (NOT per sidecar write). Stamped into every video # sidecar via StepRunner._write_sidecar so we can detect prompt-engine # regressions across runs by diff'ing sidecar provenance.prompt_engine_version. PROMPT_ENGINE_SCHEMA_VERSION = _hashlib.sha256( _pathlib.Path(__file__).read_bytes() ).hexdigest()[:12] # Pre-compiled regex for non-human character detection (android, chassis, etc.) _NON_HUMAN_RE = re.compile( r"\b(?:android|chassis|combat\s+chassis|synthetic|robotic|mechanical|" r"alloy|armored\s+shoulders|metallic\s+bone)\b", re.I, ) def _maybe_hydrate( skeleton: dict, bible: dict, episode: int = 1, asset_data: dict | None = None ) -> dict: """JIT hydrate a prompt skeleton if jit_prompt is available. Resolves {char_*} and {loc_*} tokens from live bible data. Returns the skeleton unchanged if jit_prompt is not installed. """ if not skeleton or not bible: return skeleton try: from recoil.pipeline._lib.jit_prompt import hydrate_skeleton return hydrate_skeleton(skeleton, bible, episode=episode, asset_data=asset_data) except ImportError: return skeleton from recoil.core.prompt_config import ( # noqa: E402 get_all_kinetic_descriptors, get_constant, get_lighting_direction, get_lighting_quality, load_prompt_file, ) from recoil.pipeline._lib.bible_loader import ( # noqa: E402 get_prompt_rules, get_optimal_word_range, get_i2v_ar_behavior, get_prompt_rule_with_global_default, load_bible, ) from recoil.pipeline._lib.cinema_loader import ( # noqa: E402 load_cinema_modes, render_camera_line, render_cinema_tokens, render_constraint_block, ) from pathlib import Path # noqa: E402 from typing import Callable # noqa: E402 _ERA_LOOK_CONSTRAINTS = frozenset({ "no_modern_color_grade", "no_digital_sharpness", "no_smooth_video_look", }) STORYBOARD_STYLE_LOCK = ( "rough working storyboard pencils — loose gestural linework, quick simple forms, " "minimal flat shading, lots of white paper showing, the economy of a board artist " "drawing fast. NOT a detailed illustration, NOT rendered, no crosshatching density, " "no photorealism. Motion arrows where movement matters." ) STORYBOARD_DIRECTOR_PREAMBLE = ( "You are the director. When a panel specifies a shot size, draw it at exactly " "that size; otherwise compose the strongest framing for the beat. Respect the " "30-degree rule, use reverse shots where dialogue alternates, keep consistent " "screen direction." ) STORYBOARD_FINISH_STYLE_LOCK = ( "PHOTOREALISTIC cinematic film frames — moody practical lighting, filmic color, " "production-plate quality. No sketch lines, no illustration. Convert the approved " "pencil staging into grounded live-action production stills with physical sets, " "practical props, realistic faces, natural lensing, and coherent continuity. " "No speech balloons or text inside panels." ) STORYBOARD_FINISH_COMPOSITION_REF = ( "The FIRST attached image is the COMPOSITION REFERENCE: a pencil storyboard of " "these exact N panels. Match its panel-by-panel framing, staging, and camera " "angles EXACTLY — change only the rendering style to photorealistic." ) STORYBOARD_FINISH_NO_TEXT_FOOTER = ( "Do not write text on the image. No subtitles. No captions. No panel numbers " "beyond small corner indices." ) def render_prop_invariant(prop_id: str, carrier: str, description: str) -> str: desc = str(description or "").strip() parenthetical = f" ({desc})" if desc else "" return ( f"PROP INVARIANT: the {prop_id}{parenthetical} is a PERMANENT attachment to " f"{carrier} -- it must always be shown physically attached to {carrier} and " "must NEVER appear free-floating, mounted, or as a fixture." ) def _storyboard_beat_lines( segments: list[dict], spatial_shots: list[dict] | None = None, bible: dict | None = None, sublocation_locked: bool = False, ) -> list[str]: """Render beat lines shared by pencil and finish storyboard prompts. REC-180: when ``spatial_shots`` (per-panel shot dicts carrying top-level spatial_data/asset_data/prompt_data, panel-aligned with ``segments``) and ``bible`` are supplied, each panel gets the explicit [SCREEN-LEFT/RIGHT] / OTS staging block the keyframe path already builds — so the board actually carries the derived screen-direction instead of leaving placement to the model's default. Omitting them preserves the prior (uninstructed) output. """ beat_lines: list[str] = [] for i, segment in enumerate(segments or [], start=1): lines = [f"{i}."] setting = str(segment.get("setting") or "").strip() if setting and not sublocation_locked: lines.append(f"Setting: {setting}") intent = str(segment.get("intent") or "").strip() if intent: lines.append(intent) if spatial_shots and i - 1 < len(spatial_shots): _ss = spatial_shots[i - 1] or {} _st = (_ss.get("prompt_data") or {}).get("shot_type") or _ss.get("shot_type") if _st: _visual = _resolve_shot_type(_st) _name = _SHOT_TYPE_NAMES.get(_visual, _visual) _art = "an" if _name[:1].lower() in "aeiou" else "a" lines.append(f"Shot size: {_name} — frame this panel as {_art} {_name.lower()}.") if spatial_shots and bible and i - 1 < len(spatial_shots): staging = build_spatial_continuity_block(spatial_shots[i - 1], bible) if staging: lines.append(staging) beat_lines.append("\n".join(lines)) return beat_lines def _ref_span_start(span: Any) -> int: if isinstance(span, (list, tuple)) and span: return int(span[0]) if isinstance(span, dict): return int(span.get("start") or span.get("index") or span.get("image") or 0) if isinstance(span, int): return span return 0 def _format_ref_span(span: Any) -> str: if isinstance(span, (list, tuple)) and len(span) >= 2: start, end = int(span[0]), int(span[1]) return f"@Image{start}" if start == end else f"@Image{start}-@Image{end}" if isinstance(span, dict): if "start" in span and "end" in span: start, end = int(span["start"]), int(span["end"]) return f"@Image{start}" if start == end else f"@Image{start}-@Image{end}" if "index" in span: return f"@Image{int(span['index'])}" if "image" in span: return f"@Image{int(span['image'])}" if isinstance(span, int): return f"@Image{span}" return "Attached image" def _indexed_ref_entries(ref_layout: dict, key: str) -> list[dict[str, Any]]: entries = (ref_layout or {}).get(key) or [] if isinstance(entries, dict): normalized = [] for slug, value in entries.items(): if isinstance(value, dict): normalized.append({"slug": slug, **value}) else: normalized.append({"slug": slug, "index": value}) entries = normalized return sorted( [entry for entry in entries if isinstance(entry, dict)], key=lambda entry: int(entry.get("index") or entry.get("image") or 999), ) def _storyboard_finish_ref_lines( char_descs: dict[str, str], ref_layout: dict, sublocation_locked: bool, ) -> list[str]: ref_lines = [STORYBOARD_FINISH_COMPOSITION_REF] identity_refs = (ref_layout or {}).get("identity_refs") or {} for char_id, span in sorted(identity_refs.items(), key=lambda item: _ref_span_start(item[1])): desc = char_descs.get(char_id) or "visual identity reference" ref_lines.append( f"{_format_ref_span(span)} are identity references for {char_id} — {desc}" ) sublocation_entries = _indexed_ref_entries(ref_layout or {}, "sublocation_refs") for entry in sublocation_entries: slug = entry.get("slug") or "the tagged sublocation" ref_lines.append( f"{_format_ref_span(entry)} is the {slug} sublocation geography reference. " "Keep its architecture, layout, scale, and practical lighting recognizable." ) if not sublocation_entries and (ref_layout or {}).get("sublocation_ref"): scope = "EVERY panel happens here." if sublocation_locked else "Use it for the tagged panels." ref_lines.append( "Attached sublocation image is the establishing geography reference. " f"{scope} Keep this exact geography identical and recognizable." ) worn_prop_slugs: set[str] = set() for _slugs in ((ref_layout or {}).get("worn_props") or {}).values(): worn_prop_slugs.update(_slugs or []) for entry in _indexed_ref_entries(ref_layout or {}, "prop_refs"): slug = entry.get("slug") or "tagged prop" if slug in worn_prop_slugs: # REC-247: worn permanent prop — match its appearance but keep it WORN on # the body (the pencil composition ref already places it; this preserves # the look without re-introducing a free-floating object). ref_lines.append( f"{_format_ref_span(entry)} is the {slug} reference — match its exact " "appearance (form, materials, scale, and any readout/markings). It is " "WORN on the character and must stay attached to their body, never a " "separate free-floating object." ) else: ref_lines.append( f"{_format_ref_span(entry)} is the {slug} prop identity reference. " "Preserve its form, materials, and scale." ) return ref_lines def build_storyboard_strip_prompt( segments: list[dict], slots: int, char_descs: dict[str, str], ref_layout: dict, sublocation_locked: bool, *, grid: tuple[int, int] | None = None, scene_context: str | None = None, fix_notes: list[str] | None = None, spatial_shots: list[dict] | None = None, bible: dict | None = None, carrier_facts: list[dict] | None = None, ) -> str: """Build the registered gpt-image-2 storyboard prompt. ``grid`` is (cols, rows); None falls back to a 1xN horizontal strip (legacy callers/tests). ``scene_context`` is a read-only script span for causality — never drawn as panels. """ n = len(segments or []) if grid: cols, rows = grid header = ( f"Create a storyboard as ONE single image: a grid of {cols} columns x " f"{rows} rows, panels numbered 1-{slots} reading left to right, top to " "bottom. Each panel is a 9:16 vertical film frame." ) else: header = ( f"Create a storyboard as ONE single horizontal image: {slots} panels in a " f"single row, left to right, panels numbered 1-{slots}. Each panel is a " "9:16 vertical film frame." ) if slots > n: blank_count = slots - n suffix = "cell" if blank_count == 1 else "cells" header += ( f" Draw only the first {n} panel(s); leave the trailing {blank_count} " f"{suffix} completely blank — white paper." ) ref_lines: list[str] = [] identity_refs = (ref_layout or {}).get("identity_refs") or {} for char_id, span in identity_refs.items(): start, end = span desc = char_descs.get(char_id) or "visual identity reference" ref_lines.append( f"Attached images {start}-{end} are identity references (front, profile) " f"for {char_id} — {desc}" ) sublocation_entries = _indexed_ref_entries(ref_layout or {}, "sublocation_refs") prop_entries = _indexed_ref_entries(ref_layout or {}, "prop_refs") if (ref_layout or {}).get("sublocation_ref"): if sublocation_locked: ref_lines.append( "Attached sublocation image is the establishing geography reference. " "EVERY panel happens here. Keep this exact geography identical and " "recognizable in every panel. Never relocate them." ) else: panels = (ref_layout or {}).get("sublocation_panels") or [] panel_text = ", ".join(str(p) for p in panels) if panels else "the tagged panels" ref_lines.append( "Attached sublocation image is the establishing geography reference. " f"For panel(s) {panel_text}, keep this exact geography identical and " "recognizable. Never relocate them." ) elif sublocation_entries: for entry in sublocation_entries: panels = entry.get("panels") or [] panel_text = ", ".join(str(p) for p in panels) if panels else "the tagged panels" slug = entry.get("slug") or "tagged sublocation" ref_lines.append( f"{_format_ref_span(entry)} is the {slug} sublocation geography reference. " f"For panel(s) {panel_text}, keep this exact geography identical and " "recognizable. Never relocate them." ) worn_prop_slugs: set[str] = set() for _slugs in ((ref_layout or {}).get("worn_props") or {}).values(): worn_prop_slugs.update(_slugs or []) for entry in prop_entries: slug = entry.get("slug") or "tagged prop" if slug in worn_prop_slugs: # REC-247: a worn permanent-attachment prop's ref is included so the model # can MATCH its exact appearance — but it must be drawn worn on the body, # never as a separate free-floating object (the REC-219 protection). ref_lines.append( f"{_format_ref_span(entry)} is the {slug} reference — match its exact " "appearance (shape, materials, and any readout/markings). It is WORN on " "the character it belongs to and must ALWAYS be drawn attached to their " "body in that exact spot, never as a separate free-floating object." ) else: ref_lines.append(f"{_format_ref_span(entry)} is the {slug} prop identity.") beat_lines = _storyboard_beat_lines( segments or [], spatial_shots=spatial_shots, bible=bible, sublocation_locked=sublocation_locked, ) carrier_block = ( "\n\nPROP INVARIANTS (binding):\n" + "\n".join( render_prop_invariant(f["prop_id"], f["carrier"], f["description"]) for f in carrier_facts ) if carrier_facts else "" ) blocks = [ header, f"STYLE:\n{STORYBOARD_STYLE_LOCK}", "REFERENCE MAPPING:\n" + ("\n".join(ref_lines) if ref_lines else "No attached references."), ] if scene_context: blocks.append( "SCENE CONTEXT (read-only — for causality, blocking, and continuity; " "do NOT draw panels for this text, only for the numbered beats):\n" + scene_context ) blocks.append( ( "AUTHORING:\n" f"{STORYBOARD_DIRECTOR_PREAMBLE}\n" f"Draw EXACTLY {n} panels, one per numbered beat below, in order. " "No invented panels, no filler.\n\n" "STORY BEATS:\n" + "\n\n".join(beat_lines) + carrier_block ) ) rendered_fix_notes = [ str(note).strip() for note in (fix_notes or []) if str(note).strip() ] if rendered_fix_notes: blocks.append( "FIX NOTES (from the previous attempt's review - obey these)\n" + "\n".join(rendered_fix_notes) ) return "\n\n".join(blocks) def build_gpt_image_2_storyboard_finish_prompt( segments: list[dict], slots: int, char_descs: dict[str, str], ref_layout: dict, sublocation_locked: bool, *, grid: tuple[int, int] | None = None, scene_context: str | None = None, carrier_facts: list[dict] | None = None, ) -> str: """Build the photoreal finish prompt for an approved pencil storyboard.""" n = len(segments or []) if grid: cols, rows = grid else: cols, rows = 1, slots if slots > n: panel_clause = f"panels numbered 1-{slots} (draw panels 1-{n}; trailing cells left blank)" else: panel_clause = f"panels numbered 1-{n}" header = ( f"Create a {n}-panel storyboard as ONE single vertical image: a grid of " f"{cols} columns x {rows} rows, {panel_clause}, each panel a 9:16 " "vertical film frame." ) blocks = [ header, f"STYLE:\n{STORYBOARD_FINISH_STYLE_LOCK}", "REFERENCE MAPPING:\n" + "\n".join( _storyboard_finish_ref_lines( char_descs, ref_layout or {}, sublocation_locked, ) ), ] if scene_context: blocks.append( "SCENE CONTEXT (read-only — for causality, blocking, and continuity; " "do NOT draw panels for this text, only for the numbered beats):\n" + scene_context ) beat_lines = _storyboard_beat_lines( segments or [], sublocation_locked=sublocation_locked, ) carrier_block = ( "\n\nPROP INVARIANTS (binding):\n" + "\n".join( render_prop_invariant(f["prop_id"], f["carrier"], f["description"]) for f in carrier_facts ) if carrier_facts else "" ) blocks.append( ( "AUTHORING:\n" f"{STORYBOARD_DIRECTOR_PREAMBLE}\n" f"Render EXACTLY {n} photoreal panels, one per numbered beat below, in order. " "No invented panels, no filler.\n\n" "STORY BEATS:\n" + "\n\n".join(beat_lines) + carrier_block ) ) blocks.append(STORYBOARD_FINISH_NO_TEXT_FOOTER) return "\n\n".join(blocks) # ══════════════════════════════════════════════════════════════════════ # BIBLE-AWARE GUARDRAILS # ══════════════════════════════════════════════════════════════════════ def _enforce_prompt_length(prompt: str, model: str, mode: str = "default") -> str: """Enforce bible-defined prompt length limits. Reads optimal word range and max_chars from the Prompt Bible for the given model/mode. Logs WARNING if word count exceeds optimal max. Truncates at last sentence boundary if char count exceeds max_chars. Args: prompt: The prompt string to check/truncate. model: Bible model key (e.g. "kling-v3", "veo-3.1"). mode: Prompt mode (e.g. "i2v", "t2v", "default"). Returns: The (possibly truncated) prompt string. """ try: word_min, word_max = get_optimal_word_range(model, mode) except (KeyError, TypeError): # Model not in bible — pass through silently return prompt word_count = len(prompt.split()) if word_count > word_max: logger.warning( "Prompt for %s/%s has %d words (optimal: %d-%d)", model, mode, word_count, word_min, word_max, ) try: max_chars = get_prompt_rules(model).get("max_chars") except KeyError: max_chars = None if max_chars and len(prompt) > max_chars: logger.error( "Prompt for %s exceeds max_chars (%d > %d), truncating at sentence boundary", model, len(prompt), max_chars, ) truncated = prompt[:max_chars] # Find last sentence boundary last_period = truncated.rfind(".") if last_period > 0: prompt = truncated[: last_period + 1] else: prompt = truncated.rstrip() return prompt def _get_shared_lexicon_token(key: str) -> str: """Read a shared Prompt Bible lexicon token by key.""" shared_lexicon = load_bible().get("shared_lexicon", {}) if not isinstance(shared_lexicon, dict) or key not in shared_lexicon: raise KeyError(f"PROMPT_BIBLE shared_lexicon missing {key!r}") return str(shared_lexicon[key]).strip() def _no_text_footer() -> str: return _get_shared_lexicon_token("no_text_footer") def apply_look(prompt: str, bundle) -> str: """Append Look/Identity prompt fragments from a LookBundle (krea2-flora). OPT-IN: invoked by the Flora image builders (Seedream / NBP / Krea) only when a resolved ``LookBundle`` is present. When ``bundle`` is ``None`` this is a complete no-op — the prompt is returned byte-identical — so every existing (look-less) code path is unaffected. What it appends, in order, as POSITIVE prose fragments: 1. ``bundle.look_pack["positive"]`` — house/series look descriptors. 2. ``bundle.triggers`` — identity textual anchors (e.g. "KARAVOSS"). 3. palette hex codes as positive descriptors ("color palette: #..."). 4. ``bundle.look_pack["avoid"]`` — **phrased POSITIVELY** ("avoiding X"). These are NEVER emitted as a negative_prompt (no Flora image hero supports one); they ride the positive prompt as exclusionary prose. Cinema tokens are NOT added here — they continue to render from ``bundle.cinema_mode_id`` via the existing ``render_cinema_tokens()`` in the cinematography path. Length enforcement (``_enforce_prompt_length``) is applied by the CALLING builder after this returns, exactly as before. """ if bundle is None: return prompt look_pack = getattr(bundle, "look_pack", None) or {} fragments: list[str] = [] # 1. Positive look descriptors. for token in look_pack.get("positive") or []: if token and str(token).strip(): fragments.append(str(token).strip().rstrip(".")) # 2. Identity triggers (textual anchors). for trig in getattr(bundle, "triggers", None) or []: if trig and str(trig).strip(): fragments.append(str(trig).strip().rstrip(".")) # 3. Palette hex codes → positive descriptors. palette = getattr(bundle, "palette", None) or {} hexes = [h for h in (palette.get("hex") or []) if h and str(h).strip()] if hexes: fragments.append("color palette " + ", ".join(str(h).strip() for h in hexes)) # 4. avoid[] phrased POSITIVELY (never a negative_prompt). avoid = [a for a in (look_pack.get("avoid") or []) if a and str(a).strip()] if avoid: fragments.append( "avoiding " + ", ".join(str(a).strip().rstrip(".") for a in avoid) ) if not fragments: return prompt suffix = ". ".join(fragments) base = prompt.rstrip() if base and not base.endswith("."): base += "." return (base + " " + suffix + ".").strip() def validate_start_frame_ar( start_frame_path: "Path | None", model: str, target_ar: str, ) -> list[str]: """Validate start frame aspect ratio against model's I2V behavior. Args: start_frame_path: Path to start frame image, or None. model: Bible model key (e.g. "kling-v3", "seeddance-2.0"). target_ar: Target aspect ratio string (e.g. "9:16", "16:9"). Returns: List of warning strings. Empty list if no issues. """ if start_frame_path is None: return [] try: behavior = get_i2v_ar_behavior(model) except KeyError: return [] if behavior != "matches_start_frame": return [] # Read image dimensions try: from PIL import Image with Image.open(start_frame_path) as img: width, height = img.size except Exception as e: return [f"Could not read start frame dimensions: {e}"] # Parse target AR try: ar_parts = target_ar.split(":") target_w = int(ar_parts[0]) target_h = int(ar_parts[1]) except (ValueError, IndexError): return [f"Could not parse target AR: {target_ar}"] # Compare ratios with tolerance image_ratio = width / height target_ratio = target_w / target_h tolerance = 0.05 if abs(image_ratio - target_ratio) > tolerance: return [ f"Start frame AR mismatch for {model}: " f"image is {width}x{height} ({image_ratio:.2f}), " f"target AR is {target_ar} ({target_ratio:.2f}). " f"{model} matches start frame AR — output will NOT match {target_ar}." ] return [] def build_prompt_from_bible(model: str, scene_description: str, **kwargs) -> str: """Build a prompt using bible-defined rules for a given model. Generic builder for simple/casting generations. Reads the model's prompt rules from PROMPT_BIBLE.yaml to determine style (prose, keywords, formula_6step, cot_prose) and applies best_practices as guidance. Complex plan-based generation still uses specialized builders. Args: model: Bible model key (e.g. "seedream-v4.5", "kling-v3"). scene_description: Natural language scene description. **kwargs: Optional overrides — characters (list), location (str), camera (str), lighting (str), action (str), style (str). Returns: Formatted prompt string, length-enforced per bible rules. """ rules = get_prompt_rules(model) style = rules.get("style", "prose") characters = kwargs.get("characters", []) location = kwargs.get("location", "") camera = kwargs.get("camera", "") lighting = kwargs.get("lighting", "") action = kwargs.get("action", "") style_override = kwargs.get("style", "") if style == "keywords": # Comma-separated tags extracted from scene_description + kwargs parts = [ t.strip() for t in scene_description.replace(".", ",").split(",") if t.strip() ] if characters: parts.extend(characters) if location: parts.append(location) if camera: parts.append(camera) if lighting: parts.append(lighting) if action: parts.append(action) if style_override: parts.append(style_override) prompt = ", ".join(parts) elif style == "formula_6step": # Structured sections: Subject -> Action -> Environment -> Camera -> Style -> Constraints sections = [] # Subject subject_parts = [scene_description] if characters: subject_parts.append(", ".join(characters)) sections.append(f"Subject: {' '.join(subject_parts)}") # Action sections.append(f"Action: {action}" if action else "Action: idle") # Environment sections.append( f"Environment: {location}" if location else "Environment: unspecified" ) # Camera sections.append(f"Camera: {camera}" if camera else "Camera: default framing") # Style sections.append( f"Style: {style_override}" if style_override else "Style: cinematic" ) # Constraints if lighting: sections.append(f"Constraints: {lighting}") prompt = ". ".join(sections) + "." elif style == "cot_prose": # Chain-of-thought prefix + prose segments = [scene_description] if characters: segments.append(f"Characters: {', '.join(characters)}.") if location: segments.append(f"Setting: {location}.") if action: segments.append(action) if camera: segments.append(camera) if lighting: segments.append(lighting) if style_override: segments.append(style_override) prompt = "Let's think step by step. " + " ".join(segments) else: # prose (default) segments = [scene_description] if characters: segments.append(f"Characters: {', '.join(characters)}.") if location: segments.append(f"Setting: {location}.") if action: segments.append(action) if camera: segments.append(camera) if lighting: segments.append(lighting) if style_override: segments.append(style_override) prompt = " ".join(segments) # Enforce bible-defined length limits BEFORE appending guidance prompt = _enforce_prompt_length(prompt, model) # Append best_practices as guidance comments (only if room within max_chars) best_practices = rules.get("best_practices") if best_practices: guidance = " | ".join(best_practices) guidance_block = f" [{guidance}]" max_chars = rules.get("max_chars") if max_chars is None or len(prompt) + len(guidance_block) <= max_chars: prompt = f"{prompt}{guidance_block}" return prompt class GridType(Enum): """Grid generation strategies.""" SCENE_COVERAGE = "scene_coverage" # Structured 3x3: map storyboard shots to cells DIRECTORS_TAKE = ( "directors_take" # Structured 2x2: 4 framing variations of same shot ) ACTION_BURST = "action_burst" # Generic 2x2: subtle pose variations SKIP = "skip" # No grid — direct to Pro render # Kinetic descriptors now loaded from config/lexicon.json via prompt_config # Kinetic fallback loaded from config/lexicon.json # Lighting direction and quality maps now loaded from config/lexicon.json via prompt_config # ── ENV Sanitization (ported from test_nbp_direct.py V4) ───────────── _HUMAN_PATTERNS = [ re.compile(r",?\s*a\s+figure'?s?\s+\w+\s+visible\s+[^,\.]*[,\.]?", re.I), re.compile( r",?\s*\b(?:a|the)\s+(?:figure|person|silhouette|someone|somebody)\b[^,\.]*[,\.]?", re.I, ), re.compile( r"\b(?:her|his|their)\s+(?:arms?|hands?|fingers?|face|body|boots?|feet|foot|legs?|torso|shoulders?)\b", re.I, ), re.compile(r"(?:^|\.\s*)\b(?:She|He)\s+[^\.]+\.", re.I), re.compile(r"\b(?:both|two)\s+(?:arms?|hands?|fingers?|legs?|feet)\b", re.I), re.compile(r"\bthe\s+figure\b", re.I), re.compile(r"\bfigure'?s?\b", re.I), ] _CLEANUP = [ (re.compile(r",\s*,"), ","), (re.compile(r"\s{2,}"), " "), (re.compile(r",\s*\."), "."), (re.compile(r"^\s*,\s*"), ""), ] # ── Wide-Shot Categories ────────────────────────────────────────────── _WIDE_SHOT_TYPES = {"WIDE", "LS", "EWS", "WS", "VLS"} _CLOSE_SHOT_TYPES = {"CU", "ECU", "MCU", "BCU"} _MEDIUM_SHOT_TYPES = {"MS", "MLS", "MWS", "MFS"} # ── Editorial → Visual Framing Map ────────────────────────────────── # Editorial shot types from plans that video models don't understand. # Mapped to visual equivalents before any prompt builder uses them. _EDITORIAL_TO_VISUAL = { "INSERT": "ECU", "OTS": "MCU", "POV": "MS", "CUTAWAY": "CU", "REACTION": "CU", "ESTABLISHING": "EWS", } def _resolve_shot_type(shot_type: str) -> str: """Map editorial shot types to visual framing codes. Returns the original type if it's already a visual framing code. """ return _EDITORIAL_TO_VISUAL.get(shot_type, shot_type) # ── Shot Type & Movement Display Names ──────────────────────────────── # Single source of truth — used by all prompt builders. _SHOT_TYPE_NAMES = { "ECU": "Extreme close-up", "CU": "Close-up", "BCU": "Big close-up", "MCU": "Medium close-up", "MS": "Medium shot", "MFS": "Medium full shot", "MLS": "Medium long shot", "MWS": "Medium wide shot", "LS": "Long shot", "FS": "Full shot", "WS": "Wide shot", "EWS": "Extreme wide shot", "VLS": "Very long shot", "WIDE": "Wide shot", "INSERT": "Insert shot", } _MOVEMENT_NAMES = { "pan": "panning", "tilt": "tilting", "push_in": "push-in", "pull_back": "pull-back", "tracking": "tracking", "crane": "crane", "handheld": "handheld", "steadicam": "Steadicam", "dolly": "dolly", "push": "push-in", "pull": "pull-back", "track": "tracking", } # Gerund forms for I2V (motion-focused prompts) _MOVEMENT_NAMES_GERUND = { "pan": "panning", "tilt": "tilting", "push_in": "pushing in", "pull_back": "pulling back", "tracking": "tracking", "crane": "crane", "handheld": "handheld", "steadicam": "Steadicam", "dolly": "dollying", "push": "pushing in", "pull": "pulling back", "track": "tracking", } # Endpoint markers for I2V prompt validation _ENDPOINT_MARKERS = ( "settles", "stabilizes", "at rest", "at full extension", "fully extended", "overhead", "at finish", "comes to", "stops", "freezes", "holds", "still", "motionless", "stationary", ) # ── Grid Panel Position Labels ──────────────────────────────────────── _GRID_POSITIONS_3x3 = [ "TOP LEFT", "TOP CENTER", "TOP RIGHT", "MIDDLE LEFT", "MIDDLE CENTER", "MIDDLE RIGHT", "BOTTOM LEFT", "BOTTOM CENTER", "BOTTOM RIGHT", ] _GRID_POSITIONS_2x2 = [ "TOP LEFT", "TOP RIGHT", "BOTTOM LEFT", "BOTTOM RIGHT", ] # ══════════════════════════════════════════════════════════════════════ # PUBLIC API # ══════════════════════════════════════════════════════════════════════ def sanitize_env_prompt(text: str) -> str: """Strip human-presence language from environment-only descriptions.""" result = text for pattern in _HUMAN_PATTERNS: result = pattern.sub("", result) for pattern, replacement in _CLEANUP: result = pattern.sub(replacement, result) return result.strip() def _dedup_phrases(text: str) -> str: """Remove duplicate comma-separated or period-separated phrases. Handles cases where the same descriptive phrase appears multiple times in a skeleton field (e.g. from override edits or plan generation bugs). """ if not text: return text # Split on sentence boundaries first, then phrase boundaries sentences = [s.strip() for s in re.split(r"\.\s*", text) if s.strip()] deduped_sentences = [] seen = set() for sentence in sentences: # Deduplicate comma-separated phrases within each sentence phrases = [p.strip() for p in sentence.split(",") if p.strip()] deduped_phrases = [] for phrase in phrases: key = phrase.lower().strip() if key not in seen: seen.add(key) deduped_phrases.append(phrase) if deduped_phrases: deduped_sentences.append(", ".join(deduped_phrases)) return ". ".join(deduped_sentences) def _is_plan_shot(shot: dict) -> bool: """Check if a shot dict is from a Starsend plan (vs legacy Recoil).""" return "prompt_data" in shot and "routing_data" in shot # ══════════════════════════════════════════════════════════════════════ # PLAN-AWARE PROMPT BUILDERS (primary path when plans exist) # ══════════════════════════════════════════════════════════════════════ _OVERRIDE_PRESETS = { "standard": set(), "stylized": {"film_stock", "quality_guard"}, "raw": {"film_stock", "quality_guard", "shot_footer", "non_human_locks"}, "passthrough": { "film_stock", "lighting", "spatial", "shot_footer", "non_human_locks", "quality_guard", }, "custom": set(), } def _resolve_bypasses(shot): overrides = shot.get("prompt_overrides", {}) mode = overrides.get("mode", "standard") if mode == "custom": return set(overrides.get("bypass", [])) return _OVERRIDE_PRESETS.get(mode, set()) # ── Layer color assignments for the prompt inspector ───────────────── _LAYER_COLORS = { "camera_line": "#E8A735", "film_stock": "#7B8A6E", "subject_line": "#5BA4CF", "environment_line": "#5BA4CF", "moodboard_text": "#4A9E8E", "character_descs": "#9B7EC8", "kinetic_action": "#D4763B", "lighting": "#C4B74A", "scene_visual_locks": "#8E7D3A", "spatial_continuity": "#4A7EC8", "emotion_line": "#C85A7E", "shot_footer": "#6B7280", "non_human_locks": "#8E4A4A", "camera_guard": "#9CA3AF", "quality_guard": "#C84A4A", "director_notes": "#E5E7EB", } _BYPASSABLE_LAYERS = { "film_stock", "lighting", "spatial", "shot_footer", "non_human_locks", "camera_guard", "quality_guard", } def build_prompt_sections_from_plan( shot: dict, bible: dict, project_config: dict, episode: int = 1, character_data_overrides: dict = None, ) -> list: """Build annotated per-layer prompt sections from a plan ShotRecord. Mirrors the exact logic of build_prompt_from_plan() but returns a list of dicts instead of a flat string. Each dict has: id: Layer identifier (e.g. "camera_line", "film_stock") label: Human-readable label (e.g. "Camera Line") text: The rendered text for this layer (may be empty) color: Hex color for the inspector UI bypassable: Whether this layer can be toggled off via overrides active: Whether this layer is currently active (not bypassed) """ project_config = project_config or {} character_data_overrides = character_data_overrides or {} bypasses = _resolve_bypasses(shot) sections = [] prompt_data = shot.get("prompt_data", {}) routing_data = shot.get("routing_data", {}) asset_data = shot.get("asset_data", {}) spatial_data = shot.get("spatial_data", {}) skeleton = prompt_data.get("prompt_skeleton", {}) # JIT hydration: resolve {char_*} and {loc_*} tokens from live bible skeleton = _maybe_hydrate(skeleton, bible, episode=episode, asset_data=asset_data) raw_shot_type = prompt_data.get("shot_type", "MS") shot_type = _resolve_shot_type(raw_shot_type) is_env = routing_data.get("is_env_only", False) characters = asset_data.get("characters", []) # Handle reverse OTS from coverage — check raw type before resolution is_reverse_ots = prompt_data.get("reverse_ots", False) if is_reverse_ots and raw_shot_type == "OTS": chars = asset_data.get("characters", []) if len(chars) >= 2: shoulder_char = ( chars[0].get("char_id", "") if isinstance(chars[0], dict) else str(chars[0]) ) face_char = ( chars[1].get("char_id", "") if isinstance(chars[1], dict) else str(chars[1]) ) skeleton["subject_line"] = ( f"Over-the-shoulder shot from behind {shoulder_char}, " f"looking at {face_char}'s face" ) def _section(layer_id, label, text, bypass_key=None): """Helper to build a section dict.""" is_bypassable = layer_id in _BYPASSABLE_LAYERS or ( bypass_key and bypass_key in _BYPASSABLE_LAYERS ) bk = bypass_key or layer_id is_active = bk not in bypasses return { "id": layer_id, "label": label, "text": text or "", "color": _LAYER_COLORS.get(layer_id, "#6B7280"), "bypassable": is_bypassable, "active": is_active, "bypass_key": bk, } # ── 1. Camera line ──────────────────────────────────────────────── # UI inspector: not bound to a specific model. Read the bible's global # default (no per-model override applies here). camera_line = _build_camera_line_plan(prompt_data, model_id=None) sections.append(_section("camera_line", "Camera Line", camera_line)) # ── 2. Film stock / cinematic baseline ──────────────────────────── camera_body = project_config.get("camera_body") or get_constant( "production", "camera_body" ) film_stock = project_config.get("film_stock") or get_constant( "production", "film_stock" ) film_style = project_config.get("film_style_suffix") or get_constant( "production", "film_style_suffix" ) film_text = ( f"Shot on {camera_body}, {film_stock}, {film_style}." if "film_stock" not in bypasses else "" ) sections.append(_section("film_stock", "Film Stock", film_text)) # ── 3. Prompt skeleton — subject + environment ──────────────────── subject = _dedup_phrases(skeleton.get("subject_line", "")) environment = _dedup_phrases(skeleton.get("environment_line", "")) action = skeleton.get("action_line", "") emotion = skeleton.get("emotion_line", "") if is_env: # Deduplicate: if environment text already appears in subject, skip it if environment and environment.strip().rstrip(".") in subject: env_text = f"{subject}." else: env_text = f"{subject}. {environment}." if environment else f"{subject}." env_text = sanitize_env_prompt(env_text) sections.append(_section("subject_line", "Subject Line (ENV)", env_text)) env_guard = get_constant("production", "env_only_guard") sections.append(_section("environment_line", "Environment Guard", env_guard)) else: sections.append( _section("subject_line", "Subject Line", (subject + ".") if subject else "") ) sections.append( _section( "environment_line", "Environment Line", (environment + ".") if environment else "", ) ) # ── 3b. Moodboard text enrichment ───────────────────────────────── moodboard_text = shot.get("_moodboard_text") mb_text = "" if moodboard_text and not is_env: mb_text = f"Location detail: {moodboard_text}" sections.append(_section("moodboard_text", "Moodboard Text", mb_text)) # ── 4. Character descriptions from bible ────────────────────────── char_descs_text = "" if not is_env and characters: char_descs_text = ( _build_character_descs_from_bible( characters, bible, episode, shot_type, spatial_data, character_data_overrides=character_data_overrides, subject_line=subject, ) or "" ) sections.append( _section("character_descs", "Character Descriptions", char_descs_text) ) # ── 5. Kinetic action ───────────────────────────────────────────── kinetic = prompt_data.get("kinetic_action", "") kinetic_parts = [] if kinetic: kinetic_clean = _strip_motion_language(kinetic) if kinetic_clean: kinetic_parts.append(kinetic_clean + ".") if action and action != kinetic: action_clean = _strip_motion_language(action) if action_clean: kinetic_parts.append(action_clean + ".") sections.append( _section("kinetic_action", "Kinetic Action", " ".join(kinetic_parts)) ) # ── 6. Lighting ─────────────────────────────────────────────────── has_spatial = bool(spatial_data.get("camera_side")) lighting_str = _build_lighting_from_plan( prompt_data, suppress_direction=has_spatial ) lighting_text = "" if lighting_str and "lighting" not in bypasses: lighting_text = f"Lighting: {lighting_str}" sections.append(_section("lighting", "Lighting", lighting_text)) # ── 6b. Scene visual locks ──────────────────────────────────────── scene_locks = shot.get("_scene_visual_locks") scene_locks_text = "" if scene_locks and not is_env: lock_parts = [] if scene_locks.get("lighting"): lock_parts.append(f"Scene lighting lock: {scene_locks['lighting']}") if scene_locks.get("atmosphere"): lock_parts.append(f"Scene atmosphere: {scene_locks['atmosphere']}") if scene_locks.get("palette"): lock_parts.append(f"Scene palette: {scene_locks['palette']}") if lock_parts: scene_locks_text = " ".join(lock_parts) sections.append( _section("scene_visual_locks", "Scene Visual Locks", scene_locks_text) ) # ── 6c. Spatial continuity block ────────────────────────────────── spatial_block = build_spatial_continuity_block( shot=shot, bible=bible, scene_shots=shot.get("_scene_shots"), prev_episode_final=shot.get("_prev_episode_final"), ) spatial_text = "" if spatial_block and "spatial" not in bypasses: spatial_text = spatial_block sections.append( _section( "spatial_continuity", "Spatial Continuity", spatial_text, bypass_key="spatial", ) ) # ── 7. Emotion line ─────────────────────────────────────────────── emotion_text = "" if emotion and not is_env: emotion_text = emotion + "." sections.append(_section("emotion_line", "Emotion Line", emotion_text)) # ── 8. Shot-type footer ─────────────────────────────────────────── footer_text = "" if not is_env and "shot_footer" not in bypasses: if shot_type in _WIDE_SHOT_TYPES: footer_text = _build_wide_shot_footer() elif shot_type in _CLOSE_SHOT_TYPES: footer_text = _build_close_shot_footer() else: footer_text = get_constant("production", "medium_shot_footer") sections.append(_section("shot_footer", "Shot Footer", footer_text)) # ── 9. Non-human identity locks ─────────────────────────────────── nh_parts = [] if not is_env and "non_human_locks" not in bypasses: for char in characters: char_id = char.get("char_id", "") bible_char = bible.get("characters", {}).get(char_id, {}) visual = bible_char.get("visual_description", "") if _visual_is_non_human(visual): name = bible_char.get("display_name", char_id) nh_parts.append( f"{name} identity lock: " + get_constant("production", "non_human_identity_lock") ) sections.append(_section("non_human_locks", "Non-Human Locks", " ".join(nh_parts))) # ── 10. Camera direction guard ──────────────────────────────────── guard_text = "" if not is_env and "camera_guard" not in bypasses and not has_spatial: screen_dir = spatial_data.get("screen_direction", "center") if screen_dir != "toward-camera": guard_text = get_constant("production", "camera_direction_guard") sections.append(_section("camera_guard", "Camera Guard", guard_text)) # ── 11. Quality guard ───────────────────────────────────────────── quality_text = "" if "quality_guard" not in bypasses: quality_guard = project_config.get("quality_guard", "") if quality_guard and not is_env: quality_text = quality_guard sections.append(_section("quality_guard", "Quality Guard", quality_text)) # ── 12. Director notes (always last, never filtered) ────────────── director_notes = shot.get("director_notes", "") dn_text = "" if director_notes: dn_text = f"DIRECTOR: {director_notes}" sections.append(_section("director_notes", "Director Notes", dn_text)) return sections def build_prompt_from_plan( shot: dict, bible: dict, project_config: dict, episode: int = 1, character_data_overrides: dict = None, ) -> str: """Build a complete cinematic prompt from a plan ShotRecord. This is the PRIMARY entry point when plan data exists. Uses structured prompt_skeleton fields — no regex parsing needed. Args: shot: ShotRecord dict from episode plan (has prompt_data, routing_data, etc.) bible: Global bible dict (characters, locations, props) project_config: project_config.json dict episode: Episode number (for wardrobe phase resolution) character_data_overrides: Optional dict keyed by char_id (lowercase) with visual/wardrobe overrides. Example: {"jinx": {"visual": "scarred face, cybernetic eye"}} Returns: Complete cinematic prompt string. """ sections = build_prompt_sections_from_plan( shot, bible, project_config, episode, character_data_overrides ) texts = [s["text"] for s in sections if s.get("text", "").strip()] prompt = " ".join(t.strip() for t in texts) # Clean up double periods, double spaces prompt = re.sub(r"\.{2,}", ".", prompt) prompt = re.sub(r"\s{2,}", " ", prompt) # Archetype trigger scrub prompt = _scrub_archetype_triggers(prompt) return prompt.strip() # Motion language that causes artifacts in still-image generation. # These describe temporal events (blur, sweep, shift) that a single frame # can't represent. Strip or convert to static pose equivalents. _MOTION_STRIP_PATTERNS = [ (re.compile(r"\bmotion blur\b[^,.]*", re.I), ""), (re.compile(r"\bsweeping background[^,.]*", re.I), ""), (re.compile(r"\bdynamic lateral shift\b[^,.]*", re.I), ""), (re.compile(r"\bsharp pivot\b", re.I), "turned sharply"), (re.compile(r"\bquick head turn\b", re.I), "head turned"), (re.compile(r"\brapid movement\b", re.I), "tense posture"), (re.compile(r"\bslow creeping push in\b", re.I), ""), (re.compile(r"\bdownward tilt following\b", re.I), ""), ] def _strip_motion_language(text: str) -> str: """Remove motion-specific language from kinetic/action text for stills.""" for pattern, replacement in _MOTION_STRIP_PATTERNS: text = pattern.sub(replacement, text) # Clean up resulting artifacts: double commas, leading commas, etc. text = re.sub(r",\s*,", ",", text) text = re.sub(r"^\s*,\s*", "", text) text = re.sub(r",\s*$", "", text) text = re.sub(r"\s{2,}", " ", text) return text.strip() # Known toxic words that trigger unwanted visual tropes in generation models. # "tactical" → tied-back hair, fingerless gloves, webbing/pouches. # Applied to final assembled prompt only, never to source data. _ARCHETYPE_SCRUBS = [ (re.compile(r"\btactical pants\b", re.I), "cargo pants"), (re.compile(r"\btactical assessment\b", re.I), "careful assessment"), (re.compile(r"\btactical gloves?\b", re.I), "gloves"), ] def _scrub_archetype_triggers(prompt: str) -> str: """Replace known toxic trigger words in the final prompt.""" for pattern, replacement in _ARCHETYPE_SCRUBS: prompt = pattern.sub(replacement, prompt) return prompt def _get_include_focal_length(model_id: str | None) -> bool: """Read `include_focal_length` from bible (per-model override on global_defaults). Returns False by default (Bug T, 2026-05-21 / pipeline-learnings §28): focal length and framing are independent — emitting `{focal}mm` in the camera line just bloats the prompt without changing the rendered framing. Models that opt back in via `.prompt.include_focal_length: true` restore the legacy "{framing}, {focal}mm" line. """ try: return bool( get_prompt_rule_with_global_default( model_id or "", "include_focal_length", default=False, ) ) except Exception: return False _FOCAL_MM_OPT_OUT_RE = re.compile( r"\b\d+(?:\.\d+)?(?:-\d+(?:\.\d+)?)?mm\b" r"(?!\s+(?:film|grain|print|stock|negative|gauge|emulsion|pushed|push|reversal)\b)", re.IGNORECASE, ) def _strip_focal_mm_tokens_for_model(text: str, model_id: str | None) -> str: """Remove lens-like mm tokens when a model opts out of focal lengths.""" if not text or _get_include_focal_length(model_id): return text cleaned = _FOCAL_MM_OPT_OUT_RE.sub("", text) cleaned = re.sub(r"\s{2,}", " ", cleaned) cleaned = re.sub(r"\s+,", ",", cleaned) return cleaned.strip(" ,") def _build_camera_line_plan(prompt_data: dict, model_id: str | None = None) -> str: """Build camera line from plan prompt_data (structured fields). The `{focal}mm` token is opt-in via `global_defaults.include_focal_length` (or per-model `.prompt.include_focal_length`). Defaults to off per pipeline-learnings §28 — pass model_id so per-model overrides apply. """ shot_type = _resolve_shot_type(prompt_data.get("shot_type", "MS")) focal_length = prompt_data.get("focal_length", "50mm") camera_movement = prompt_data.get("camera_movement", "static") type_name = _SHOT_TYPE_NAMES.get(shot_type, f"{shot_type} shot") include_focal = _get_include_focal_length(model_id) if include_focal: line = f"{type_name}, {focal_length}" else: line = f"{type_name}" if camera_movement and camera_movement != "static": movement = _MOVEMENT_NAMES.get(camera_movement, camera_movement) line += f", {movement}" return line + "." def _build_lighting_from_plan( prompt_data: dict, suppress_direction: bool = False ) -> str: """Build lighting string from plan's structured lighting data. Args: suppress_direction: If True, omit directional text (e.g. "from LEFT") to avoid conflict with the spatial block's camera-relative lighting. Gemini bug-stomp finding: abstract direction here + camera-relative in spatial block = mushy/conflicting lighting. """ lighting = prompt_data.get("lighting", {}) sources = lighting.get("sources", []) if not sources: return "" dominant_idx = lighting.get("dominant_source_index", 0) source = sources[min(dominant_idx, len(sources) - 1)] motivator = source.get("motivator", "") direction = source.get("direction", "ABOVE") quality = source.get("quality", "hard") color_temp = source.get("color_temp", "neutral") intensity = source.get("intensity", "moderate") parts = [] if color_temp and color_temp != "neutral": parts.append(f"{color_temp} light") else: parts.append("light") if suppress_direction: parts.append(f"casting {quality} shadows") else: parts.append(f"casting {quality} shadows from {direction}") if motivator: parts.append(f"(source: {motivator})") if intensity and intensity != "moderate": parts.append(f"[{intensity}]") return " ".join(parts) # ══════════════════════════════════════════════════════════════════════ # SPATIAL CONTINUITY ENGINE # Gemini-approved spatial injection (3 rounds) + Opus architecture (2 rounds) # See consultations/spatial-continuity/SYNTHESIS.md for full decisions. # ══════════════════════════════════════════════════════════════════════ # Direction mapping: abstract lighting → camera-relative _LIGHTING_FLIP_MAP = { "LEFT": {"A": ("LEFT", "RIGHT"), "B": ("RIGHT", "LEFT")}, "RIGHT": {"A": ("RIGHT", "LEFT"), "B": ("LEFT", "RIGHT")}, "SIDE": {"A": ("LEFT", "RIGHT"), "B": ("RIGHT", "LEFT")}, } _INVARIANT_DIRECTIONS = {"ABOVE", "BELOW", "SELF_ILLUMINATED"} _POSITION_ORDER = { "left": 0, "center-left": 0, "center": 1, "center-right": 2, "right": 2, } def _resolve_camera_relative_lighting( lighting_data: dict, camera_side: str, ) -> tuple[str, str]: """Resolve abstract lighting direction to camera-relative (light_side, shadow_side). Rules (Gemini consultation): - ABOVE, BELOW, SELF_ILLUMINATED: invariant → default LEFT/RIGHT - LEFT, RIGHT, SIDE: become camera-left/camera-right, flip on Side B """ if not lighting_data: return ("LEFT", "RIGHT") sources = lighting_data.get("sources", []) if not sources: return ("LEFT", "RIGHT") dominant_idx = lighting_data.get("dominant_source_index", 0) source = sources[min(dominant_idx, len(sources) - 1)] direction = source.get("direction", "LEFT").upper() if direction in _INVARIANT_DIRECTIONS: return ("LEFT", "RIGHT") mapping = _LIGHTING_FLIP_MAP.get(direction, _LIGHTING_FLIP_MAP["LEFT"]) return mapping.get(camera_side, ("LEFT", "RIGHT")) def _normalize_camera_side(raw: str) -> str: """Normalize camera_side to 'A' or 'B'. Handles bad data like 'left'/'right'/'LEFT'.""" if not raw: return "A" val = str(raw).strip().lower() if val in ("left", "a", "side_a"): return "A" if val in ("right", "b", "side_b"): return "B" return "A" def _resolve_display_name(char_entry: dict, bible: dict) -> str: """Get display name for a character from bible.""" if not isinstance(char_entry, dict): return str(char_entry) char_id = char_entry.get("char_id", "") bible_char = bible.get("characters", {}).get(char_id, {}) return bible_char.get("display_name", char_id) def _is_moving(screen_dir: str, prompt_data: dict) -> bool: """Detect if character is moving (not static center).""" if screen_dir in ("left-to-right", "right-to-left"): return True camera_movement = prompt_data.get("camera_movement", "static") return camera_movement not in ("static", "", None) def _resolve_facing(screen_dir: str, camera_side: str) -> str: """Resolve facing direction from screen direction + camera side.""" if screen_dir == "left-to-right": return "RIGHT" if camera_side == "A" else "LEFT" elif screen_dir == "right-to-left": return "LEFT" if camera_side == "A" else "RIGHT" return "RIGHT" def _resolve_lr_assignment( characters: list[dict], spatial_data: dict, camera_side: str, bible: dict, ) -> tuple[str, str]: """Resolve which character is screen-LEFT vs screen-RIGHT. Camera Side A = stage matches screen (stage-left = screen-left). Camera Side B = stage inverts (stage-left = screen-right). """ char_a = characters[0] char_b = characters[1] pos_a = ( char_a.get("screen_position", "left") if isinstance(char_a, dict) else "left" ) pos_b = ( char_b.get("screen_position", "right") if isinstance(char_b, dict) else "right" ) name_a = _resolve_display_name(char_a, bible) name_b = _resolve_display_name(char_b, bible) a_is_stage_left = _POSITION_ORDER.get(pos_a, 0) < _POSITION_ORDER.get(pos_b, 2) if camera_side == "A": return (name_a, name_b) if a_is_stage_left else (name_b, name_a) else: return (name_b, name_a) if a_is_stage_left else (name_a, name_b) def _resolve_ots_assignment( characters: list[dict], spatial_data: dict, camera_side: str, bible: dict, ) -> tuple[str, str, str]: """Resolve foreground/background for OTS shots. Side A: camera favors char_b (char_a is FG shoulder). Side B: camera favors char_a (char_b is FG shoulder). Returns (fg_char_name, bg_char_name, fg_side). """ char_a = characters[0] char_b = characters[1] name_a = _resolve_display_name(char_a, bible) name_b = _resolve_display_name(char_b, bible) pos_a = ( char_a.get("screen_position", "left") if isinstance(char_a, dict) else "left" ) if camera_side == "A": if _POSITION_ORDER.get(pos_a, 0) <= 1: return (name_a, name_b, "LEFT") else: return (name_b, name_a, "RIGHT") else: if _POSITION_ORDER.get(pos_a, 0) <= 1: return (name_a, name_b, "RIGHT") else: return (name_b, name_a, "LEFT") def _is_punch_in(shot: dict, scene_shots: list[dict] | None) -> bool: """Detect if this shot is a punch-in from a wider shot.""" spatial = shot.get("spatial_data", {}) if spatial.get("punch_in_from"): return True if not scene_shots: return False shot_type = _resolve_shot_type(shot.get("prompt_data", {}).get("shot_type", "")) if shot_type not in ("ECU", "BCU"): return False shot_id = shot.get("shot_id", "") current_idx = next( (i for i, s in enumerate(scene_shots) if s.get("shot_id") == shot_id), -1 ) if current_idx <= 0: return False prev = scene_shots[current_idx - 1] prev_type = _resolve_shot_type(prev.get("prompt_data", {}).get("shot_type", "")) if prev_type not in _WIDE_SHOT_TYPES and prev_type not in ("MS", "MCU"): return False current_chars = { c.get("char_id") if isinstance(c, dict) else c for c in shot.get("asset_data", {}).get("characters", []) } prev_chars = { c.get("char_id") if isinstance(c, dict) else c for c in prev.get("asset_data", {}).get("characters", []) } return bool(current_chars & prev_chars) def _resolve_punch_in_source( shot: dict, scene_shots: list[dict] | None, bible: dict, ) -> tuple[str, str]: """Return (wider_shot_id, character_display_name) for punch-in formatting.""" spatial = shot.get("spatial_data", {}) wider_id = spatial.get("punch_in_from", "") if not wider_id and scene_shots: shot_id = shot.get("shot_id", "") idx = next( (i for i, s in enumerate(scene_shots) if s.get("shot_id") == shot_id), 0 ) if idx > 0: wider_id = scene_shots[idx - 1].get("shot_id", "") chars = shot.get("asset_data", {}).get("characters", []) char_name = _resolve_display_name(chars[0], bible) if chars else "subject" return wider_id, char_name def _macro_to_micro_lighting(light_side: str) -> tuple[str, str]: """Convert frame-level light side to face-level for ECU. Camera-left key → lights LEFT side of face. Returns (lit_face_side, shadow_face_side). """ if light_side == "RIGHT": return ("RIGHT", "LEFT") return ("LEFT", "RIGHT") def _resolve_eyeline(spatial_data: dict, camera_side: str) -> str: """Resolve eyeline direction for ECU punch-in. Solo characters default to slightly off-camera (toward light side) rather than direct-to-camera, which produces more cinematic results. Only explicit 'toward-camera' screen_direction triggers direct gaze. """ screen_dir = spatial_data.get("screen_direction", "center") if screen_dir == "toward-camera": return "camera" elif screen_dir in ("left-to-right", "right"): return "RIGHT" if camera_side == "A" else "LEFT" elif screen_dir in ("right-to-left", "left"): return "LEFT" if camera_side == "A" else "RIGHT" else: relationships = spatial_data.get("character_relationships", {}) interaction = relationships.get("interaction_type", "solo") if interaction in ("dialogue", "confrontation"): return "LEFT" if camera_side == "B" else "RIGHT" # Solo/center: look slightly off-camera toward light side # (more cinematic than direct-to-camera stare) return "RIGHT" if camera_side == "A" else "LEFT" def _resolve_insert_facing(spatial_data: dict, camera_side: str) -> str: """Resolve object facing direction for INSERT shots.""" screen_dir = spatial_data.get("screen_direction", "center") if screen_dir in ("left-to-right",): return "RIGHT" if camera_side == "A" else "LEFT" elif screen_dir in ("right-to-left",): return "LEFT" if camera_side == "A" else "RIGHT" return "RIGHT" # ── Format Templates (pure string interpolation) ──────────────────── def _format_env(light_side: str, shadow_side: str) -> str: return ( f"SPATIAL CONTINUITY (Environment):\n" f"Primary light source originating from SCREEN-{light_side}, " f"casting shadows toward SCREEN-{shadow_side}." ) def _format_insert( facing: str, light_side: str, shadow_side: str, camera_side: str ) -> str: return ( f"SPATIAL CONTINUITY (Insert):\n" f"[OBJECT ORIENTATION]: Object angled/directed toward screen-{facing}.\n" f"Key light striking from CAMERA-{light_side}, casting shadows to the {shadow_side}." ) def _format_punch_in( wider_shot_id: str, camera_side: str, char_name: str, lit_face: str, shadow_face: str, eyeline: str, ) -> str: # Gemini bug-stomp: removed shot ID from header — meaningless noise to model if eyeline == "camera": gaze_line = "Eyes looking directly into camera." else: gaze_line = ( f"Gaze fixed on a point beyond the {eyeline} edge of the frame. " f"Three-quarter profile, eyes locked on something off-screen." ) return ( f"SPATIAL CONTINUITY (Extreme Close-Up Punch-In):\n" f"EXTREME CLOSE UP. {char_name}'s face fills the frame.\n" f"Key light striking the {lit_face} side of the face, " f"casting deep shadows on the {shadow_face} side.\n" f"{gaze_line}" ) def _format_solo_moving( camera_side: str, position: str, facing: str, char_name: str, light_side: str, shadow_side: str, ) -> str: return ( f"SPATIAL CONTINUITY:\n" f"[POSITIONED SCREEN-{position.upper()}, PROFILING {facing}]: " f"{char_name} angled toward screen-{facing}.\n" f"Leave negative space / lead room on the screen-{facing} side of the frame.\n" f"Key light from CAMERA-{light_side}, casting shadows to the {shadow_side}." ) def _format_solo_static( camera_side: str, char_name: str, light_side: str, shadow_side: str, ) -> str: return ( f"SPATIAL CONTINUITY:\n" f"[CENTER OF FRAME]: {char_name} facing forward, eyes looking slightly off-camera to the {light_side}.\n" f"Key light from CAMERA-{light_side}, casting shadows to the {shadow_side}." ) def _format_two_char_wide( camera_side: str, left_char: str, right_char: str, light_side: str, shadow_side: str, ) -> str: # Gemini bug-stomp: [SCREEN-LEFT/RIGHT] not [LEFT SIDE OF FRAME] for 9:16 return ( f"SPATIAL CONTINUITY (maintain across cuts):\n" f"[SCREEN-LEFT]: {left_char}, body angled toward screen-right.\n" f"[SCREEN-RIGHT]: {right_char}, body angled toward screen-left.\n" f"Key light from CAMERA-{light_side}, casting shadows to the {shadow_side}." ) def _format_two_char_ots( camera_side: str, fg_side: str, bg_position: str, fg_char: str, bg_char: str, light_side: str, shadow_side: str, ) -> str: return ( f"SPATIAL CONTINUITY (Over-the-shoulder / Z-depth):\n" f"[FOREGROUND, SCREEN-{fg_side}, BLURRED]: {fg_char}'s shoulder " f"and back of head framing the {fg_side} edge.\n" f"[BACKGROUND, {bg_position}, IN FOCUS]: {bg_char} facing camera, " f"eyeline directed toward screen-{fg_side}.\n" f"Key light from CAMERA-{light_side}, casting shadows to the {shadow_side}." ) # ── Main Router ────────────────────────────────────────────────────── def build_spatial_continuity_block( shot: dict, bible: dict, scene_shots: list[dict] | None = None, prev_episode_final: dict | None = None, ) -> str: """Build the SPATIAL CONTINUITY injection block for a shot. Routes to the appropriate format based on shot type and character count. Returns empty string if spatial_data is missing. """ spatial_data = shot.get("spatial_data", {}) if not spatial_data: return "" prompt_data = shot.get("prompt_data", {}) routing_data = shot.get("routing_data", {}) asset_data = shot.get("asset_data", {}) camera_side = _normalize_camera_side(spatial_data.get("camera_side", "A")) shot_type = _resolve_shot_type(prompt_data.get("shot_type", "MS")) is_env = routing_data.get("is_env_only", False) characters = asset_data.get("characters", []) char_count = len(characters) # Inherit camera side from previous episode for cliffhanger resolutions if prev_episode_final and not spatial_data.get("camera_side"): prev_shot = prev_episode_final.get("shot", {}) camera_side = prev_shot.get("spatial_data", {}).get("camera_side", "A") # Resolve lighting for all formats light_side, shadow_side = _resolve_camera_relative_lighting( prompt_data.get("lighting", {}), camera_side ) # Route to format if is_env or (char_count == 0 and shot_type != "INSERT"): return _format_env(light_side, shadow_side) if shot_type == "INSERT": facing = _resolve_insert_facing(spatial_data, camera_side) return _format_insert(facing, light_side, shadow_side, camera_side) if _is_punch_in(shot, scene_shots): wider_shot_id, char_name = _resolve_punch_in_source(shot, scene_shots, bible) lit_face, shadow_face = _macro_to_micro_lighting(light_side) eyeline = _resolve_eyeline(spatial_data, camera_side) return _format_punch_in( wider_shot_id, camera_side, char_name, lit_face, shadow_face, eyeline ) if char_count == 1: char_name = _resolve_display_name(characters[0], bible) screen_dir = spatial_data.get("screen_direction", "center") position = ( characters[0].get("screen_position", "center") if isinstance(characters[0], dict) else "center" ) if _is_moving(screen_dir, prompt_data): facing = _resolve_facing(screen_dir, camera_side) pos_upper = position.upper() # Collapse to 3 positions if pos_upper in ("CENTER-LEFT", "CENTER-RIGHT"): pos_upper = "CENTER" return _format_solo_moving( camera_side, pos_upper, facing, char_name, light_side, shadow_side ) else: return _format_solo_static(camera_side, char_name, light_side, shadow_side) if char_count >= 2: if shot_type in _WIDE_SHOT_TYPES: left_char, right_char = _resolve_lr_assignment( characters, spatial_data, camera_side, bible ) return _format_two_char_wide( camera_side, left_char, right_char, light_side, shadow_side ) else: fg_char, bg_char, fg_side = _resolve_ots_assignment( characters, spatial_data, camera_side, bible ) bg_position = "CENTER" return _format_two_char_ots( camera_side, fg_side, bg_position, fg_char, bg_char, light_side, shadow_side, ) return "" def _build_character_anchor(bible_char: dict, wardrobe: str = "") -> str: """Build affirmative physical anchors to prevent archetype drift. Extracts hair style and hand state from the visual description AND resolved wardrobe, returning explicit positive constraints. This prevents the generation model from defaulting to trope-driven assumptions (e.g., "tactical context" → tied-back hair + fingerless gloves). Uses affirmative language ("bare hands, visible skin") instead of negative ("no gloves") — affirmative works better with Gemini 3 Pro. """ visual = bible_char.get("visual_description", "") if not visual: return "" anchors = [] low = visual.lower() # Check both visual_description AND wardrobe for glove/hand references combined_low = f"{low} {wardrobe.lower()}" if wardrobe else low # Hair anchor — extract hair descriptor and make it explicit hair_match = re.search( r"((?:long|short|medium|wavy|curly|straight|braided|shaved|buzzed)" r"[^\.]*hair[^\.]*(?:loose|flowing|down|over|shoulder|back|bun|ponytail|tied)[^\.]*)", low, ) if hair_match: anchors.append(hair_match.group(0).strip()) elif "loose" in low and "hair" in low: # Fallback: just note loose hair anchors.append("hair worn loose, not tied back") # Hand anchor — only add when NEITHER visual NOR wardrobe mention gloves. # Positive constraint only — Gemini responds better to affirmative language. if "glove" not in combined_low and "gauntlet" not in combined_low: anchors.append("bare hands with visible skin") return ". ".join(anchors).capitalize() if anchors else "" def _build_character_descs_from_bible( characters: list[dict], bible: dict, episode: int, shot_type: str, spatial_data: dict, character_data_overrides: dict = None, subject_line: str = "", ) -> str: """Build character descriptions from bible data for plan shots.""" bible_chars = bible.get("characters", {}) character_data_overrides = character_data_overrides or {} descs = [] subject_lower = (subject_line or "").lower() for char in characters: char_id = char.get("char_id", "") phase_id = char.get("wardrobe_phase_id", "") bible_char = bible_chars.get(char_id, {}) if not bible_char: continue display_name = bible_char.get("display_name", char_id) visual = bible_char.get("visual_description", "") scale = bible_char.get("scale_prompt_fragment", "") # Apply character_data_overrides if present override = character_data_overrides.get(char_id.lower(), {}) if override.get("visual"): visual = override["visual"] # Find wardrobe from the matching phase wardrobe = "" hair_makeup = "" marks = "" for phase in bible_char.get("phases", []): if phase.get("phase_id") == phase_id: wardrobe = phase.get("wardrobe_description", "") hair_makeup = phase.get("hair_makeup", "") marks = phase.get("distinguishing_marks", "") break # Fallback: flat wardrobe.default when no phases exist if not wardrobe: wd = bible_char.get("wardrobe", {}) if isinstance(wd, dict): wardrobe = wd.get(phase_id or "default", wd.get("default", "")) elif isinstance(wd, str): wardrobe = wd # Apply wardrobe override after phase lookup if override.get("wardrobe"): wardrobe = override["wardrobe"] # ── Build Character Anchor ───────────────────────────────── # Extract key physical traits for affirmative anchoring. # This fights archetype drift (e.g., "tactical" triggering # tied-back hair and gloves). anchor = _build_character_anchor(bible_char, wardrobe=wardrobe) # Wardrobe dedup: skip wardrobe when subject_line already covers it. # Storyboard LLM writes wardrobe into subject_line from the same bible, # causing double-pass duplication (Gemini consultation finding). wardrobe_redundant = False if wardrobe and subject_lower: # Check if key wardrobe nouns already appear in subject_line wardrobe_words = {w for w in wardrobe.lower().split() if len(w) > 4} if wardrobe_words: overlap = sum(1 for w in wardrobe_words if w in subject_lower) wardrobe_redundant = overlap >= len(wardrobe_words) * 0.4 if shot_type in _WIDE_SHOT_TYPES: # Wide: silhouette + posture + wardrobe only parts = [] if scale: parts.append(scale) if wardrobe and not wardrobe_redundant: parts.append(wardrobe) descs.append(f"{display_name}: " + ". ".join(parts) + ".") elif shot_type in _CLOSE_SHOT_TYPES: # Close: full visual + hair/makeup + marks + anchor parts = [visual] if hair_makeup: parts.append(hair_makeup) if marks: parts.append(marks) if anchor: parts.append(anchor) descs.append(f"{display_name}: " + ". ".join(parts) + ".") else: # Medium: visual + wardrobe + anchor parts = [visual] if wardrobe and not wardrobe_redundant: # Don't double-prefix if wardrobe already starts with "[Name] wears" if wardrobe.lower().startswith( display_name.lower() ) or wardrobe.lower().startswith("wearing"): parts.append(wardrobe) else: parts.append(f"Wearing {wardrobe}") if anchor: parts.append(anchor) descs.append(f"{display_name}: " + ". ".join(parts) + ".") # For two-character shots, add spatial separation if len(characters) >= 2: relationships = spatial_data.get("character_relationships", {}) interaction = relationships.get("interaction_type", "solo") if interaction == "confrontation": descs.append("Characters face each other across the frame.") elif interaction == "dialogue": descs.append("Characters in conversation, shared frame.") return " ".join(descs) def _visual_is_non_human(visual_desc: str) -> bool: """Check if a visual description indicates a non-human character.""" return bool(_NON_HUMAN_RE.search(visual_desc)) def build_video_prompt_from_plan( shot: dict, bible: dict, project_config: dict, episode: int = 1, ) -> str: """Build a flattened video prompt from plan data. Uses dense natural language paragraphs — NO section headers (CAMERA:, SUBJECT:, etc.) as video models render text overlays from them. Includes Visual Anchors block for wardrobe/environment/lighting consistency (critical — without this, video will morph). """ project_config = project_config or {} routing_data = shot.get("routing_data", {}) prompt_data = shot.get("prompt_data", {}) audio_data = shot.get("audio_data", {}) asset_data = shot.get("asset_data", {}) skeleton = prompt_data.get("prompt_skeleton", {}) # JIT hydration skeleton = _maybe_hydrate(skeleton, bible, episode=episode, asset_data=asset_data) duration = routing_data.get("target_editorial_duration_s", 5) # ── Build flattened natural language prompt ─────────────────────── # Generic plan-based video prompt — no single bound model. Use the # bible's global default for include_focal_length. camera_line = _build_camera_line_plan(prompt_data, model_id=None) movement = prompt_data.get("camera_movement", "static") if movement and movement != "static": camera_line = camera_line.rstrip(".") + f", camera {movement}." subject = skeleton.get("subject_line", "") action = skeleton.get("action_line", "") emotion = skeleton.get("emotion_line", "") # Dense natural language paragraph (no section headers) prompt_parts = [f"Cinematic video, {duration} seconds.", camera_line] if subject: prompt_parts.append(f"{subject}.") if action: prompt_parts.append(f"{action}.") if emotion: prompt_parts.append(f"{emotion}.") # Audio context (subtle, not as a header) ambient = audio_data.get("ambient_sfx", "") foley = audio_data.get("foley_action", "") audio_parts = [p for p in [ambient, foley] if p] if audio_parts: prompt_parts.append(f"Audio context: {'. '.join(audio_parts)}.") prompt_parts.append(get_constant("production", "cinematic_quality_baseline")) # quality_guard intentionally omitted — it's an image-gen anatomy guard # that adds noise to video model prompts (Kling, SeedDance, Veo) # ── Director notes (always last in prompt parts, before visual anchors) ── director_notes = shot.get("director_notes", "") if director_notes and director_notes.strip(): prompt_parts.append(f"DIRECTOR: {director_notes.strip()}") main_prompt = " ".join(s.strip() for s in prompt_parts if s and s.strip()) # ── Visual Anchors (CRITICAL for identity/wardrobe consistency) ── # Build anchors using the same function as image prompts, adapted for video anchors = build_visual_anchors(shot, shot, bible, project_config) if anchors: # Adapt header for video context anchors = anchors.replace( "VISUAL ANCHORS (MUST REMAIN CONSTANT IN ALL PANELS):", "VISUAL ANCHORS (MUST REMAIN CONSTANT THROUGHOUT CLIP):", ) return f"{main_prompt}\n\n{anchors}" return main_prompt def build_visual_anchors( storyboard: dict, shot: dict, breakdown: dict, project_config: dict, character_data: dict = None, ) -> str: """Build the VISUAL ANCHORS block from storyboard + breakdown data. This is the critical consistency enforcement mechanism. Lists 3-6 traits that MUST remain constant across all grid panels. Args: storyboard: Full storyboard JSON (top-level dict with location, atmosphere, etc.) shot: Single shot dict from storyboard['shots'] breakdown: Full breakdown.json dict (characters, locations, etc.) project_config: project_config.json dict character_data: Optional override for character visual data Returns: Formatted VISUAL ANCHORS block string. """ anchors = [] # 1. Environment anchor — from storyboard location location = storyboard.get("location", "") atmosphere = shot.get("atmosphere", "") or storyboard.get("atmosphere", "") if location: env_line = location.rstrip(".") if atmosphere: # Keep it concise — first sentence of atmosphere atmos_first = atmosphere.split(".")[0].strip() env_line += f". {atmos_first}" anchors.append(f"Environment: {env_line}") # 2. Lighting anchor — explicit directional vector lighting_vector = _build_lighting_vector(shot, storyboard) if lighting_vector: anchors.append(f"Lighting: {lighting_vector}") # 3. Character anchor(s) — from storyboard characters or breakdown chars_in_shot = shot.get("characters_in_shot", []) if chars_in_shot: for char_key in chars_in_shot: char_info = _resolve_character_visual( char_key, storyboard, breakdown, character_data ) if char_info: anchors.append(f"Character ({char_key.title()}): {char_info}") # 4. Props anchor — extract signature props from shot action/description props = _extract_props(shot, storyboard, project_config) if props: anchors.append(f"Props: {props}") # 5. Color palette anchor — from shot or storyboard palette palette = shot.get("color_palette", []) or storyboard.get("color_palette", []) hex_map = project_config.get("hex_object_map", {}) if palette and hex_map: palette_items = [] for hex_code in palette[:4]: # Cap at 4 for brevity clean_hex = hex_code.split("(")[0].strip() if "(" in hex_code else hex_code desc = hex_map.get(clean_hex, "") if desc: palette_items.append(f"{clean_hex} ({desc})") else: palette_items.append(clean_hex) anchors.append(f"Palette: {', '.join(palette_items)}") # Format the block if not anchors: return "" lines = ["VISUAL ANCHORS (MUST REMAIN CONSTANT IN ALL PANELS):"] for anchor in anchors: lines.append(f"- {anchor}") return "\n".join(lines) def build_cinematic_prompt( shot: dict, storyboard: dict, character_data: dict = None, project_config: dict = None, is_env: bool = False, bible: dict = None, episode: int = 1, ) -> str: """Build a complete cinematic prompt for a single frame. Auto-detects plan format: if shot has prompt_data/routing_data, routes to build_prompt_from_plan() (no regex needed). Otherwise falls through to legacy regex-based path. Args: shot: Single shot dict (plan ShotRecord or legacy storyboard) storyboard: Full storyboard JSON (used for legacy path) character_data: Optional dict of character visual data keyed by name project_config: project_config.json dict is_env: Force ENV sanitization (overrides characters_in_shot check) bible: Global bible dict (used for plan path) episode: Episode number (used for plan path) Returns: Complete cinematic prompt string. """ # Plan-first: if shot has structured data, use the clean path if _is_plan_shot(shot): return build_prompt_from_plan( shot=shot, bible=bible or storyboard, # bible may be passed as storyboard project_config=project_config or {}, episode=episode, ) project_config = project_config or {} character_data = character_data or {} sections = [] shot_type = _resolve_shot_type(shot.get("shot_type", "MS")) chars_in_shot = shot.get("characters_in_shot", []) is_env_shot = is_env or len(chars_in_shot) == 0 # ── 1. Camera line (always first — framing-first architecture) ──── camera_line = _build_camera_line(shot) sections.append(camera_line) # ── 2. Film stock / cinematic baseline ──────────────────────────── cinematic = storyboard.get("cinematic", "") if cinematic: sections.append(cinematic) # ── 3. Scene context ────────────────────────────────────────────── hero_frame = shot.get("hero_frame", "") if hero_frame: # Use hero_frame as the primary description — it is the richest scene_text = hero_frame else: # Fallback to first_frame + action scene_text = shot.get("first_frame", "") or shot.get("action", "") if is_env_shot: scene_text = sanitize_env_prompt(scene_text) sections.append(scene_text) sections.append(get_constant("production", "env_only_guard")) else: sections.append(scene_text) # ── 4. Character descriptions (if not ENV) ──────────────────────── if not is_env_shot: for char_key in chars_in_shot: char_desc = _build_character_description( char_key, shot, storyboard, character_data, shot_type ) if char_desc: sections.append(char_desc) # ── 5. Kinetic descriptors from action/emotion ──────────────────── kinetic = _build_kinetic_layer(shot) if kinetic: sections.append(kinetic) # ── 6. Lighting vector lock ─────────────────────────────────────── lighting = _build_lighting_vector(shot, storyboard) if lighting: sections.append(f"Lighting: {lighting}") # ── 7. Atmosphere ───────────────────────────────────────────────── atmosphere = shot.get("atmosphere", "") if atmosphere and atmosphere not in scene_text: sections.append(atmosphere) # ── 8. Shot-type footer (wide vs close constraints) ─────────────── # Skip anatomical constraints entirely for ENV-only shots if not is_env_shot: if shot_type in _WIDE_SHOT_TYPES: sections.append(_build_wide_shot_footer()) elif shot_type in _CLOSE_SHOT_TYPES: sections.append(_build_close_shot_footer()) else: # Medium shots get positive anatomical constraints but no facial strip sections.append(get_constant("production", "medium_shot_footer")) # ── 9. Identity lock for non-human characters ───────────────────── if not is_env_shot: for char_key in chars_in_shot: if _is_non_human(char_key, character_data, storyboard): sections.append( f"{char_key.title()} identity lock: " + get_constant("production", "non_human_identity_lock") ) # ── 10. Camera direction guard ──────────────────────────────────── if not is_env_shot: spatial = shot.get("spatial", {}) blocking = spatial.get("blocking", {}) has_direct_to_camera = any( b.get("facing") == "toward-camera" for b in blocking.values() ) # Only add guard if character is NOT explicitly facing camera if not has_direct_to_camera: sections.append(get_constant("production", "camera_direction_guard")) # ── 11. Quality guard from project config ───────────────────────── quality_guard = project_config.get("quality_guard", "") if quality_guard and not is_env_shot: sections.append(quality_guard) # Assemble prompt = " ".join(s.strip() for s in sections if s and s.strip()) # Final cleanup — collapse double spaces, fix trailing punctuation prompt = re.sub(r"\s{2,}", " ", prompt) prompt = re.sub(r"\.\s*\.", ".", prompt) return prompt.strip() def build_grid_prompt( shots: list[dict], storyboard: dict, breakdown: dict, project_config: dict, grid_type: GridType, character_data: dict = None, grid_size: str = "3x3", ) -> str: """Build a structured grid prompt for multi-panel generation. For SCENE_COVERAGE: each panel gets a different shot from the storyboard. For DIRECTORS_TAKE: 4 framing variations of one shot. For ACTION_BURST: subtle pose variations of one shot. Uses the same reinforcement pattern as casting/wardrobe grids: - Diegetic framing (director's shot selection contact sheet) - Explicit grid structure with BLACK borders - Visual Anchors block - Per-panel assignments with shot-specific data - Mid-prompt grid reinforcement (combats recency bias) Args: shots: List of shot dicts to assign to panels. storyboard: Full storyboard JSON. breakdown: Full breakdown.json dict. project_config: project_config.json dict. grid_type: Which grid strategy to use. character_data: Optional character visual overrides. grid_size: "3x3" or "2x2". Returns: Complete structured grid prompt string. """ character_data = character_data or {} rows, cols = _parse_grid_size(grid_size) total_panels = rows * cols positions = _GRID_POSITIONS_3x3 if grid_size == "3x3" else _GRID_POSITIONS_2x2 sections = [] # ── Grid directive + diegetic frame ─────────────────────────────── sections.append( f"CRITICAL DIRECTIVE: Generate a single image containing a {cols}x{rows} grid of photos.\n" f"DIEGETIC FRAMING: A director's shot selection contact sheet — " f"printed stills from a day's shoot, arranged on a lightbox for scene coverage review.\n" f"Exactly {total_panels} panels arranged in {rows} rows by {cols} columns.\n" f"GRID STRUCTURE: {cols} columns by {rows} rows. Panels edge-to-edge with no gaps, " f"no borders, no margins, no white space. No text, no labels, no numbering, no captions." ) # ── Photographic anchors ────────────────────────────────────────── sections.append("") sections.append( "PHOTOGRAPHIC ANCHORS:\n" f"- Camera: {get_constant('production', 'camera_body')}, {get_constant('production', 'film_stock')}. " f"{get_constant('production', 'film_style_suffix')}.\n" f"- Texture: {get_constant('production', 'grid_texture_guard')}\n" "- Each panel is a separate still photograph from the same scene.\n" "- Consistent lighting, color temperature, and environment across all panels." ) # ── Visual Anchors ──────────────────────────────────────────────── # Use the first shot as the anchor source anchor_shot = shots[0] if shots else {} anchors_block = build_visual_anchors( storyboard, anchor_shot, breakdown, project_config, character_data ) if anchors_block: sections.append("") sections.append(anchors_block) # ── Cinematic baseline ──────────────────────────────────────────── cinematic = storyboard.get("cinematic", "") if cinematic: sections.append("") sections.append(f"Cinematic style: {cinematic}") # ── Panel assignments ───────────────────────────────────────────── sections.append("") sections.append("PANEL ASSIGNMENTS:") if grid_type == GridType.SCENE_COVERAGE: sections.extend( _build_scene_coverage_panels(shots, positions, total_panels, storyboard) ) elif grid_type == GridType.DIRECTORS_TAKE: sections.extend(_build_directors_take_panels(shots, positions)) elif grid_type == GridType.ACTION_BURST: sections.extend(_build_action_burst_panels(shots, positions)) else: # SKIP should not reach here, but handle gracefully sections.append("(Single frame — no grid generation)") # ── Grid reinforcement (mid-prompt, combats recency bias) ───────── sections.append("") sections.append( f"GRID REINFORCEMENT: This image MUST be a {cols}x{rows} grid " f"with exactly {total_panels} panels. Edge-to-edge, no gaps, no borders, " f"no text, no labels. Each panel is a distinct still photograph. " f"Do not merge panels. Do not use any other layout." ) return "\n".join(sections) def build_two_character_prompt( shot: dict, storyboard: dict, char_a_data: dict, char_b_data: dict, project_config: dict = None, ) -> str: """Build prompt for two-character shots with spatial separation. Uses: "LEFT SIDE OF FRAME: [char_a]. RIGHT SIDE OF FRAME: [char_b]." Character descriptions separated by periods, not commas. Args: shot: Single shot dict. storyboard: Full storyboard JSON. char_a_data: Dict with keys 'name', 'visual', 'wardrobe', 'identity_type'. char_b_data: Dict with keys 'name', 'visual', 'wardrobe', 'identity_type'. project_config: project_config.json dict. Returns: Complete two-character prompt string. """ project_config = project_config or {} sections = [] shot_type = _resolve_shot_type(shot.get("shot_type", "MS")) # ── Camera line ─────────────────────────────────────────────────── camera_line = _build_camera_line(shot) sections.append(camera_line) # ── Cinematic baseline ──────────────────────────────────────────── cinematic = storyboard.get("cinematic", "") if cinematic: sections.append(cinematic) # ── Scene context (without character specifics) ─────────────────── atmosphere = shot.get("atmosphere", "") location = storyboard.get("location", "") if location: sections.append(location) if atmosphere: sections.append(atmosphere) # ── Spatial character separation ────────────────────────────────── # Determine spatial assignment from blocking data or default L/R spatial = shot.get("spatial", {}) blocking = spatial.get("blocking", {}) char_a_name = char_a_data.get("name", "Character A") char_b_name = char_b_data.get("name", "Character B") # Determine which character goes on which side char_a_pos = blocking.get(char_a_name.lower(), {}).get("position", "left") char_b_pos = blocking.get(char_b_name.lower(), {}).get("position", "right") # If both default to the same side, force L/R split if char_a_pos == char_b_pos: char_a_pos = "left" char_b_pos = "right" side_label_a = ( "LEFT SIDE OF FRAME" if "left" in char_a_pos else "RIGHT SIDE OF FRAME" ) side_label_b = ( "RIGHT SIDE OF FRAME" if "right" in char_b_pos else "LEFT SIDE OF FRAME" ) # Avoid both on same side label if side_label_a == side_label_b: side_label_a = "LEFT SIDE OF FRAME" side_label_b = "RIGHT SIDE OF FRAME" # Build character A description char_a_visual = char_a_data.get("visual", "") char_a_wardrobe = char_a_data.get("wardrobe", "") char_a_desc = f"{char_a_name}. {char_a_visual}." if char_a_wardrobe: char_a_desc += f" Wearing {char_a_wardrobe}." # Build character B description char_b_visual = char_b_data.get("visual", "") char_b_wardrobe = char_b_data.get("wardrobe", "") char_b_desc = f"{char_b_name}. {char_b_visual}." if char_b_wardrobe: char_b_desc += f" Wearing {char_b_wardrobe}." sections.append(f"{side_label_a}: {char_a_desc}") sections.append(f"{side_label_b}: {char_b_desc}") # ── Action / kinetic layer ──────────────────────────────────────── action = shot.get("action", "") if action: sections.append(action) kinetic = _build_kinetic_layer(shot) if kinetic: sections.append(kinetic) # ── Lighting vector lock ────────────────────────────────────────── lighting = _build_lighting_vector(shot, storyboard) if lighting: sections.append(f"Lighting: {lighting}") # ── Shot-type footer ────────────────────────────────────────────── if shot_type in _WIDE_SHOT_TYPES: sections.append(_build_wide_shot_footer()) elif shot_type in _CLOSE_SHOT_TYPES: sections.append(_build_close_shot_footer()) else: sections.append(get_constant("production", "medium_shot_footer")) # ── Identity locks ──────────────────────────────────────────────── if char_a_data.get("identity_type") == "non_human": sections.append( f"{char_a_name} identity lock: " + get_constant("production", "non_human_identity_lock") ) if char_b_data.get("identity_type") == "non_human": sections.append( f"{char_b_name} identity lock: " + get_constant("production", "non_human_identity_lock") ) # ── Quality guard ───────────────────────────────────────────────── quality_guard = project_config.get("quality_guard", "") if quality_guard: sections.append(quality_guard) sections.append(_no_text_footer()) # Assemble with period-separation (NOT commas) between character blocks prompt = " ".join(s.strip() for s in sections if s and s.strip()) prompt = re.sub(r"\s{2,}", " ", prompt) prompt = re.sub(r"\.\s*\.", ".", prompt) return prompt.strip() # ══════════════════════════════════════════════════════════════════════ # LOCATION / EXPRESSION REFERENCE PROMPTS # ══════════════════════════════════════════════════════════════════════ def build_location_ref_prompt( location_desc: str, lighting_notes: list[str], config: dict, ) -> str: """Build prompt for generating location reference images. Generates a single ENV shot with no people (aspect ratio from config). Used to populate the empty location reference library. Args: location_desc: Location description text (from breakdown.json samples). lighting_notes: Lighting notes from breakdown.json. config: Project config dict. Returns: Complete ENV prompt string. """ sections = [] camera = config.get("camera_body") or get_constant("production", "camera_body") film = config.get("film_stock") or get_constant("production", "film_stock") film_style = config.get("film_style_suffix") or get_constant( "production", "film_style_suffix" ) sections.append(f"Shot on {camera}, {film}, {film_style}") # ENV sanitization clean_desc = sanitize_env_prompt(location_desc) sections.append( f"ENVIRONMENT SHOT — completely empty, no people, no figures. {clean_desc}" ) if lighting_notes: sections.append(f"LIGHTING: {'. '.join(lighting_notes)}") sections.append( "Photorealistic. Cinematic composition. Environmental depth and " "atmospheric haze. Visible grain consistent with film stock. " "No people, no figures, no silhouettes. No text, no labels." ) return "\n\n".join(s for s in sections if s) def build_universal_expression_matrix( emotion_1: str = "joy", emotion_2: str = "anger", emotion_3: str = "sorrow", ) -> str: """Build a 3x3 expression matrix prompt for universal expression library. ADR-C05: Generates a 3×3 grid with 3 emotions (columns) at 3 intensity levels (rows: subtle, active, extreme). Uses a generic, bald, androgynous actor — NOT character-specific — to prevent identity over-baking when expression refs are later combined with character identity refs. Run once per emotion trio for the entire show. 3 trios = 27 expression refs. Cost: $0.039 per grid via Flash 3.1, temperature 0.2. Args: emotion_1: First emotion (column 1). emotion_2: Second emotion (column 2). emotion_3: Third emotion (column 3). Returns: Complete expression matrix prompt string. """ grid_framing = get_constant("casting", "grid_diegetic_framing") expression_subject = get_constant("shared", "universal_expression_subject") background = get_constant("casting", "casting_background") return ( f"CRITICAL DIRECTIVE: Generate a single 3x3 {grid_framing} of facial expressions. " f"Diegetic framing: A clinical facial muscle articulation test.\n\n" f"SUBJECT: {expression_subject} " f"The subject's identity must remain completely sterile and constant across all panels.\n\n" f"GRID STRUCTURE (3x3 MATRIX):\n" f"COLUMN 1: {emotion_1.capitalize()} | COLUMN 2: {emotion_2.capitalize()} | COLUMN 3: {emotion_3.capitalize()}\n" f"ROW 1 (Top): Subtle / Internalized (micro-expressions, eyes only).\n" f"ROW 2 (Middle): Active / Conversational (clear facial engagement).\n" f"ROW 3 (Bottom): Peak / Extreme (maximum physical manifestation).\n\n" f"PHOTOGRAPHIC ANCHORS:\n" f"- Medium: Pure grayscale/monochrome photograph. Close-up strictly on the face.\n" f"- Lighting: Flat, even ring-light. {background}.\n" f"- Texture: Unretouched, extreme detail, visible pores. No color grading. Modern studio lighting.\n\n" f"CRITICAL: Do NOT render any text, labels, captions, or annotations on or around the images. " f"The grid must contain only photographs — no words, no titles, no emotion names, no row/column headers." ) # ══════════════════════════════════════════════════════════════════════ # VIDEO PROMPT BUILDERS # ══════════════════════════════════════════════════════════════════════ def build_video_prompt( shot: dict, storyboard: dict, breakdown: dict, project_config: dict, duration_s: int = 5, character_data: Optional[dict] = None, bible: dict = None, episode: int = 1, ) -> str: """Build a single-shot video prompt (T2V / I2V). Auto-detects plan format and routes to plan-aware builder. Args: shot: Single shot dict (plan or legacy). storyboard: Full storyboard JSON. breakdown: Full breakdown.json dict. project_config: project_config.json dict. duration_s: Video duration in seconds. character_data: Optional character visual overrides. bible: Global bible dict (plan path). episode: Episode number (plan path). Returns: Complete video prompt string. """ # Plan-first if _is_plan_shot(shot): return build_video_prompt_from_plan( shot=shot, bible=bible or breakdown, project_config=project_config or {}, episode=episode, ) character_data = character_data or {} sections = [] # ── Duration header ────────────────────────────────────────────── sections.append(f"CINEMATIC VIDEO SHOT — {duration_s} SECONDS.") # ── Visual Anchors ─────────────────────────────────────────────── anchors = build_visual_anchors( storyboard, shot, breakdown, project_config, character_data ) if anchors: # Replace header to emphasize constancy during motion anchors = anchors.replace( "VISUAL ANCHORS (MUST REMAIN CONSTANT IN ALL PANELS):", "VISUAL ANCHORS (MUST REMAIN CONSTANT):", ) sections.append(anchors) # ── Camera line + movement ─────────────────────────────────────── camera_line = _build_camera_line(shot) camera_movement = shot.get("camera_movement", "") if camera_movement: camera_line += f" Camera movement: {camera_movement}." else: # Infer subtle movement from shot type shot_type = _resolve_shot_type(shot.get("shot_type", "MS")).upper() if shot_type in {"WIDE", "LS", "EWS", "VLS", "ELS"}: camera_line += " Slow, steady establishing move." elif shot_type in {"CU", "ECU", "MCU", "BCU"}: camera_line += " Subtle handheld drift." sections.append(f"CAMERA: {camera_line}") # ── Action (verb-first, no artifact language) ──────────────────── action = shot.get("action", "") if action: sections.append(f"ACTION: {action}") # ── Emotion / performance direction ────────────────────────────── emotion = shot.get("emotion", "") if emotion: sections.append(f"EMOTION: {emotion}") # ── Audio block ────────────────────────────────────────────────── sound_design = shot.get("sound_design", "") atmosphere = shot.get("atmosphere", "") audio_parts = [] if sound_design: audio_parts.append(sound_design) if atmosphere and "sound" in atmosphere.lower(): audio_parts.append(atmosphere) if audio_parts: sections.append(f"AUDIO: {'. '.join(audio_parts)}") # ── Quality footer ─────────────────────────────────────────────── sections.append( "Photorealistic. Cinematic. No text overlays. " "Consistent character identity throughout clip." ) # quality_guard intentionally omitted — image-gen anatomy text is noise for video models prompt = "\n\n".join(s for s in sections if s and s.strip()) _video_model = shot.get("model", "") if _video_model: prompt = _enforce_prompt_length(prompt, _video_model) return prompt def build_multi_shot_prompt( shots: list[dict], storyboard: dict, breakdown: dict, project_config: dict, character_data: Optional[dict] = None, ) -> str: """Build a multi-shot scene batch prompt (SeedDance 2.0). Generates a continuous scene as N distinct shots with shared visual anchors. Used when is_batch_eligible() returns True. Structure (from Gemini consultation Round 3): - Total duration header with shot count - Shared Visual Anchors block - Per-shot sections with type, action, audio Args: shots: List of shot dicts in the scene batch (3-8 shots). storyboard: Full storyboard JSON. breakdown: Full breakdown.json dict. project_config: project_config.json dict. character_data: Optional character visual overrides. Returns: Complete multi-shot prompt string. """ character_data = character_data or {} sections = [] # Calculate total duration (default 5s per shot) default_dur = project_config.get("video", {}).get("default_duration_s", 5) total_duration = sum(shot.get("duration_s", default_dur) for shot in shots) # ── Header ─────────────────────────────────────────────────────── sections.append( f"GENERATE A {total_duration}-SECOND CINEMATIC SCENE " f"IN {len(shots)} DISTINCT SHOTS." ) # ── Shared Visual Anchors ──────────────────────────────────────── anchor_shot = shots[0] if shots else {} anchors = build_visual_anchors( storyboard, anchor_shot, breakdown, project_config, character_data ) if anchors: anchors = anchors.replace( "VISUAL ANCHORS (MUST REMAIN CONSTANT IN ALL PANELS):", "VISUAL ANCHORS (MUST REMAIN CONSTANT THROUGHOUT):", ) sections.append(anchors) # ── Per-shot sections ──────────────────────────────────────────── sections.append("---") for i, shot in enumerate(shots, 1): shot_dur = shot.get("duration_s", default_dur) shot_type = _resolve_shot_type(shot.get("shot_type", "MS")) camera_movement = shot.get("camera_movement", "") action = shot.get("action", "") sound_design = shot.get("sound_design", "") shot_lines = [f"[SHOT {i} - DURATION: {shot_dur}s]"] # Shot type + camera type_line = f"- SHOT TYPE: {shot_type}" if camera_movement: type_line += f", {camera_movement}" shot_lines.append(type_line) # Action if action: shot_lines.append(f"- ACTION: {action}") # Emotion emotion = shot.get("emotion", "") if emotion: shot_lines.append(f"- EMOTION: {emotion}") # Audio if sound_design: shot_lines.append(f"- AUDIO: {sound_design}") sections.append("\n".join(shot_lines)) # ── Quality footer ─────────────────────────────────────────────── sections.append( "Photorealistic. Cinematic. Consistent character identity " "and location across all shots. No text overlays." ) prompt = "\n\n".join(s for s in sections if s and s.strip()) prompt = _enforce_prompt_length(prompt, "seeddance-2.0") return prompt # ══════════════════════════════════════════════════════════════════════ # ENGINE-SPECIFIC VIDEO PROMPT BUILDERS # ══════════════════════════════════════════════════════════════════════ def _truncate_at_natural_break(text: str, max_words: int) -> str: """Truncate text at the last comma, period, or sentence boundary within budget. Avoids mid-phrase cuts like "sharp features and prominent" (missing noun). Prefers ending at ". " then ", " then raw word boundary. """ words = text.split() if len(words) <= max_words: return text.strip().rstrip(".,;") truncated = " ".join(words[:max_words]) for sep in [". ", ", "]: idx = truncated.rfind(sep) if idx > len(truncated) // 3: return truncated[:idx].rstrip(".,; ") return truncated.rstrip(".,; ") def _flatten_lighting_to_prose( prompt_data: dict, max_sources: int = 2, model: str = "", ) -> str: """Flatten structured lighting data into dense prose. Unlike _build_lighting_from_plan() which returns a labeled format, this returns natural language suitable for embedding in prompts. When model is "seeddance-2.0", appends Seedance-preferred lighting anchor phrases derived from the structured source data. Args: prompt_data: Dict containing 'lighting' with 'sources' list. max_sources: Maximum number of light sources to include. model: Target model key. When "seeddance-2.0", appends ranked anchors. """ lighting = prompt_data.get("lighting", {}) sources = lighting.get("sources", []) if not sources: return "" parts = [] for source in sources[:max_sources]: motivator = source.get("motivator", "") direction = source.get("direction", "").lower() quality = source.get("quality", "hard") color_temp = source.get("color_temp", "") intensity = source.get("intensity", "") phrase_parts = [] if color_temp and color_temp != "neutral": phrase_parts.append(color_temp) if quality: phrase_parts.append(quality) phrase_parts.append("light") if direction: phrase_parts.append(f"from {direction}") if motivator: phrase_parts.append(f"({motivator})") if intensity and intensity != "moderate": phrase_parts.append(f"at {intensity} intensity") parts.append(" ".join(phrase_parts)) base_prose = ", ".join(parts) # Seedance-specific: append ranked lighting anchors if model == "seeddance-2.0": anchors = _seedance_lighting_anchors(sources) if anchors: base_prose = f"{base_prose}. {anchors}" return base_prose def _seedance_lighting_anchors(sources: list[dict]) -> str: """Derive Seedance-preferred lighting anchor phrases from structured source data. Analyzes source motivators, quality, and color temperature to select the most relevant ranked terms from the PROMPT_BIBLE lighting vocabulary. Returns at most 2 anchor phrases to preserve word budget. Args: sources: List of lighting source dicts with motivator, direction, quality, color_temp keys. Returns: Comma-separated anchor phrases, or empty string if no signals match. """ try: prompt_rules = get_prompt_rules("seeddance-2.0") lighting_vocab = prompt_rules.get("lighting_vocabulary", {}) signal_map = lighting_vocab.get("signal_map", {}) ranked_terms = lighting_vocab.get("ranked_terms", []) except (KeyError, AttributeError): return "" if not signal_map: return "" signals: set[str] = set() for source in sources: motivator = source.get("motivator", "").lower() quality = source.get("quality", "").lower() color_temp = source.get("color_temp", "").lower() direction = source.get("direction", "").lower() # hard + directional = motivated lighting if quality == "hard" and direction and direction != "ambient": signals.add("hard_directional") # named real-world motivator = practical source if motivator and motivator not in ("ambient", "natural", "available"): signals.add("practical_source") # warm color temperature if color_temp in ("warm", "tungsten", "amber", "golden"): signals.add("warm_color_temp") # soft ambient = atmospheric if quality == "soft" and direction == "ambient": signals.add("atmospheric") # directional from side/back/below = shadow shaping if quality == "hard" and direction in ("side", "back", "below"): signals.add("shadow_shaping") if not signals: return "" # Map signals to anchor phrases, preserve rank order from bible rank_lookup = {term: i for i, term in enumerate(ranked_terms)} anchors = [] for signal_key in signals: phrase = signal_map.get(signal_key, "") if phrase and phrase not in anchors: anchors.append(phrase) # Sort by rank (lower index = higher priority), take top 2 anchors.sort(key=lambda p: rank_lookup.get(p, 999)) return ", ".join(anchors[:2]) def _get_seeddance_film_stock(project_config: dict) -> str: """Return film stock string for Seedance builders. Uses project_config.film_stock if set, otherwise falls back to the PROMPT_BIBLE film_stock_default for seeddance-2.0. """ stock = project_config.get("film_stock", "") if stock: return stock try: return get_prompt_rules("seeddance-2.0").get("film_stock_default", "") except KeyError: return "" def _enforce_single_verb_action(action: str) -> str: """Detect compound actions and simplify to dominant (first) clause. Reads compound_markers from PROMPT_BIBLE action_enforcement config. Soft enforcement: logs warning and returns the first clause. Only active when action_enforcement.mode == "soft" in the bible. Args: action: Raw action string from shot skeleton or kinetic_action. Returns: Simplified single-action string, or original if no compound detected. """ if not action: return action try: enforcement = get_prompt_rules("seeddance-2.0").get("action_enforcement", {}) except KeyError: return action if enforcement.get("mode") != "soft": return action markers = enforcement.get("compound_markers", []) if not markers: return action for marker in markers: idx = action.find(marker) if idx > 0: simplified = action[:idx].strip().rstrip(",") logger.warning( "Seedance T2V: compound action simplified — " "'%s' → '%s' (split at '%s')", action[:80], simplified[:80], marker.strip(), ) return simplified return action # ══════════════════════════════════════════════════════════════════════ # CHARACTER / WARDROBE / SCENE HELPERS (compressed for video budgets) # ══════════════════════════════════════════════════════════════════════ def _build_character_descs_brief( characters: list[dict], bible: dict, episode: int, max_words: int = 30, ) -> str: """Compressed character descriptions for video prompt budgets. Extracts key visual traits from bible (physical appearance, not wardrobe). Returns a prose string under max_words. """ bible_chars = bible.get("characters", {}) descs = [] for char in characters[:2]: # max 2 characters char_id = char.get("char_id", "") bible_char = bible_chars.get(char_id, {}) if not bible_char: continue display_name = bible_char.get("display_name", char_id) visual = bible_char.get("visual_description", "") if visual: per_char = max_words // max(len(characters[:2]), 1) descs.append( f"{display_name}, {_truncate_at_natural_break(visual, per_char)}" ) return ". ".join(descs) + "." if descs else "" def _resolve_wardrobe_brief( characters: list[dict], bible: dict, episode: int, max_words: int = 20, ) -> str: """Resolve wardrobe phase for characters, compressed for video budgets.""" bible_chars = bible.get("characters", {}) wardrobe_parts = [] for char in characters[:2]: char_id = char.get("char_id", "") phase_id = char.get("wardrobe_phase_id", "") bible_char = bible_chars.get(char_id, {}) if not bible_char: continue wardrobe = "" for phase in bible_char.get("phases", []): if phase.get("phase_id") == phase_id: wardrobe = phase.get("wardrobe_description", "") break if not wardrobe: wd = bible_char.get("wardrobe", {}) if isinstance(wd, dict): wardrobe = wd.get(phase_id or "default", wd.get("default", "")) elif isinstance(wd, str): wardrobe = wd if wardrobe: display_name = bible_char.get("display_name", char_id) # Strip character name from start if wardrobe text repeats it w = wardrobe.strip() for prefix in [ f"{display_name}'s ", f"{display_name} ", f"{char_id}'s ", f"{char_id} ", ]: if w.lower().startswith(prefix.lower()): w = w[len(prefix) :] break # Truncate at sentence boundary within word budget per_char = max_words // max(len(characters[:2]), 1) w = _truncate_at_natural_break(w, per_char) wardrobe_parts.append(f"{display_name} — {w}") return ". ".join(wardrobe_parts) + "." if wardrobe_parts else "" def _build_character_anchor_brief( characters: list[dict], bible: dict, max_words: int = 10, ) -> str: """Anti-archetype-drift traits from bible (bare hands, loose hair, etc.).""" bible_chars = bible.get("characters", {}) anchors = [] for char in characters[:1]: # primary character only char_id = char.get("char_id", "") bible_char = bible_chars.get(char_id, {}) if not bible_char: continue # Extract scale_prompt_fragment if available (anti-drift traits) anchor = bible_char.get("scale_prompt_fragment", "") if anchor: anchor_words = anchor.split()[:max_words] anchors.append(" ".join(anchor_words)) return ". ".join(anchors) + "." if anchors else "" def _get_scene_visual_locks_compressed( shot: dict, bible: dict, max_words: int = 15, ) -> str: """Extract scene palette/atmosphere from shot or bible location. Checks shot._scene_visual_locks first (populated by scene planner), then falls back to bible location entry. """ # Try pre-populated scene locks first scene_locks = shot.get("_scene_visual_locks") if scene_locks: lock_parts = [] if scene_locks.get("lighting"): lock_parts.append(scene_locks["lighting"]) if scene_locks.get("atmosphere"): lock_parts.append(scene_locks["atmosphere"]) if scene_locks.get("palette"): lock_parts.append(scene_locks["palette"]) if lock_parts: text = ", ".join(lock_parts) words = text.split()[:max_words] return " ".join(words) # Fallback: derive from bible location asset_data = shot.get("asset_data", {}) location_id = asset_data.get("location_id", "") if location_id and bible: locations = bible.get("locations", {}) loc = locations.get(location_id, {}) atmosphere = loc.get("atmosphere", "") palette = loc.get("color_palette", "") if isinstance(palette, list): palette = ", ".join(str(p) for p in palette) if isinstance(atmosphere, list): atmosphere = ", ".join(str(a) for a in atmosphere) if atmosphere or palette: parts = [p for p in [atmosphere, palette] if p] text = ", ".join(parts) words = text.split()[:max_words] return " ".join(words) return "" # ══════════════════════════════════════════════════════════════════════ # TRANSFORM REGISTRY — Named model-specific post-processors # ══════════════════════════════════════════════════════════════════════ _TRANSFORMS: dict[str, callable] = { "seedance_anchors": _seedance_lighting_anchors, "single_verb_enforce": _enforce_single_verb_action, } # To register a new transform: # 1. Write the function (takes raw data, returns formatted string) # 2. Add it to _TRANSFORMS: _TRANSFORMS["my_transform"] = my_function # 3. Document it in PROMPT_BIBLE enrichment_profile for the relevant model # 4. The builder calls _TRANSFORMS["my_transform"](data) where needed # # Transforms are model-specific formatting applied AFTER CoreSemantics # extraction. They live in the builder code, not in the extraction layer. # ══════════════════════════════════════════════════════════════════════ # CORE SEMANTICS EXTRACTOR (Phase 4 — enrichment parity) # Single extraction point for all builders. Builders become consumers. # ══════════════════════════════════════════════════════════════════════ def extract_core_semantics( shot: dict, bible: dict, project_config: dict, episode: int = 1, ) -> dict: """Extract all semantic content from a shot ONCE for all builders. Returns a flat dict with every enrichment extracted and ready to use. Builders select which keys they need and apply model-specific formatting. This is the single extraction point — no builder should do its own bible lookups or JIT hydration. Args: shot: Plan ShotRecord dict. bible: Global bible dict. project_config: Project config dict. episode: Episode number for wardrobe phase resolution. Returns: Dict with ~20 keys covering all enrichment categories. """ prompt_data = shot.get("prompt_data", {}) skeleton = prompt_data.get("prompt_skeleton", {}) asset_data = shot.get("asset_data", {}) routing_data = shot.get("routing_data", {}) # JIT hydration (resolves {char_*} and {loc_*} tokens from bible) try: from recoil.pipeline._lib.jit_prompt import hydrate_skeleton if skeleton and bible: skeleton = hydrate_skeleton( skeleton, bible, episode=episode, asset_data=asset_data, ) except ImportError: pass # Character enrichments characters = asset_data.get("characters", []) char_desc = "" wardrobe = "" char_anchor = "" if characters and bible: char_desc = _build_character_descs_brief(characters, bible, episode) wardrobe = _resolve_wardrobe_brief(characters, bible, episode) char_anchor = _build_character_anchor_brief(characters, bible) # Scene enrichments scene_locks = "" if bible: scene_locks = _get_scene_visual_locks_compressed(shot, bible) # Film stock (project config override → bible default) film_stock = project_config.get("film_stock", "") if not film_stock: try: film_stock = get_prompt_rules("seeddance-2.0").get("film_stock_default", "") except KeyError: pass # Camera shot_type = _resolve_shot_type(prompt_data.get("shot_type", "MS")) camera_movement = prompt_data.get("camera_movement", "static") focal_length = prompt_data.get("focal_length", "50mm") return { # Core skeleton (hydrated) "subject_line": skeleton.get("subject_line", ""), "action_line": skeleton.get("action_line", ""), "environment_line": skeleton.get("environment_line", ""), "emotion_line": skeleton.get("emotion_line", ""), # Action variants "kinetic_action": prompt_data.get("kinetic_action", ""), # Camera "shot_type": shot_type, "camera_movement": camera_movement, "focal_length": focal_length, # Timing "duration_s": routing_data.get("target_editorial_duration_s", 5), # Identity (T2V/R2V use these; I2V skips them) "character_descs": char_desc, "wardrobe": wardrobe, "character_anchor": char_anchor, "characters": characters, # Scene consistency "scene_visual_locks": scene_locks, # Style (raw data — builders apply model-specific formatting) "film_stock": film_stock, "lighting_data": prompt_data.get("lighting", {}), # Audio "allow_music": project_config.get("allow_music", False), "audio_data": shot.get("audio_data", {}), # Coverage pass context (empty unless injected by caller) "arc_preamble": skeleton.get("arc_context", ""), # Creative direction "director_notes": shot.get("director_notes", ""), # Start/end frames (for I2V routing decisions) "start_frame": ( routing_data.get("start_frame_path") or routing_data.get("start_frame_url") or asset_data.get("start_frame_path") or asset_data.get("start_frame_url") or "" ), "end_frame": ( routing_data.get("end_frame_path") or routing_data.get("end_frame_url") or "" ), # Raw shot dict (for builders that need deep access) "_shot": shot, "_bible": bible, "_project_config": project_config, "_episode": episode, } def build_seedream_prompt( shot: dict, bible: dict, project_config: dict, episode: int = 1, _core_semantics: dict | None = None, look_bundle=None, ) -> str: """Build a concise keyframe prompt for Seedream v4.5/v5 (40-80 words). Seedream uses finite attention — long prompts dilute focus. Uses labeled-roles syntax for multi-character shots. Wardrobe is critical because the keyframe is the visual reference for all downstream I2V and R2V generation. Args: shot: Plan ShotRecord dict. bible: Global bible dict. project_config: Project config dict. episode: Episode number. Returns: Concise prose prompt, 40-80 words. """ cs = _core_semantics or extract_core_semantics( shot, bible, project_config, episode, ) parts = [] # Labeled roles for multi-character shots characters = cs.get("characters", []) if len(characters) > 1 and bible: bible_chars = bible.get("characters", {}) for i, char in enumerate(characters[:3], 1): char_id = char.get("char_id", "") display = bible_chars.get(char_id, {}).get("display_name", char_id) parts.append(f"Figure {i} is {display}.") # Subject (compressed — most important for Seedream attention) if cs["subject_line"]: # Take first 20 words of subject subj_words = cs["subject_line"].split()[:20] parts.append(" ".join(subj_words).rstrip(".") + ".") # Wardrobe (critical — keyframe wardrobe cascades to all video) if cs["wardrobe"]: parts.append(cs["wardrobe"].strip().rstrip(".") + ".") # Action (compressed) action = cs["action_line"] or cs["kinetic_action"] if action: action_words = action.split()[:12] parts.append(" ".join(action_words).rstrip(".") + ".") # Environment (compressed) if cs["environment_line"]: env_words = cs["environment_line"].split()[:15] parts.append(" ".join(env_words).rstrip(".") + ".") # Film stock (anchors color science for downstream video) if cs["film_stock"]: parts.append(f"Shot on {cs['film_stock']}.") # Lighting (brief) lighting = _flatten_lighting_to_prose( {"lighting": cs["lighting_data"]}, max_sources=1, ) if lighting: parts.append(lighting.rstrip(".") + ".") prompt = " ".join(p.strip() for p in parts if p and p.strip()) prompt = re.sub(r"\.{2,}", ".", prompt) # krea2-flora Look/Identity injection (opt-in; no-op when look_bundle is None). prompt = apply_look(prompt, look_bundle) prompt = _enforce_prompt_length(prompt, "seedream-v4.5") return f"{prompt.rstrip()} {_no_text_footer()}".strip() def build_veo_prompt( shot: dict, bible: dict, project_config: dict, episode: int = 1, _core_semantics: dict | None = None, ) -> str: """Build a dense cinematic prose prompt for Veo 3.1 T2V. Dense natural language, max 1500 chars. NO section headers — Veo renders text overlays from them. Camera control embedded in natural language ("slow pull back", "85mm lens"). Uses CoreSemantics for all enrichment data. The _core_semantics param allows the router to pass a pre-extracted dict (avoiding re-extraction when compiling multiple model prompts for the same shot). Args: shot: Plan shot dict with routing_data, prompt_data, etc. bible: Global bible dict. project_config: project_config.json dict. episode: Episode number. _core_semantics: Pre-extracted CoreSemantics dict (optional). Returns: Dense prose prompt string, max 1500 chars. """ cs = _core_semantics or extract_core_semantics( shot, bible, project_config, episode, ) prompt_data = shot.get("prompt_data", {}) # Build camera in natural language (no labels) camera_line = _build_camera_line_plan(prompt_data, model_id="veo-3.1") # Strip trailing period for embedding camera_prose = camera_line.rstrip(".") # Lens-type from cinema mode (Phase 2a) — always emit, even for i2v. _veo_cinema_block = shot.get("cinematography") or {} _veo_mode = None _veo_mode_id = _veo_cinema_block.get("mode") or cs["_project_config"].get("cinema_mode") if _veo_mode_id: try: _veo_mode = load_cinema_modes().get("modes", {}).get(_veo_mode_id) except Exception: # noqa: BLE001 pass _veo_camera_line = render_camera_line(shot, _veo_mode, "veo-3.1") if _veo_camera_line: camera_prose = _veo_camera_line.rstrip(".") # Build dense prose — no section headers parts = [] parts.append(f"Cinematic {cs['duration_s']}-second video clip, {camera_prose}") # Phase 9 (Bug T): focal-length opt-in per PROMPT_BIBLE.yaml. # Default false globally; per-model opt-in via include_focal_length: true. if cs["focal_length"] and _get_include_focal_length("veo-3.1"): parts.append(f"{cs['focal_length']} lens") if cs["environment_line"]: parts.append(cs["environment_line"]) if cs["character_descs"]: parts.append(cs["character_descs"]) if cs["wardrobe"]: parts.append(cs["wardrobe"]) if cs["action_line"]: parts.append(cs["action_line"]) # Lighting flattened to prose lighting_prose = _flatten_lighting_to_prose( {"lighting": cs["lighting_data"]}, ) if lighting_prose: parts.append(lighting_prose) if cs["kinetic_action"]: parts.append(cs["kinetic_action"]) if cs["emotion_line"]: parts.append(cs["emotion_line"]) # Scene visual locks (consistency anchors) if cs["scene_visual_locks"]: parts.append(cs["scene_visual_locks"]) # Arc context from coverage pass if cs["arc_preamble"]: parts.append(cs["arc_preamble"]) # Director notes if cs["director_notes"]: parts.append(cs["director_notes"]) # Film stock / quality (natural language) — overridden by cinema mode. # WHY the start_frame guard: build_veo_prompt is dispatched for BOTH # ("veo-3.1", "t2v") AND ("veo-3.1", "i2v") via the BUILDERS table # (prompt_engine.py ~line 6031-6032). The i2v-skip rule (Opus §Q3 / # PROMPT_BIBLE best_practices line 494) says start-frame shots already # carry the visual look from the image, so cinema tokens (camera/lens/ # filtration/stock/grain/grade) must not be injected. Veo's # `cinema_token_map` currently nulls out body/lens/aperture/shutter # (only color-science tokens survive), but that's a soft mitigation — # if a future contributor adds lens support to Veo's token map this # guard is what enforces the contract. The 5 other builders are # modality-specific (kling_t2v vs kling_i2v are separate functions); # only Veo shares one function across both modalities, hence this guard. _has_start_frame = bool(shot.get("routing_data", {}).get("start_frame_path")) _cinema_block = shot.get("cinematography") or {} _cinema_tokens = "" if not _has_start_frame: _cinema_tokens = render_cinema_tokens( mode_id=_cinema_block.get("mode") or cs["_project_config"].get("cinema_mode"), model_id="veo-3.1", shot_overrides=_cinema_block.get("overrides"), ) if _cinema_tokens: # Veo uses lowercase comma-fragment prose parts.append(_cinema_tokens.lower()) elif cs["film_stock"]: parts.append(f"shot on {cs['film_stock']}") parts.append("photorealistic, high production value") # Constraint block (Phase 2a) — for i2v, filter out era/look constraints. _veo_mode_constraints = (_veo_mode or {}).get("default_constraints") or [] if _veo_mode_constraints: if _has_start_frame: _veo_mode_constraints = [ c for c in _veo_mode_constraints if c not in _ERA_LOOK_CONSTRAINTS ] if _veo_mode_constraints: _pos_suffix, _neg_phrases = render_constraint_block(_veo_mode_constraints, "veo-3.1") if not _pos_suffix and _neg_phrases: _ban = ["; ".join(f"no {frag.strip()}" for frag in p.split(",")) for p in _neg_phrases] _pos_suffix = "Constraints: " + "; ".join(_ban) + "." if _pos_suffix: parts.append(_pos_suffix.lower().rstrip(".")) # Join as dense prose prompt = ", ".join(p.strip().rstrip(".,") for p in parts if p and p.strip()) prompt = prompt[0].upper() + prompt[1:] if prompt else "" prompt += "." # Hard cap at 1500 chars if len(prompt) > 1500: prompt = prompt[:1497] + "..." prompt = _enforce_prompt_length(prompt, "veo-3.1") return prompt def build_kling_i2v_prompt( shot: dict, bible: dict | None = None, project_config: dict | None = None, include_audio_cues: bool = False, ) -> str: """Build an I2V prompt for Kling 3.0 (target 18-25 words, 30 hard max). Motion + camera + director notes. Does NOT include environment_line or subject_line — the NBP keyframe provides all visual context. Text encoder fights image encoder on overlap, causing background melting. Kling I2V optimal range: 18-25 words (30 hard max, API limit: 2500 chars). Focus on what moves and how — the image carries everything else. Audio cues appended after || delimiter, outside the word cap. Args: shot: Plan shot dict with routing_data, prompt_data. include_audio_cues: If True, append audio cues after || delimiter to guide native audio generation. Returns: Prompt string, hard-capped at 30 words (audio after || delimiter). """ from recoil.pipeline._lib.verb_calibration import calibrate_verbs prompt_data = shot.get("prompt_data", {}) skeleton = prompt_data.get("prompt_skeleton", {}) shot_type = _resolve_shot_type(prompt_data.get("shot_type", "MS")) camera_movement = prompt_data.get("camera_movement", "static") kinetic = prompt_data.get("kinetic_action", "") action_line = skeleton.get("action_line", "") sentences = [] # Opening sentence: combine shot type + camera movement naturally shot_label = _SHOT_TYPE_NAMES.get(shot_type, shot_type) if camera_movement and camera_movement != "static": move = _MOVEMENT_NAMES_GERUND.get(camera_movement, camera_movement) sentences.append(f"{shot_label}, {move}.") else: sentences.append(f"{shot_label}.") # Action as a natural sentence if action_line: a = action_line.strip().rstrip(".") sentences.append(a + ".") # Kinetic action if kinetic: k = kinetic.strip().rstrip(".") sentences.append(k + ".") # Director notes (motion choreography — highest priority for I2V) director_notes = shot.get("director_notes", "") if director_notes and director_notes.strip(): dn = director_notes.strip().rstrip(".") sentences.append(dn + ".") # Join as natural language and truncate to word limit from recoil.core.prompt_config import load_constants _limits = load_constants().get("formatter_limits", {}).get("kling_i2v", {}) _word_cap = _limits.get("soft_words") or 30 prompt = " ".join(s for s in sentences if s) # Apply verb calibration (narrative verbs → physics-grounded equivalents) prompt = calibrate_verbs(prompt) words = prompt.split() if len(words) > _word_cap: words = words[:_word_cap] prompt = " ".join(words) # End cleanly on a sentence boundary if possible if "." in prompt: prompt = prompt[: prompt.rfind(".") + 1] else: prompt = prompt.rstrip(".,") + "." else: prompt = " ".join(words) # Endpoint validation — warn if prompt lacks a motion endpoint if not any(m in prompt.lower() for m in _ENDPOINT_MARKERS): logger.debug("I2V prompt lacks explicit endpoint: %s", prompt[:60]) # Append audio cues after || delimiter (outside word cap) if include_audio_cues: audio_data = shot.get("audio_data", {}) ambient = (audio_data.get("ambient_sfx", "") or "").strip() foley = (audio_data.get("foley_action", "") or "").strip() # Foley first (more specific), then ambient — max ~8 words audio_parts = [p for p in (foley, ambient) if p] if audio_parts: prompt += f" || Sound: {', '.join(audio_parts)}." # A2 leak fix (R4) — strip character proper nouns. Single-shot kling i2v # had never run through Option C; identical content-policy risk. characters = shot.get("asset_data", {}).get("characters", []) or [] bible_chars = (bible or {}).get("characters", {}) if isinstance(bible, dict) else {} prompt = _strip_character_names(prompt, characters, bible_chars) prompt = _enforce_prompt_length(prompt, "kling-v3", "i2v") return prompt def build_multi_prompt_sequence( batch: list[dict], batch_char_ids: list[str] = None, has_location_element: bool = False, total_elements: int = 0, prompt_overrides: dict = None, duration_overrides: dict = None, ) -> list[dict]: """Build a multi-prompt sequence from a batch of plan shots. Each shot becomes an {index, prompt, duration} entry for the Kling multi_prompt array. Format confirmed by API probe. If batch_char_ids is provided, injects @Element1/@Element2 references into prompts for fal.ai Elements support. When has_location_element is True, also appends the location @Element ref to every shot's prompt. Args: batch: List of plan shot dicts (with _api_duration from batcher). batch_char_ids: Sorted char IDs for the batch (determines @Element numbering). has_location_element: If True, a location element is the last element in the payload. Its @Element ref will be appended to every prompt. total_elements: Total number of elements in the payload (chars + location). prompt_overrides: Optional dict {shot_id: user_edited_prompt}. When provided, uses the override text instead of building from plan data. duration_overrides: Optional dict {shot_id: duration_seconds}. When provided, uses the override duration instead of _api_duration. Returns: List of {index: int, prompt: str, duration: int} dicts. """ sequence = [] for i, shot in enumerate(batch): shot_id = shot.get("shot_id", "") # Use override prompt if provided, otherwise build from plan if prompt_overrides and shot_id in prompt_overrides: prompt = prompt_overrides[shot_id] else: prompt = build_kling_i2v_prompt(shot) # Inject @Element references if using Elements if batch_char_ids or has_location_element: shot_char_ids = [ c.get("char_id", "") for c in shot.get("asset_data", {}).get("characters", []) if c.get("char_id") ] from recoil.pipeline._lib.elements import ElementManager prompt = ElementManager.inject_element_refs( prompt, shot_char_ids, batch_char_ids or [], has_location_element=has_location_element, total_elements=total_elements, ) # Use override duration if provided, otherwise _api_duration if duration_overrides and shot_id in duration_overrides: duration = duration_overrides[shot_id] else: duration = shot.get("_api_duration") or 5 sequence.append({"index": i + 1, "prompt": prompt, "duration": duration}) return sequence def _apply_editorial_priors( prompt_text: str, generation_config: dict, project: str, focus_character: str = "", shot_type: str = "", location_id: str = "", is_env: bool = False, ) -> tuple[str, dict]: """Apply editorial priors to a prompt and generation config. Returns (modified_prompt, modified_config). Phase 1 scaffolding — no call site yet. Wire into build_coverage_prompts() or the coverage pass prompt builder when editorial_priors.json has data. """ try: from recoil.pipeline._lib.feedback_logger import ( get_editorial_priors, match_priors, ) except ImportError: return prompt_text, generation_config priors = get_editorial_priors(project) if not priors.get("prompt_modifiers_learned") and not priors.get( "parameter_overrides" ): return prompt_text, generation_config prompt_mods, param_overrides = match_priors( priors, focus_character=focus_character, shot_type=shot_type, location_id=location_id, is_env=is_env, ) if prompt_mods: prompt_text = prompt_text.rstrip(". ") + ". " + ". ".join(prompt_mods) + "." if param_overrides: config = dict(generation_config) config.update(param_overrides) return prompt_text, config return prompt_text, generation_config def build_coverage_prompts( shot: dict, framings: list[str] = None, bible: dict = None, project_config: dict = None, include_audio_cues: bool = False, ) -> list[dict]: """Build enriched coverage variant prompts from plan data + Bible. Includes subject, action, environment, emotion, aesthetic tone, and focal length — not just framing + action. Args: shot: Plan shot dict with prompt_data. framings: List of framing codes. Defaults to ["WS", "MS", "CU"]. bible: Global bible dict (for aesthetic_directives). project_config: Project config dict (for film_stock). include_audio_cues: If True, append ambient SFX from audio_data to guide native audio generation. Ambient sound only. Returns: List of {framing: str, prompt: str, duration: int} dicts. """ if framings is None: framings = ["WS", "MS", "CU"] prompt_data = shot.get("prompt_data", {}) skeleton = prompt_data.get("prompt_skeleton", {}) # Extract all available context subject = (skeleton.get("subject_line", "") or "").strip().rstrip(".,;") action = (skeleton.get("action_line", "") or "").strip().rstrip(".,;") environment = (skeleton.get("environment_line", "") or "").strip().rstrip(".,;") emotion = (skeleton.get("emotion_line", "") or "").strip().rstrip(".,;") kinetic = (prompt_data.get("kinetic_action", "") or "").strip().rstrip(".,;") focal_length = prompt_data.get("focal_length", "") # Aesthetic tone from Bible tone = "" if bible: ad = bible.get("aesthetic_directives", {}) tone = (ad.get("tone", "") or "").strip().rstrip(".,;") # Film stock from project config film_stock = "" if project_config: film_stock = (project_config.get("film_stock", "") or "").strip() # Build the core description (subject + action/kinetic) core = subject or action or "character in scene" if kinetic and kinetic != action: core = f"{core}, {kinetic}" # Framing-specific prompt templates — each includes the full context framing_data = { "WS": { "prefix": "Wide shot", "lens": "24mm", "extra": "full environment visible, establishing", }, "FS": { "prefix": "Full shot", "lens": "35mm", "extra": "head to toe, full body visible", }, "MS": {"prefix": "Medium shot", "lens": "50mm", "extra": "waist up"}, "MCU": { "prefix": "Medium close-up", "lens": "65mm", "extra": "chest and face, intimate framing", }, "CU": { "prefix": "Close-up", "lens": "85mm", "extra": "face fills the frame, shallow depth of field", }, "ECU": { "prefix": "Extreme close-up", "lens": "100mm macro", "extra": "eyes and expression only, macro detail", }, "OTS": { "prefix": "Over-the-shoulder shot", "lens": focal_length or "50mm", "extra": "depth layering, foreground shoulder", }, "LOW": { "prefix": "Low angle shot", "lens": "35mm", "extra": "camera below eye level, looking up, imposing", }, "HIGH": { "prefix": "High angle shot", "lens": "35mm", "extra": "camera above, looking down, diminishing", }, "PROFILE": { "prefix": "Profile shot", "lens": "85mm", "extra": "side view, silhouette potential", }, } prompts = [] for framing in framings: fd = framing_data.get( framing, {"prefix": f"{framing} shot", "lens": "50mm", "extra": ""} ) # Assemble the full prompt parts = [fd["prefix"]] # Lens — use framing-appropriate lens, not the shot's original focal length # (a WS at 24mm is more natural than forcing the plan's 50mm on a wide) lens = fd.get("lens", focal_length or "50mm") if lens: parts.append(lens) # Framing-specific description if fd.get("extra"): parts.append(fd["extra"]) # Subject + action parts.append(core) # Emotion/performance if emotion: parts.append(emotion) # Environment (shortened for tighter framings) if environment and framing in ("WS", "FS", "MS"): # Full environment for wider shots parts.append(f"Setting: {environment}") elif environment and framing in ("OTS", "MCU"): # Abbreviated for medium framings env_short = environment.split(",")[0] if "," in environment else environment parts.append(env_short) # CU/ECU skip environment — face is everything # Aesthetic tone if tone: parts.append(tone) # Film stock if film_stock: parts.append(f"shot on {film_stock}") parts.append("photorealistic, cinematic") # Join and clean prompt_text = ", ".join(p.strip() for p in parts if p and p.strip()) prompt_text = prompt_text[0].upper() + prompt_text[1:] if prompt_text else "" prompt_text = prompt_text.rstrip(".,") + "." # Append audio cues if requested if include_audio_cues: audio_data = shot.get("audio_data", {}) ambient = (audio_data.get("ambient_sfx", "") or "").strip() foley = (audio_data.get("foley_action", "") or "").strip() audio_parts = [p for p in (ambient, foley) if p] if audio_parts: prompt_text += f" Sound: {', '.join(audio_parts)}. Ambient sound only." prompts.append( { "framing": framing, "prompt": prompt_text, "duration": 5, } ) return prompts def build_kling_t2v_prompt( shot: dict, bible: dict | None = None, project_config: dict | None = None, episode: int = 1, _core_semantics: dict | None = None, ) -> str: """Build a T2V prompt for Kling 3.0 (75-100 words). Subject + action + kinetic + environment. Active verbs. Direct labeling ("Setting: ...") works well for Kling. Uses CoreSemantics for all enrichment data including character descriptions and film stock. Uses CoreSemantics for all enrichment data. The _core_semantics param allows the router to pass a pre-extracted dict (avoiding re-extraction when compiling multiple model prompts for the same shot). Args: shot: Plan shot dict with routing_data, prompt_data. bible: Global bible dict (optional for backward compat). project_config: Project config dict (optional). episode: Episode number for wardrobe phase resolution. _core_semantics: Pre-extracted CoreSemantics dict (optional). Returns: Prompt string targeting 75-100 words. """ cs = _core_semantics or extract_core_semantics( shot, bible or {}, project_config or {}, episode, ) parts = [] # Shot framing — with lens-type from cinema mode (Phase 2a) _kling_cinema_block = shot.get("cinematography") or {} _kling_mode_id = _kling_cinema_block.get("mode") or cs["_project_config"].get("cinema_mode") _kling_mode = None if _kling_mode_id: try: _kling_mode = load_cinema_modes().get("modes", {}).get(_kling_mode_id) except Exception: # noqa: BLE001 pass _kling_camera_line = render_camera_line(shot, _kling_mode, "kling-v3") if _kling_camera_line: parts.append(f"{_kling_camera_line.rstrip('.')} {cs['duration_s']} seconds.") else: type_name = _SHOT_TYPE_NAMES.get(cs["shot_type"], cs["shot_type"]) frame_line = f"{type_name}, {cs['duration_s']} seconds" if cs["camera_movement"] and cs["camera_movement"] != "static": frame_line += ( f", {_MOVEMENT_NAMES.get(cs['camera_movement'], cs['camera_movement'])}" ) parts.append(frame_line + ".") # Subject (direct labeling) if cs["subject_line"]: parts.append(f"Subject: {cs['subject_line']}.") # Action (active verbs) if cs["action_line"]: parts.append(f"Action: {cs['action_line']}.") # Kinetic layer if cs["kinetic_action"]: parts.append(f"Motion: {cs['kinetic_action']}.") # Environment if cs["environment_line"]: parts.append(f"Setting: {cs['environment_line']}.") # Emotion/mood if cs["emotion_line"]: parts.append(f"Mood: {cs['emotion_line']}.") # Arc context from coverage pass (if present and budget allows) if cs["arc_preamble"]: parts.append(f"Arc: {cs['arc_preamble'].strip().rstrip('.')}.") # Character description from CoreSemantics if cs["character_descs"]: parts.append(f"Character: {cs['character_descs']}.") # Wardrobe from CoreSemantics if cs["wardrobe"]: parts.append(f"Wearing: {cs['wardrobe']}.") # Film stock from CoreSemantics — overridden by cinema mode when active. _cinema_block = shot.get("cinematography") or {} _cinema_tokens = render_cinema_tokens( mode_id=_cinema_block.get("mode") or cs["_project_config"].get("cinema_mode"), model_id="kling-v3", shot_overrides=_cinema_block.get("overrides"), ) if _cinema_tokens: parts.append(f"{_cinema_tokens}.") elif cs["film_stock"]: parts.append(f"Shot on {cs['film_stock']}.") # Lighting (brief) lighting_prose = _flatten_lighting_to_prose( {"lighting": cs["lighting_data"]}, ) if lighting_prose: parts.append(f"Lighting: {lighting_prose}.") # Scene visual locks from CoreSemantics if cs["scene_visual_locks"]: parts.append(f"Scene: {cs['scene_visual_locks']}.") # Director notes if cs["director_notes"]: parts.append(cs["director_notes"].strip().rstrip(".") + ".") parts.append("Cinematic, photorealistic, high production value.") # Constraint block (Phase 2a) _kling_mode_constraints = (_kling_mode or {}).get("default_constraints") or [] if _kling_mode_constraints: _pos_suffix, _neg_phrases = render_constraint_block(_kling_mode_constraints, "kling-v3") if not _pos_suffix and _neg_phrases: _ban = ["; ".join(f"no {frag.strip()}" for frag in p.split(",")) for p in _neg_phrases] _pos_suffix = "Constraints: " + "; ".join(_ban) + "." if _pos_suffix: parts.append(_pos_suffix) prompt = " ".join(p.strip() for p in parts if p and p.strip()) # Clean up double periods prompt = re.sub(r"\.{2,}", ".", prompt) prompt = _enforce_prompt_length(prompt, "kling-v3", "t2v") return prompt # ══════════════════════════════════════════════════════════════════════ # PROMPTPACKAGE ROUTER (ADR-R05) # Single entry point that compiles all model-specific prompts at once. # ══════════════════════════════════════════════════════════════════════ def compile_core_semantics(shot: dict, bible: dict) -> dict: """Extract unified semantic dict from a plan shot's 5 consumer groups. This is the single source of truth that all model-specific builders read from. Avoids re-parsing the same data in each builder. Args: shot: Plan ShotRecord dict (has prompt_data, routing_data, etc.) bible: Global bible dict. Returns: Unified semantic dict with normalized fields. """ prompt_data = shot.get("prompt_data", {}) routing_data = shot.get("routing_data", {}) asset_data = shot.get("asset_data", {}) spatial_data = shot.get("spatial_data", {}) audio_data = shot.get("audio_data", {}) skeleton = prompt_data.get("prompt_skeleton", {}) # Resolve characters from bible characters = asset_data.get("characters", []) resolved_chars = [] bible_chars = bible.get("characters", {}) for char in characters: char_id = char.get("char_id", "") if isinstance(char, dict) else str(char) bible_char = bible_chars.get(char_id, {}) resolved_chars.append( { "char_id": char_id, "display_name": bible_char.get("display_name", char_id), "visual_description": bible_char.get("visual_description", ""), "wardrobe_phase_id": char.get("wardrobe_phase_id", "") if isinstance(char, dict) else "", "emotion_keyword": char.get("emotion_keyword", "neutral") if isinstance(char, dict) else "neutral", "screen_position": char.get("screen_position", "center") if isinstance(char, dict) else "center", "is_non_human": _visual_is_non_human( bible_char.get("visual_description", "") ), } ) return { # Routing "shot_id": shot.get("shot_id", ""), "pipeline": shot.get("pipeline", "still"), "model": shot.get("model", ""), "is_env": routing_data.get("is_env_only", False), "has_dialogue": routing_data.get("has_dialogue", False), "duration_s": routing_data.get("target_editorial_duration_s", 5), "num_characters": routing_data.get("num_characters", 0), # Camera "shot_type": _resolve_shot_type(prompt_data.get("shot_type", "MS")), "camera_movement": prompt_data.get("camera_movement", "static"), "focal_length": prompt_data.get("focal_length", "50mm"), "kinetic_action": prompt_data.get("kinetic_action", ""), # Skeleton "subject_line": skeleton.get("subject_line", ""), "environment_line": skeleton.get("environment_line", ""), "action_line": skeleton.get("action_line", ""), "emotion_line": skeleton.get("emotion_line", ""), "camera_line": skeleton.get("camera_line", ""), # Lighting "lighting": prompt_data.get("lighting", {}), # Characters (resolved) "characters": resolved_chars, # Spatial "screen_direction": spatial_data.get("screen_direction", "center"), "character_relationships": spatial_data.get("character_relationships", {}), # Audio "dialogue": audio_data.get("dialogue", []), "ambient_sfx": audio_data.get("ambient_sfx", ""), "foley_action": audio_data.get("foley_action", ""), # Asset "location_id": asset_data.get("location_id", ""), "time_of_day": asset_data.get("time_of_day", "day"), "visual_mode": asset_data.get("visual_mode", "reality"), "props": asset_data.get("props", []), } def build_wan_i2v_prompt( shot: dict, bible: dict | None = None, project_config: dict | None = None, episode: int = 1, has_end_frame: bool = False, include_audio_cues: bool = False, prompt_style: str = "balanced", _core_semantics: dict | None = None, ) -> str: """Build an I2V prompt for Wan 2.7 (target 150-300 words, 1800 char cap). Unlike Kling I2V (30 words max due to encoder fighting), Wan uses DiT with T5-XXL — craves dense text. Include environment and lighting as reinforcement to prevent background drift. Uses CoreSemantics for all enrichment data. The _core_semantics param allows the router to pass a pre-extracted dict (avoiding re-extraction when compiling multiple model prompts for the same shot). When has_end_frame=True (In Between mode): describes the transition path between two known visual states, not the destination. prompt_style controls density: - "directed": 250-300 words, all fields, explicit endpoints - "balanced": 150-200 words, camera+action+emotion+lighting (default) - "open": 50-80 words, shot type + action_line only, evocative Args: shot: Plan shot dict with routing_data, prompt_data. bible: Global bible dict (optional for backward compat). project_config: Project config dict (optional). episode: Episode number for wardrobe phase resolution. has_end_frame: If True, switch to transition-path mode. include_audio_cues: If True, append audio cues after || delimiter. prompt_style: "directed", "balanced", or "open". _core_semantics: Pre-extracted CoreSemantics dict (optional). Returns: Prompt string, soft-capped at word limit per style. """ try: from recoil.pipeline._lib.verb_calibration import calibrate_verbs except ImportError: def calibrate_verbs(x): # No-op if not available return x from recoil.core.prompt_config import load_constants cs = _core_semantics or extract_core_semantics( shot, bible or {}, project_config or {}, episode, ) prompt_data = shot.get("prompt_data", {}) # Camera line (still built from prompt_data — model-specific formatting) camera_line = _build_camera_line_plan(prompt_data, model_id="wan-2.7-i2v") # Lighting lighting_prose = _flatten_lighting_to_prose( {"lighting": cs["lighting_data"]}, ) # Style-dependent word limits _limits = load_constants().get("formatter_limits", {}) if has_end_frame: limit_key = "wan_between" else: limit_key = "wan_i2v" limits = _limits.get(limit_key, {}) hard_words = limits.get("hard_words", 300) sections = [] if has_end_frame: # ── IN BETWEEN MODE: transition path description ── sections.append(f"Cinematic {cs['duration_s']}-second video transition.") sections.append(f"Camera: {camera_line}") if prompt_style != "open": # Transition directive (not action destination) if cs["action_line"]: a = cs["action_line"].strip().rstrip(".") sections.append( f"Continuous smooth motion. {a}. " "The movement bridges the starting and ending visual states." ) if cs["kinetic_action"] and prompt_style == "directed": sections.append( f"Transition dynamics: {cs['kinetic_action'].strip().rstrip('.')}." ) if cs["emotion_line"]: sections.append( f"Mood shifts: {cs['emotion_line'].strip().rstrip('.')}." ) if cs["environment_line"]: sections.append( f"Setting remains constant throughout: {cs['environment_line'].strip().rstrip('.')}." ) if lighting_prose and prompt_style == "directed": sections.append(f"Lighting: {lighting_prose}") else: # Open style — minimal if cs["action_line"]: sections.append(cs["action_line"].strip()) sections.append("Cinematic, smooth temporal interpolation.") else: # ── STANDARD I2V MODE: action description ── sections.append(f"Cinematic {cs['duration_s']}-second video clip.") sections.append(f"Camera: {camera_line}") if cs["action_line"]: sections.append(cs["action_line"].strip().rstrip(".") + ".") if prompt_style in ("balanced", "directed"): if cs["kinetic_action"]: sections.append(cs["kinetic_action"].strip().rstrip(".") + ".") if cs["emotion_line"]: sections.append(cs["emotion_line"].strip().rstrip(".") + ".") if prompt_style == "directed": if cs["director_notes"]: sections.append( f"Direction: {cs['director_notes'].strip().rstrip('.')}." ) if cs["subject_line"]: sections.append(cs["subject_line"].strip().rstrip(".") + ".") if prompt_style in ("balanced", "directed"): # Environment as 1-sentence drift anchor (frame carries primary visuals) if cs["environment_line"]: # Truncate to first sentence for reinforcement env_first = cs["environment_line"].split(".")[0].strip() sections.append(f"Setting: {env_first}.") if lighting_prose: sections.append(f"Lighting: {lighting_prose}") # Arc context from coverage pass (Wan has budget for this) if cs["arc_preamble"]: sections.append(cs["arc_preamble"]) # Character reinforcement (Wan's T5-XXL encoder benefits from text that # matches image content — prevents identity/wardrobe drift during motion) # Unlike Kling/Seedance I2V, Wan benefits from text reinforcement. if cs["character_descs"]: sections.append( f"The figure is {_truncate_at_natural_break(cs['character_descs'], 20)}." ) if cs["wardrobe"]: sections.append(f"{_truncate_at_natural_break(cs['wardrobe'], 12)}.") # Film stock reinforcement if cs["film_stock"]: sections.append(f"Shot on {cs['film_stock']}.") sections.append("Cinematic, high production value.") prompt = " ".join(s for s in sections if s and s.strip()) # Apply verb calibration prompt = calibrate_verbs(prompt) # Word cap words = prompt.split() if len(words) > hard_words: words = words[:hard_words] prompt = " ".join(words) if "." in prompt: prompt = prompt[: prompt.rfind(".") + 1] else: prompt = prompt.rstrip(".,") + "." # Character cap hard_chars = limits.get("hard_chars", 1800) if len(prompt) > hard_chars: prompt = prompt[:hard_chars] if "." in prompt: prompt = prompt[: prompt.rfind(".") + 1] else: prompt = prompt.rstrip(".,") + "." # Audio cues (outside word cap, after || delimiter) if include_audio_cues: audio = cs["audio_data"] ambient = audio.get("ambient_sfx", "") foley = audio.get("foley_action", "") audio_parts = [p for p in [ambient, foley] if p] if audio_parts: prompt += f" || Audio: {'; '.join(audio_parts)}" prompt = _enforce_prompt_length(prompt, "wan-2.7-i2v") return prompt # ══════════════════════════════════════════════════════════════════════ # WAN 2.7 R2V PROMPT (Reference-to-Video — no frame, prompt carries ALL) # ══════════════════════════════════════════════════════════════════════ def build_wan_r2v_prompt( shots: list[dict], bible: dict, project_config: dict, episode: int = 1, multi_shots: bool = False, _core_semantics: dict | None = None, ) -> str: """Build an R2V prompt for Wan 2.7 (250-400 words, 2400 char cap). R2V has NO start frame — identity from reference images. The prompt must carry ALL visual context: environment, lighting, action, wardrobe. Uses CoreSemantics for all enrichment data. The _core_semantics param allows the router to pass a pre-extracted dict for the primary shot (avoiding re-extraction when compiling multiple model prompts for the same shot). For multi-shot, per-shot data (action, camera, duration) is still extracted per-shot from the shots list. When multi_shots=True, compiles a scene-level prompt from multiple shots using beat structure (model decides camera changes internally). Args: shots: List of plan shot dicts. Single-shot: [shot]. Multi: [shot1, shot2, ...]. bible: Global bible dict (character visual descriptions, locations). project_config: Project config dict (film_stock, etc.). episode: Episode number. multi_shots: If True, build scene-level beat prompt. _core_semantics: Pre-extracted CoreSemantics dict for the primary shot (optional). Used for wardrobe, scene_visual_locks, film_stock, lighting, arc_preamble, environment_line, character_descs, etc. Returns: Dense prompt string, soft-capped at 400 words / 2400 chars. """ try: from recoil.pipeline._lib.verb_calibration import calibrate_verbs except ImportError: def calibrate_verbs(x): # No-op if not available return x from recoil.core.prompt_config import load_constants if not shots: return "" _limits = load_constants().get("formatter_limits", {}).get("wan_r2v", {}) hard_words = _limits.get("hard_words", 400) primary_shot = shots[0] cs = _core_semantics or extract_core_semantics( primary_shot, bible, project_config, episode, ) prompt_data = primary_shot.get("prompt_data", {}) lighting_prose = _flatten_lighting_to_prose(prompt_data) sections = [] if multi_shots and len(shots) > 1: # ── MULTI-SHOT SCENE PROMPT ── total_duration = sum( s.get("routing_data", {}).get("target_editorial_duration_s", 5) for s in shots ) sections.append(f"Cinematic {total_duration}-second multi-shot sequence.") sections.append( f"Setting: {cs['environment_line']}" if cs["environment_line"] else "" ) if lighting_prose: sections.append(f"Lighting: {lighting_prose}") # Character descriptions from bible char_lines = _build_character_lines(shots, bible) if char_lines: sections.append("Characters:") sections.extend(char_lines) # Wardrobe from CoreSemantics if cs["wardrobe"]: sections.append(cs["wardrobe"].strip().rstrip(".") + ".") # Scene visual locks from CoreSemantics if cs["scene_visual_locks"]: sections.append(cs["scene_visual_locks"]) # Beat structure (hydrate each beat's skeleton individually) sections.append("Action sequence:") for i, s in enumerate(shots): s_skeleton = s.get("prompt_data", {}).get("prompt_skeleton", {}) # JIT hydrate per-beat skeleton s_skeleton = _maybe_hydrate( s_skeleton, bible, episode=episode, asset_data=s.get("asset_data") ) s_action = s_skeleton.get("action_line", "") s_kinetic = s.get("prompt_data", {}).get("kinetic_action", "") s_camera = _build_camera_line_plan( s.get("prompt_data", {}), model_id="wan-2.7-r2v" ) beat_parts = [f"Beat {i + 1} ({s_camera}):"] if s_action: beat_parts.append(s_action.strip().rstrip(".") + ".") if s_kinetic: beat_parts.append(s_kinetic.strip().rstrip(".") + ".") sections.append(" ".join(beat_parts)) # Camera flow summary shot_types = [ _resolve_shot_type(s.get("prompt_data", {}).get("shot_type", "MS")) for s in shots ] type_names = [_SHOT_TYPE_NAMES.get(st, st) for st in shot_types] sections.append(f"Camera progresses through: {', then '.join(type_names)}.") else: # ── SINGLE-SHOT PROMPT ── camera_line = _build_camera_line_plan(prompt_data, model_id="wan-2.7-r2v") sections.append(f"Cinematic {cs['duration_s']}-second video clip.") sections.append(f"Camera: {camera_line}") sections.append( f"Setting: {cs['environment_line']}" if cs["environment_line"] else "" ) # Character descriptions (R2V needs these — no frame carries identity) char_lines = _build_character_lines(shots, bible) if char_lines: sections.append("Characters:") sections.extend(char_lines) # Wardrobe from CoreSemantics if cs["wardrobe"]: sections.append(cs["wardrobe"].strip().rstrip(".") + ".") # Scene visual locks from CoreSemantics if cs["scene_visual_locks"]: sections.append(cs["scene_visual_locks"]) if cs["action_line"]: sections.append(cs["action_line"].strip()) if cs["kinetic_action"]: sections.append(cs["kinetic_action"].strip()) if cs["emotion_line"]: sections.append(f"Mood: {cs['emotion_line'].strip()}") if lighting_prose: sections.append(f"Lighting: {lighting_prose}") # Arc context from coverage pass (CoreSemantics) if cs["arc_preamble"]: sections.insert(1, cs["arc_preamble"]) # After opening, before beats # Director notes (aggregated, deduplicated) all_notes = list( dict.fromkeys( s.get("director_notes", "").strip() for s in shots if s.get("director_notes", "").strip() ) ) if all_notes: sections.append(f"Direction: {' '.join(all_notes)}") # Film stock from CoreSemantics — overridden by cinema mode when active. _cinema_block = (shots[0].get("cinematography") if shots else None) or {} _cinema_tokens = render_cinema_tokens( mode_id=_cinema_block.get("mode") or cs["_project_config"].get("cinema_mode"), model_id="wan-2.7-r2v", shot_overrides=_cinema_block.get("overrides"), ) if _cinema_tokens: sections.append(f"{_cinema_tokens}.") elif cs["film_stock"]: sections.append(f"Shot on {cs['film_stock']}.") sections.append("Photorealistic, cinematic, high production value.") # Constraint block (Phase 2a) _wan_mode_constraints = [] _wan_mode_id = _cinema_block.get("mode") or cs["_project_config"].get("cinema_mode") if _wan_mode_id: try: _wan_mode = load_cinema_modes().get("modes", {}).get(_wan_mode_id) if _wan_mode: _wan_mode_constraints = _wan_mode.get("default_constraints") or [] except Exception: # noqa: BLE001 pass if _wan_mode_constraints: _pos_suffix, _neg_phrases = render_constraint_block(_wan_mode_constraints, "wan-2.7-r2v") if not _pos_suffix and _neg_phrases: _ban = ["; ".join(f"no {frag.strip()}" for frag in p.split(",")) for p in _neg_phrases] _pos_suffix = "Constraints: " + "; ".join(_ban) + "." if _pos_suffix: sections.append(_pos_suffix) prompt = " ".join(s for s in sections if s and s.strip()) # Apply verb calibration prompt = calibrate_verbs(prompt) # Word cap words = prompt.split() if len(words) > hard_words: words = words[:hard_words] prompt = " ".join(words) if "." in prompt: prompt = prompt[: prompt.rfind(".") + 1] else: prompt = prompt.rstrip(".,") + "." # Character cap hard_chars = _limits.get("hard_chars", 2400) if len(prompt) > hard_chars: prompt = prompt[:hard_chars] if "." in prompt: prompt = prompt[: prompt.rfind(".") + 1] else: prompt = prompt.rstrip(".,") + "." prompt = _enforce_prompt_length(prompt, "wan-2.7-r2v") return prompt def _build_character_lines(shots: list[dict], bible: dict) -> list[str]: """Build character description lines for R2V prompts from bible data. Iterates ALL shots to capture characters introduced in later beats. Deduplicates by char_id — first occurrence's emotion wins. """ lines = [] seen_chars = set() bible_chars = bible.get("characters", {}) if bible else {} for s in shots: characters = s.get("asset_data", {}).get("characters", []) for char_entry in characters: char_id = char_entry.get("char_id", "") if not char_id or char_id in seen_chars: continue seen_chars.add(char_id) emotion = char_entry.get("emotion_keyword", "") bible_char = bible_chars.get(char_id, {}) display_name = bible_char.get("display_name", char_id.capitalize()) visual_desc = bible_char.get("visual_description", "") parts = [f"- {display_name}:"] if visual_desc: parts.append(visual_desc.strip().rstrip(".") + ".") if emotion: parts.append(f"Emotion: {emotion}.") lines.append(" ".join(parts)) return lines # Alias for external callers expecting scene-level compilation compile_wan_r2v_scene = build_wan_r2v_prompt # ══════════════════════════════════════════════════════════════════════ # SEEDDANCE R2V PROMPT (Reference-to-Video — 3-zone prompt) # ══════════════════════════════════════════════════════════════════════ def build_seeddance_r2v_prompt( shots: list[dict], bible: dict, project_config: dict, episode: int = 1, ref_manifest: dict | None = None, anchor_duration_s: int = 1, _core_semantics: dict | None = None, ) -> str: """Build a 3-zone R2V prompt for SeedDance from plan shots + bible data. Uses CoreSemantics for enrichment data from the primary shot. The _core_semantics param allows the router to pass a pre-extracted dict (avoiding re-extraction when compiling multiple model prompts for the same shot). Zone A — Reference declarations (@ImageN anchors) + wardrobe Zone B — Shot list (camera, action, duration per shot) Zone C — Global constraints (style, ambient audio, scene visual locks) When ref_manifest is provided, @Image tokens resolve to concrete indices. When None, emits placeholder tokens (@Image{identity_1}, @Image{scene_1}) for the assembler to resolve later. Args: shots: List of plan shot dicts. Single-shot: [shot]. Multi: [shot1, ...]. bible: Global bible dict (characters, locations). project_config: Project config dict (film_stock, style, etc.). episode: Episode number. ref_manifest: Maps ref_type to @ImageN index, e.g. {"identity_1": 1, "identity_2": 2, "scene_1": 3}. If None, placeholder tokens are emitted. anchor_duration_s: Duration override for Shot 1 (default 1s to lock identity without copying refs). 0 or None uses natural duration. _core_semantics: Pre-extracted CoreSemantics dict for the primary shot (optional). Used for wardrobe, scene_visual_locks, film_stock, lighting, arc_preamble, environment_line, and allow_music. Returns: Prompt string with Zone A, Zone B, Zone C sections. """ if not shots: return "" primary_shot = shots[0] cs = _core_semantics or extract_core_semantics( primary_shot, bible, project_config, episode, ) prompt_data = primary_shot.get("prompt_data", {}) skeleton = prompt_data.get("prompt_skeleton", {}) asset_data = primary_shot.get("asset_data", {}) # JIT hydration (same pattern as build_wan_r2v_prompt) skeleton = _maybe_hydrate(skeleton, bible, episode=episode, asset_data=asset_data) bible_chars = bible.get("characters", {}) if bible else {} bible_locs = bible.get("locations", {}) if bible else {} # ── Zone A — Reference declarations ── zone_a_lines = [] seen_chars = [] seen_locs = [] # Collect characters across all shots for s in shots: characters = s.get("asset_data", {}).get("characters", []) for char_entry in characters: char_id = char_entry.get("char_id", "") if not char_id or char_id in seen_chars: continue seen_chars.append(char_id) bible_char = bible_chars.get(char_id, {}) display_name = bible_char.get( "display_name", char_id.replace("-", " ").title() ) visual_desc = bible_char.get("visual_description", "") # Truncate to 2-3 key traits traits = visual_desc.strip().rstrip(".") if visual_desc else "character" ref_type_key = f"identity_{len(seen_chars)}" if ref_manifest and ref_type_key in ref_manifest: token = f"@Image{ref_manifest[ref_type_key]}" else: token = f"@Image{{{ref_type_key}}}" zone_a_lines.append(f"{token} is {display_name} — {traits}.") # Wardrobe from CoreSemantics (enriches Zone A character declarations) if cs["wardrobe"]: zone_a_lines.append(cs["wardrobe"].strip().rstrip(".") + ".") # Collect locations location_id = primary_shot.get("asset_data", {}).get("location_id", "") if not location_id: # Try to extract from environment_line env_line = skeleton.get("environment_line", "") if env_line: location_id = ( env_line.split(",")[0].strip().lower().replace(" ", "-") if env_line else "" ) for s in shots: s_location = s.get("asset_data", {}).get("location_id", location_id) if not s_location or s_location in seen_locs: continue seen_locs.append(s_location) bible_loc = bible_locs.get(s_location, {}) loc_name = bible_loc.get("display_name", s_location.replace("-", " ").title()) loc_desc = bible_loc.get( "spatial_description", bible_loc.get("description", "") ) spatial = loc_desc.strip().rstrip(".") if loc_desc else "location" loc_count = len(seen_locs) ref_type_key = f"scene_{loc_count}" if ref_manifest and ref_type_key in ref_manifest: token = f"@Image{ref_manifest[ref_type_key]}" else: token = f"@Image{{{ref_type_key}}}" zone_a_lines.append(f"{token} is {loc_name} — {spatial}.") # ── Zone B — Shot list ── zone_b_lines = [] missing_motion_line_shots: list[int] = [] # Pre-load cinema modes once for the entire shot loop (Phase 2a). try: _all_cinema_modes = load_cinema_modes().get("modes", {}) except Exception: # noqa: BLE001 _all_cinema_modes = {} for i, s in enumerate(shots): s_prompt_data = s.get("prompt_data", {}) s_skeleton = s_prompt_data.get("prompt_skeleton", {}) s_routing = s.get("routing_data", {}) # JIT hydrate per-shot skeleton (skip i==0 — already hydrated above) if i > 0: s_skeleton = _maybe_hydrate( s_skeleton, bible, episode=episode, asset_data=s.get("asset_data") ) shot_type = _resolve_shot_type(s_prompt_data.get("shot_type", "MS")) type_name = _SHOT_TYPE_NAMES.get(shot_type, f"{shot_type} shot") focal_length = s_prompt_data.get("focal_length", "50mm") camera_movement = s_prompt_data.get("camera_movement", "static") movement_name = ( _MOVEMENT_NAMES.get(camera_movement, camera_movement) if camera_movement and camera_movement != "static" else "Static" ) action_line = s_skeleton.get("action_line", "") motion_line = s_skeleton.get("motion_line", "") or "" motion_line_text = str(motion_line).strip() emotion_line = s_skeleton.get("emotion_line", "") subject_line = s_skeleton.get("subject_line", "") if not motion_line_text: missing_motion_line_shots.append(i + 1) # Duration: anchor override for Shot 1 natural_duration = s_routing.get("target_editorial_duration_s", 5) if i == 0 and anchor_duration_s: duration = anchor_duration_s else: duration = natural_duration # Build subject reference tokens s_chars = s.get("asset_data", {}).get("characters", []) subject_parts = [] for char_entry in s_chars: cid = _char_id_from_entry(char_entry).strip().upper() if cid: bc = bible_chars.get(cid, {}) subject_parts.append( bc.get("display_name", cid.replace("-", " ").title()) ) # Find location token for this shot s_loc = s.get("asset_data", {}).get("location_id", location_id) loc_ref = "" if s_loc: loc_idx = list(seen_locs).index(s_loc) + 1 if s_loc in seen_locs else 0 if loc_idx: ref_type_key = f"scene_{loc_idx}" if ref_manifest and ref_type_key in ref_manifest: loc_ref = f"@Image{ref_manifest[ref_type_key]}" else: loc_ref = f"@Image{{{ref_type_key}}}" # Build shot line shot_num = i + 1 parts = [f"Shot {shot_num}:"] # Camera line with lens-type from cinema mode (Phase 2a). _s_cinema_cam = s.get("cinematography") or {} _s_mode_id_cam = _s_cinema_cam.get("mode") or cs["_project_config"].get("cinema_mode") _s_mode_for_cam = _all_cinema_modes.get(_s_mode_id_cam) if _s_mode_id_cam else None _s_camera_line = render_camera_line(s, _s_mode_for_cam, "seeddance-2.0") if _s_camera_line: parts.append(_s_camera_line) elif camera_movement and camera_movement != "static": focal_clause = ( f", {focal_length}" if _get_include_focal_length("seeddance-2.0") else "" ) parts.append(f"{movement_name} {type_name}{focal_clause}.") else: focal_clause = ( f", {focal_length}" if _get_include_focal_length("seeddance-2.0") else "" ) parts.append(f"Static {type_name}{focal_clause}.") # Transition for non-first shots if i > 0: subject_str = subject_parts[0] if subject_parts else "subject" parts.append(f"Cut to {subject_str}.") elif subject_parts and loc_ref: # Anchor shot: character in location char_tokens = [] for idx, cid in enumerate(seen_chars): ref_key = f"identity_{idx + 1}" if ref_manifest and ref_key in ref_manifest: char_tokens.append(f"@Image{ref_manifest[ref_key]}") else: char_tokens.append(f"@Image{{{ref_key}}}") if char_tokens: parts.append(f"{subject_parts[0]} from {char_tokens[0]} in {loc_ref}.") # Subject (emitted first, per PROMPT_BIBLE multi_shot order) if subject_line: parts.append(subject_line.strip().rstrip(".") + ".") # Action (one action per shot) if action_line: parts.append(action_line.strip().rstrip(".") + ".") if motion_line_text: parts.append(motion_line_text.rstrip(".") + ".") # Emotion (one per shot) if emotion_line: parts.append(emotion_line.strip().rstrip(".") + ".") # Duration at end parts.append(f"{duration}s.") zone_b_lines.append(" ".join(parts)) if missing_motion_line_shots: logger.warning( "r2v: %d/%d shots missing motion_line (shot numbers: %s)", len(missing_motion_line_shots), len(shots), ", ".join(str(n) for n in missing_motion_line_shots), ) # Arc context from CoreSemantics (coverage pass context) arc_context = cs["arc_preamble"] if arc_context: zone_b_lines.insert(0, arc_context.strip().rstrip(".") + ".") # ── Zone C — Global constraints ── zone_c_parts = [] film_stock = cs["film_stock"] lighting_prose = _flatten_lighting_to_prose( {"lighting": cs["lighting_data"]}, model="seeddance-2.0", ) style_parts = [] # Cinema mode (new) — overrides the film_stock baseline when active. # Falls back to the existing `Shot on {film_stock}` baseline when empty. _cinema_block = (shots[0].get("cinematography") if shots else None) or {} _cinema_tokens = render_cinema_tokens( mode_id=_cinema_block.get("mode") or cs["_project_config"].get("cinema_mode"), model_id="seeddance-2.0", shot_overrides=_cinema_block.get("overrides"), ) if _cinema_tokens: style_parts.append( _strip_focal_mm_tokens_for_model( str(_cinema_tokens).lower(), "seeddance-2.0", ) ) elif film_stock: style_parts.append(f"shot on {str(film_stock).lower()}") if lighting_prose: style_parts.append(lighting_prose) if cs["scene_visual_locks"]: style_parts.append(cs["scene_visual_locks"]) style_parts.append("Cinematic, photorealistic") zone_c_parts.append(". ".join(style_parts) + ".") # Quality suffix zone_c_parts.append( "4k, hd, rich details, sharp clarity, cinematic texture, natural colors, stable picture." ) # Constraint block (Phase 2a) _r2v_mode_constraints = [] _r2v_cinema_block = (shots[0].get("cinematography") if shots else None) or {} _r2v_mode_id = _r2v_cinema_block.get("mode") or cs["_project_config"].get("cinema_mode") if _r2v_mode_id: _r2v_mode = _all_cinema_modes.get(_r2v_mode_id) if _r2v_mode: _r2v_mode_constraints = _r2v_mode.get("default_constraints") or [] if _r2v_mode_constraints: _pos_suffix, _neg_phrases = render_constraint_block(_r2v_mode_constraints, "seeddance-2.0") if not _pos_suffix and _neg_phrases: _ban = ["; ".join(f"no {frag.strip()}" for frag in p.split(",")) for p in _neg_phrases] _pos_suffix = "Constraints: " + "; ".join(_ban) + "." if _pos_suffix: zone_c_parts.append(_pos_suffix) # Diegetic audio — physically-present sounds only (footsteps, fabric, breath, # body movement, weather, mechanical). Not "ambient" — that's narrower than what # the scene physically produces. environment_line = cs["environment_line"] if environment_line: zone_c_parts.append( f"Diegetic audio only: {environment_line.strip().rstrip('.')} sounds." ) else: zone_c_parts.append("Diegetic audio only.") # Music directive — matches seeddance_t2v / seeddance_i2v phrasing if not cs["allow_music"]: zone_c_parts.append("No music, no score.") # ── Assemble 3-zone prompt ── sections = [] if zone_a_lines: sections.extend(zone_a_lines) if zone_b_lines: sections.extend(zone_b_lines) if zone_c_parts: sections.extend(zone_c_parts) prompt = " ".join(sections) # Enforce prompt length (4-8 sentences for multi-shot from PROMPT_BIBLE) prompt = _enforce_prompt_length(prompt, "seeddance-2.0", mode="r2v") return prompt def _resolve_identity_token( ref_manifest: dict | None, char_ordinal: int, ) -> str: """Resolve `@Image{identity_N}` to a hydrated `@ImageM` token. Fallback (no manifest, or key missing) is the LITERAL `@Image1` — never `@Image{identity_N}`, which is Bug P. A literal token is a valid, no-op-safe placeholder that won't trip the audit gate's "no @Image{ literal" assertion (#3). """ rm = ref_manifest or {} key = f"identity_{char_ordinal}" idx = rm.get(key) if idx is None: return "@Image1" return f"@Image{idx}" def _resolve_scene_token( ref_manifest: dict | None, loc_ordinal: int, ) -> str | None: """Resolve `@Image{scene_N}` to a hydrated `@ImageM` token, defensively. Returns None when the manifest has no `scene_N` entry (unhydrated location — no ref image exists, so no token should be emitted). Callers must check for None and omit the location declaration / in-shot clause entirely. When hydrated, returns `@Image{idx}` where idx is the 1-based position of the location ref in the refs array. """ rm = ref_manifest or {} key = f"scene_{loc_ordinal}" idx = rm.get(key) if idx is None: return None return f"@Image{idx}" def _character_name_candidates(char_id: str, bible_char: dict | None = None) -> set[str]: """Return configured character name forms, including display names and aliases.""" candidates: set[str] = set() if char_id: candidates.add(str(char_id)) candidates.add(str(char_id).replace("-", " ").replace("_", " ").title()) candidates.add(str(char_id).title()) if not isinstance(bible_char, dict): return {c for c in candidates if c} for key in ("display_name", "name"): value = bible_char.get(key) if value: candidates.add(str(value)) for key in ("aliases", "alias", "also_known_as", "names"): value = bible_char.get(key) if not value: continue values = value if isinstance(value, (list, tuple, set)) else [value] for alias in values: if alias: candidates.add(str(alias)) return {c.strip() for c in candidates if c and c.strip()} def _char_id_from_entry(entry: Any) -> str: """Read a character id from dict, dataclass-like, or bare string entries.""" if entry is None: return "" if isinstance(entry, dict): return str(entry.get("char_id") or entry.get("id") or entry.get("name") or "") for attr in ("char_id", "id", "name"): value = getattr(entry, attr, None) if value: return str(value) return str(entry) def _strip_character_names( prompt: str, characters: list, bible_chars: dict | None = None, ) -> str: """Replace character proper nouns with role-scoped anchors. Generalizes _render_action_no_proper_nouns to operate on arbitrary prompt text (not just one action line). Use for: - Single-shot i2v / r2v builders (A2 leak coverage) - Per-segment scoping inside build_seeddance_r2v_prompt_multi (B2 fix) The `characters` arg accepts the same shapes as _render_action_no_proper_nouns: list[dict] with "char_id"/"role" keys, OR list[str] of bare char_ids (no role information — defaults to "the subject"). Returns the input unchanged if there are no characters or the prompt is empty. """ if not prompt or not characters: return prompt out = prompt bible_chars = bible_chars or {} for entry in characters: if isinstance(entry, str): cid = entry role = "" elif isinstance(entry, dict): cid = entry.get("char_id", "") role = (entry.get("role") or "").lower() else: cid = getattr(entry, "char_id", "") or getattr(entry, "name", "") role = (getattr(entry, "role", "") or "").lower() if not cid: continue replacement = "the protagonist" if role == "protagonist" else "the subject" bible_char = bible_chars.get(cid, {}) if bible_chars else {} candidates = _character_name_candidates(cid, bible_char) # Longer first so "Jade Cooper" goes before "Jade". for name in sorted(candidates, key=len, reverse=True): if not name: continue out = re.sub(rf"\b{re.escape(name)}\b", replacement, out) return out def _render_action_no_proper_nouns( action_line: str, characters: list[dict], bible_chars: dict, ) -> str: """Strip character proper nouns from an action line. Replaces: - protagonist role -> "the protagonist" - other roles -> "the subject" Reads names from BOTH the asset-data `char_id` field AND the bible `display_name` (the human-readable form the planner uses in action text — e.g. "Jade" not "JADE"). Locations and other non-person nouns are left alone. Returns the action line unchanged if no substitution is needed (no characters, empty action, etc.). """ if not action_line or not characters: return action_line out = action_line for entry in characters: cid = entry.get("char_id", "") if isinstance(entry, dict) else "" if not cid: continue role = (entry.get("role") or "").lower() if isinstance(entry, dict) else "" replacement = "the protagonist" if role == "protagonist" else "the subject" # Collect every form of the character's name we might see: # raw char_id ("JADE"), title-cased ("Jade"), and bible display_name. bible_char = bible_chars.get(cid, {}) if bible_chars else {} candidates = _character_name_candidates(cid, bible_char) # Longer names first so "Jade Cooper" replaces before "Jade". for name in sorted(candidates, key=len, reverse=True): if not name: continue out = re.sub( rf"\b{re.escape(name)}\b", replacement, out, ) return out def _render_action_with_tokens( action_line: str, characters: list, bible_chars: dict, name_to_token: dict, ) -> str: """Replace character proper nouns in an action line with @ImageN tokens. For multi-character shots, replaces each character's name forms with the corresponding @ImageN token from name_to_token (keyed by char_id). The \\b word boundary leaves possessive clitics attached, so "@Image2's throat" is produced correctly. MANDATORY off-frame fallback: after token replacements, any character name NOT in name_to_token (off-frame mention) is collapsed via _render_action_no_proper_nouns so bare proper nouns never survive. """ if not action_line or not characters: return action_line out = action_line for entry in characters: cid = entry.get("char_id", "") if isinstance(entry, dict) else "" if not cid or cid not in name_to_token: continue token = name_to_token[cid] bible_char = bible_chars.get(cid, {}) if bible_chars else {} candidates = _character_name_candidates(cid, bible_char) # Longer names first so "Jade Cooper" replaces before "Jade". for name in sorted(candidates, key=len, reverse=True): if not name: continue out = re.sub( rf"\b{re.escape(name)}\b", token, out, ) # Off-frame fallback: collapse any surviving proper nouns for characters # named in the action but absent from name_to_token (off-frame mentions). # _render_action_no_proper_nouns only collapses characters in the list it is # handed, so pass ALL bible characters — in-frame names are already replaced # by @ImageN tokens (nothing left to match), while off-frame names collapse # to "the subject"/"the protagonist". Without this, an off-frame bible name # leaks as a bare proper noun (content-policy + audit-#4 regression). fallback_chars = ( [ {"char_id": cid, "role": data.get("role", "") if isinstance(data, dict) else ""} for cid, data in bible_chars.items() ] if bible_chars else characters ) out = _render_action_no_proper_nouns(out, fallback_chars, bible_chars) return out def _bind_generic_subject_to_shot_tokens( text: str, focus_token: str, addressed_token: str | None = None, ) -> str: """Bind Option-C generic subject prose to per-shot @ImageN tokens.""" if not text or not focus_token: return text other_character = addressed_token or "the other character" out = re.sub( r"\bspeaks to the subject\b", f"speaks to {other_character}", text, flags=re.I, ) out = re.sub(r"\bthe subject\b", focus_token, out, flags=re.I) out = re.sub(r"\bthe protagonist\b", focus_token, out, flags=re.I) return out _AUTHORED_TIMECODE_RE = re.compile(r"\[\d+:\d{2}\s*-\s*\d+:\d{2}\]") _BOUND_IMAGEN_RE = re.compile(r"@Image(\d+)") def bind_named_prose( authored_text: str, primitive: Any, ref_manifest: dict | None, *, modality: str, ) -> BoundPrompt: """Bind real-name authored prose to the payload shape for a modality.""" manifest = dict(ref_manifest or _primitive_ref_value(primitive, "manifest") or {}) character_entries, bible_chars = _configured_character_entries(primitive) if modality == "r2v_multi": text = _bind_r2v_named_prose( authored_text, primitive, manifest, character_entries, bible_chars, ) payload_refs: dict[str, Any] = {} elif modality == "video_i2v": text = _strip_character_names( authored_text, character_entries, bible_chars, ) payload_refs = {} start_frame = _primitive_ref_value(primitive, "start_frame") image_tail = _primitive_ref_value(primitive, "image_tail") if image_tail is None: image_tail = _primitive_ref_value(primitive, "end_frame") if start_frame is not None: payload_refs["start_frame"] = start_frame if image_tail is not None: payload_refs["image_tail"] = image_tail else: raise BindAssertionError(f"unsupported bind modality {modality!r}") _assert_bound_prompt( text, primitive, manifest, character_entries, bible_chars, modality=modality, ) return BoundPrompt(text=text, payload_refs=payload_refs, manifest=manifest) def _bind_r2v_named_prose( authored_text: str, primitive: Any, ref_manifest: dict, character_entries: list[dict[str, str]], bible_chars: dict[str, dict], ) -> str: seen_chars = _primitive_char_ids(primitive, bible_chars) # This is the extracted pass-level binding map from # build_seeddance_r2v_prompt_multi. Keep it pass-scoped so cross-segment # character mentions bind before the proper-noun tripwire runs. pass_token_chars = [{"char_id": _cid} for _cid in seen_chars] pass_name_to_token = { _cid: _resolve_identity_token(ref_manifest, seen_chars.index(_cid) + 1) for _cid in seen_chars } bound_segments: list[str] = [] for segment_text in _split_authored_segments(authored_text): # Preserve the builder's per-segment alias to the pass-level map. name_to_token = pass_name_to_token if name_to_token: bound = _render_action_with_tokens( segment_text, pass_token_chars, bible_chars, name_to_token, ) else: bound = _render_action_no_proper_nouns( segment_text, character_entries, bible_chars, ) bound_segments.append(bound) return "\n".join(bound_segments) def _split_authored_segments(authored_text: str) -> list[str]: lines = str(authored_text or "").splitlines() if len(lines) <= 1: return [str(authored_text or "")] return lines def _assert_bound_prompt( text: str, primitive: Any, manifest: dict, character_entries: list[dict[str, str]], bible_chars: dict[str, dict], *, modality: str, ) -> None: leaked_names = _configured_character_name_leaks(text, character_entries, bible_chars) if leaked_names: raise BindAssertionError( "post-bind character name leak: " + ", ".join(sorted(leaked_names)) ) expected_beats = len(getattr(primitive, "timing_segments", None) or []) if expected_beats: actual_beats = len(_AUTHORED_TIMECODE_RE.findall(text or "")) if actual_beats != expected_beats: raise BindAssertionError( f"post-bind beat count {actual_beats} != timing_segments {expected_beats}" ) image_numbers = [int(n) for n in _BOUND_IMAGEN_RE.findall(text or "")] if modality == "video_i2v" and image_numbers: raise BindAssertionError( "video_i2v bound prose must not contain @ImageN tokens" ) backed_numbers = {int(v) for v in manifest.values() if _is_intish(v)} unbacked = sorted({n for n in image_numbers if n not in backed_numbers}) if unbacked: raise BindAssertionError( "post-bind @ImageN token(s) not backed by manifest: " + ", ".join(f"@Image{n}" for n in unbacked) ) def _configured_character_name_leaks( text: str, character_entries: list[dict[str, str]], bible_chars: dict[str, dict], ) -> set[str]: leaks: set[str] = set() for entry in character_entries: cid = entry.get("char_id", "") for name in _character_name_candidates(cid, bible_chars.get(cid, {})): if len(name) < 2: continue if re.search(rf"\b{re.escape(name)}\b", text or ""): leaks.add(name) return leaks def _configured_character_entries(primitive: Any) -> tuple[list[dict[str, str]], dict[str, dict]]: bible_chars: dict[str, dict] = {} refs = getattr(primitive, "refs", None) if isinstance(refs, dict): for source_key in ("characters", "character_refs", "character_manifest"): _merge_character_config(bible_chars, refs.get(source_key)) bible = refs.get("bible") or refs.get("prompt_bible") if isinstance(bible, dict): _merge_character_config(bible_chars, bible.get("characters")) for segment in getattr(primitive, "timing_segments", None) or []: if not isinstance(segment, dict): continue for source_key in ("characters", "character_refs"): _merge_character_config(bible_chars, segment.get(source_key)) for cid in getattr(primitive, "char_ids", None) or []: cid = str(cid) bible_chars.setdefault( cid, {"display_name": cid.replace("-", " ").replace("_", " ").title()}, ) entries = [{"char_id": cid} for cid in _primitive_char_ids(primitive, bible_chars)] return entries, bible_chars def _merge_character_config(out: dict[str, dict], source: Any) -> None: if not source: return if isinstance(source, dict): iterable = [] for key, value in source.items(): if isinstance(value, dict): merged = dict(value) merged.setdefault("char_id", key) iterable.append(merged) else: iterable.append({"char_id": key, "display_name": value}) elif isinstance(source, (list, tuple, set)): iterable = source else: iterable = [source] for entry in iterable: if isinstance(entry, dict): cid = entry.get("char_id") or entry.get("id") or entry.get("name") if not cid: continue cid = str(cid) existing = out.setdefault(cid, {}) existing.update({k: v for k, v in entry.items() if k != "char_id"}) else: cid = str(entry) out.setdefault( cid, {"display_name": cid.replace("-", " ").replace("_", " ").title()}, ) def _primitive_char_ids(primitive: Any, bible_chars: dict[str, dict]) -> list[str]: seen: set[str] = set() out: list[str] = [] for cid in list(getattr(primitive, "char_ids", None) or []) + list(bible_chars): cid = str(cid) if not cid or cid in seen: continue seen.add(cid) out.append(cid) return out def _primitive_ref_value(primitive: Any, key: str) -> Any: refs = getattr(primitive, "refs", None) if isinstance(refs, dict): return refs.get(key) return None def _is_intish(value: Any) -> bool: try: int(value) except (TypeError, ValueError): return False return True def build_seeddance_r2v_prompt_multi( shots: list[dict], bible: dict, project_config: dict, episode: int = 1, coverage_pass_dict: dict | None = None, ref_manifest: dict | None = None, anchor_duration_s: int | None = None, format_mode: str = "shot_label", brief_declarations: bool = True, casting_state: dict | None = None, segment_timestamps: list[float] | None = None, _core_semantics: dict | None = None, ) -> str: """Build a multi-shot SeedDance R2V prompt from coverage pass segments. Uses CoreSemantics from the primary shot for global enrichment (wardrobe, scene_visual_locks, film_stock, lighting, allow_music). Per-shot data (action, camera, duration) is still extracted per-shot from the shots list. The _core_semantics param allows passing a pre-extracted dict for the primary shot. Takes a plain dict (NOT the CoveragePass class — prompt_engine must not import from orchestrator). The dict has keys: segments, arc_preamble, focus_character, location_id. Zone A — Reference declarations from bible + coverage pass context + wardrobe Zone B — Shot list built from coverage pass segments Zone C — Global constraints (style, ambient audio, scene visual locks) Args: shots: List of plan shot dicts (fallback if coverage_pass_dict is None). bible: Global bible dict. project_config: Project config dict. episode: Episode number. coverage_pass_dict: Coverage pass as plain dict with keys: segments, arc_preamble, focus_character, location_id. ref_manifest: Maps ref_type to @ImageN index. anchor_duration_s: Duration override for Shot 1. Default None = use natural segment duration (normal pattern). Set to 1 (or any int) to force Shot 1 short — the "ref anchor" pattern: let refs prime identity/location briefly, then let shots 2+ carry the action. One of several selectable patterns; good fallback choice after a rejected take. format_mode: Segment marker format. One of: "shot_label" — "Shot 1: ... 3s." (default, community standard) "timeline" — "[0s] ... [3s] ..." (timestamp markers) "paren" — "(0s) ... (3s) ..." (parenthetical timestamps) "hybrid" — "Shot 1 (0-3s): ..." (labels + timestamps) _core_semantics: Pre-extracted CoreSemantics dict for the primary shot (optional). Used for wardrobe, scene_visual_locks, film_stock, lighting, and allow_music. The arc_preamble comes from coverage_pass_dict, not CoreSemantics. Returns: Prompt string with Zone A, Zone B, Zone C sections. """ if not shots and not coverage_pass_dict: return "" # If coverage_pass_dict provided, extract shots from segments if coverage_pass_dict: segments = coverage_pass_dict.get("segments", []) focus_character = coverage_pass_dict.get("focus_character", "") location_id_override = coverage_pass_dict.get("location_id", "") arc_preamble = coverage_pass_dict.get("arc_preamble", "") # Build shot list from segments if no shots provided if not shots and segments: shots = segments else: focus_character = "" location_id_override = "" arc_preamble = "" if not shots: return "" primary_shot = shots[0] cs = _core_semantics or extract_core_semantics( primary_shot, bible, project_config, episode, ) prompt_data = primary_shot.get("prompt_data", {}) skeleton = prompt_data.get("prompt_skeleton", {}) asset_data = primary_shot.get("asset_data", {}) # JIT hydration skeleton = _maybe_hydrate(skeleton, bible, episode=episode, asset_data=asset_data) bible_chars = bible.get("characters", {}) if bible else {} # ── Zone A — Reference declarations ── zone_a_lines = [] seen_chars = [] seen_locs = [] # Max words for character trait descriptions in Zone A. # brief_declarations=True: ~15 words (ref image carries the rest) # brief_declarations=False: full bible visual_description _BRIEF_CHAR_WORDS = 15 # Resolve character traits: prefer brief_visual from casting_state (prompt-optimized), # fall back to truncated bible visual_description, or full bible if brief_declarations=False. cast_chars = (casting_state or {}).get("characters", {}) def _resolve_char_traits(char_id: str, bible_char: dict) -> str: # 1. casting_state brief_visual (hand-curated, prompt-optimized) brief = cast_chars.get(char_id, {}).get("brief_visual", "") if brief and brief_declarations: return brief.strip().rstrip(".") # 2. Truncated bible (fallback when no brief_visual) visual_desc = bible_char.get("visual_description", "") if brief_declarations and visual_desc: return _truncate_at_natural_break(visual_desc, _BRIEF_CHAR_WORDS) # 3. Full bible return visual_desc.strip().rstrip(".") if visual_desc else "" # Bug U fix (2026-05-20): Option C — zero proper nouns. # Anchor lines use diegetic role descriptors ("the protagonist character", # "another character") instead of the character display_name. Identity # is bound via the @ImageN ref; the text "the protagonist character" is # a role descriptor that does not trip fal's content-policy filter the # way repeated character names do. def _role_descriptor(char_id: str, bible_char: dict, is_first: bool) -> str: role = (bible_char.get("role") or "").lower() if role == "protagonist" or is_first: return "the protagonist character" if role: return f"a {role} character" return "another character" # Focus character first (from coverage pass) if focus_character and focus_character not in seen_chars: seen_chars.append(focus_character) bible_char = bible_chars.get(focus_character, {}) descriptor = _role_descriptor(focus_character, bible_char, is_first=True) traits = _resolve_char_traits(focus_character, bible_char) token = _resolve_identity_token(ref_manifest, len(seen_chars)) line = f"{token} is {descriptor}" if traits: line += f" — {traits}" line += "." zone_a_lines.append(line) # Remaining characters from shots for s in shots: characters = s.get("asset_data", {}).get("characters", []) for char_entry in characters: char_id = _char_id_from_entry(char_entry).strip().upper() if not char_id or char_id in seen_chars: continue seen_chars.append(char_id) bible_char = bible_chars.get(char_id, {}) descriptor = _role_descriptor( char_id, bible_char, is_first=(len(seen_chars) == 1) ) traits = _resolve_char_traits(char_id, bible_char) token = _resolve_identity_token(ref_manifest, len(seen_chars)) line = f"{token} is {descriptor}" if traits: line += f" — {traits}" line += "." zone_a_lines.append(line) # Wardrobe — prefer brief_wardrobe from casting_state (prompt keywords), # fall back to CoreSemantics cs["wardrobe"] (truncated bible prose). # Bug U: wardrobe lines anchor on the @ImageN token, not the character # display_name (which would re-introduce the proper noun the anchor # line strips). _wardrobe_resolved = False if casting_state and brief_declarations: for char_id in seen_chars: cast_char = cast_chars.get(char_id, {}) brief_wd = cast_char.get("brief_wardrobe", {}) if isinstance(brief_wd, dict): phase_id = "" for s in shots: for ce in s.get("asset_data", {}).get("characters", []): if ce.get("char_id") == char_id and ce.get("wardrobe_phase_id"): phase_id = ce["wardrobe_phase_id"] break if phase_id: break wd_text = brief_wd.get(phase_id, "") if wd_text: char_ordinal = seen_chars.index(char_id) + 1 wd_token = _resolve_identity_token(ref_manifest, char_ordinal) zone_a_lines.append( f"{wd_token} wardrobe — {wd_text.strip().rstrip('.')}." ) _wardrobe_resolved = True if not _wardrobe_resolved and cs["wardrobe"]: zone_a_lines.append(cs["wardrobe"].strip().rstrip(".") + ".") # Location from coverage pass or shots. # Locations are NOT person names — keep the display_name; only character # proper nouns trigger fal's content filter. eff_location = location_id_override or primary_shot.get("asset_data", {}).get( "location_id", "" ) if eff_location and eff_location not in seen_locs: seen_locs.append(eff_location) token = _resolve_scene_token(ref_manifest, len(seen_locs)) if token is not None: zone_a_lines.append(f"{token} is the location reference.") # else: unhydrated location — no ref image, omit declaration entirely # Additional locations from shots for s in shots: s_loc = s.get("asset_data", {}).get("location_id", "") if s_loc and s_loc not in seen_locs: seen_locs.append(s_loc) token = _resolve_scene_token(ref_manifest, len(seen_locs)) if token is not None: zone_a_lines.append(f"{token} is the location reference.") # else: unhydrated location — no ref image, omit declaration entirely # ── Zone B — Shot list ── zone_b_lines = [] missing_motion_line_shots: list[int] = [] # Optional arc preamble from coverage pass (NOT from CoreSemantics — # coverage_pass_dict carries multi-shot-specific arc context) if arc_preamble: zone_b_lines.append(arc_preamble.strip().rstrip(".") + ".") # Bug Q (Phase 8): when `segment_timestamps` is supplied by # build_dispatch_payload, the prompt's editorial cuts come from the # SAME cumulative-sum used for `expected_segment_timestamps`. Without # it, fall back to the legacy per-shot accumulation from # `target_editorial_duration_s` (unchanged behavior for direct callers). cumulative_s = 0 # running timestamp for timeline/hybrid modes # B2 fix (R4 Phase 7) — Per-segment prompt scoping. Collect the union of # every character loaded across ALL segments of this pass. Each segment # text below gets stripped of proper nouns belonging to characters NOT # in THAT segment's local roster (prompt_bleed warnings from generate.py). # Strip is character-entity scoped — locations (Int_Lower_Decks_Corridor, # etc.) are NOT in pass_chars, so they survive the strip per spec § 1.7. pass_chars: list[dict] = [] pass_char_ids_seen: set[str] = set() for _pass_shot in shots: for _entry in (_pass_shot.get("asset_data", {}) or {}).get("characters") or []: _cid = ( _entry.get("char_id") if isinstance(_entry, dict) else str(_entry) ) or "" _cid_norm = _cid.strip().upper().replace("-", "_") if _cid_norm and _cid_norm not in pass_char_ids_seen: pass_char_ids_seen.add(_cid_norm) pass_chars.append( _entry if isinstance(_entry, dict) else {"char_id": _cid} ) pass_token_chars = [{"char_id": _cid} for _cid in seen_chars] pass_name_to_token = { _cid: _resolve_identity_token(ref_manifest, seen_chars.index(_cid) + 1) for _cid in seen_chars } # Pre-load cinema modes once for the entire shot loop (Phase 2a). try: _all_cinema_modes_m = load_cinema_modes().get("modes", {}) except Exception: # noqa: BLE001 _all_cinema_modes_m = {} for i, s in enumerate(shots): s_prompt_data = s.get("prompt_data", {}) s_skeleton = s_prompt_data.get("prompt_skeleton", {}) s_routing = s.get("routing_data", {}) # JIT hydrate per-shot skeleton (skip i==0 — already hydrated above) if i > 0: s_skeleton = _maybe_hydrate( s_skeleton, bible, episode=episode, asset_data=s.get("asset_data") ) shot_type = _resolve_shot_type(s_prompt_data.get("shot_type", "MS")) type_name = _SHOT_TYPE_NAMES.get(shot_type, f"{shot_type} shot") focal_length = s_prompt_data.get("focal_length", "50mm") camera_movement = s_prompt_data.get("camera_movement", "static") movement_name = ( _MOVEMENT_NAMES.get(camera_movement, camera_movement) if camera_movement and camera_movement != "static" else "Static" ) action_line = s_skeleton.get("action_line", "") motion_line = s_skeleton.get("motion_line", "") or "" motion_line_text = str(motion_line).strip() subject_line = s_skeleton.get("subject_line", "") emotion_line = s_skeleton.get("emotion_line", "") if not motion_line_text: missing_motion_line_shots.append(i + 1) # Duration: anchor override for Shot 1 natural_duration = s_routing.get("target_editorial_duration_s", 5) if i == 0 and anchor_duration_s: duration = anchor_duration_s else: duration = natural_duration # Bug Q (Phase 8): if segment_timestamps was supplied, ride the # SAME cumulative-sum the payload uses. This keeps the prompt's # [Ns] / (Ns) / Shot N (A-Bs) markers aligned with # expected_segment_timestamps[i] exactly — no second timing model. if segment_timestamps is not None and i < len(segment_timestamps): cumulative_s = segment_timestamps[i] if i + 1 < len(segment_timestamps): duration = segment_timestamps[i + 1] - cumulative_s # Per-shot character ordinals — used to resolve the @ImageN token # for "Cut to ..." transitions. Bug U: NEVER use display_names here. s_chars = s.get("asset_data", {}).get("characters", []) shot_char_tokens: list[str] = [] for char_entry in s_chars: cid = _char_id_from_entry(char_entry).strip().upper() if cid and cid in seen_chars: ordinal = seen_chars.index(cid) + 1 shot_char_tokens.append(_resolve_identity_token(ref_manifest, ordinal)) shot_focus_token = shot_char_tokens[0] if shot_char_tokens else "" shot_addressed_token = ( shot_char_tokens[1] if len(shot_char_tokens) >= 2 else None ) # Location token — only set if the ref is hydrated (scene_N in manifest). # Unhydrated: no ref image exists, so omit the "in @ImageN" clause. s_loc = s.get("asset_data", {}).get("location_id", eff_location) loc_ref = "" if s_loc and s_loc in seen_locs: loc_idx = list(seen_locs).index(s_loc) + 1 _resolved_loc = _resolve_scene_token(ref_manifest, loc_idx) loc_ref = _resolved_loc if _resolved_loc is not None else "" # Build shot line — format depends on format_mode shot_num = i + 1 end_s = cumulative_s + duration if format_mode == "timeline": parts = [f"[{cumulative_s}s]"] elif format_mode == "hybrid": parts = [f"Shot {shot_num} ({cumulative_s}-{end_s}s):"] elif format_mode == "paren": parts = [f"({cumulative_s}s)"] else: parts = [f"Shot {shot_num}:"] # Camera line with lens-type from cinema mode (Phase 2a). _s_cinema_cam_m = s.get("cinematography") or {} _s_mode_id_cam_m = _s_cinema_cam_m.get("mode") or cs["_project_config"].get("cinema_mode") _s_mode_for_cam_m = _all_cinema_modes_m.get(_s_mode_id_cam_m) if _s_mode_id_cam_m else None _s_camera_line_m = render_camera_line(s, _s_mode_for_cam_m, "seeddance-2.0") if _s_camera_line_m: parts.append(_s_camera_line_m) elif camera_movement and camera_movement != "static": focal_clause = ( f", {focal_length}" if _get_include_focal_length("seeddance-2.0") else "" ) parts.append(f"{movement_name} {type_name}{focal_clause}.") else: focal_clause = ( f", {focal_length}" if _get_include_focal_length("seeddance-2.0") else "" ) parts.append(f"Static {type_name}{focal_clause}.") # Transition for non-first shots — Bug U: cut to the @ImageN token, # never to the character display_name. # S1: if the shot raw carries a meaningful "transition" key (non-empty, # not the default "smooth"), use it; otherwise fall back to "Cut to". if i > 0: cut_token = shot_char_tokens[0] if shot_char_tokens else "@Image1" seg_transition = s.get("transition") if isinstance(s, dict) else None if seg_transition and seg_transition.strip() and seg_transition.lower() != "smooth": parts.append(f"{seg_transition.strip().capitalize()} to {cut_token}.") else: parts.append(f"Cut to {cut_token}.") elif shot_char_tokens and loc_ref: # Anchor shot — bind first character + location via tokens only. parts.append(f"{shot_char_tokens[0]} in {loc_ref}.") # A1 — use pass-level token substitution for character names. Shot # focus/cut tokens stay local, but cross-shot mentions like "Wren" in a # JADE-only shot must bind to the pass-level @ImageN before the # proper-noun scrub runs. name_to_token = pass_name_to_token # A4 — subject_line (blocking) emitted FIRST per PROMPT_BIBLE multi_shot # ordering: [subject_line, action_line, environment_line, emotion_line]. # Carries blocking/posture, not camera; run through name strip/token. if subject_line: if name_to_token: sanitized_subject = _render_action_with_tokens( subject_line, pass_token_chars, bible_chars, name_to_token, ) else: sanitized_subject = _render_action_no_proper_nouns( subject_line, s_chars, bible_chars, ) sanitized_subject = _bind_generic_subject_to_shot_tokens( sanitized_subject, shot_focus_token, shot_addressed_token, ) parts.append(sanitized_subject.strip().rstrip(".") + ".") # Action — A1: bind character names to pass-level @ImageN tokens. if action_line: if name_to_token: sanitized = _render_action_with_tokens( action_line, pass_token_chars, bible_chars, name_to_token, ) else: sanitized = _render_action_no_proper_nouns( action_line, s_chars, bible_chars, ) sanitized = _bind_generic_subject_to_shot_tokens( sanitized, shot_focus_token, shot_addressed_token, ) parts.append(sanitized.strip().rstrip(".") + ".") if motion_line_text: if name_to_token: sanitized_motion = _render_action_with_tokens( motion_line_text, pass_token_chars, bible_chars, name_to_token, ) else: sanitized_motion = _render_action_no_proper_nouns( motion_line_text, s_chars, bible_chars, ) sanitized_motion = _bind_generic_subject_to_shot_tokens( sanitized_motion, shot_focus_token, shot_addressed_token, ) parts.append(sanitized_motion.strip().rstrip(".") + ".") # R5 A2 fix (2026-05-21)—per-segment scripted dialogue emission # mirroring build_seeddance_i2v_prompt. Each segment's # audio_data.dialogue becomes a quoted clause appended after that # segment's action line. Per-segment Option C strip already wired # below (B2 R4 Phase 7) scrubs speaker proper nouns; the quoted # line survives. pipeline-learnings §23. s_audio = s.get("audio_data") or {} s_dialogue = s_audio.get("dialogue") or [] s_has_dialogue = bool( (s.get("routing_data") or {}).get("has_dialogue") or s.get("has_dialogue") ) if s_dialogue and s_has_dialogue: s_first = s_dialogue[0] if isinstance(s_first, dict): s_line = str(s_first.get("text") or "").strip() else: s_line = str(s_first).strip() s_line = s_line.strip('"').strip("'").strip() if s_line: if name_to_token: s_line = _render_action_with_tokens( s_line, pass_token_chars, bible_chars, name_to_token, ) s_line = _bind_generic_subject_to_shot_tokens( s_line, shot_focus_token, shot_addressed_token, ) speaker = shot_focus_token or "The character" parts.append(f'{speaker} speaks: "{s_line}".') # Emotion — A4: drop the `and not brief_declarations` gate so emotion_line # survives in brief mode. brief_declarations defaults True and the live # caller passes no override, so emotion was always dropped before this fix. if emotion_line: sanitized_emotion = _render_action_no_proper_nouns( emotion_line, s_chars, bible_chars, ) parts.append(sanitized_emotion.strip().rstrip(".") + ".") # Duration — only append explicit duration for shot_label mode # timeline/hybrid/paren already encode timing in the marker if format_mode == "shot_label": parts.append(f"{duration}s.") segment_text = " ".join(parts) # B2 fix (R4 Phase 7) — scope the proper-noun strip to characters NOT # in THIS segment. The pass-level Option C anchor already handles # names of THIS segment's characters via _render_action_no_proper_nouns # above; here we scrub names of other characters in the pass roster # that leaked into this segment's text (e.g. WREN mentioned in a # JADE-only segment). Character-entity scoped — locations untouched. segment_char_ids: set[str] = set() for _ce in s_chars: _cid_raw = (_ce.get("char_id") if isinstance(_ce, dict) else str(_ce)) or "" _cid_norm = _cid_raw.strip().upper().replace("-", "_") if _cid_norm: segment_char_ids.add(_cid_norm) non_segment_chars = [ c for c in pass_chars if ((c.get("char_id") if isinstance(c, dict) else str(c)) or "") .strip() .upper() .replace("-", "_") not in segment_char_ids ] if non_segment_chars: segment_text = _strip_character_names( segment_text, non_segment_chars, bible_chars, ) zone_b_lines.append(segment_text) cumulative_s = end_s if missing_motion_line_shots: logger.warning( "r2v_multi: %d/%d shots missing motion_line (shot numbers: %s)", len(missing_motion_line_shots), len(shots), ", ".join(str(n) for n in missing_motion_line_shots), ) # ── Zone C — Global constraints ── zone_c_parts = [] film_stock = cs["film_stock"] lighting_prose = _flatten_lighting_to_prose( {"lighting": cs["lighting_data"]}, model="seeddance-2.0", ) style_parts = [] # Cinema mode (new) — primary shot's cinema metadata applies to the # whole batch (consistent with extract_core_semantics primary_shot pattern). _cinema_block = primary_shot.get("cinematography") or {} _cinema_tokens = render_cinema_tokens( mode_id=_cinema_block.get("mode") or cs["_project_config"].get("cinema_mode"), model_id="seeddance-2.0", shot_overrides=_cinema_block.get("overrides"), ) if _cinema_tokens: style_parts.append( _strip_focal_mm_tokens_for_model( str(_cinema_tokens).lower(), "seeddance-2.0", ) ) elif film_stock: style_parts.append(f"shot on {str(film_stock).lower()}") if lighting_prose: style_parts.append(lighting_prose) if cs["scene_visual_locks"]: style_parts.append(cs["scene_visual_locks"]) style_parts.append("Cinematic, photorealistic") zone_c_parts.append(". ".join(style_parts) + ".") # Quality suffix zone_c_parts.append( "4k, hd, rich details, sharp clarity, cinematic texture, natural colors, stable picture." ) # Constraint block (Phase 2a) — injected into the LIVE r2v_multi builder's zone-C. _multi_mode_constraints = [] _multi_mode_id = _cinema_block.get("mode") or cs["_project_config"].get("cinema_mode") if _multi_mode_id: _multi_mode = _all_cinema_modes_m.get(_multi_mode_id) if _multi_mode: _multi_mode_constraints = _multi_mode.get("default_constraints") or [] if _multi_mode_constraints: _pos_suffix, _neg_phrases = render_constraint_block(_multi_mode_constraints, "seeddance-2.0") if not _pos_suffix and _neg_phrases: _ban = ["; ".join(f"no {frag.strip()}" for frag in p.split(",")) for p in _neg_phrases] _pos_suffix = "Constraints: " + "; ".join(_ban) + "." if _pos_suffix: zone_c_parts.append(_pos_suffix) # Diegetic audio — brief mode truncates since location ref carries the visual. # "Diegetic" is broader than "ambient": includes footsteps, fabric, breath, body # movement, weather, mechanical — anything the scene physically produces. environment_line = cs["environment_line"] if environment_line: if brief_declarations: env_words = environment_line.split()[:15] zone_c_parts.append(f"Diegetic: {' '.join(env_words).rstrip('.,;')}.") else: zone_c_parts.append( f"Diegetic audio only: {environment_line.strip().rstrip('.')} sounds." ) else: zone_c_parts.append("Diegetic audio only.") # Music directive — matches seeddance_t2v / seeddance_i2v phrasing if not cs["allow_music"]: zone_c_parts.append("No music, no score.") # ── Assemble 3-zone prompt ── sections = [] if zone_a_lines: sections.extend(zone_a_lines) if zone_b_lines: sections.extend(zone_b_lines) if zone_c_parts: sections.extend(zone_c_parts) prompt = " ".join(sections) prompt = _enforce_prompt_length(prompt, "seeddance-2.0", mode="multi_shot") return prompt # ══════════════════════════════════════════════════════════════════════ # SEEDDANCE T2V PROMPT (Text-to-Video — five-block prose, no refs) # ══════════════════════════════════════════════════════════════════════ def build_seeddance_t2v_prompt( shot: dict, bible: dict, project_config: dict, episode: int = 1, _core_semantics: dict | None = None, ) -> str: """Build a five-block prose T2V prompt for SeedDance (no image refs). Uses CoreSemantics for all enrichment data. The _core_semantics param allows the router to pass a pre-extracted dict (avoiding re-extraction when compiling multiple model prompts for the same shot). Five blocks merged into flowing prose (NO section headers): 1. SUBJECT — character + wardrobe + anchor + environment + emotion 2. ARC CONTEXT — scene-level emotional direction (coverage pass only) 3. ACTION — single primary motion (verb-enforced) 4. CAMERA — shot type + movement 5. STYLE — film stock + Seedance lighting anchors + scene visual locks Followed by: quality suffix, director notes, audio directive. Target: 120-250 words. """ cs = _core_semantics or extract_core_semantics( shot, bible, project_config, episode, ) parts = [] # ── Block 1: SUBJECT ── subject_parts = [] if cs["subject_line"]: subject_parts.append(cs["subject_line"].strip().rstrip(".")) if cs["character_descs"]: subject_parts.append(cs["character_descs"].strip().rstrip(".")) if cs["wardrobe"]: subject_parts.append(cs["wardrobe"].strip().rstrip(".")) if cs["character_anchor"]: subject_parts.append(cs["character_anchor"].strip().rstrip(".")) if cs["environment_line"]: subject_parts.append(cs["environment_line"].strip().rstrip(".")) if cs["emotion_line"]: subject_parts.append(cs["emotion_line"].strip().rstrip(".")) if subject_parts: parts.append(". ".join(subject_parts) + ".") # ── Arc context (scene-level emotional direction from coverage pass) ── if cs["arc_preamble"]: parts.append(cs["arc_preamble"].strip().rstrip(".") + ".") # ── Block 2: ACTION (one primary motion, verb-enforced) ── action = cs["action_line"] or cs["kinetic_action"] if action: action = _enforce_single_verb_action(action) parts.append(action.strip().rstrip(".") + ".") # ── Block 3: CAMERA (with lens-type from cinema mode) ── _cinema_block_cam = shot.get("cinematography") or {} _mode_for_cam = None _mode_id_cam = _cinema_block_cam.get("mode") or cs["_project_config"].get("cinema_mode") if _mode_id_cam: try: _all_modes = load_cinema_modes().get("modes", {}) _mode_for_cam = _all_modes.get(_mode_id_cam) except Exception: # noqa: BLE001 pass _camera_line = render_camera_line(shot, _mode_for_cam, "seeddance-2.0") if _camera_line: parts.append(_camera_line) else: type_name = _SHOT_TYPE_NAMES.get(cs["shot_type"], f"{cs['shot_type']} shot") if cs["camera_movement"] and cs["camera_movement"] != "static": movement = _MOVEMENT_NAMES.get(cs["camera_movement"], cs["camera_movement"]) parts.append(f"{type_name}, {movement}.") else: parts.append(f"Static {type_name}.") # ── Block 4: STYLE (film stock + lighting + scene locks) ── lighting_prose = _flatten_lighting_to_prose( {"lighting": cs["lighting_data"]}, model="seeddance-2.0", ) style_parts = [] # Cinema mode (new) — overrides film_stock baseline when active. _cinema_block = shot.get("cinematography") or {} _cinema_tokens = render_cinema_tokens( mode_id=_cinema_block.get("mode") or cs["_project_config"].get("cinema_mode"), model_id="seeddance-2.0", shot_overrides=_cinema_block.get("overrides"), ) if _cinema_tokens: style_parts.append( _strip_focal_mm_tokens_for_model( str(_cinema_tokens).lower(), "seeddance-2.0", ) ) elif cs["film_stock"]: style_parts.append(f"shot on {str(cs['film_stock']).lower()}") if lighting_prose: style_parts.append(lighting_prose) if cs["scene_visual_locks"]: style_parts.append(cs["scene_visual_locks"]) if style_parts: parts.append(". ".join(style_parts) + ".") # ── Quality suffix ── parts.append( "4k, hd, rich details, sharp clarity, " "cinematic texture, natural colors, stable picture." ) # ── Constraint block (Phase 2a) ── _mode_constraints = [] if _mode_for_cam is not None: _mode_constraints = _mode_for_cam.get("default_constraints") or [] if _mode_constraints: _pos_suffix, _neg_phrases = render_constraint_block(_mode_constraints, "seeddance-2.0") if not _pos_suffix and _neg_phrases: _ban = ["; ".join(f"no {frag.strip()}" for frag in p.split(",")) for p in _neg_phrases] _pos_suffix = "Constraints: " + "; ".join(_ban) + "." if _pos_suffix: parts.append(_pos_suffix) # ── Director notes ── if cs["director_notes"]: parts.append(cs["director_notes"].strip().rstrip(".") + ".") # ── Audio directive ── if not cs["allow_music"]: parts.append("No music, no score.") prompt = " ".join(p.strip() for p in parts if p and p.strip()) prompt = re.sub(r"\.{2,}", ".", prompt) return _enforce_prompt_length(prompt, "seeddance-2.0", mode="t2v") # ══════════════════════════════════════════════════════════════════════ # SEEDDANCE I2V PROMPT (Image-to-Video — motion-only, start frame ref) # ══════════════════════════════════════════════════════════════════════ def build_seeddance_i2v_prompt( shot: dict, bible: dict | None = None, project_config: dict | None = None, episode: int = 1, _core_semantics: dict | None = None, ) -> str: """Build a motion-only I2V prompt for SeedDance (50-75 words). Uses CoreSemantics for all enrichment data. The _core_semantics param allows the router to pass a pre-extracted dict (avoiding re-extraction when compiling multiple model prompts for the same shot). Strips all character visual descriptions. Uses "@Image1 as the first frame" as the anchor and "The subject in @Image1" as pronoun — never character names. Keeps motion/action, camera movement, and environment changes only. Pre-flight: raises ValueError if no start frame path in shot data. Args: shot: Plan ShotRecord dict. project_config: Project config dict. bible: Global bible dict (for CoreSemantics extraction). episode: Episode number. _core_semantics: Pre-extracted CoreSemantics dict (optional). Returns: Motion-only prompt string for SeedDance I2V. Raises: ValueError: If no start frame is present in the shot data. """ cs = _core_semantics or extract_core_semantics( shot, bible or {}, project_config or {}, episode, ) # Pre-flight: verify start frame exists start_frame = cs["start_frame"] if not start_frame: raise ValueError( "build_seeddance_i2v_prompt requires a start frame in " "routing_data.start_frame_path/url or asset_data.start_frame_path/url" ) parts = [] # Anchor declaration parts.append("@Image1 as the first frame.") # R5 A1 carry-over (2026-05-21)—multi-character @ImageN anchor emission. # When the shot carries 2+ characters, declare each additional @ImageN so # the model binds identity for every character in the frame. Option C # (pipeline-learnings §29) requires no proper nouns—the anchor uses the # ordinal descriptor "the second character" / "the third character" etc., # never the bible display_name. Single-char shots keep the legacy # `@Image1 as the first frame` line only—no change. _shot_chars = shot.get("asset_data", {}).get("characters", []) or [] _n_chars = len(_shot_chars) if _n_chars >= 2: _ordinals = [ "first", "second", "third", "fourth", "fifth", "sixth", "seventh", "eighth", "ninth", ] _anchor_bits = ["@Image1 is the first character."] for _i in range(2, _n_chars + 1): _ord = _ordinals[_i - 1] if _i - 1 < len(_ordinals) else f"#{_i}" _anchor_bits.append(f"@Image{_i} is the {_ord} character.") parts.append(" ".join(_anchor_bits)) # Action — use "The subject in @Image1" as pronoun action = cs["action_line"] or cs["kinetic_action"] if action: # Strip any character names — use generic subject reference parts.append(f"The subject in @Image1 {action.strip().rstrip('.')}.") else: parts.append("The subject in @Image1 remains in frame.") # R5 A2 fix (2026-05-21)—emit scripted dialogue as a quoted clause per # pipeline-learnings §23. CoreSemantics exposes shot.audio_data on # cs["audio_data"]; we pull dialogue from there. Position matters— # between action and camera so the model lip-syncs the line in # temporal alignment with the action. # _strip_character_names below replaces speaker names with "the subject"; # the \b{name}\b regex matches the bare name token, not quoted content, # so the scripted line survives the strip verbatim. _cs_audio = cs.get("audio_data") or {} dialogue_list = _cs_audio.get("dialogue") or [] has_dialogue_flag = bool( (shot.get("routing_data") or {}).get("has_dialogue") or shot.get("has_dialogue") ) if dialogue_list and has_dialogue_flag: first = dialogue_list[0] if isinstance(first, dict): line_text = str(first.get("text") or "").strip() else: line_text = str(first).strip() # Strip outer matched quotes so we can wrap consistently in doubles. line_text = line_text.strip('"').strip("'").strip() if line_text: parts.append(f'The subject speaks: "{line_text}"') # R6 Phase 9 (c3) — gaze cue. Push eyeline three-quarters away # from lens to address A3 (smile-morph + direct-to-camera). # UNCONDITIONAL within the dialogue branch (no OTS exemption); # SYNTHESIS Phase 9 ambiguity resolution. parts.append( "Subject looks off-camera, eyeline three-quarters away from lens." ) # Camera movement if cs["camera_movement"] and cs["camera_movement"] != "static": movement = _MOVEMENT_NAMES.get(cs["camera_movement"], cs["camera_movement"]) parts.append(f"{movement.capitalize()}.") # Environment change (if meaningful) if cs["environment_line"]: env_stripped = cs["environment_line"].strip().rstrip(".") if env_stripped: parts.append(f"{env_stripped}.") # Identity preservation parts.append( "Same person as @Image1. Do not alter facial proportions, eye shape, or hairstyle." ) # Audio directive parts.append("No music, no score.") # Quality suffix removed per consultation recommendation (22% budget reclamation) prompt = " ".join(p.strip() for p in parts if p and p.strip()) # Clean up double periods prompt = re.sub(r"\.{2,}", ".", prompt) # A2 leak fix (R4) — strip character proper nouns. Single-shot i2v had # never run through Option C; identical content-policy risk to r2v_multi. characters = shot.get("asset_data", {}).get("characters", []) or [] bible_chars = (bible or {}).get("characters", {}) if isinstance(bible, dict) else {} prompt = _strip_character_names(prompt, characters, bible_chars) prompt = _enforce_prompt_length(prompt, "seeddance-2.0", mode="i2v") return prompt def compile_all_prompts( shot: dict, bible: dict, project_config: dict, episode: int = 1, ) -> dict: """Compile all model-specific prompts for a shot. Extracts CoreSemantics ONCE, then passes to each builder as a pre-extracted dict. Each builder formats the same semantic content for its model's preferences (word budget, structure, enrichments). Args: shot: Plan ShotRecord dict. bible: Global bible dict. project_config: Project config dict. episode: Episode number. Returns: Dict: { "previs_flash": str, # Flash 3.1 previs prompt "keyframe_nbp": str, # NBP production keyframe prompt "keyframe_seedream": str, # Seedream concise keyframe (40-80 words) "kling_i2v": str, # Kling I2V prompt (30 words max) "kling_t2v": str, # Kling T2V prompt (75-100 words) "seeddance_r2v": str, # SeedDance R2V 3-zone prompt "seeddance_t2v": str, # SeedDance T2V five-block prose (120-250 words) "seeddance_i2v": str, # SeedDance I2V motion-only (50-75 words) "veo_t2v": str, # Veo long-shot/ENV prompt "video_plan": str, # Generic video prompt from plan "wan_i2v": str, # Wan 2.7 I2V prompt (150-300 words) "wan_between": str, # Wan 2.7 In Between transition prompt "wan_r2v": str, # Wan 2.7 R2V prompt (250-400 words) } """ # ── Extract CoreSemantics ONCE ── cs = extract_core_semantics(shot, bible, project_config, episode) compiled = {} # 1. Keyframe (NBP — legacy, full 12-layer structured prompt) compiled["keyframe_nbp"] = build_prompt_from_plan( shot, bible, project_config, episode, ) # 1a. Keyframe (Seedream — concise 40-80 words) compiled["keyframe_seedream"] = build_seedream_prompt( shot, bible, project_config, episode, _core_semantics=cs, ) # 1b. Previs (same as keyframe for quality gate) compiled["previs_flash"] = compiled["keyframe_nbp"] # 2. SeedDance T2V — five-block prose (120-250 words) compiled["seeddance_t2v"] = build_seeddance_t2v_prompt( shot=shot, bible=bible, project_config=project_config, episode=episode, _core_semantics=cs, ) # 3. SeedDance I2V — motion-only (50-75 words) try: compiled["seeddance_i2v"] = build_seeddance_i2v_prompt( shot=shot, project_config=project_config, bible=bible, episode=episode, _core_semantics=cs, ) except ValueError: compiled["seeddance_i2v"] = "" # 4. SeedDance R2V — 3-zone reference-to-video compiled["seeddance_r2v"] = build_seeddance_r2v_prompt( shots=[shot], bible=bible, project_config=project_config, episode=episode, _core_semantics=cs, ) # 5. Kling I2V — minimal motion (30 words) compiled["kling_i2v"] = build_kling_i2v_prompt(shot) # 6. Kling T2V — labeled sections (75-100 words) compiled["kling_t2v"] = build_kling_t2v_prompt( shot, bible, project_config, episode, _core_semantics=cs, ) # 7. Veo — dense cinematic prose (1500 char cap) compiled["veo_t2v"] = build_veo_prompt( shot=shot, bible=bible, project_config=project_config, episode=episode, _core_semantics=cs, ) # 8. Generic video from plan (fallback for unknown models) compiled["video_plan"] = build_video_prompt_from_plan( shot=shot, bible=bible, project_config=project_config, episode=episode, ) # 9. Wan I2V — dense DiT conditioning (150-300 words) compiled["wan_i2v"] = build_wan_i2v_prompt( shot, bible=bible, project_config=project_config, episode=episode, has_end_frame=False, _core_semantics=cs, ) # 10. Wan In Between compiled["wan_between"] = build_wan_i2v_prompt( shot, bible=bible, project_config=project_config, episode=episode, has_end_frame=True, _core_semantics=cs, ) # 11. Wan R2V compiled["wan_r2v"] = build_wan_r2v_prompt( [shot], bible, project_config, episode=episode, multi_shots=False, _core_semantics=cs, ) return compiled # ══════════════════════════════════════════════════════════════════════ # PREVIS PROMPT (ADR H02 — Flash 3.1 previs gate) # ══════════════════════════════════════════════════════════════════════ def build_previs_prompt( shot: dict, bible: dict | None = None, config: dict | None = None, look_bundle=None ) -> str: """Build an enriched previz prompt for Flash 3.1 image generation. Uses the pre-compiled previs_flash prompt from the plan if available (generated during Stage 2 plan pass with full context). Otherwise builds from shot data + bible, including lighting, atmosphere, and location detail — NOT just mechanical framing. With character ref images now injected alongside the prompt, Flash benefits from richer environmental/mood context. Args: shot: Plan shot dict (ShotRecord format with 5 consumer groups). bible: Optional global bible dict for location enrichment. Returns: Enriched prompt string for Flash previz. """ # ── JIT hydration: resolve tokens from live bible data ── skeleton_raw = shot.get("prompt_data", {}).get("prompt_skeleton", {}) if skeleton_raw and bible: hydrated = _maybe_hydrate( skeleton_raw, bible, asset_data=shot.get("asset_data", {}) ) if hydrated is not skeleton_raw: shot.setdefault("prompt_data", {})["prompt_skeleton"] = hydrated # ── Build from shot data + bible ── prompt_data = shot.get("prompt_data", {}) routing_data = shot.get("routing_data", {}) asset_data = shot.get("asset_data", {}) skeleton = prompt_data.get("prompt_skeleton", {}) shot_type = _resolve_shot_type(prompt_data.get("shot_type", "MS")) is_env = routing_data.get("is_env_only", False) characters = asset_data.get("characters", []) location_id = asset_data.get("location_id", "unknown location") parts = [] # 1. Shot type + focal length type_name = _SHOT_TYPE_NAMES.get(shot_type, shot_type) focal = prompt_data.get("focal_length", "") if focal: parts.append(f"{type_name}, {focal}") else: camera_movement = prompt_data.get("camera_movement", "static") if camera_movement and camera_movement != "static": parts.append(f"{type_name}, {camera_movement}") else: parts.append(type_name) # 2. Characters with positions if is_env: parts.append("ENVIRONMENT ONLY, no people") elif characters: for char in characters: char_id = char.get("char_id", "character") position = char.get("screen_position", "center") parts.append(f"{char_id} on {position} side") # 3. Action — use full subject line, not just first sentence subject = skeleton.get("subject_line", "") if subject: parts.append(subject) # 4. Environment line from skeleton (has more detail than just location name) env_line = skeleton.get("environment_line", "") if env_line: parts.append(env_line) else: location_display = ( location_id.replace("_", " ").replace("int ", "").replace("ext ", "") ) parts.append(location_display) # 5. Lighting from shot data lighting_data = prompt_data.get("lighting", {}) sources = lighting_data.get("sources", []) if sources: light_parts = [] for src in sources[:2]: # Max 2 sources to keep prompt concise motivator = src.get("motivator", "") quality = src.get("quality", "") color = src.get("color_temp", "") if motivator: desc = f"{color} {quality} light from {motivator}".strip() light_parts.append(desc) if light_parts: parts.append(", ".join(light_parts)) # 6. Bible enrichment — location mood and lighting if bible: loc_data = bible.get("locations", {}).get(location_id, {}) mood = loc_data.get("mood", "") if mood: parts.append(mood) bible_lighting = loc_data.get("lighting", "") if bible_lighting and not sources: parts.append(bible_lighting) # 7. Emotion line emotion = skeleton.get("emotion_line", "") if emotion: parts.append(emotion) # 8. Aspect ratio ar = config.get("aspect_ratio") if config else None if not ar: from recoil.core.paths import get_config ar = get_config().get("production_aspect_ratio", "9:16") orientation = "vertical" if ar == "9:16" else "horizontal" parts.append(f"{ar} {orientation} frame") prompt = ". ".join(p.strip() for p in parts if p and p.strip()) + "." prompt = prompt.replace("..", ".") # krea2-flora Look/Identity injection (opt-in; no-op when look_bundle is None). prompt = apply_look(prompt, look_bundle) prompt = _enforce_prompt_length(prompt, "gemini-3.1-flash-image-preview") return f"{prompt.rstrip()} {_no_text_footer()}".strip() # ══════════════════════════════════════════════════════════════════════ # FLASH ENRICHMENT WRAPPER # ══════════════════════════════════════════════════════════════════════ def _call_flash_enrichment( system_prompt: str, user_input: str, temperature: float = 0.3 ) -> str: """Call Flash for prompt enrichment. Returns enriched text or empty string on failure.""" import os try: from google import genai from google.genai import types as genai_types api_key = os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY") if not api_key: return "" client = genai.Client(api_key=api_key) config = genai_types.GenerateContentConfig( temperature=temperature, systemInstruction=system_prompt, ) response = client.models.generate_content( model="gemini-2.5-flash", contents=user_input, config=config, ) if response and response.text: return response.text.strip() return "" except Exception as e: logger.warning("Flash enrichment failed: %s", e) return "" def enrich_prompt( base_prompt: str, model_target: str, locked_terms: list[str], kinetic_descriptor: str = "", is_env: bool = False, project_config: dict = None, system_prompt_version: str = "1.0", ) -> tuple[str, str]: """Flash-enrich a compiled prompt for a target generation model. Args: base_prompt: The compiled prompt from Python builders. model_target: Target model key (e.g. "nbp", "kling_i2v", "veo"). locked_terms: Strings that MUST appear verbatim in output. kinetic_descriptor: Pre-matched kinetic descriptor to inject. is_env: True if this is an ENV-only shot (bypass enrichment). project_config: Project config dict (may contain skip_flash_enrichment). system_prompt_version: Version of the system prompt file to load. Returns: (enriched_prompt, version_tag) tuple. Falls back to base_prompt if enrichment fails. """ # ENV bypass — Flash hallucinates humans into empty rooms if is_env: return base_prompt, "bypass_env" # Global bypass flag if project_config and project_config.get("skip_flash_enrichment"): return base_prompt, "bypass_config" # Load system prompt base_instructions = load_prompt_file("flash_base_instructions.txt") model_instructions = load_prompt_file( f"flash_to_{model_target}_v{system_prompt_version}.txt" ) if not model_instructions: return base_prompt, "bypass_no_prompt_file" system_prompt = base_instructions + "\n\n" + model_instructions # Build Flash input flash_input = "LOCKED_TERMS:\n" for term in locked_terms: flash_input += f"- {term}\n" if kinetic_descriptor: flash_input += f"\nKINETIC_DESCRIPTOR: {kinetic_descriptor}\n" flash_input += f"\nSHOT DATA TO REWRITE:\n{base_prompt}" # Call Flash enriched = _call_flash_enrichment(system_prompt, flash_input) if not enriched: return base_prompt, "fallback_error" # Validate locked terms preserved missing = [t for t in locked_terms if t not in enriched] if missing: # Retry at temp=0 enriched = _call_flash_enrichment(system_prompt, flash_input, temperature=0.0) if not enriched: return base_prompt, "fallback_retry_error" missing = [t for t in locked_terms if t not in enriched] if missing: return base_prompt, "fallback_validation" return enriched, f"v{system_prompt_version}" # ══════════════════════════════════════════════════════════════════════ # PRIVATE HELPERS # ══════════════════════════════════════════════════════════════════════ def _build_kinetic_layer(shot: dict) -> str: """Convert action/emotion fields to kinetic descriptors. Maps semantic descriptions to camera-artifact language: - Physical strain -> "muscles taut, unbalanced dynamic pose, off-axis framing" - Impact -> "motion blur, kinetic energy, dust kicked into lens" - Stillness -> "still, static framing, held breath" Args: shot: Single shot dict with 'action' and 'emotion' fields. Returns: Kinetic descriptor string, or empty string. """ action = shot.get("action", "") emotion = shot.get("emotion", "") combined = f"{action} {emotion}" if not combined.strip(): return "" descriptors = get_all_kinetic_descriptors(combined, max_count=3) # get_all_kinetic_descriptors returns the fallback list when nothing # matches, so we always have at least one entry. return ", ".join(descriptors) + "." def _build_lighting_vector(shot: dict, storyboard: dict) -> str: """Extract and format explicit lighting direction from shot data. Converts generic "amber light" to locked vectors like "amber light casting hard shadows from TOP LEFT" Args: shot: Single shot dict with 'lighting' field. storyboard: Full storyboard dict for fallback lighting_strategy. Returns: Lighting vector string with explicit directional coords. """ lighting_text = shot.get("lighting", "") if not lighting_text: lighting_text = storyboard.get("visual_vocabulary", {}).get( "lighting_strategy", "" ) if not lighting_text: return "" # Determine direction (from config/lexicon.json via prompt_config) direction = get_lighting_direction(lighting_text) or "SIDE" # Determine quality (from config/lexicon.json via prompt_config) quality = get_lighting_quality(lighting_text) or "hard" # Extract color/source from the original text # Look for color words near "light" color_match = re.search( r"\b(amber|blue|cold\s+blue|electric\s+blue|red|white|warm|orange|cyan|" r"green|crimson|golden|silver|violet|magenta|teal)\b", lighting_text, re.I, ) color = color_match.group(1).lower() if color_match else "" # Build the locked vector parts = [] if color: parts.append(f"{color} light") else: parts.append("light") parts.append(f"casting {quality} shadows from {direction}") # Append any specific sources mentioned source_match = re.search( r"\b(emergency\s+strip\w*|debt\s+counter|headlamp|cryo[\s-]?pod|" r"strip\s+light\w*|LCD|counter\s+display|chest\s+port|eye\s+glow)", lighting_text, re.I, ) if source_match: parts.append(f"(source: {source_match.group(1).lower()})") return " ".join(parts) def _build_camera_line(shot: dict) -> str: """Build camera/lens description from shot data. Format: "{shot_type} shot, {camera_angle} angle, {focal_length} at {aperture}" Args: shot: Single shot dict with shot_type, camera_angle, focal_length, aperture. Returns: Camera line string. """ shot_type = _resolve_shot_type(shot.get("shot_type", "MS")) camera_angle = shot.get("camera_angle", "eye") focal_length = shot.get("focal_length", "50mm") aperture = shot.get("aperture", "f/2.0") camera_movement = shot.get("camera_movement", "") type_name = _SHOT_TYPE_NAMES.get(shot_type, f"{shot_type} shot") # Expand angle angle_names = { "eye": "eye-level", "low": "low angle", "high": "high angle", "dutch": "dutch angle", "overhead": "overhead", "worm": "worm's eye", "bird": "bird's eye", } angle_name = angle_names.get(camera_angle, camera_angle) line = f"{type_name}, {angle_name}, {focal_length} at {aperture}" if camera_movement and camera_movement != "static": movement = _MOVEMENT_NAMES.get(camera_movement, camera_movement) line += f", {movement}" return line + "." def _build_wide_shot_footer() -> str: """Footer for WIDE/LS/EWS shots -- strips facial demands.""" return get_constant("production", "wide_shot_footer") def _build_close_shot_footer() -> str: """Footer for MS/MCU/CU/ECU shots -- demands anatomical accuracy.""" return get_constant("production", "close_shot_footer") def _resolve_character_visual( char_key: str, storyboard: dict, breakdown: dict, character_data: dict = None ) -> str: """Resolve character visual description from available sources. Priority: character_data override > storyboard characters > breakdown. Args: char_key: Character key (e.g. 'jinx', 'kian'). storyboard: Full storyboard JSON. breakdown: Full breakdown.json dict. character_data: Optional override dict keyed by character name. Returns: Character visual description string. """ # Check override first if character_data and char_key in character_data: return character_data[char_key].get("visual", "") # Check storyboard characters block sb_chars = storyboard.get("characters", {}) if char_key in sb_chars: return sb_chars[char_key].get("visual", "") # Check breakdown (uppercase keys) bd_chars = breakdown.get("characters", {}) if breakdown else {} upper_key = char_key.upper() if upper_key in bd_chars: return bd_chars[upper_key].get("visual_description", "") return "" def _build_character_description( char_key: str, shot: dict, storyboard: dict, character_data: dict, shot_type: str ) -> str: """Build character description appropriate for the shot type. For wide shots: only silhouette + posture + wardrobe. For close shots: full visual detail + skin texture. For medium shots: balanced description. Args: char_key: Character key. shot: Single shot dict. storyboard: Full storyboard JSON. character_data: Character visual overrides. shot_type: The shot type abbreviation. Returns: Character description string. """ # Get base visual from storyboard characters block sb_chars = storyboard.get("characters", {}) char_info = sb_chars.get(char_key, {}) visual = char_info.get("visual", "") wardrobe = char_info.get("wardrobe", "") hair_makeup = char_info.get("hair_makeup", "") scale_fragment = char_info.get("scale_prompt_fragment", "") # Override from character_data if available if character_data and char_key in character_data: override = character_data[char_key] visual = override.get("visual", visual) wardrobe = override.get("wardrobe", wardrobe) if not visual: return "" if shot_type in _WIDE_SHOT_TYPES: # Wide: silhouette, posture, scale. Strip facial details. parts = [] if scale_fragment: parts.append(scale_fragment) elif visual: # Extract only build/body descriptors build_match = re.search( r"\b(?:lean|wiry|massive|tall|broad|armored|slim|muscular)\b[^\.]*", visual, re.I, ) if build_match: parts.append(build_match.group(0).strip()) if wardrobe: parts.append(wardrobe) return ". ".join(parts) + "." if parts else "" elif shot_type in _CLOSE_SHOT_TYPES: # Close: full visual detail, emphasize skin, eyes, texture parts = [visual] if hair_makeup: parts.append(hair_makeup) return ". ".join(parts) + "." else: # Medium: full visual + wardrobe parts = [visual] if wardrobe: parts.append(f"Wearing {wardrobe}") return ". ".join(parts) + "." def _extract_props(shot: dict, storyboard: dict, project_config: dict = None) -> str: """Extract signature props mentioned in shot action/description. Reads prop patterns from project_config["prop_patterns"]. Each entry is {"pattern": , "label": }. If no patterns are defined in config, returns empty string (graceful no-op). Args: shot: Single shot dict. storyboard: Full storyboard JSON. project_config: Project config dict (from Recoil project_config.json). Returns: Comma-separated props string. """ project_config = project_config or {} raw_patterns = project_config.get("prop_patterns", []) if not raw_patterns: return "" action = shot.get("action", "") description = shot.get("description", "") combined = f"{action} {description}" found = [] for entry in raw_patterns: pattern = re.compile(entry["pattern"], re.I) if pattern.search(combined): label = entry["label"] if label not in found: found.append(label) return ", ".join(found) def _is_non_human(char_key: str, character_data: dict, storyboard: dict) -> bool: """Check if a character is non-human (android, chassis, etc). Args: char_key: Character key (lowercase). character_data: Optional character visual data dict. storyboard: Full storyboard JSON. Returns: True if character is non-human. """ # Check character_data for explicit identity_type if character_data and char_key in character_data: if character_data[char_key].get("identity_type") == "non_human": return True # Check storyboard character visual for mechanical keywords sb_chars = storyboard.get("characters", {}) char_info = sb_chars.get(char_key, {}) visual = char_info.get("visual", "") return _visual_is_non_human(visual) def _parse_grid_size(grid_size: str) -> tuple[int, int]: """Parse grid size string into (rows, cols). Args: grid_size: String like "3x3" or "2x2". Returns: Tuple of (rows, cols). """ parts = grid_size.lower().split("x") if len(parts) == 2: return int(parts[0]), int(parts[1]) return 3, 3 def _build_scene_coverage_panels( shots: list[dict], positions: list[str], total_panels: int, storyboard: dict ) -> list[str]: """Build SCENE_COVERAGE panel assignments. Maps actual storyboard shots to grid positions. Supports both legacy flat-key format and plan-format (nested prompt_data/asset_data/etc.). Args: shots: List of shot dicts (flat or plan-format). positions: Grid position labels. total_panels: Total number of panels. storyboard: Full storyboard JSON. Returns: List of panel assignment strings. """ lines = [] first_environment = None # Track for dedup across panels # Pre-compute storyboard-level lighting fallback once (avoids regex per panel) _storyboard_lighting_fallback = _build_lighting_vector({"lighting": ""}, storyboard) for i in range(min(total_panels, len(shots))): shot = shots[i] pos = positions[i] panel_num = i + 1 # Support both flat-key and plan-format schemas prompt_data = shot.get("prompt_data", {}) skeleton = prompt_data.get("prompt_skeleton", {}) shot_name = ( shot.get("shot_id") or shot.get("name") or f"shot_{shot.get('id', i + 1)}" ) shot_type = _resolve_shot_type( prompt_data.get("shot_type") or shot.get("shot_type", "MS") ) camera_movement = prompt_data.get("camera_movement", "static") kinetic_action = prompt_data.get("kinetic_action", "") # Action: prefer skeleton subject_line > kinetic_action > flat action subject_line = skeleton.get("subject_line", "") action_line = skeleton.get("action_line", "") action = subject_line or action_line or kinetic_action or shot.get("action", "") # Environment from skeleton environment = skeleton.get("environment_line", "") # Emotion from skeleton or flat emotion = skeleton.get("emotion_line", "") or shot.get("emotion", "") lines.append("") lines.append(f"PANEL {panel_num} ({pos}) — {shot_name}:") lines.append(f"Shot Type: {shot_type}. Camera: {camera_movement}.") if action: lines.append(f"Subject: {action}.") # Environment: full description on first panel, "same environment" after if environment: if first_environment is None: first_environment = environment lines.append(f"Environment: {environment}.") elif environment != first_environment: lines.append(f"Environment: {environment}.") else: lines.append("Environment: Same as Panel 1.") if kinetic_action and kinetic_action != action: lines.append(f"Cinematic treatment: {kinetic_action}.") if emotion: lines.append(f"Emotion: {emotion}.") # Lighting: use shared flattener for structured data, fallback to storyboard lighting = _flatten_lighting_to_prose(prompt_data, max_sources=3) if not lighting: lighting = _storyboard_lighting_fallback if lighting: lines.append(f"Lighting: {lighting}.") # If we have fewer shots than panels, fill with variations if len(shots) < total_panels: for i in range(len(shots), total_panels): pos = positions[i] panel_num = i + 1 lines.append("") lines.append( f"PANEL {panel_num} ({pos}) — VARIATION: " "Subtle reframing of the best composition from previous panels." ) return lines def _build_directors_take_panels(shots: list[dict], positions: list[str]) -> list[str]: """Build DIRECTORS_TAKE panel assignments. 4 specific cinematic variations of the same shot. Args: shots: List of shot dicts (typically 1 shot repeated). positions: Grid position labels (2x2). Returns: List of panel assignment strings. """ lines = [] shot = shots[0] shot_type = _resolve_shot_type(shot.get("shot_type", "MS")) action = shot.get("action", "") emotion = shot.get("emotion", "") shot_name = shot.get("name", f"shot_{shot.get('id', 1)}") # Four genuinely different editorial framings — not subtle rotations tighter = _tighten_shot_type(shot_type) variations = [ { "label": "Wide Establishing", "framing": "WS, eye-level, full environment visible, subject placed in context of the space", "note": "Scene geography. Where are we, who is here.", }, { "label": "Medium Coverage", "framing": f"{shot_type}, slightly off-center, three-quarter angle, natural conversational distance", "note": "The workhorse shot. Action and reaction visible.", }, { "label": "Tight Close-Up", "framing": f"{tighter}, eye-level, face fills the frame, shallow depth of field", "note": "Emotion and micro-expression. Background falls away.", }, { "label": "Over-the-Shoulder / Low Angle", "framing": f"{shot_type}, low angle looking up, or over-the-shoulder if two characters present", "note": "Power dynamic or POV framing. Completely different spatial relationship.", }, ] for i, var in enumerate(variations[: len(positions)]): pos = positions[i] panel_num = i + 1 lines.append("") lines.append(f"PANEL {panel_num} ({pos}) - {var['label']} of {shot_name}:") lines.append(f"Framing: {var['framing']}.") lines.append(f"Note: {var['note']}") if action: lines.append(f"Action (same in all panels): {action}") if emotion: lines.append(f"Emotion (same in all panels): {emotion}.") return lines def _build_action_burst_panels(shots: list[dict], positions: list[str]) -> list[str]: """Build ACTION_BURST panel assignments. Subtle pose variations of the same action. Args: shots: List of shot dicts (typically 1 shot). positions: Grid position labels (2x2). Returns: List of panel assignment strings. """ lines = [] shot = shots[0] shot_type = _resolve_shot_type(shot.get("shot_type", "MS")) camera_angle = shot.get("camera_angle", "eye") action = shot.get("action", "") shot_name = shot.get("name", f"shot_{shot.get('id', 1)}") pose_labels = [ "Pose A: Peak of the action. Maximum exertion.", "Pose B: Just before the peak. Building tension.", "Pose C: Just after the peak. Follow-through.", "Pose D: Recovery beat. Breath between actions.", ] for i, pose in enumerate(pose_labels[: len(positions)]): pos = positions[i] panel_num = i + 1 lines.append("") lines.append(f"PANEL {panel_num} ({pos}) - {shot_name}:") lines.append(f"Shot Type: {shot_type}. Camera Angle: {camera_angle}.") lines.append(pose) if action: lines.append(f"Base action: {action}") return lines def _tighten_shot_type(shot_type: str) -> str: """Return a tighter framing of the given shot type. MS -> MCU, MCU -> CU, CU -> ECU, WIDE -> MS, etc. Args: shot_type: Original shot type abbreviation. Returns: Tighter shot type abbreviation. """ tighten_map = { "WIDE": "MS", "EWS": "LS", "LS": "MS", "WS": "MS", "MLS": "MS", "MWS": "MS", "MS": "MCU", "MFS": "MS", "MCU": "CU", "CU": "ECU", "ECU": "ECU", "BCU": "ECU", } return tighten_map.get(shot_type, shot_type) # ══════════════════════════════════════════════════════════════════════ # DEMO / CLI # ══════════════════════════════════════════════════════════════════════ if __name__ == "__main__": import json import sys from pathlib import Path # Resolve paths from recoil.core.paths import RECOIL_ROOT, DEFAULT_PROJECT storyboard_path = ( RECOIL_ROOT / DEFAULT_PROJECT / "storyboards" / "storyboard_ep_001.json" ) config_path = RECOIL_ROOT / DEFAULT_PROJECT / "visual" / "project_config.json" breakdown_path = RECOIL_ROOT / DEFAULT_PROJECT / "visual" / "breakdown.json" if not storyboard_path.exists(): print(f"ERROR: Storyboard not found at {storyboard_path}") sys.exit(1) # Load data storyboard = json.loads(storyboard_path.read_text()) project_config = json.loads(config_path.read_text()) if config_path.exists() else {} breakdown = ( json.loads(breakdown_path.read_text()) if breakdown_path.exists() else {} ) shots = storyboard["shots"] print("=" * 70) print("PROMPT ENGINE DEMO — EP001 Shots 1-3") print("=" * 70) # ── Shot 1: ENV shot (corridor_dolly_to_clang) ─────────────────── shot_1 = shots[0] print(f"\n{'─' * 70}") print(f"SHOT 1: {shot_1['name']} (id={shot_1['id']})") print(f" Type: {shot_1['shot_type']} | Characters: {shot_1['characters_in_shot']}") print(" ENV shot: True (no characters)") print(f"{'─' * 70}\n") prompt_1 = build_cinematic_prompt( shot=shot_1, storyboard=storyboard, project_config=project_config, is_env=True, ) print(prompt_1) # ── Shot 2: Jinx character shot (jinx_wedges_hook) ─────────────── shot_2 = shots[1] print(f"\n{'─' * 70}") print(f"SHOT 2: {shot_2['name']} (id={shot_2['id']})") print(f" Type: {shot_2['shot_type']} | Characters: {shot_2['characters_in_shot']}") print(f"{'─' * 70}\n") prompt_2 = build_cinematic_prompt( shot=shot_2, storyboard=storyboard, project_config=project_config, ) print(prompt_2) # ── Shot 3: ECU character shot (rebreather_fog) ────────────────── shot_3 = shots[2] print(f"\n{'─' * 70}") print(f"SHOT 3: {shot_3['name']} (id={shot_3['id']})") print(f" Type: {shot_3['shot_type']} | Characters: {shot_3['characters_in_shot']}") print(f"{'─' * 70}\n") prompt_3 = build_cinematic_prompt( shot=shot_3, storyboard=storyboard, project_config=project_config, ) print(prompt_3) # ── Grid prompt: Scene Coverage 3x3 for shots 1-9 ──────────────── print(f"\n{'=' * 70}") print("GRID PROMPT: Scene Coverage 3x3 (Shots 1-9)") print(f"{'=' * 70}\n") grid_shots = shots[:9] grid_prompt = build_grid_prompt( shots=grid_shots, storyboard=storyboard, breakdown=breakdown, project_config=project_config, grid_type=GridType.SCENE_COVERAGE, grid_size="3x3", ) print(grid_prompt) # ── Grid prompt: Director's Take 2x2 for shot 2 ────────────────── print(f"\n{'=' * 70}") print("GRID PROMPT: Director's Take 2x2 (Shot 2 — Jinx wedges hook)") print(f"{'=' * 70}\n") dt_prompt = build_grid_prompt( shots=[shot_2], storyboard=storyboard, breakdown=breakdown, project_config=project_config, grid_type=GridType.DIRECTORS_TAKE, grid_size="2x2", ) print(dt_prompt) # ── ENV sanitization demo ───────────────────────────────────────── print(f"\n{'=' * 70}") print("ENV SANITIZATION DEMO") print(f"{'=' * 70}\n") raw_text = ( "The corridor stretches into darkness, a figure's silhouette visible " "against amber strip lights. Her boots scrape against corroded grating. " "She moves through the haze, both hands gripping the salvage hook. " "The ship groans. Condensation drips from exposed pipe bundles." ) print(f"BEFORE:\n{raw_text}\n") print(f"AFTER:\n{sanitize_env_prompt(raw_text)}") # ── Visual Anchors demo ─────────────────────────────────────────── print(f"\n{'=' * 70}") print("VISUAL ANCHORS DEMO (Shot 2)") print(f"{'=' * 70}\n") anchors = build_visual_anchors( storyboard=storyboard, shot=shot_2, breakdown=breakdown, project_config=project_config, ) print(anchors) # ══════════════════════════════════════════════════════════════════════ # COVERAGE PROMPT DERIVATION # Generates synthetic shot dicts for coverage angles (reactions, cutaways, # wide safeties) that flow through the existing build_prompt_from_plan() # pipeline unchanged. # # Consultation: consultations/starsend/coverage-density-dramatic-peaks/SYNTHESIS.md # ══════════════════════════════════════════════════════════════════════ # ── Emotion inversion for reaction shots ── _WITNESS_EMOTION_MAP = { "shock": "alarm, frozen disbelief", "terror": "raw fear, survival instinct", "terrifying": "raw fear, instinctive recoil", "rage": "flinching retreat, self-preservation", "fury": "wide-eyed alarm, backing away", "activation": "startled vigilance, bracing", "violence": "defensive fear, tensing", "violent": "defensive recoil, protective stance", "lethal": "helpless terror, frozen", "confrontation": "defiant resistance, gritting teeth", "defiance": "grudging respect, wary assessment", "betrayal": "heartbreak, stunned disbelief", "revelation": "dawning realization, processing", "power": "intimidation, involuntary step back", "overwhelming": "shrinking, overwhelmed", "brutal": "visceral horror, looking away", "cold": "unease, wariness", "calculating": "suspicion, guard raised", "panic": "helpless dread, frozen", "desperation": "pity, concern", "grief": "empathetic sorrow, reaching out", "helplessness": "anguished witnessing, unable to act", } def _invert_emotion_v1(hero_emotion: str) -> str: """Keyword-based emotion inversion for coverage reactions. Transforms the hero shot's emotion into what a witness would feel. Uses max 3 emotion beats to avoid prompt bloat. """ if not hero_emotion: return "watchful tension, processing" parts = [p.strip() for p in hero_emotion.split(",")] inverted = [] for part in parts: matched = False for keyword, reaction in _WITNESS_EMOTION_MAP.items(): if keyword in part.lower(): inverted.append(reaction) matched = True break if not matched: inverted.append(f"witnessing {part}") # Deduplicate and cap at 3 beats seen = set() unique = [] for item in inverted: if item not in seen: seen.add(item) unique.append(item) joined = ", ".join(unique[:3]) beats = [b.strip() for b in joined.split(",")] return ", ".join(beats[:3]) # ── Cutaway target extraction ── _SKIP_ATMOSPHERE_WORDS = { "light", "lighting", "shadow", "shadows", "atmosphere", "mood", "tone", "ambient", "glow", "haze", "fog", "dark", "dimly", "dim", "lit", } def _extract_cutaway_target(shot: dict) -> dict | None: """Pick a cutaway subject from props or environment. Priority: shot props > environment detail phrases. Returns dict with 'subject' and 'description', or None. """ asset_data = shot.get("asset_data", {}) # Strategy 1: Named props props = asset_data.get("props", {}) if isinstance(props, list): # props might be a list of strings or dicts for p in props: name = p.get("prop_id") if isinstance(p, dict) else str(p) if name: return {"subject": name, "description": name} elif isinstance(props, dict): for prop_name, prop_data in props.items(): desc = ( prop_data.get("description", prop_name) if isinstance(prop_data, dict) else prop_name ) return {"subject": prop_name, "description": desc} # Strategy 2: Environment line detail extraction skeleton = shot.get("prompt_data", {}).get("prompt_skeleton", {}) env_line = skeleton.get("environment_line", "") if not env_line: return None phrases = [p.strip() for p in env_line.replace(";", ",").split(",")] for phrase in phrases: words = phrase.lower().split() # Skip pure atmosphere descriptors meaningful = [ w for w in words if w not in _SKIP_ATMOSPHERE_WORDS and len(w) > 2 ] if not meaningful: continue if len(phrase) > 5: return {"subject": phrase.strip(), "description": phrase.strip()} return None # ── Main derivation function ── def derive_coverage_shot( hero_shot: dict, coverage_type: str, all_characters: list[str] | None = None, ) -> dict | None: """Derive a synthetic shot dict for a coverage angle. The returned dict has the same schema as a real plan shot, so it can be passed directly to build_prompt_from_plan() for full prompt compilation. Args: hero_shot: The anchor shot dict from the episode plan. coverage_type: One of "reaction", "cutaway", "wide". all_characters: List of all char_ids in the scene (for reaction targeting). Returns: A synthetic shot dict, or None if the coverage type isn't viable (e.g., reaction with no other characters). """ import copy derived = copy.deepcopy(hero_shot) skeleton = derived.setdefault("prompt_data", {}).setdefault("prompt_skeleton", {}) hero_sid = hero_shot.get("shot_id", "") if coverage_type == "reaction": # Identify the hero character and find the reactor hero_chars = derived.get("asset_data", {}).get("characters", []) hero_char_ids = [] for c in hero_chars: cid = (c.get("char_id", "") if isinstance(c, dict) else str(c)).upper() if cid: hero_char_ids.append(cid) # Determine hero subject from subject_line subject_line = skeleton.get("subject_line", "").upper() hero_char = "" for cid in hero_char_ids: if cid in subject_line: hero_char = cid break if not hero_char and hero_char_ids: hero_char = hero_char_ids[0] # Find reactor (non-hero character) scene_chars = all_characters or hero_char_ids reactors = [c for c in scene_chars if c.upper() != hero_char] if not reactors: return None # Single-character scene, no reaction possible reactor = reactors[0] # Build reaction skeleton — CLEAN SINGLE (hero NOT in frame) hero_emotion = skeleton.get("emotion_line", "") hero_action = skeleton.get("action_line", "") skeleton["subject_line"] = f"{reactor}, reacting" skeleton["action_line"] = ( f"witnessing: {hero_action}" if hero_action else "watching, processing" ) skeleton["emotion_line"] = _invert_emotion_v1(hero_emotion) # environment_line stays IDENTICAL (locked agreement) derived["prompt_data"]["shot_type"] = "CU" derived["shot_id"] = f"{hero_sid}.rx.{reactor.lower()}" # Strip to reactor only — clean single derived.setdefault("asset_data", {})["characters"] = [ { "char_id": reactor, "emotion_keyword": "reacting", "screen_position": "center", "visibility": "in_frame", } ] elif coverage_type == "cutaway": target = _extract_cutaway_target(hero_shot) if target is None: return None # No viable cutaway subject skeleton["subject_line"] = target["description"] skeleton["action_line"] = "" skeleton["emotion_line"] = "" # environment_line stays IDENTICAL derived["prompt_data"]["shot_type"] = "ECU" derived.setdefault("routing_data", {})["is_env_only"] = True derived.setdefault("asset_data", {})["characters"] = [] # Build cutaway shot_id target_slug = target["subject"].lower().replace(" ", "_")[:12] derived["shot_id"] = f"{hero_sid}.cut.{target_slug}" elif coverage_type == "wide": # Keep everything, just change shot type to WS derived["prompt_data"]["shot_type"] = "WS" derived.setdefault("prompt_data", {})["focal_length"] = "24mm" derived["shot_id"] = f"{hero_sid}.ws" # camera_side goes neutral for wide derived.setdefault("spatial_data", {})["camera_side"] = "N" else: return None # Apply editorial priors (prompt modifiers only, no parameter overrides) try: from recoil.pipeline._lib.feedback_logger import ( get_editorial_priors, match_priors, ) project = hero_shot.get("_project", "") if project: priors = get_editorial_priors(project) focus_char = "" if coverage_type == "reaction" and reactors: focus_char = reactors[0] mods, _ = match_priors( priors, focus_character=focus_char, shot_type=derived["prompt_data"]["shot_type"], location_id=derived.get("asset_data", {}).get("location_id", ""), is_env=derived.get("routing_data", {}).get("is_env_only", False), ) if mods: derived["_editorial_modifiers"] = mods except (ImportError, Exception): pass # Editorial priors are optional return derived # ══════════════════════════════════════════════════════════════════════ # BUILDERS DISPATCH (CP-3, Phase 2) — single lookup table for # (model_id, modality) → builder callable. Locked design — no class # hierarchy, no clever indirection. Adding a new model = one line: # BUILDERS[("new-model-v1", "i2v")] = build_new_model_i2v_prompt # # Located at the END of the module so all builder functions are bound # at dict-literal evaluation time. `Callable` is imported up top. # ══════════════════════════════════════════════════════════════════════ # Phase 2 — populated with EXISTING pipeline/lib builders only. # Phase 3 — adds keyframe entries (nbp/keyframe, flash/previz) + # gemini-* image variants from visual/prompt_engine.py. # Phase 4 — adds T2I keyframe entries from tools/prompt_engine.py # (or empty if Phase 1 audit confirms zero external callers). # # NOTE (Phase 2 audit finding): build_seeddance_i2v_multishot_prompt does # NOT exist in pipeline/lib/prompt_engine.py — the broken import in # pipeline/tools/test_via_steprunner.py:625 references a non-existent # symbol (single-d "seedance" typo). The closest extant builders are # build_multi_shot_prompt and build_multi_prompt_sequence. The # ("seeddance-2.0", "i2v_multishot") key is therefore omitted from # Phase 2 BUILDERS. Phase 3 may add it once the underlying builder # is written or the broken import in tools/ is fixed. BUILDERS: dict[tuple[str, str], Callable] = { # Kling — I2V variants (kling-o3 doubles as elements-R2V via payload hints, # not a separate builder; see api_client.py:985–1005 endpoint resolution). ("kling-o3", "i2v"): build_kling_i2v_prompt, ("kling-v3", "i2v"): build_kling_i2v_prompt, ("kling-v3-i2v", "i2v"): build_kling_i2v_prompt, ("kling-v3", "t2v"): build_kling_t2v_prompt, # Seedance — full grid (i2v, r2v, t2v, multi-shot r2v). ("seeddance-2.0", "i2v"): build_seeddance_i2v_prompt, ("seeddance-2.0", "r2v"): build_seeddance_r2v_prompt, ("seeddance-2.0", "r2v_multi"): build_seeddance_r2v_prompt_multi, ("seeddance-2.0", "t2v"): build_seeddance_t2v_prompt, # Wan. ("wan-2.7-i2v", "i2v"): build_wan_i2v_prompt, ("wan-2.7-r2v", "r2v"): build_wan_r2v_prompt, # HappyHorse-1.0 (Alibaba) — backup retry tier (post 2026-05-07 test). # Reuses wan builders as a placeholder. Token syntax differs (HappyHorse # uses character1/character2 positional, not @Image1). Production retry # paths bypass these via explicit --prompt; rewrite if HappyHorse # promotes to a primary slot. ("happy-horse-i2v", "i2v"): build_wan_i2v_prompt, ("happy-horse-r2v", "r2v"): build_wan_r2v_prompt, # Veo (deprioritized 2026-04-09 per memory note, kept for keyframe-trap fix). ("veo-3.1", "i2v"): build_veo_prompt, ("veo-3.1", "t2v"): build_veo_prompt, # Image / keyframe / previz — wired in Phase 3. # Phase 1 audit confirmed 0 visual-only ALIVE symbols; pipeline/lib already # holds the canonical builder (`build_previs_prompt`) for these modalities. # No function migration needed — only dispatch-table wiring. # # Per audit § Modality enumeration, the orchestrator dispatches image # generation under both `"keyframe"` (production frames, see # pipeline/orchestrator/pipeline.py + production_loop.py) and `"previz"` # (gut-check pre-render, see pipeline/orchestrator/production_types.py). # We register BOTH modalities for every image-gen model_id so callers can # dispatch either string and resolve to the same builder. `build_previs_prompt` # is the only keyframe-style builder in pipeline/lib (per audit confirmed # `build_keyframe_prompt` does not exist anywhere). ("nbp", "keyframe"): build_previs_prompt, ("nbp", "previz"): build_previs_prompt, ("flash", "keyframe"): build_previs_prompt, ("flash", "previz"): build_previs_prompt, ("gemini-2.5-flash-image", "keyframe"): build_previs_prompt, ("gemini-2.5-flash-image", "previz"): build_previs_prompt, ("gemini-3-pro-image-preview", "keyframe"): build_previs_prompt, ("gemini-3-pro-image-preview", "previz"): build_previs_prompt, ("gemini-3.1-flash-image-preview", "keyframe"): build_previs_prompt, ("gemini-3.1-flash-image-preview", "previz"): build_previs_prompt, # Seedream (ByteDance) image models — docstring covers v4.5/v5. ("seedream-v4.5", "keyframe"): build_seedream_prompt, ("seedream-v4.5", "previz"): build_seedream_prompt, ("seedream-v5-lite", "keyframe"): build_seedream_prompt, ("seedream-v5-lite", "previz"): build_seedream_prompt, # GPT Image 2 (OpenAI via fal.ai) — t2i + single-ref edit. Uses the # generic previz builder; gpt-image-2 prefers natural-language prompts # without weight syntax, which build_previs_prompt already emits. Wired # 2026-05-25 to close the test_keys_match_provider_strategy_snapshot # gap left by the 2026-05-19 build (Phase 4b deferral). ("gpt-image-2", "keyframe"): build_previs_prompt, ("gpt-image-2", "previz"): build_previs_prompt, ("gpt-image-2", "storyboard"): build_storyboard_strip_prompt, ("gpt-image-2", "storyboard_finish"): build_gpt_image_2_storyboard_finish_prompt, ("gemini-3.1-flash-image-preview", "storyboard"): build_storyboard_strip_prompt, ("gemini-3.1-flash-image-preview", "storyboard_finish"): build_gpt_image_2_storyboard_finish_prompt, # krea2-flora image models (Phase 0) routed to Flora. Kept in sync with # provider_strategy.json — enforced by # test_get_builder::test_keys_match_provider_strategy_snapshot. Builder # choice mirrors Phase 2's apply_look wiring: Seedream family → # build_seedream_prompt; Krea-2 / Nano-Banana / Gemini-image → # build_previs_prompt (natural-language, no weight syntax). ("seedream-v5", "keyframe"): build_seedream_prompt, ("seedream-v5", "previz"): build_seedream_prompt, ("krea-2", "keyframe"): build_previs_prompt, ("krea-2", "previz"): build_previs_prompt, # krea-2-turbo — LOCAL ComfyUI (open-weights), natural-language prompts # like the other Krea/Nano-Banana image models → build_previs_prompt. ("krea-2-turbo", "keyframe"): build_previs_prompt, ("krea-2-turbo", "previz"): build_previs_prompt, ("krea-2-references", "keyframe"): build_previs_prompt, ("krea-2-references", "previz"): build_previs_prompt, ("nano-banana", "keyframe"): build_previs_prompt, ("nano-banana", "previz"): build_previs_prompt, ("gemini-3-pro", "keyframe"): build_previs_prompt, ("gemini-3-pro", "previz"): build_previs_prompt, # Coverage / multi-shot composer entries — these are dispatch convenience # so callers can do get_builder(model, "coverage") instead of importing # build_coverage_prompts directly. ("seeddance-2.0", "coverage"): build_coverage_prompts, ("kling-o3", "coverage"): build_coverage_prompts, ("kling-v3", "coverage"): build_coverage_prompts, } IMAGE_PROMPT_BUILDERS: tuple[Callable, ...] = ( build_previs_prompt, build_seedream_prompt, build_grid_prompt, build_two_character_prompt, build_location_ref_prompt, build_universal_expression_matrix, ) # Storyboard prompts are text-bearing by contract: they request rendered panel # numbers and caption strips, so they are exempt from the no-text invariant. TEXT_BEARING_BUILDERS: tuple[Callable, ...] = ( build_storyboard_strip_prompt, ) def get_builder(model_id: str, modality: str) -> Callable: """Resolve the prompt builder for (model_id, modality). Raises KeyError with a clear message listing supported keys for that model_id if the lookup fails. Phase 2 populates the table with existing pipeline/lib builders only; Phases 3-4 add keyframe + T2I entries. Args: model_id: Model identifier as it appears in provider_strategy.json (e.g. "kling-o3", "seeddance-2.0", "wan-2.7-i2v", "nbp"). modality: One of {"i2v", "r2v", "r2v_multi", "i2v_multishot", "t2v", "keyframe", "previz", "coverage"}. See recoil/docs/prompt-engine-audit.md § Modality enumeration. Returns: The builder callable. Call it with the same args you'd pass to the direct import — no signature change. Raises: KeyError: if no builder is registered for the (model_id, modality) tuple. Error message includes the supported modalities for that model_id, plus the canonical full key list. """ key = (model_id, modality) if key in BUILDERS: return BUILDERS[key] supported_for_model = sorted({m for (mid, m) in BUILDERS if mid == model_id}) if supported_for_model: raise KeyError( f"No builder for ({model_id!r}, {modality!r}). " f"Supported modalities for {model_id!r}: {supported_for_model}. " f"See recoil/docs/prompt-engine-audit.md for the full key list." ) raise KeyError( f"No builder for ({model_id!r}, {modality!r}). " f"Model {model_id!r} is not registered in BUILDERS. " f"Add via BUILDERS[({model_id!r}, {modality!r})] = build_<…>_prompt " f"in pipeline/lib/prompt_engine.py." ) __all_builders__ = sorted({fn.__name__ for fn in BUILDERS.values()})