#!/usr/bin/env python3 """Internal UI for screenplay-to-webbook prototype runs.""" from __future__ import annotations import json import sqlite3 import subprocess import sys import threading from datetime import datetime from pathlib import Path from typing import Dict, List, Sequence from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile from fastapi.responses import RedirectResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates ROOT_DIR = Path(__file__).resolve().parents[1] UI_DIR = Path(__file__).resolve().parent DB_PATH = UI_DIR / "prototype_ui.sqlite3" SCREENPLAY_DIR = ROOT_DIR / "screenplay" UPLOADS_DIR = SCREENPLAY_DIR / "uploads" PROTOTYPE_DIR = ROOT_DIR / "prototype" PROMPT_MANIFEST_PATH = SCREENPLAY_DIR / "prototype_prompts" / "manifest.json" PIPELINE_SCRIPT = SCREENPLAY_DIR / "prototype_episode_novel.py" STATIC_DIR = UI_DIR / "static" TEMPLATES_DIR = UI_DIR / "templates" app = FastAPI(title="TheRoad Scene Prototype UI") app.mount("/ui-static", StaticFiles(directory=str(STATIC_DIR)), name="ui-static") app.mount("/prototype-assets", StaticFiles(directory=str(PROTOTYPE_DIR)), name="prototype-assets") templates = Jinja2Templates(directory=str(TEMPLATES_DIR)) RUN_LOCK = threading.Lock() RUN_THREADS: Dict[int, threading.Thread] = {} REVIEW_STATUSES = {"pending", "approved", "needs_fix"} def now_iso() -> str: return datetime.now().isoformat(timespec="seconds") def db_connect() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def init_db() -> None: UI_DIR.mkdir(parents=True, exist_ok=True) with db_connect() as conn: conn.executescript( """ CREATE TABLE IF NOT EXISTS runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_file TEXT NOT NULL, output_dir TEXT NOT NULL UNIQUE, prompt_version TEXT, web_episode_count INTEGER NOT NULL, sections_per_episode INTEGER NOT NULL, status TEXT NOT NULL, created_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, return_code INTEGER, log_path TEXT, summary_path TEXT, notes TEXT, detected_from_files INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS scene_reviews ( run_id INTEGER NOT NULL, scene_id TEXT NOT NULL, status TEXT NOT NULL, notes TEXT NOT NULL DEFAULT '', updated_at TEXT NOT NULL, PRIMARY KEY (run_id, scene_id), FOREIGN KEY (run_id) REFERENCES runs(id) ); """ ) def available_screenplays() -> List[str]: files = [path.relative_to(ROOT_DIR).as_posix() for path in sorted(SCREENPLAY_DIR.glob("*.pdf"))] files.extend(path.relative_to(ROOT_DIR).as_posix() for path in sorted(UPLOADS_DIR.glob("*.pdf"))) return files def prompt_versions() -> Dict[str, object]: manifest = json.loads(PROMPT_MANIFEST_PATH.read_text(encoding="utf-8")) return manifest def slug_from_path(value: str) -> str: cleaned = "".join(ch if ch.isalnum() else "_" for ch in Path(value).stem).strip("_").lower() return cleaned or "run" def make_output_dir(source_file: str) -> Path: stamp = datetime.now().strftime("%Y%m%d_%H%M%S") return PROTOTYPE_DIR / f"ui_{stamp}_{slug_from_path(source_file)}" def remap_workspace_path(path_value: str | Path, anchor_dir: str, target_root: Path) -> Path: path = Path(path_value) if path.exists(): return path.resolve() parts = path.parts if anchor_dir in parts: anchor_index = parts.index(anchor_dir) candidate = target_root / Path(*parts[anchor_index + 1 :]) if candidate.exists(): return candidate.resolve() return path def resolve_output_dir_path(output_dir_value: str | Path) -> Path: return remap_workspace_path(output_dir_value, "prototype", PROTOTYPE_DIR) def resolve_source_file_path(source_file: str) -> Path: source_path = Path(source_file) candidates = [] if source_path.is_absolute(): candidates.append(source_path) else: candidates.extend( [ ROOT_DIR / source_path, SCREENPLAY_DIR / source_path, SCREENPLAY_DIR / source_path.name, UPLOADS_DIR / source_path, UPLOADS_DIR / source_path.name, ] ) for candidate in candidates: if candidate.exists(): return candidate.resolve() raise FileNotFoundError(f"Unable to resolve source screenplay: {source_file}") def resolve_run_assets(run_row: sqlite3.Row) -> Dict[str, object]: output_dir = resolve_output_dir_path(run_row["output_dir"]) summary_path = output_dir / "metadata" / "prototype_summary.json" package_path = output_dir / "metadata" / "webbook_package.json" image_manifest_path = output_dir / "metadata" / "image_generation_manifest.json" scene_stills_path = output_dir / "analysis" / "scene_stills.json" result: Dict[str, object] = { "summary": None, "package": None, "image_manifest": None, "scene_cards": [], "pdf_urls": [], "reference_urls": [], } if summary_path.exists(): result["summary"] = json.loads(summary_path.read_text(encoding="utf-8")) if package_path.exists(): result["package"] = json.loads(package_path.read_text(encoding="utf-8")) if image_manifest_path.exists(): result["image_manifest"] = json.loads(image_manifest_path.read_text(encoding="utf-8")) scene_lookup: Dict[str, Dict[str, object]] = {} if scene_stills_path.exists(): scene_payload = json.loads(scene_stills_path.read_text(encoding="utf-8")) scene_lookup = {item["still_id"]: item for item in scene_payload.get("scene_stills", [])} if result["summary"]: for pdf_path in result["summary"].get("rendered_episode_pdfs", []): result["pdf_urls"].append( { "label": Path(pdf_path).name, "url": prototype_asset_url(pdf_path), } ) if result["package"] and result["image_manifest"]: scene_manifest = result["image_manifest"].get("scenes", {}) for episode in result["package"].get("episodes", []): for section in episode.get("sections", []): still_id = section["still_id"] still = scene_lookup.get(still_id, {}) scene_entry = scene_manifest.get(still_id, {}) result["scene_cards"].append( { "episode_number": episode["episode_number"], "episode_title": episode["title"], "still_id": still_id, "section_title": section["section_title"], "image_url": prototype_asset_url(scene_entry.get("path")), "beat_title": still.get("beat_title", ""), "screenplay_scene_heading": still.get("screenplay_scene_heading", ""), "still_prompt": still.get("still_frame_prompt_raw", ""), } ) for entity_id, entry in list(result["image_manifest"].get("entities", {}).items())[:24]: result["reference_urls"].append( { "entity_id": entity_id, "url": prototype_asset_url(entry.get("path")), } ) return result def prototype_asset_url(path_value: str | None) -> str | None: if not path_value: return None path = remap_workspace_path(path_value, "prototype", PROTOTYPE_DIR) if not path.is_absolute(): path = ROOT_DIR / path try: rel = path.relative_to(PROTOTYPE_DIR) except ValueError: return None return f"/prototype-assets/{rel.as_posix()}" def run_is_busy(run_row: sqlite3.Row) -> bool: return run_row["status"] in {"queued", "running"} def review_counts(scene_cards: Sequence[Dict[str, object]], reviews: Dict[str, sqlite3.Row]) -> Dict[str, int]: counts = {status: 0 for status in REVIEW_STATUSES} for card in scene_cards: scene_id = str(card["still_id"]) status = reviews.get(scene_id, {}).get("status", "pending") counts[status] = counts.get(status, 0) + 1 counts["total"] = len(scene_cards) return counts def write_review_notes_payload(output_dir: Path, reviews: Dict[str, sqlite3.Row], scene_ids: Sequence[str]) -> Path: payload = { scene_id: reviews.get(scene_id, {}).get("notes", "") for scene_id in scene_ids } notes_path = output_dir / "metadata" / f"scene_review_notes_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" notes_path.parent.mkdir(parents=True, exist_ok=True) notes_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return notes_path def sync_discovered_runs() -> None: init_db() summary_paths = sorted(PROTOTYPE_DIR.glob("*/metadata/prototype_summary.json")) with db_connect() as conn: for summary_path in summary_paths: summary = json.loads(summary_path.read_text(encoding="utf-8")) output_dir = summary_path.parents[1] package_path = output_dir / "metadata" / "webbook_package.json" prompt_version = None if package_path.exists(): package = json.loads(package_path.read_text(encoding="utf-8")) prompt_version = package.get("generation_metadata", {}).get("prompt_version") existing = conn.execute("SELECT id, status FROM runs WHERE output_dir = ?", (str(output_dir),)).fetchone() values = ( summary.get("source_file", output_dir.name), str(output_dir), prompt_version, summary.get("web_episode_count", 0), summary.get("sections_per_episode", 0), "completed", datetime.fromtimestamp(summary_path.stat().st_mtime).isoformat(timespec="seconds"), datetime.fromtimestamp(summary_path.stat().st_mtime).isoformat(timespec="seconds"), 0, str(output_dir / "run.log"), str(summary_path), 1, ) if existing: if existing["status"] != "running": conn.execute( """ UPDATE runs SET source_file=?, prompt_version=?, web_episode_count=?, sections_per_episode=?, status=?, finished_at=?, return_code=?, log_path=?, summary_path=?, detected_from_files=? WHERE output_dir=? """, ( values[0], values[2], values[3], values[4], values[5], values[7], values[8], values[9], values[10], values[11], str(output_dir), ), ) else: conn.execute( """ INSERT INTO runs ( source_file, output_dir, prompt_version, web_episode_count, sections_per_episode, status, created_at, finished_at, return_code, log_path, summary_path, detected_from_files ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, values, ) conn.commit() def list_runs() -> List[sqlite3.Row]: sync_discovered_runs() with db_connect() as conn: rows = conn.execute("SELECT * FROM runs ORDER BY created_at DESC, id DESC").fetchall() return rows def create_run(source_file: str, web_episode_count: int, sections_per_episode: int, prompt_version: str) -> int: output_dir = make_output_dir(source_file) output_dir.mkdir(parents=True, exist_ok=True) log_path = output_dir / "run.log" created_at = now_iso() with db_connect() as conn: cursor = conn.execute( """ INSERT INTO runs ( source_file, output_dir, prompt_version, web_episode_count, sections_per_episode, status, created_at, log_path, detected_from_files ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0) """, ( source_file, str(output_dir), prompt_version, web_episode_count, sections_per_episode, "queued", created_at, str(log_path), ), ) conn.commit() run_id = int(cursor.lastrowid) return run_id def update_run_status(run_id: int, **fields: object) -> None: if not fields: return columns = ", ".join(f"{key} = ?" for key in fields) values = list(fields.values()) + [run_id] with db_connect() as conn: conn.execute(f"UPDATE runs SET {columns} WHERE id = ?", values) conn.commit() def fetch_run(run_id: int) -> sqlite3.Row: sync_discovered_runs() with db_connect() as conn: row = conn.execute("SELECT * FROM runs WHERE id = ?", (run_id,)).fetchone() if not row: raise HTTPException(status_code=404, detail="Run not found.") return row def read_log_tail(log_path: str | None, line_limit: int = 120) -> str: if not log_path: return "" path = remap_workspace_path(log_path, "prototype", PROTOTYPE_DIR) if not path.exists(): return "" lines = path.read_text(encoding="utf-8", errors="replace").splitlines() return "\n".join(lines[-line_limit:]) def fetch_scene_reviews(run_id: int) -> Dict[str, sqlite3.Row]: with db_connect() as conn: rows = conn.execute("SELECT * FROM scene_reviews WHERE run_id = ?", (run_id,)).fetchall() return {row["scene_id"]: dict(row) for row in rows} def launch_run_thread( run_id: int, *, extra_args: Sequence[str] | None = None, append_log: bool = False, log_label: str = "full run", ) -> None: thread = threading.Thread( target=run_pipeline, args=(run_id, list(extra_args or []), append_log, log_label), daemon=True, ) with RUN_LOCK: RUN_THREADS[run_id] = thread thread.start() def run_pipeline(run_id: int, extra_args: List[str], append_log: bool, log_label: str) -> None: run = fetch_run(run_id) output_dir = resolve_output_dir_path(run["output_dir"]) log_path = remap_workspace_path(run["log_path"], "prototype", PROTOTYPE_DIR) try: source_path = resolve_source_file_path(run["source_file"]) except FileNotFoundError as exc: update_run_status(run_id, status="failed", finished_at=now_iso(), notes=str(exc), return_code=1) return update_run_status(run_id, status="running", started_at=now_iso(), finished_at=None, return_code=None) cmd = [ sys.executable, str(PIPELINE_SCRIPT), "--input", str(source_path), "--output-dir", str(output_dir), "--web-episode-count", str(run["web_episode_count"]), "--sections-per-episode", str(run["sections_per_episode"]), "--prompt-version", run["prompt_version"] or prompt_versions()["current_version"], ] cmd.extend(extra_args) output_dir.mkdir(parents=True, exist_ok=True) if append_log: with log_path.open("a", encoding="utf-8") as log_file: log_file.write(f"\n=== {now_iso()} | {log_label} ===\n") else: log_path.write_text("", encoding="utf-8") try: with log_path.open("a", encoding="utf-8") as log_file: process = subprocess.Popen( cmd, cwd=str(ROOT_DIR), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, ) assert process.stdout is not None for line in process.stdout: log_file.write(line) log_file.flush() return_code = process.wait() except Exception as exc: update_run_status(run_id, status="failed", finished_at=now_iso(), notes=str(exc), return_code=1) return summary_path = output_dir / "metadata" / "prototype_summary.json" if return_code == 0 and summary_path.exists(): update_run_status( run_id, status="completed", finished_at=now_iso(), return_code=0, summary_path=str(summary_path), ) else: update_run_status( run_id, status="failed", finished_at=now_iso(), return_code=return_code, ) sync_discovered_runs() @app.on_event("startup") def startup() -> None: init_db() UPLOADS_DIR.mkdir(parents=True, exist_ok=True) PROTOTYPE_DIR.mkdir(parents=True, exist_ok=True) sync_discovered_runs() @app.get("/") def index(request: Request): manifest = prompt_versions() runs = list_runs() return templates.TemplateResponse( "index.html", { "request": request, "screenplays": available_screenplays(), "runs": runs, "manifest": manifest, "current_prompt_version": manifest["current_version"], }, ) @app.post("/screenplays/upload") async def upload_screenplay(file: UploadFile = File(...)) -> RedirectResponse: if not file.filename or not file.filename.lower().endswith(".pdf"): raise HTTPException(status_code=400, detail="Only PDF files are supported.") UPLOADS_DIR.mkdir(parents=True, exist_ok=True) target = UPLOADS_DIR / f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{Path(file.filename).name}" target.write_bytes(await file.read()) return RedirectResponse(url="/", status_code=303) @app.post("/runs") def start_run( source_file: str = Form(...), web_episode_count: int = Form(...), sections_per_episode: int = Form(...), prompt_version: str = Form(...), ) -> RedirectResponse: if source_file not in available_screenplays(): raise HTTPException(status_code=404, detail="Screenplay not found.") run_id = create_run( source_file=source_file, web_episode_count=web_episode_count, sections_per_episode=sections_per_episode, prompt_version=prompt_version, ) launch_run_thread(run_id) return RedirectResponse(url=f"/runs/{run_id}", status_code=303) @app.get("/runs/{run_id}") def run_detail(request: Request, run_id: int): run = fetch_run(run_id) assets = resolve_run_assets(run) reviews = fetch_scene_reviews(run_id) return templates.TemplateResponse( "run_detail.html", { "request": request, "run": run, "assets": assets, "resolved_output_dir": str(resolve_output_dir_path(run["output_dir"])), "log_tail": read_log_tail(run["log_path"]), "reviews": reviews, "review_counts": review_counts(assets["scene_cards"], reviews), }, ) @app.post("/runs/{run_id}/scene-review") def save_scene_review( run_id: int, scene_id: str = Form(...), status: str = Form(...), notes: str = Form(""), ) -> RedirectResponse: fetch_run(run_id) if status not in REVIEW_STATUSES: raise HTTPException(status_code=400, detail="Invalid review status.") with db_connect() as conn: conn.execute( """ INSERT INTO scene_reviews (run_id, scene_id, status, notes, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(run_id, scene_id) DO UPDATE SET status=excluded.status, notes=excluded.notes, updated_at=excluded.updated_at """, (run_id, scene_id, status, notes, now_iso()), ) conn.commit() return RedirectResponse(url=f"/runs/{run_id}#scene-{scene_id}", status_code=303) @app.post("/runs/{run_id}/regenerate-needs-fix") def regenerate_needs_fix(run_id: int) -> RedirectResponse: run = fetch_run(run_id) if run_is_busy(run): raise HTTPException(status_code=409, detail="This run is already active.") assets = resolve_run_assets(run) if not assets["scene_cards"]: raise HTTPException(status_code=400, detail="This run has no reviewable scenes yet.") reviews = fetch_scene_reviews(run_id) scene_ids = [ card["still_id"] for card in assets["scene_cards"] if reviews.get(card["still_id"], {}).get("status") == "needs_fix" ] if not scene_ids: raise HTTPException(status_code=400, detail="No scenes are currently marked needs_fix.") notes_path = write_review_notes_payload(resolve_output_dir_path(run["output_dir"]), reviews, scene_ids) launch_run_thread( run_id, extra_args=["--only-scene-ids", ",".join(scene_ids), "--review-notes-path", str(notes_path)], append_log=True, log_label=f"needs_fix regeneration ({len(scene_ids)} scenes)", ) return RedirectResponse(url=f"/runs/{run_id}", status_code=303) @app.post("/runs/{run_id}/scene-regenerate") def regenerate_single_scene( run_id: int, scene_id: str = Form(...), ) -> RedirectResponse: run = fetch_run(run_id) if run_is_busy(run): raise HTTPException(status_code=409, detail="This run is already active.") assets = resolve_run_assets(run) valid_scene_ids = {card["still_id"] for card in assets["scene_cards"]} if scene_id not in valid_scene_ids: raise HTTPException(status_code=404, detail="Scene not found in this run.") reviews = fetch_scene_reviews(run_id) notes_path = write_review_notes_payload(resolve_output_dir_path(run["output_dir"]), reviews, [scene_id]) launch_run_thread( run_id, extra_args=["--only-scene-ids", scene_id, "--review-notes-path", str(notes_path)], append_log=True, log_label=f"scene regeneration ({scene_id})", ) return RedirectResponse(url=f"/runs/{run_id}#scene-{scene_id}", status_code=303)