# api/routes/prompt_inspector.py """Prompt Inspector API — annotated layer breakdown for the inspector drawer.""" import copy import json import re from pathlib import Path from typing import Optional from fastapi import APIRouter, Body, Depends from fastapi.responses import JSONResponse from ..deps import get_project, get_paths router = APIRouter(tags=["prompt-inspector"]) def _find_shot_in_plans(plans_dir: Path, shot_id: str) -> tuple: """Search all plan files for a shot by ID. Returns (shot_dict, episode_number) or (None, None). """ ep_match = re.match(r"EP(\d+)", shot_id) if ep_match: # Fast path: extract episode from shot_id ep_num = int(ep_match.group(1)) plan_path = plans_dir / f"ep_{ep_num:03d}_plan.json" if plan_path.exists(): try: plan = json.loads(plan_path.read_text(encoding="utf-8")) for s in plan.get("shots", []): if s.get("shot_id") == shot_id: return copy.deepcopy(s), ep_num except (json.JSONDecodeError, IOError): pass # Slow path: scan all plan files for plan_path in sorted(plans_dir.glob("ep_*_plan.json")): try: plan = json.loads(plan_path.read_text(encoding="utf-8")) # Extract episode number from filename fname_match = re.search(r"ep_(\d+)_plan", plan_path.name) if not fname_match: continue ep_num = int(fname_match.group(1)) for s in plan.get("shots", []): if s.get("shot_id") == shot_id: return copy.deepcopy(s), ep_num except (json.JSONDecodeError, IOError): continue return None, None def _load_bible(bible_path: Path) -> dict: """Load the global bible, returning empty dict on failure.""" if bible_path.exists(): try: return json.loads(bible_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError): pass return {} def _load_project_config(project_dir: Path) -> dict: """Load project_config.json, returning empty dict on failure.""" config_path = project_dir / "project_config.json" if config_path.exists(): try: return json.loads(config_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError): pass return {} @router.post("/api/compile-prompt-preview") def compile_prompt_preview( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), ): """Compile and return annotated prompt layers + formatter outputs. Request body: shot_id (str, required): Shot identifier (e.g. "EP001_SC001_001") mode (str, optional): Override preset mode (standard/stylized/raw/passthrough/custom) draft_prompt (str, optional): Override prompt text (not yet wired — future phase) bypasses (list[str], optional): Layer IDs to bypass (only used when mode=custom) shot_type (str, optional): Override shot type (e.g. "CU", "WS") Returns: JSON with annotated layers, formatter outputs, blocking data, and metadata. """ shot_id = body.get("shot_id") if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) # ── Load shot from plan ────────────────────────────────────────── shot_data, episode = _find_shot_in_plans(paths["plans_dir"], shot_id) if not shot_data: return JSONResponse( {"error": f"Shot not found in any plan: {shot_id}"}, status_code=404, ) # ── Load and apply saved overrides ──────────────────────────────── from .overrides import _load_overrides all_ov = _load_overrides(paths["plans_dir"], episode) saved_overrides = all_ov.get(shot_id, {}) if saved_overrides: from recoil.pipeline._lib.previz_context import apply_overrides shot_data = apply_overrides(shot_data, saved_overrides) # ── Load bible + project config ────────────────────────────────── bible = _load_bible(paths["bible_path"]) project_config = _load_project_config(paths["project_dir"]) # ── Apply optional overrides from request ──────────────────────── warnings = [] mode = body.get("mode") bypasses_list = body.get("bypasses") override_shot_type = body.get("shot_type") valid_preset_modes = {"standard", "stylized", "raw", "passthrough", "custom"} if mode or bypasses_list: overrides = shot_data.setdefault("prompt_overrides", {}) if mode and mode in valid_preset_modes: overrides["mode"] = mode if bypasses_list and mode == "custom": overrides["bypass"] = bypasses_list if override_shot_type: pd = shot_data.setdefault("prompt_data", {}) original_type = pd.get("shot_type", "MS") pd["shot_type"] = override_shot_type if original_type != override_shot_type: warnings.append( f"Shot type overridden: {original_type} -> {override_shot_type}" ) # ── Director notes override ─────────────────────────────────── draft_director_notes = body.get("director_notes") if draft_director_notes is not None: shot_data["director_notes"] = draft_director_notes.strip() # ── Build annotated sections ───────────────────────────────────── from recoil.pipeline._lib.prompt_engine import ( build_prompt_sections_from_plan, build_prompt_from_plan, build_video_prompt_from_plan, build_kling_i2v_prompt, build_kling_t2v_prompt, _resolve_bypasses, _OVERRIDE_PRESETS, ) sections = build_prompt_sections_from_plan( shot_data, bible, project_config, episode=episode, ) # ── Flat compiled prompt ───────────────────────────────────────── compiled_prompt = build_prompt_from_plan( shot_data, bible, project_config, episode=episode, ) # ── Formatter outputs ──────────────────────────────────────────── formatters = {} try: formatters["video"] = build_video_prompt_from_plan( shot_data, bible, project_config, episode=episode, ) except Exception as e: formatters["video"] = f"[error: {e}]" warnings.append(f"video formatter failed: {e}") try: formatters["kling_i2v"] = build_kling_i2v_prompt(shot_data) except Exception as e: formatters["kling_i2v"] = f"[error: {e}]" warnings.append(f"kling_i2v formatter failed: {e}") try: formatters["kling_t2v"] = build_kling_t2v_prompt(shot_data) except Exception as e: formatters["kling_t2v"] = f"[error: {e}]" warnings.append(f"kling_t2v formatter failed: {e}") # ── Formatter overrides (manual edits saved per-shot) ──────────── formatter_overrides = {} for fmt_key in ("video", "kling_i2v", "kling_t2v"): ov_key = f"formatter_{fmt_key}" ov_val = saved_overrides.get(ov_key) if ov_val is not None: formatter_overrides[fmt_key] = ov_val # ── Blocking data ──────────────────────────────────────────────── blocking = shot_data.get("blocking_metadata") # ── Bypass metadata ────────────────────────────────────────────── active_bypasses = sorted(_resolve_bypasses(shot_data)) bypass_mode = shot_data.get("prompt_overrides", {}).get("mode", "standard") available_presets = {k: sorted(v) for k, v in _OVERRIDE_PRESETS.items()} # ── Word count ─────────────────────────────────────────────────── word_count = len(compiled_prompt.split()) if compiled_prompt else 0 # ── Formatter limits from constants (single source of truth) ───── from recoil.core.prompt_config import load_constants formatter_limits = load_constants().get("formatter_limits", {}) return JSONResponse({ "shot_id": shot_id, "compiled_prompt": compiled_prompt, "layers": sections, "formatters": formatters, "formatter_overrides": formatter_overrides, "formatter_limits": formatter_limits, "blocking": blocking, "bypass_mode": bypass_mode, "active_bypasses": active_bypasses, "available_presets": available_presets, "word_count": word_count, "warnings": warnings, })