# api/routes/overrides.py """Shot override CRUD + context preview endpoints.""" import copy import json import re from pathlib import Path from fastapi import APIRouter, Body, Depends, Query from fastapi.responses import JSONResponse from ..deps import get_project, get_paths, get_store router = APIRouter(tags=["overrides"]) def _overrides_path(plans_dir: Path, ep_num: int) -> Path: return plans_dir / f"ep_{ep_num:03d}_overrides.json" def _load_overrides(plans_dir: Path, ep_num: int) -> dict: p = _overrides_path(plans_dir, ep_num) if p.exists(): try: return json.loads(p.read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError): pass return {} def _save_overrides(plans_dir: Path, ep_num: int, all_overrides: dict): p = _overrides_path(plans_dir, ep_num) p.write_text(json.dumps(all_overrides, indent=2), encoding="utf-8") @router.post("/api/shot-override") def save_shot_override( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), ): """Save Tier 1 overrides for a shot.""" shot_id = body.get("shot_id") overrides = body.get("overrides", {}) if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) ep_match = re.match(r"EP(\d+)", shot_id) if not ep_match: return JSONResponse({"error": f"Invalid shot_id: {shot_id}"}, status_code=400) ep_num = int(ep_match.group(1)) all_ov = _load_overrides(paths["plans_dir"], ep_num) existing = all_ov.get(shot_id, {}) existing.update(overrides) # Remove keys set to None (director clearing an override) existing = {k: v for k, v in existing.items() if v is not None} all_ov[shot_id] = existing _save_overrides(paths["plans_dir"], ep_num, all_ov) return JSONResponse({"status": "saved", "shot_id": shot_id, "overrides": existing}) @router.get("/api/shot-override") def get_shot_override( shot_id: str = Query(...), project: str = Depends(get_project), paths: dict = Depends(get_paths), ): """Get overrides for a shot.""" ep_match = re.match(r"EP(\d+)", shot_id) if not ep_match: return JSONResponse({"error": f"Invalid shot_id: {shot_id}"}, status_code=400) ep_num = int(ep_match.group(1)) all_ov = _load_overrides(paths["plans_dir"], ep_num) return JSONResponse({ "shot_id": shot_id, "overrides": all_ov.get(shot_id, {}), }) @router.post("/api/preview-context") def preview_context( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), ): """Dry-run context assembly — returns slot previews + conflict flags.""" shot_id = body.get("shot_id") if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) ep_match = re.match(r"EP(\d+)", shot_id) if not ep_match: return JSONResponse({"error": f"Invalid shot_id: {shot_id}"}, status_code=400) ep_num = int(ep_match.group(1)) # Load plan + shot data plan_path = paths["plans_dir"] / f"ep_{ep_num:03d}_plan.json" shot_data = None 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: shot_data = copy.deepcopy(s) break except (json.JSONDecodeError, IOError): pass if not shot_data: return JSONResponse({"error": f"Shot not found in plan: {shot_id}"}, status_code=404) # Apply overrides use_overrides = body.get("use_overrides", True) overrides_applied = {} if use_overrides: all_ov = _load_overrides(paths["plans_dir"], ep_num) overrides_applied = all_ov.get(shot_id, {}) if overrides_applied: from recoil.pipeline._lib.previz_context import apply_overrides shot_data = apply_overrides(shot_data, overrides_applied) # Build previews from recoil.pipeline._lib.previz_context import ( build_behavioral_preamble, build_generative_directive, resolve_previz_character_refs, resolve_location_refs, ) # Load bible for directive bible = None if paths["bible_path"].exists(): try: bible = json.loads(paths["bible_path"].read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError): pass preamble = build_behavioral_preamble(shot_data) directive = build_generative_directive(shot_data, bible=bible) # Resolve refs (metadata only — no bytes) asset_data = shot_data.get("asset_data", {}) routing_data = shot_data.get("routing_data", {}) is_env = routing_data.get("is_env_only", False) char_refs = resolve_previz_character_refs(shot_data, project=project) char_summary = [] for _, mime, label in char_refs: char_summary.append(label) location_id = asset_data.get("location_id", "") loc_refs = resolve_location_refs(location_id, project=project) if location_id else [] sec_loc_refs = [] for sec_id in asset_data.get("secondary_location_ids", []): sec_loc_refs.extend(resolve_location_refs(sec_id, project=project, max_refs=1)) # Conflict detection conflicts = [] chars = asset_data.get("characters", []) if is_env and len(chars) > 0: conflicts.append({ "level": "error", "msg": "is_env_only is true but characters are assigned. " "Preamble will include ENV restrictions. Directive will say 'no people in frame'." }) if not is_env and len(chars) == 0: shot_type = shot_data.get("prompt_data", {}).get("shot_type", "") if shot_type not in ("INSERT", "EST", "EWS"): conflicts.append({ "level": "warning", "msg": f"Non-ENV shot (type: {shot_type}) with no characters. No identity refs will be injected." }) # Check prompt_override vs characters prompt_override = body.get("prompt_text", "") or overrides_applied.get("prompt_override", "") if prompt_override: # Load casting state for character name detection casting_path = paths["plans_dir"].parent / "casting_state.json" if casting_path.exists(): try: casting = json.loads(casting_path.read_text(encoding="utf-8")) known_chars = list(casting.get("characters", {}).keys()) char_ids_in_shot = { (c.get("char_id", "") if isinstance(c, dict) else str(c)).upper() for c in chars } for name in known_chars: if name.lower() in prompt_override.lower() and name.upper() not in char_ids_in_shot: conflicts.append({ "level": "warning", "msg": f"Prompt mentions '{name}' but character is not in the shot. No identity refs for {name}." }) except (json.JSONDecodeError, IOError): pass # Check for "no people" in directive when not env if not is_env and "no people" in directive.lower(): conflicts.append({ "level": "error", "msg": "is_env_only is false but directive still contains 'no people'. Override may not have propagated." }) return JSONResponse({ "shot_id": shot_id, "overrides_applied": overrides_applied, "is_env_only": is_env, "shot_type": shot_data.get("prompt_data", {}).get("shot_type", ""), "characters": [ (c.get("char_id") if isinstance(c, dict) else str(c)) for c in chars ], "locations": { "primary": location_id, "secondary": asset_data.get("secondary_location_ids", []), }, "ref_counts": { "character_refs": len(char_refs), "location_refs": len(loc_refs), "secondary_location_refs": len(sec_loc_refs), }, "preamble_preview": preamble[:800], "directive_preview": directive[:800], "conflicts": conflicts, })