#!/usr/bin/env python3 """Editor Hub — local dev server for Recoil visual editors. Serves index.html, editor HTML files, and project data APIs on 127.0.0.1:8420. """ import json import os import socket import subprocess import sys from http.server import HTTPServer, SimpleHTTPRequestHandler from pathlib import Path from urllib.parse import unquote, urlparse # Add recoil root to path so core.model_profiles resolves cleanly. _recoil_root = str(Path(__file__).resolve().parent.parent) if _recoil_root not in sys.path: sys.path.insert(0, _recoil_root) from recoil.core.model_profiles import get_model from recoil.core.paths import ProjectPaths PORT = 8420 HOST = "127.0.0.1" # ── Project root detection ────────────────────────────────────────── # Engine root: walk up from editors/ → find tools/ + editors/ siblings EDITORS_DIR = Path(__file__).resolve().parent PROJECT_ROOT = None candidate = EDITORS_DIR for _ in range(10): # New layout: editors/ and tools/ are siblings at engine root if (candidate / "tools").is_dir() and (candidate / "editors").is_dir(): PROJECT_ROOT = candidate break # Legacy layout: marker if (candidate / "_engine").is_dir(): PROJECT_ROOT = candidate break candidate = candidate.parent if PROJECT_ROOT is None: print("ERROR: Could not locate engine root.", file=sys.stderr) sys.exit(1) # Projects directory: env var > sibling projects/ > PROJECT_ROOT itself (legacy) _projects_env = os.environ.get("RECOIL_PROJECTS_DIR", "") PROJECTS_DIR = Path(_projects_env) if _projects_env else Path("__nonexistent__") if not PROJECTS_DIR.is_dir(): PROJECTS_DIR = PROJECT_ROOT.parent / "projects" if not PROJECTS_DIR.is_dir(): PROJECTS_DIR = PROJECT_ROOT # fallback for legacy layout # Directories to skip when scanning for projects SKIP_DIRS = { "_engine", "_development", "_docs", "archives", "archive", "Pitch", "Posters", "inbox", ".claude", ".git", "tools", "lib", "agents", "editors", "templates", "schemas", "skills", "lenses", "evaluation", "calibration", "docs", "config", } IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp"} SKIP_FILENAMES = {"comparison", "triptych", "grid", "contact_sheet", "side_by_side"} ANGLE_KEYWORDS = [ ("closeup_front", "closeup_front"), ("closeup_three_quarter", "closeup_three_quarter"), ("three_quarter_right", "three_quarter_right"), ("three_quarter_left", "three_quarter_left"), ("full_body_three_quarter", "full_body_three_quarter"), ("full_body", "full_body"), ("profile_right", "profile_right"), ("profile_left", "profile_left"), ("back_right", "back_right"), ("back_left", "back_left"), ("back_center", "back"), ("back", "back"), ("low_angle", "low_angle"), ("high_angle", "high_angle"), ("front", "front"), ] EXPRESSION_KEYWORDS = ["neutral", "tired", "focused", "wary", "exhausted", "angry", "determined"] def _build_manifest(cand_dir, character): """Auto-generate a picker manifest by scanning shootout/, picks/, and keystones/. Includes file modification timestamps and deduplicates across runs: when multiple files share the same (angle, expression, engine) combo, only the newest file is kept. """ from datetime import datetime, timezone raw = [] scan_dirs = [ ("shootout", cand_dir / "shootout"), ("picks", cand_dir / "picks"), ("keystones", cand_dir / "keystones"), ("imported", cand_dir / "imported"), ] for source, scan_dir in scan_dirs: if not scan_dir.is_dir(): continue for root, _dirs, files in os.walk(scan_dir): for fname in sorted(files): fpath = Path(root) / fname if fpath.suffix.lower() not in IMAGE_EXTS: continue if any(skip in fname.lower() for skip in SKIP_FILENAMES): continue stat = fpath.stat() if stat.st_size < 10_000: continue rel = str(fpath.relative_to(cand_dir)) mtime = stat.st_mtime modified = datetime.fromtimestamp(mtime, tz=timezone.utc).isoformat() # Infer angle name_lower = fpath.stem.lower().replace("-", "_") angle = "unknown" for kw, val in ANGLE_KEYWORDS: if kw in name_lower: angle = val break # Infer expression expression = "unknown" for expr in EXPRESSION_KEYWORDS: if expr in name_lower: expression = expr break # Infer engine engine = "unknown" path_lower = str(fpath).lower() if "nbp" in path_lower or "gemini" in path_lower: engine = "gemini" elif "qwen" in path_lower: engine = "qwen" elif "seedvr2" in path_lower: engine = "seedvr2" elif source == "keystones": engine = "midjourney" elif source == "shootout": engine = "shootout" raw.append({ "filename": rel, "angle": angle, "expression": expression, "pass": 0, "engine": engine, "source": source, "modified": modified, "_mtime": mtime, }) # Deduplicate: for each (angle, expression, engine) combo, keep only the # newest file. keystones and picks are exempt from dedup (always kept). best = {} exempt = [] for entry in raw: if entry["source"] in ("keystones", "picks", "imported"): exempt.append(entry) continue key = (entry["angle"], entry["expression"], entry["engine"]) if key not in best or entry["_mtime"] > best[key]["_mtime"]: best[key] = entry deduped = exempt + list(best.values()) # Sort by mtime descending (newest first) then assign indices deduped.sort(key=lambda e: e["_mtime"], reverse=True) candidates = [] for idx, entry in enumerate(deduped): entry.pop("_mtime", None) entry["index"] = idx candidates.append(entry) return { "character": character, "target_model": "z_image", "generated": "auto", "pipeline": "auto_scan", "candidate_count": len(candidates), "candidates": candidates, } def scan_projects(): """Scan PROJECTS_DIR for project directories containing episodes/, treatment.md, or *.fountain.""" projects = [] for entry in sorted(PROJECTS_DIR.iterdir()): if not entry.is_dir(): continue if entry.name in SKIP_DIRS or entry.name.startswith("."): continue has_episodes = (entry / "episodes").is_dir() has_treatment = (entry / "treatment.md").is_file() has_breakdown = (entry / "visual" / "breakdown.json").is_file() fountain_files = sorted(f.name for f in entry.glob("*.fountain")) storyboard_files = sorted(f.name for f in (entry / "storyboards").glob("storyboard_*.json")) if (entry / "storyboards").is_dir() else [] episode_count = 0 if has_episodes: episode_count = len(list((entry / "episodes").glob("ep_*.md"))) has_development = (entry / "development").is_dir() # Only include dirs that look like actual projects if not (has_episodes or has_treatment or fountain_files or has_development): continue # Detect LoRA candidate directories lora_candidates = [] lora_dir = entry / "visual" / "lora_candidates" if lora_dir.is_dir(): for char_dir in sorted(lora_dir.iterdir()): if char_dir.is_dir() and (char_dir / "manifest.json").is_file(): lora_candidates.append(char_dir.name) projects.append({ "name": entry.name, "episodes": episode_count, "treatment": has_treatment, "breakdown": has_breakdown, "fountain_files": fountain_files, "storyboard_files": storyboard_files, "lora_candidates": lora_candidates, }) # Sort: projects with breakdown first, then alphabetically projects.sort(key=lambda p: (not p["breakdown"], p["name"])) return projects class EditorHubHandler(SimpleHTTPRequestHandler): """Custom handler for the editor hub server.""" def __init__(self, *args, **kwargs): # Serve static files from the editors directory super().__init__(*args, directory=str(EDITORS_DIR), **kwargs) def end_headers(self): self.send_header("Access-Control-Allow-Origin", "http://127.0.0.1:8430") self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, PATCH, DELETE, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.send_header("Cache-Control", "no-cache, no-store, must-revalidate") super().end_headers() def log_message(self, format, *args): # Compact logging print(f" {args[0]}" if args else "") def _proxy_starsend(self, method="GET", body=None): """Proxy requests to the local Starsend server so remote clients can reach it.""" import urllib.request parsed = urlparse(self.path) # Strip /starsend-proxy prefix, forward the rest to localhost:8430 target_path = parsed.path[len("/starsend-proxy"):] if parsed.query: target_path += "?" + parsed.query target_url = f"http://127.0.0.1:8430{target_path}" try: req = urllib.request.Request(target_url, data=body, method=method) req.add_header("Content-Type", "application/json") with urllib.request.urlopen(req, timeout=30) as resp: data = resp.read() self.send_response(resp.status) self.send_header("Content-Type", resp.getheader("Content-Type", "application/json")) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) except Exception as e: self._error(502, f"Starsend proxy error: {e}") def do_GET(self): parsed = urlparse(self.path) path = unquote(parsed.path).rstrip("/") # ── Starsend Proxy ───────────────────────────────────── if path.startswith("/starsend-proxy/"): self._proxy_starsend("GET") return # ── API Routes ────────────────────────────────────────── if path == "/api/projects": self._json_response(scan_projects()) return # /api/project//... if path.startswith("/api/project/"): parts = path[len("/api/project/"):].split("/", 3) project_name = parts[0] project_dir = PROJECTS_DIR / project_name if not project_dir.is_dir(): self._error(404, f"Project not found: {project_name}") return if len(parts) >= 2: resource = parts[1] # GET /api/project//lora-candidates/[/selection] if resource == "lora-candidates" and len(parts) >= 3: character = parts[2] cand_dir = project_dir / "visual" / "lora_candidates" / character try: cand_dir_resolved = cand_dir.resolve() if not str(cand_dir_resolved).startswith(str(project_dir.resolve())): self._error(403, "Path traversal not allowed") return except Exception: self._error(400, "Invalid path") return sub_resource = parts[3] if len(parts) == 4 else None if sub_resource == "selection": sel_path = cand_dir / "selection.json" if sel_path.is_file(): self._file_response(sel_path, "application/json") else: self._json_response({"character": character, "selections": {}}) elif sub_resource is None: # Always auto-generate manifest (fresh timestamps + dedup) manifest = _build_manifest(cand_dir, character) self._json_response(manifest) else: self._error(400, f"Unknown sub-resource: {sub_resource}") return if resource == "breakdown": bd_path = project_dir / "visual" / "breakdown.json" if bd_path.is_file(): self._file_response(bd_path, "application/json") else: self._error(404, "No breakdown.json found") return if resource == "storyboard" and len(parts) == 3: filename = parts[2] sb_path = project_dir / "storyboards" / filename if sb_path.is_file(): self._file_response(sb_path, "application/json") else: self._error(404, f"Storyboard not found: {filename}") return # GET /api/project//review/ if resource == "review" and len(parts) == 3: filename = parts[2] review_path = project_dir / "storyboards" / "reviews" / filename try: review_path = review_path.resolve() if not str(review_path).startswith(str(project_dir.resolve())): self._error(403, "Path traversal not allowed") return except Exception: self._error(400, "Invalid path") return if review_path.is_file(): self._file_response(review_path, "application/json") else: self._error(404, f"Review not found: {filename}") return # GET /api/project//dailies/ if resource == "dailies" and len(parts) == 3: try: episode = int(parts[2]) except ValueError: self._error(400, "Invalid episode number") return ep_str = str(episode).zfill(3) assets_dir = project_dir / "storyboards" / "assets" / f"ep_{ep_str}" files = {} if assets_dir.is_dir(): for f in sorted(assets_dir.iterdir()): if f.is_file(): files[f.name] = { "path": f"storyboards/assets/ep_{ep_str}/{f.name}", "size": f.stat().st_size, "ext": f.suffix.lstrip("."), } self._json_response({ "episode": episode, "assets_dir": f"storyboards/assets/ep_{ep_str}", "file_count": len(files), "files": files, }) return # GET /api/project//corpus-summary if resource == "corpus-summary": try: summary = self._build_corpus_summary() self._json_response(summary) except Exception as e: self._error(500, f"Corpus summary failed: {e}") return # GET /api/project//score/ if resource == "score" and len(parts) == 3: try: episode = int(parts[2]) except ValueError: self._error(400, "Invalid episode number") return try: score = self._build_episode_score(project_dir, episode) self._json_response(score) except Exception as e: self._error(500, f"Score calculation failed: {e}") return # GET /api/project//pipeline-status if resource == "pipeline-status": try: status = self._build_pipeline_status(project_dir, project_name) self._json_response(status) except Exception as e: self._error(500, f"Pipeline status failed: {e}") return # GET /api/project//visual-bible if resource == "visual-bible": vb_path = project_dir / "visual_bible.md" config_path = project_dir / "visual" / "project_config.json" result = { "has_visual_bible": vb_path.is_file(), "has_project_config": config_path.is_file(), "markdown": "", "project_config": {}, } if vb_path.is_file(): result["markdown"] = vb_path.read_text(encoding="utf-8") if config_path.is_file(): try: result["project_config"] = json.loads(config_path.read_text(encoding="utf-8")) except json.JSONDecodeError: result["project_config"] = {} self._json_response(result) return if resource == "annotations": ann_path = ProjectPaths.from_root(project_dir).state_dir / "script_doctor_annotations.json" if ann_path.is_file(): self._file_response(ann_path, "application/json") else: self._error(404, "No annotations file found") return if resource == "fountain" and len(parts) == 3: filename = parts[2] ft_path = project_dir / filename if ft_path.is_file() and ft_path.suffix == ".fountain": self._file_response(ft_path, "text/plain; charset=utf-8") else: self._error(404, f"Fountain file not found: {filename}") return # GET /api/project//shootouts/ if resource == "shootouts" and len(parts) >= 3: character = parts[2] shootout_dir = project_dir / "visual" / "lora_candidates" / character / "shootout" try: shootout_resolved = shootout_dir.resolve() if not str(shootout_resolved).startswith(str(project_dir.resolve())): self._error(403, "Path traversal not allowed") return except Exception: self._error(400, "Invalid path") return runs = [] if shootout_dir.is_dir(): for d in shootout_dir.iterdir(): if d.is_dir() and d.name != "__pycache__": results_path = d / "results.json" if results_path.is_file(): try: data = json.loads(results_path.read_text()) if not isinstance(data, dict): continue except (json.JSONDecodeError, ValueError): continue data["dirname"] = d.name data["images"] = [ f.name for f in sorted(d.iterdir()) if f.suffix.lower() in (".png", ".jpeg", ".jpg", ".webp") ] runs.append(data) # Sort by timestamp (newest first) runs.sort(key=lambda r: r.get("timestamp", ""), reverse=True) # Load notes if they exist notes_path = shootout_dir / "shootout_notes.json" notes = {} if notes_path.is_file(): try: notes = json.loads(notes_path.read_text()) except json.JSONDecodeError: pass self._json_response({ "character": character, "runs": runs, "notes": notes, }) return # GET /api/project//shot-lab/ if resource == "shot-lab" and len(parts) >= 3: try: episode = int(parts[2]) except ValueError: self._error(400, "Invalid episode number") return sub_resource = parts[3] if len(parts) == 4 else None # GET /api/project//shot-lab//status if sub_resource == "status": self._json_response(self._shot_lab_status(project_dir, episode)) return # GET /api/project//shot-lab/ — load storyboard + assets if sub_resource is None: try: data = self._shot_lab_load(project_dir, project_name, episode) self._json_response(data) except Exception as e: self._error(500, f"Shot Lab load failed: {e}") return self._error(400, f"Unknown shot-lab sub-resource: {sub_resource}") return # GET /api/project//episodes//content if resource == "episodes" and len(parts) >= 4 and parts[3] == "content": ep_str = parts[2].zfill(3) ep_file = project_dir / "episodes" / f"ep_{ep_str}.md" if ep_file.is_file(): self._json_response({"content": ep_file.read_text(encoding="utf-8")}) else: self._error(404, f"Episode not found: ep_{ep_str}") return # GET /api/project//episodes//annotations if resource == "episodes" and len(parts) >= 4 and parts[3] == "annotations": ep_str = parts[2].zfill(3) ann_file = project_dir / "annotations" / f"ep_{ep_str}.json" if ann_file.is_file(): try: data = json.loads(ann_file.read_text(encoding="utf-8")) self._json_response(data) except json.JSONDecodeError: self._json_response({"annotations": []}) else: self._json_response({"annotations": []}) return # Serve project files (images, manifests, etc.) # /api/project//file/ if resource == "file" and len(parts) >= 3: rel_path = "/".join(parts[2:]) file_path = project_dir / rel_path # Security: ensure path stays within project dir try: file_path = file_path.resolve() if not str(file_path).startswith(str(project_dir.resolve())): self._error(403, "Path traversal not allowed") return except Exception: self._error(400, "Invalid path") return if file_path.is_file(): mime_map = { ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp", ".json": "application/json", ".svg": "image/svg+xml", ".mp4": "video/mp4", ".html": "text/html; charset=utf-8", } mime = mime_map.get(file_path.suffix.lower(), "application/octet-stream") self._file_response(file_path, mime) else: self._error(404, f"File not found: {rel_path}") return self._error(400, "Invalid API path") return # ── /files/ — serve any file from PROJECTS_DIR or PROJECT_ROOT ──── if path.startswith("/files/"): rel_path = path[len("/files/"):] # Try PROJECTS_DIR first, fall back to PROJECT_ROOT file_path = PROJECTS_DIR / rel_path if not file_path.exists(): file_path = PROJECT_ROOT / rel_path try: file_path = file_path.resolve() allowed_roots = [str(PROJECTS_DIR.resolve()), str(PROJECT_ROOT.resolve())] if not any(str(file_path).startswith(r) for r in allowed_roots): self._error(403, "Path traversal not allowed") return except Exception: self._error(400, "Invalid path") return if file_path.is_file(): mime_map = { ".html": "text/html; charset=utf-8", ".css": "text/css", ".js": "application/javascript", ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp", ".svg": "image/svg+xml", ".json": "application/json", ".mp4": "video/mp4", ".woff2": "font/woff2", } mime = mime_map.get(file_path.suffix.lower(), "application/octet-stream") self._file_response(file_path, mime) else: self._error(404, f"File not found: {rel_path}") return # ── Static files ──────────────────────────────────────── if path == "" or path == "/": self.path = "/prepro-console.html" # Fall through to _standalone/ for files not in editors root file_path = urlparse(self.path).path requested = EDITORS_DIR / file_path.lstrip("/") if not requested.is_file(): standalone = EDITORS_DIR / "_standalone" / file_path.lstrip("/") if standalone.is_file(): self.path = "/_standalone" + self.path super().do_GET() def do_POST(self): parsed = urlparse(self.path) path = unquote(parsed.path).rstrip("/") # ── Starsend Proxy ───────────────────────────────────── if path.startswith("/starsend-proxy/"): length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length) if length > 0 else None self._proxy_starsend("POST", body) return # ── POST /api/ai/rewrite ────────────────────────────────────── if path == "/api/ai/rewrite": content_length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(content_length)) if content_length > 0 else {} self._handle_ai_rewrite(body) return if path.startswith("/api/project/"): parts = path[len("/api/project/"):].split("/", 3) project_name = parts[0] project_dir = PROJECTS_DIR / project_name if not project_dir.is_dir(): self._error(404, f"Project not found: {project_name}") return # POST /api/project//lora-candidates//selection if (len(parts) == 4 and parts[1] == "lora-candidates" and parts[3] == "selection"): character = parts[2] cand_dir = project_dir / "visual" / "lora_candidates" / character sel_path = cand_dir / "selection.json" try: sel_path_resolved = (cand_dir / "selection.json").resolve() if not str(sel_path_resolved).startswith(str(project_dir.resolve())): self._error(403, "Path traversal not allowed") return except Exception: self._error(400, "Invalid path") return if not cand_dir.is_dir(): self._error(404, f"No candidates dir for {character}") return content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: json.loads(body) sel_path.write_bytes(body) self._json_response({"status": "saved", "character": character}) except json.JSONDecodeError: self._error(400, "Invalid JSON body") except Exception as e: self._error(500, str(e)) return # POST /api/project//episodes//annotations if len(parts) >= 4 and parts[1] == "episodes" and parts[3] == "annotations": ep_str = parts[2].zfill(3) ann_dir = project_dir / "annotations" ann_file = ann_dir / f"ep_{ep_str}.json" content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: data = json.loads(body) ann_dir.mkdir(parents=True, exist_ok=True) ann_file.write_text(json.dumps(data, indent=2), encoding="utf-8") self._json_response({"status": "saved", "path": str(ann_file.relative_to(PROJECTS_DIR))}) except json.JSONDecodeError: self._error(400, "Invalid JSON body") except Exception as e: self._error(500, str(e)) return # POST /api/project//review/ if len(parts) >= 3 and parts[1] == "review": filename = parts[2] review_dir = project_dir / "storyboards" / "reviews" review_path = review_dir / filename # Security: ensure path stays within project dir try: review_dir.mkdir(parents=True, exist_ok=True) review_path = review_path.resolve() if not str(review_path).startswith(str(project_dir.resolve())): self._error(403, "Path traversal not allowed") return except Exception: self._error(400, "Invalid path") return # Read body content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: # Validate JSON json.loads(body) review_path.write_bytes(body) self._json_response({"status": "saved", "path": str(review_path.relative_to(PROJECT_ROOT))}) except json.JSONDecodeError: self._error(400, "Invalid JSON body") except Exception as e: self._error(500, str(e)) return # POST /api/project//run-gate/ if len(parts) == 3 and parts[1] == "run-gate": try: episode = int(parts[2]) except ValueError: self._error(400, "Invalid episode number") return gate_script = PROJECT_ROOT / "tools" / "visual_gate.py" if not gate_script.is_file(): self._error(404, "visual_gate.py not found") return try: result = subprocess.run( [sys.executable, str(gate_script), "batch", "--project", project_name, "--episode", str(episode)], capture_output=True, text=True, timeout=600, cwd=str(PROJECT_ROOT), ) # Read the output file if it was created ep_str = str(episode).zfill(3) output_path = project_dir / "storyboards" / "reviews" / f"visual_gate_ep_{ep_str}.json" if output_path.is_file(): with open(output_path) as f: gate_data = json.load(f) self._json_response({"status": "completed", "results": gate_data}) else: self._json_response({ "status": "completed", "stdout": result.stdout[-2000:] if result.stdout else "", "stderr": result.stderr[-2000:] if result.stderr else "", "returncode": result.returncode, }) except subprocess.TimeoutExpired: self._error(504, "Gate run timed out (10 min limit)") except Exception as e: self._error(500, f"Gate run failed: {e}") return # POST /api/project//save-annotations/ if len(parts) == 3 and parts[1] == "save-annotations": try: episode = int(parts[2]) except ValueError: self._error(400, "Invalid episode number") return ep_str = str(episode).zfill(3) sb_dir = project_dir / "storyboards" sb_file = sb_dir / f"storyboard_ep_{ep_str}.json" ann_file = sb_dir / f"annotations_ep_{ep_str}.json" archive_dir = sb_dir / "archive" content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: data = json.loads(body) # Archive current storyboard if it exists archived_path = None if sb_file.is_file(): archive_dir.mkdir(parents=True, exist_ok=True) from datetime import datetime ts = datetime.now().strftime("%Y%m%d_%H%M%S") archive_name = f"storyboard_ep_{ep_str}_{ts}.json" archived = archive_dir / archive_name import shutil shutil.copy2(str(sb_file), str(archived)) archived_path = str(archived.relative_to(PROJECT_ROOT)) # Save annotations sb_dir.mkdir(parents=True, exist_ok=True) ann_file.write_text(json.dumps(data, indent=2)) self._json_response({ "status": "saved", "annotations_path": str(ann_file.relative_to(PROJECT_ROOT)), "archived_storyboard": archived_path, }) except json.JSONDecodeError: self._error(400, "Invalid JSON body") except Exception as e: self._error(500, str(e)) return # POST /api/project//save-episode/ if len(parts) == 3 and parts[1] == "save-episode": try: episode = int(parts[2]) except ValueError: self._error(400, "Invalid episode number") return ep_str = str(episode).zfill(3) ep_file = project_dir / "episodes" / f"ep_{ep_str}.md" content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: data = json.loads(body) content = data.get("content", "") ep_file.parent.mkdir(parents=True, exist_ok=True) ep_file.write_text(content, encoding="utf-8") self._json_response({ "status": "saved", "path": str(ep_file.relative_to(PROJECT_ROOT)), }) except json.JSONDecodeError: self._error(400, "Invalid JSON body") except Exception as e: self._error(500, str(e)) return # POST /api/project//shootouts//notes if (len(parts) >= 4 and parts[1] == "shootouts" and parts[3] == "notes"): character = parts[2] shootout_dir = project_dir / "visual" / "lora_candidates" / character / "shootout" try: shootout_resolved = shootout_dir.resolve() if not str(shootout_resolved).startswith(str(project_dir.resolve())): self._error(403, "Path traversal not allowed") return except Exception: self._error(400, "Invalid path") return if not shootout_dir.is_dir(): self._error(404, f"No shootout dir for {character}") return content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: new_data = json.loads(body) notes_path = shootout_dir / "shootout_notes.json" # Merge with existing notes existing = {} if notes_path.is_file(): try: existing = json.loads(notes_path.read_text()) except json.JSONDecodeError: pass run_dirname = new_data.get("run_dirname") if run_dirname: from datetime import datetime existing[run_dirname] = { "winner": new_data.get("winner"), "notes": new_data.get("notes", ""), "updated": datetime.now().isoformat(), } notes_path.write_text(json.dumps(existing, indent=2)) self._json_response({"status": "saved", "character": character}) except json.JSONDecodeError: self._error(400, "Invalid JSON body") except Exception as e: self._error(500, str(e)) return # POST /api/project//preview-prompt if len(parts) >= 2 and parts[1] == "preview-prompt": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: data = json.loads(body) except json.JSONDecodeError: self._error(400, "Invalid JSON body") return episode = data.get("episode") shot_id = data.get("shot_id") model = data.get("model", "z_image") frame_type = data.get("frame_type", "hero") storyboard_file = data.get("storyboard_file") if episode is None or shot_id is None: self._error(400, "Missing required fields: episode, shot_id") return try: result = self._preview_prompt( project_dir, episode, shot_id, model, frame_type, storyboard_file ) self._json_response(result) except Exception as e: self._error(500, f"Preview failed: {e}") return # POST /api/project//visual-bible if len(parts) >= 2 and parts[1] == "visual-bible": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: data = json.loads(body) except json.JSONDecodeError: self._error(400, "Invalid JSON body") return try: markdown = data.get("markdown", "") project_config_updates = data.get("project_config", {}) # Write visual_bible.md vb_path = project_dir / "visual_bible.md" vb_path.write_text(markdown, encoding="utf-8") # Update project_config.json if provided if project_config_updates: config_path = project_dir / "visual" / "project_config.json" existing_config = {} if config_path.is_file(): try: existing_config = json.loads(config_path.read_text(encoding="utf-8")) except json.JSONDecodeError: pass existing_config.update(project_config_updates) config_path.parent.mkdir(parents=True, exist_ok=True) config_path.write_text(json.dumps(existing_config, indent=2), encoding="utf-8") self._json_response({ "status": "saved", "visual_bible_path": str(vb_path.relative_to(PROJECT_ROOT)), }) except Exception as e: self._error(500, str(e)) return # POST /api/project//annotations # Full sync: replaces annotations array in script_doctor_annotations.json if len(parts) >= 2 and parts[1] == "annotations": content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: data = json.loads(body) new_annotations = data.get("annotations", []) state_dir = ProjectPaths.from_root(project_dir).state_dir state_dir.mkdir(parents=True, exist_ok=True) ann_path = state_dir / "script_doctor_annotations.json" # Preserve existing metadata if file exists existing = {} if ann_path.is_file(): with open(ann_path) as f: existing = json.load(f) existing["annotations"] = new_annotations new_edits = data.get("edits") if new_edits is not None: existing["edits"] = new_edits existing.setdefault("project", project_name) from datetime import datetime existing["updated_at"] = datetime.now().isoformat() ann_path.write_text(json.dumps(existing, indent=2)) self._json_response({"status": "saved", "count": len(new_annotations), "edits": len(new_edits) if new_edits else 0}) except json.JSONDecodeError: self._error(400, "Invalid JSON body") except Exception as e: self._error(500, str(e)) return # POST /api/project//shot-lab//compile if (len(parts) >= 4 and parts[1] == "shot-lab" and parts[3] == "compile"): try: episode = int(parts[2]) except ValueError: self._error(400, "Invalid episode number") return content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: data = json.loads(body) shot_ids = data.get("shot_ids", []) model = data.get("model", "z_image") result = self._shot_lab_compile(project_dir, episode, shot_ids, model) self._json_response(result) except json.JSONDecodeError: self._error(400, "Invalid JSON body") except Exception as e: self._error(500, f"Compile failed: {e}") return # POST /api/project//shot-lab//generate if (len(parts) >= 4 and parts[1] == "shot-lab" and parts[3] == "generate"): try: episode = int(parts[2]) except ValueError: self._error(400, "Invalid episode number") return content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length) try: data = json.loads(body) self._json_response({ "status": "not_implemented", "message": "Generation trigger not yet wired — use CLI for now: " f"python3 generate_storyboard_keyframes.py {project_name} " f"-e {episode} --shots {','.join(str(s) for s in data.get('shot_ids', []))}", }) except json.JSONDecodeError: self._error(400, "Invalid JSON body") except Exception as e: self._error(500, f"Generate failed: {e}") return self._error(404, "POST not supported for this path") def do_OPTIONS(self): self.send_response(204) self.end_headers() def _json_response(self, data): body = json.dumps(data, indent=2).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _handle_ai_rewrite(self, body): """Call Gemini 2.5 Pro to generate 2 rewrite proposals for selected text.""" import urllib.request api_key = os.environ.get("GOOGLE_API_KEY") if not api_key: self._error(500, "GOOGLE_API_KEY not set") return selected_text = body.get("selected_text", "") line_type = body.get("line_type", "action") character = body.get("character") scene_context = body.get("scene_context", "") project = body.get("project", "") system_prompt = ( f"You are an expert script doctor for the vertical microdrama '{project}'.\n" f"Provide EXACTLY TWO rewrite proposals for the selected {line_type} text.\n" f"Rules:\n" f"1. Preserve the line type. If the original is DIALOGUE, write DIALOGUE. If ACTION, write ACTION.\n" f"2. Proposal 1: subtle polish (tighter, more natural).\n" f"3. Proposal 2: stronger swing (more subtext, punchier, more stylized).\n" f"4. Keep word counts similar to the original. Vertical microdramas require fast pacing.\n" f"5. Output ONLY valid JSON: {{\"proposals\": [\"rewrite 1\", \"rewrite 2\"]}}\n" ) user_content = json.dumps({ "scene_context": scene_context[:2000], "selected_text": selected_text, "line_type": line_type, "character": character, }) try: url = f"https://generativelanguage.googleapis.com/v1beta/models/{get_model('pro', 'text')}:generateContent?key={api_key}" req_data = json.dumps({ "systemInstruction": {"parts": [{"text": system_prompt}]}, "contents": [{"role": "user", "parts": [{"text": user_content}]}], "generationConfig": { "maxOutputTokens": 300, "responseMimeType": "application/json", "responseSchema": { "type": "OBJECT", "properties": { "proposals": { "type": "ARRAY", "items": {"type": "STRING"} } }, "required": ["proposals"] } } }).encode("utf-8") req = urllib.request.Request( url, headers={"content-type": "application/json"}, data=req_data, ) with urllib.request.urlopen(req, timeout=30) as resp: result = json.loads(resp.read().decode("utf-8")) content_text = result["candidates"][0]["content"]["parts"][0]["text"] parsed = json.loads(content_text) self._json_response(parsed) except json.JSONDecodeError: import re try: m = re.search(r'\{[^}]*"proposals"[^}]*\}', content_text) if m: self._json_response(json.loads(m.group())) else: self._error(500, "AI returned invalid JSON") except Exception: self._error(500, "AI returned invalid JSON") except Exception as e: self._error(500, str(e)) def _file_response(self, filepath, content_type): try: body = filepath.read_bytes() self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) except Exception as e: self._error(500, str(e)) def _build_corpus_summary(self): """Build aggregate corpus stats from Visual Grammar Bible JSON files.""" corpus_dir = PROJECT_ROOT.parent / "docs" / "research" / "visual_grammar_bible" / "corpus" if not corpus_dir.is_dir(): return {"error": "No corpus directory found", "scenes": 0} scenes = [] all_shot_types = {} all_angles = {} all_movements = {} all_dof = {} all_lens = {} total_shots = 0 medium_asl = {} # medium -> [shot_durations] for f in sorted(corpus_dir.glob("*.json")): try: with open(f) as fh: data = json.load(fh) except Exception: continue scene_meta = data.get("scene", {}) shots = data.get("shots", []) source = scene_meta.get("source", {}) medium = source.get("medium", "unknown") duration = scene_meta.get("duration_seconds", 0) shot_count = len(shots) if shot_count == 0: continue total_shots += shot_count asl = round(duration * 1000 / shot_count) if shot_count > 0 else 0 scenes.append({ "id": f.stem, "title": scene_meta.get("title", f.stem), "medium": medium, "duration": duration, "shots": shot_count, "asl_ms": asl, "genres": scene_meta.get("genre_tags", []), }) if medium not in medium_asl: medium_asl[medium] = [] medium_asl[medium].append(asl) for shot in shots: st = (shot.get("shot_type") or "").upper() if st: all_shot_types[st] = all_shot_types.get(st, 0) + 1 angle = shot.get("camera_angle", "") if angle: all_angles[angle] = all_angles.get(angle, 0) + 1 move = shot.get("camera_movement", "") if move: all_movements[move] = all_movements.get(move, 0) + 1 dof = shot.get("depth_of_field", "") if dof: all_dof[dof] = all_dof.get(dof, 0) + 1 lens = shot.get("lens_feel", "") if lens: all_lens[lens] = all_lens.get(lens, 0) + 1 # Compute medium averages medium_avg = {} for m, vals in medium_asl.items(): medium_avg[m] = round(sum(vals) / len(vals)) if vals else 0 return { "total_scenes": len(scenes), "total_shots": total_shots, "scenes": scenes, "shot_types": all_shot_types, "camera_angles": all_angles, "camera_movements": all_movements, "depth_of_field": all_dof, "lens_feel": all_lens, "asl_by_medium": medium_avg, } def _build_episode_score(self, project_dir, episode): """Compare a storyboard's shot distribution against corpus targets.""" ep_str = str(episode).zfill(3) sb_dir = project_dir / "storyboards" # Find storyboard file sb_file = None if sb_dir.is_dir(): for f in sb_dir.glob("*.json"): if f"ep_{ep_str}" in f.name or f"ep{ep_str}" in f.name: sb_file = f break if not sb_file or not sb_file.is_file(): return {"episode": episode, "has_storyboard": False, "deviations": []} with open(sb_file) as fh: sb_data = json.load(fh) shots = sb_data.get("shots", []) total = len(shots) if total == 0: return {"episode": episode, "has_storyboard": True, "total_shots": 0, "deviations": []} # Count shot types type_counts = {} static_count = 0 single_count = 0 for s in shots: direction = s.get("direction", {}) st = (direction.get("shot_type") or s.get("shot_type") or "").upper() type_counts[st] = type_counts.get(st, 0) + 1 movement = (direction.get("camera_movement") or s.get("camera_movement") or "").lower() if movement in ("static", ""): static_count += 1 subjects = direction.get("subjects", s.get("subjects", 1)) if isinstance(subjects, (int, float)) and subjects <= 1: single_count += 1 # Corpus targets targets = { "MCU": 41.3, "MS": 36.3, "CU": 10.7, "MLS": 6.1, "WIDE": 2.2, "INSERT": 2.2, } deviations = [] for stype, target_pct in targets.items(): actual_pct = (type_counts.get(stype, 0) / total) * 100 delta = actual_pct - target_pct threshold = 10 if target_pct > 10 else 5 if abs(delta) > threshold: deviations.append({ "metric": f"shot_type_{stype}", "actual": round(actual_pct, 1), "target": target_pct, "delta": round(delta, 1), "severity": "warn" if abs(delta) < threshold * 2 else "error", }) static_pct = (static_count / total) * 100 if static_pct < 60: deviations.append({ "metric": "static_camera", "actual": round(static_pct, 1), "target": 77.0, "delta": round(static_pct - 77.0, 1), "severity": "warn", }) single_pct = (single_count / total) * 100 if single_pct < 60: deviations.append({ "metric": "single_subject", "actual": round(single_pct, 1), "target": 75.5, "delta": round(single_pct - 75.5, 1), "severity": "warn", }) return { "episode": episode, "has_storyboard": True, "total_shots": total, "shot_type_distribution": {k: round((v / total) * 100, 1) for k, v in type_counts.items()}, "static_camera_pct": round(static_pct, 1), "single_subject_pct": round(single_pct, 1), "deviations": deviations, } def _build_pipeline_status(self, project_dir, project_name): """Build per-episode pipeline status.""" episodes_dir = project_dir / "episodes" sb_dir = project_dir / "storyboards" has_breakdown = (project_dir / "visual" / "breakdown.json").is_file() status = {} if not episodes_dir.is_dir(): return {"project": project_name, "episodes": status} ep_files = sorted(episodes_dir.glob("ep_*.md")) for ep_file in ep_files: try: ep_num = int(ep_file.stem.split("_")[1]) except (IndexError, ValueError): continue ep_str = str(ep_num).zfill(3) has_sb = False shot_count = 0 if sb_dir.is_dir(): for f in sb_dir.glob("*.json"): if f"ep_{ep_str}" in f.name or f"ep{ep_str}" in f.name: has_sb = True try: with open(f) as fh: sb = json.load(fh) shot_count = len(sb.get("shots", [])) except Exception: pass break # Check dailies assets_dir = sb_dir / "assets" / f"ep_{ep_str}" if sb_dir.is_dir() else None has_dailies = assets_dir.is_dir() and any(assets_dir.iterdir()) if assets_dir else False status[ep_num] = { "breakdown": has_breakdown, "storyboard": has_sb, "dailies": has_dailies, "shots": shot_count, } return {"project": project_name, "episodes": status} def _preview_prompt(self, project_dir, episode, shot_id, model, frame_type, storyboard_file): """Run the prompt compiler and return the resolved prompt for a shot.""" from prompt_compiler import ( compile as compile_prompt, _load_breakdown, _load_project_config, OverrideStore, PreviousShotContext, build_prev_context, ) # Load breakdown + config breakdown = _load_breakdown(project_dir) project_config = _load_project_config(project_dir) override_store = OverrideStore(project_dir) # Load LoRA registry lora_registry = {} lora_path = project_dir / "visual" / "lora_registry.json" if lora_path.is_file(): with open(lora_path) as f: lora_registry = json.load(f) # Find storyboard ep_str = f"{episode:03d}" if storyboard_file: sb_path = project_dir / "storyboards" / storyboard_file else: sb_path = project_dir / "storyboards" / f"storyboard_ep_{ep_str}.json" if not sb_path.is_file(): return {"error": f"Storyboard not found: {sb_path.name}"} with open(sb_path) as f: storyboard = json.load(f) # Find the shot shot = None for s in storyboard.get("shots", []): if s.get("id") == shot_id: shot = s break if shot is None: return {"error": f"Shot {shot_id} not found in storyboard"} # Build previous shot context for transition-aware prompts prev_shot = None for s in storyboard.get("shots", []): if s.get("id") == shot_id - 1: prev_shot = s break prev_context = None if prev_shot and not shot.get("scene_break_before"): prev_context = build_prev_context( prev_shot, breakdown, episode, storyboard, model, lora_registry, project_config, ) overrides = override_store.list_all() if override_store else None result = compile_prompt( shot=shot, breakdown=breakdown, episode=episode, storyboard=storyboard, model=model, lora_registry=lora_registry, overrides=overrides, project_config=project_config, previous_shot_context=prev_context, frame_type=frame_type, ) return result # ── Shot Lab Helpers ────────────────────────────────────────────── def _shot_lab_load(self, project_dir, project_name, episode): """Load storyboard + compiled prompts + existing assets for Shot Lab.""" ep_str = f"{episode:03d}" sb_path = project_dir / "storyboards" / f"storyboard_ep_{ep_str}.json" if not sb_path.is_file(): return {"error": f"Storyboard not found: storyboard_ep_{ep_str}.json"} with open(sb_path) as f: storyboard = json.load(f) shots = storyboard.get("shots", []) # Load LoRA registry lora_registry = {} lora_path = project_dir / "visual" / "lora_registry.json" if lora_path.is_file(): with open(lora_path) as f: lora_data = json.load(f) lora_registry = lora_data.get("characters", lora_data) # Check existing assets assets_dir = project_dir / "storyboards" / "assets" / f"ep_{ep_str}" existing_assets = {} if assets_dir.is_dir(): for f in sorted(assets_dir.iterdir()): if f.is_file(): existing_assets[f.name] = { "path": f"storyboards/assets/ep_{ep_str}/{f.name}", "size": f.stat().st_size, "ext": f.suffix.lstrip("."), } # Build per-shot summary shot_summaries = [] for shot in shots: shot_id = shot.get("id") gen_approach = shot.get("generation_approach", "standard_flf") chars = [c.lower() for c in shot.get("characters_in_shot", [])] asset_base = shot.get("asset_name", f"shot_{shot_id:02d}") # Determine status from existing assets has_strip = f"{asset_base}_strip.png" in existing_assets has_first = f"{asset_base}_first.png" in existing_assets has_last = f"{asset_base}_last.png" in existing_assets has_mid = f"{asset_base}_mid.png" in existing_assets has_first_hq = f"{asset_base}_first_hq.png" in existing_assets has_video = any(k.startswith(asset_base) and k.endswith(".mp4") for k in existing_assets) if has_video: status = "complete" elif has_first_hq: status = "upscale_done" elif has_first or has_strip: status = "keyframes_done" else: status = "pending" # LoRA info per character lora_info = {} for char in chars: char_reg = lora_registry.get(char, {}) lora_info[char] = { "trigger": char_reg.get("trigger", ""), "has_z_image": bool(char_reg.get("z_image_t2i", {}).get("path")), "has_flux2": bool(char_reg.get("t2i", {}).get("path")), "has_video": bool( char_reg.get("video", {}).get("high_noise_path") or char_reg.get("video", {}).get("low_noise_path") ), } shot_summaries.append({ "id": shot_id, "name": shot.get("name", ""), "beat": shot.get("beat", ""), "generation_approach": gen_approach, "characters": chars, "status": status, "lora_info": lora_info, "has_hero_frame": bool(shot.get("hero_frame")), "has_triptych_prompt": bool(shot.get("triptych_prompt")), "first_frame": shot.get("first_frame", ""), "hero_frame": shot.get("hero_frame", ""), "last_frame": shot.get("last_frame", ""), "action": shot.get("action", ""), "motion_prompt": shot.get("motion_prompt", ""), "shot_type": shot.get("shot_type", ""), "camera_movement": shot.get("camera_movement", ""), "focal_length": shot.get("focal_length", ""), "continuity_from": shot.get("continuity_from"), "same_angle_from": shot.get("same_angle_from"), "reference_image_url": shot.get("reference_image_url"), "seed": shot.get("seed"), "asset_name": asset_base, }) return { "project": project_name, "episode": episode, "storyboard_file": sb_path.name, "total_shots": len(shots), "shots": shot_summaries, "existing_assets": existing_assets, "lora_registry": { char: { "trigger": data.get("trigger", ""), "has_z_image": bool(data.get("z_image_t2i", {}).get("path")), "has_flux2": bool(data.get("t2i", {}).get("path")), } for char, data in lora_registry.items() }, "storyboard_meta": { "version": storyboard.get("version"), "location": storyboard.get("location", ""), "cinematic": storyboard.get("cinematic", ""), }, } def _shot_lab_status(self, project_dir, episode): """Check generation status for an episode.""" ep_str = f"{episode:03d}" assets_dir = project_dir / "storyboards" / "assets" / f"ep_{ep_str}" result = {"episode": episode, "assets_dir_exists": assets_dir.is_dir(), "files": {}} if assets_dir.is_dir(): for f in sorted(assets_dir.iterdir()): if f.is_file(): result["files"][f.name] = { "size": f.stat().st_size, "modified": f.stat().st_mtime, } return result def _shot_lab_compile(self, project_dir, episode, shot_ids, model): """Compile prompts for specific shots.""" from prompt_compiler import ( compile as compile_prompt, _load_breakdown, _load_project_config, OverrideStore, ) breakdown = _load_breakdown(project_dir) project_config = _load_project_config(project_dir) override_store = OverrideStore(project_dir) lora_registry = {} lora_path = project_dir / "visual" / "lora_registry.json" if lora_path.is_file(): with open(lora_path) as f: lora_registry = json.load(f) ep_str = f"{episode:03d}" sb_path = project_dir / "storyboards" / f"storyboard_ep_{ep_str}.json" if not sb_path.is_file(): return {"error": f"Storyboard not found"} with open(sb_path) as f: storyboard = json.load(f) overrides = override_store.list_all() if override_store else None results = {} for shot in storyboard.get("shots", []): if shot_ids and shot.get("id") not in shot_ids: continue shot_id = shot.get("id") shot_results = {} for frame_type in ["first", "mid", "last", "hero"]: try: result = compile_prompt( shot=shot, breakdown=breakdown, episode=episode, storyboard=storyboard, model=model, lora_registry=lora_registry, overrides=overrides, project_config=project_config, frame_type=frame_type, ) shot_results[frame_type] = { "prompt": result.get("prompt", ""), "negative_prompt": result.get("negative_prompt", ""), "word_count": len(result.get("prompt", "").split()), } except Exception as e: shot_results[frame_type] = {"error": str(e)} results[shot_id] = shot_results return {"episode": episode, "model": model, "compiled": results} def _error(self, code, message): body = json.dumps({"error": message}).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def port_in_use(port, host="127.0.0.1"): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: return s.connect_ex((host, port)) == 0 def main(): if port_in_use(PORT, HOST): print(f"Port {PORT} already in use — server may already be running.") sys.exit(0) server = HTTPServer((HOST, PORT), EditorHubHandler) print(f"Editor Hub serving on http://{HOST}:{PORT}") print(f"Engine root: {PROJECT_ROOT}") print(f"Projects dir: {PROJECTS_DIR}") print(f"Press Ctrl+C to stop.\n") try: server.serve_forever() except KeyboardInterrupt: print("\nShutting down.") server.shutdown() if __name__ == "__main__": main()