DaVinci Resolve MCP

#!/usr/bin/env python3 """Local dashboard for single-user media analysis batch jobs.""" from __future__ import annotations import argparse import functools import hashlib import json import os import re import sqlite3 import sys import threading import time import webbrowser from concurrent.futures import ThreadPoolExecutor from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any, Dict, List, Optional, Tuple from urllib.parse import parse_qs, unquote, urlparse from src.utils.media_analysis import ( _write_json as _atomic_write_json, analysis_index_status, analysis_root_coverage, build_analysis_index, detect_capabilities, query_analysis_index, resolve_output_root, clip_directory_hash, load_clip_index, stable_clip_directory, stable_clip_match_hashes, ) from src.utils.media_analysis_jobs import ( MEDIA_EXTENSIONS, batch_job_status, cancel_batch_job, create_batch_job_from_paths, list_batch_jobs, project_root_for_dashboard, resume_batch_job, run_batch_job_slice, ) from src.utils.platform import setup_environment from src.utils.analysis_memory import read_panel_state, write_panel_state from src.utils import brain_edits as _brain_edits from src.utils import timeline_versioning as _timeline_versioning from src.utils import timeline_brain_db as _timeline_brain_db HTML = r""" DaVinci Resolve MCP

Overview

Waiting for first refresh

Active Project Checking Resolve connection pending

Source Clips 0 Source clips only

Sequences 0 Read-only context

Media Status Checking Inventory pending

Waiting for Resolve media inventory.

Resolve Projects

Open the full database view to inspect projects outside the current Project Manager folder.

Current Project Checking Current folder pending

Open Projects 0 Shown in the project dropdown

Database Projects 0 Database pending

Analysis Contexts 0 Existing local reports and indexes

All Database Projects

This view walks Resolve Project Manager folders read-only, then uses confirmation before loading a different project in Resolve.

Filter

Open Projects to scan the current Resolve database.

Inventory

Resolve media is inventoried read-only and filtered to source clips so timelines, compounds, titles, and generated items stay out of analysis queues.

Resolve connection pending

Find Bin Media type Clip status Analysis status Poll

Auto refresh Resolve clips

waiting for first refresh

Checking Resolve media pool.

0 selected No analyzable clips chosen yet.

Review

Bin overview · click a clip to drill into shot detail.

Loading analyzed clips…

Resolve 21 AI Console

Run Resolve's local AI operations on the current Media Pool folder or a specific clip. These run on Resolve's GPU/AI engine — the analysis and slate ops are safe and reversible; motion-deblur and speech generation create new media files and ask for confirmation first. Source media is never modified. Every run is recorded in the Resolve 21 AI ops ledger (Preferences → Caps + Safety).

Checking which AI methods this Resolve build exposes…

Governance

Per-session limits for the two media-creating ops (deblur, speech). In Advisory mode you're warned in the confirm dialog but never blocked; in Enforce mode an over-tier run is refused until you raise the tier, relax the mode, or consciously override. Pick the tier that matches the job.

Target

Folder ops run on the current Media Pool folder in Resolve. Choose Specific clip and paste a clip id (from media_pool.get_clips) to target one clip.

Current folder Specific clip Clip id

Analysis

Safe, reversible. IntelliSearch and Slate require their AI Extras (Extras Download Manager).

Identify faces Better mode

Marker

Speaker detection

Motion deblur

Renders new deblurred media. Creates files; asks for confirmation. Leave fields blank for Resolve defaults.

Format Codec Extreme mode Use mark in/out Render at source res

Speech generator

AI text-to-speech. Requires the AI Speech Generator Extra. Creates a new audio item; asks for confirmation.

Text Voice model Speed Pitch Variation Add to timeline Timecode Audio track

Session

Quiet Resolve's background tasks for this session before heavy work. Resets on restart.

Result

No op run yet.

Resolve

Current read-only Resolve connection, product/version, project inventory, and any warnings from the media pool probe.

Resolve diagnostics pending.

OK Healthy Warn Needs attention Error Action required Idle Not yet probed

MCP

DaVinci Resolve MCP server identity, detected Resolve scripting paths, and per-client install status for every supported harness (Claude Desktop, Claude Code, Cursor, VS Code, Windsurf, Cline, Roo Code, Zed, Continue, Antigravity).

MCP diagnostics pending.

Clients

Install writes a managed davinci-resolve entry into the client's MCP config file. Remove reverses it. Existing entries in the file are preserved.

Scanning clients…

Analysis Storage

Project analysis paths used for reports, the durable job database, and the local search index.

Storage diagnostics pending.

Tools

Runtime helpers detected by the dashboard. Missing tools may reduce technical metadata or media probing depth.

Tool diagnostics pending.

Media Pool History

Provenance log for destructive media-pool operations — deletes, replaces, relinks. Each row captures the action, the target clip or folder, the analysis run that triggered it, and the initiator. Kept separate from timeline brain_edits because the addressable entity is a media pool item, not a timeline.

loading…

Reference

Sections

Load a document.

Loading README

Choose a document.

Server Defaults

These preferences are server-wide defaults for this local MCP install. Dashboard convenience settings stay in this browser.

Loading setup defaults

Analysis

Defaults used when agents run analysis or publish analysis-backed outputs without specifying every option.

Vision default Transcription default Slate detection Source trust Default analysis depth Default sample frames Frame sampling mode

Frames / minute Frame floor Frame ceiling

Persistence Summary style Report format

Caps + Safety

Token + frame budgets for analysis, plus safety rails for destructive edits. Usage is tracked per project in _soul/timeline_brain.sqlite.

Budget preset

Pick the preset that matches the job. Override individual fields below if needed.

Vision token usage

Live consumption against the active caps. Green = comfortable, amber = approaching, red = at or over budget.

Per‑clip —

Per‑job —

Today —

Daily vision-token usage (last 30 days)

loading…

Resolve 21 AI ops

Local Resolve AI operations (audio classification, IntelliSearch, slate, motion-deblur, speech). These run on Resolve's GPU/AI engine and do not consume the Claude analysis token budget above — tracked here for invocations, time, and files created.

loading…

Safety

Auto-archive timelines before destructive edits and refuse high-blast-radius ops when archive fails.

Auto-save Resolve project after each archive When on, project.SaveProject() fires after every version-on-mutate archive. Protects against Resolve crashes losing history.

Strict mode — always on for:

timeline.delete_timelines
timeline.delete_track
timeline.delete_clips with ripple=true

Strict mode REFUSES the underlying call if the pre-mutation archive can't be created. Other destructive ops degrade silently — pass strict=true in any action's params to opt in to refusal for that call.

Advanced — per-field overrides Leave blank to use the preset value shown in the placeholder. Type a number or unlimited to override.

Response chars Vision tokens / clip Frames / clip Vision tokens / job Vision tokens / day Wall clock seconds / call Max frame dimension (px)

Inspector & debug Look up usage for a specific clip or batch, browse refusals, or reset today's day-scope rollup.

Clip id or Job id

enter a clip_id or job_id and press Look up

Recent caps refusals

loading…

Metadata And Markers

Controls for Resolve metadata writes and source-time marker suggestions. Source media remains read-only.

Timed markers Overwrite policy Max markers per clip

Marker custom data Metadata fields

Marker types

Marker colors

Include confidence scores Include source time notes Ask before metadata publish Dry run first

Paths And Workflow

Optional locations and Resolve navigation behavior after operations complete.

Preferred analysis root

Generated media folder

Inventory limit Inventory exclude bins (comma separated) Post-operation page

Where files live

Locations the dashboard uses for server defaults and browser-local preferences. Read-only diagnostics.

Loading preference storage.

MCP Updates

Server-wide update-check behavior for the MCP package. Checks are best-effort and should not block startup.

Update policy Channel Check interval hours Snooze hours

Apply update

Loading status…

Stash local changes if dirty Override active-job lock

Update history

loading…

""" def _safe_call(obj: Any, method_name: str, *args: Any) -> Tuple[Any, Optional[str]]: if obj is None or not hasattr(obj, method_name): return None, f"{method_name} unavailable" try: return getattr(obj, method_name)(*args), None except Exception as exc: return None, str(exc) def _safe_name(obj: Any, fallback: str = "Untitled") -> str: value, _ = _safe_call(obj, "GetName") return str(value or fallback) def _safe_id(obj: Any) -> Optional[str]: value, _ = _safe_call(obj, "GetUniqueId") return str(value) if value else None # ── Resolve scripting API serialization ───────────────────────────────────── # The dashboard runs on a ThreadingHTTPServer, so /api/boot, /api/projects and # /api/resolve/media can land on separate threads concurrently (especially at # startup). DaVinci's scripting API is not thread-safe, so every entry point that # talks to it acquires this re-entrant lock for the full duration of its calls. _RESOLVE_API_LOCK = threading.RLock() _RESOLVE_ENV_READY = False def _serialize_resolve(func): """Decorator: hold the Resolve API lock for the whole call.""" @functools.wraps(func) def wrapper(*args, **kwargs): with _RESOLVE_API_LOCK: return func(*args, **kwargs) return wrapper def _connect_resolve_read_only() -> Tuple[Any, Optional[str]]: global _RESOLVE_ENV_READY with _RESOLVE_API_LOCK: # Environment + sys.path setup is pure overhead and never goes stale, so # run it once per process rather than on every connection. if not _RESOLVE_ENV_READY: try: setup_environment() modules_path = os.environ.get("RESOLVE_SCRIPT_API") if modules_path: candidate = os.path.join(modules_path, "Modules") if candidate not in sys.path: sys.path.append(candidate) _RESOLVE_ENV_READY = True except Exception as exc: return None, f"Resolve scripting API unavailable: {exc}" try: import DaVinciResolveScript as dvr_script # type: ignore except Exception as exc: return None, f"Resolve scripting API unavailable: {exc}" try: resolve = dvr_script.scriptapp("Resolve") except Exception as exc: return None, f"Resolve connection failed: {exc}" if resolve is None: return None, "DaVinci Resolve is not connected. Open Resolve Studio with a project loaded." return resolve, None @_serialize_resolve def _current_resolve_project_id() -> Tuple[Optional[str], Optional[str]]: """(project_id, error) for the currently-open Resolve project. A handful of cheap API calls — used by the media-poll reuse path to detect when the user has switched projects in Resolve since the inventory was cached, without paying for a full Media Pool walk. """ resolve, error = _connect_resolve_read_only() if error or resolve is None: return None, error or "Resolve unavailable" pm, pm_error = _safe_call(resolve, "GetProjectManager") if not pm or pm_error: return None, pm_error or "Project manager unavailable" project, _ = _safe_call(pm, "GetCurrentProject") if not project: return None, "No Resolve project open" return _safe_id(project), None @_serialize_resolve def _resolve_ai_features(resolve: Any) -> Dict[str, Any]: """Report which Resolve 21.0 AI scripting methods are available on the connected build, plus the Extra each AI-gated method requires. Presence is detected via getattr (no Resolve round-trips beyond fetching the handles), so this stays cheap enough for the boot handshake. """ def has(obj: Any, name: str) -> bool: return bool(obj) and callable(getattr(obj, name, None)) project = folder = None try: pm = resolve.GetProjectManager() project = pm.GetCurrentProject() if pm else None mp = project.GetMediaPool() if project else None folder = mp.GetRootFolder() if mp else None except Exception: pass features = { "disable_background_tasks": has(resolve, "DisableBackgroundTasksForCurrentResolveSession"), "generate_speech": has(project, "GenerateSpeech"), "perform_audio_classification": has(folder, "PerformAudioClassification"), "clear_audio_classification": has(folder, "ClearAudioClassification"), "analyze_for_intellisearch": has(folder, "AnalyzeForIntellisearch"), "analyze_for_slate": has(folder, "AnalyzeForSlate"), "remove_motion_blur": has(folder, "RemoveMotionBlur"), } return { "features": features, "available_count": sum(1 for v in features.values() if v), # Methods that additionally need an Extras download to actually run. "requires_extra": { "analyze_for_intellisearch": "AI IntelliSearch", "analyze_for_slate": "AI Slate ID", "generate_speech": "AI Speech Generator", }, } def _resolve_identity() -> Dict[str, Any]: resolve, error = _connect_resolve_read_only() if not resolve: return {"available": False, "error": error} product, _ = _safe_call(resolve, "GetProductName") version_string, _ = _safe_call(resolve, "GetVersionString") version_tuple, _ = _safe_call(resolve, "GetVersion") page, _ = _safe_call(resolve, "GetCurrentPage") return { "available": True, "product": str(product) if product else "DaVinci Resolve", "version_string": str(version_string) if version_string else None, "version": list(version_tuple) if isinstance(version_tuple, (list, tuple)) else None, "page": str(page) if page else None, "ai_features": _resolve_ai_features(resolve), } # ── Resolve 21 AI Console: op dispatch ────────────────────────────────────── # Folder/clip-level ops are routed to the consolidated `folder` / # `media_pool_item` tools; project/resolve-level ops to their tools. The # consolidated tools own the confirm-token gate for the two media-creators, so # this dispatcher just relays params (incl. confirm_token) and the result. _AI_CONSOLE_FOLDER_OPS = frozenset({ "perform_audio_classification", "clear_audio_classification", "analyze_for_intellisearch", "analyze_for_slate", "remove_motion_blur", "transcribe_audio", "clear_transcription", }) def _run_resolve_ai_op(body: Dict[str, Any]) -> Dict[str, Any]: """Dispatch one AI Console op to the right consolidated server tool. body = {op, target?, params?}. target is 'folder' (current Media Pool folder, default) or 'clip' (params.clip_id required). Returns the tool's response verbatim — including a {status:'confirmation_required', confirm_token, preview} shape for the gated media-creating ops. """ op = (body.get("op") or "").strip() target = (body.get("target") or "folder").strip() params = dict(body.get("params") or {}) if not op: return {"success": False, "error": "op is required"} try: from src.server import ( folder as _folder_tool, media_pool_item as _mpi_tool, project_settings as _ps_tool, resolve_control as _rc_tool, ) except Exception as exc: # pragma: no cover - import guard return {"success": False, "error": f"server tools unavailable: {exc}"} if op == "disable_background_tasks": return _rc_tool("disable_background_tasks_for_current_session", {}) if op == "generate_speech": return _ps_tool("generate_speech", params) if op not in _AI_CONSOLE_FOLDER_OPS: return {"success": False, "error": f"unknown op {op!r}"} if target == "clip": clip_id = params.get("clip_id") or body.get("clip_id") if not clip_id: return {"success": False, "error": "clip target requires a clip_id"} params["clip_id"] = clip_id return _mpi_tool(op, params) # default: operate on the current Media Pool folder return _folder_tool(op, params) def _clip_props(clip: Any) -> Dict[str, Any]: props, _ = _safe_call(clip, "GetClipProperty", "") return props if isinstance(props, dict) else {} def _first_prop(props: Dict[str, Any], keys: Tuple[str, ...]) -> Any: for key in keys: value = props.get(key) if value not in (None, ""): return value return None # ── File-existence probing ────────────────────────────────────────────────── # stat() calls on mounted network storage dominate inventory time (300+ source # clips on a Z:\ share can take tens of seconds serially). We probe paths in a # thread pool and memoize results for a short TTL so the recurring media poll # does not re-stat unchanged paths every few seconds. _PATH_EXISTS_TTL = 60.0 _PATH_PROBE_WORKERS = 16 _PATH_EXISTS_CACHE: Dict[str, Tuple[float, bool]] = {} _PATH_EXISTS_LOCK = threading.Lock() def _cached_path_exists(path: str, now: float, ttl: float) -> Optional[bool]: with _PATH_EXISTS_LOCK: entry = _PATH_EXISTS_CACHE.get(path) if entry is not None and (now - entry[0]) <= ttl: return entry[1] return None def _store_path_exists(path: str, exists: bool, now: float) -> None: with _PATH_EXISTS_LOCK: _PATH_EXISTS_CACHE[path] = (now, exists) def _probe_paths_exist(paths: Any, *, probe: bool = True, ttl: float = _PATH_EXISTS_TTL) -> Dict[str, bool]: """Resolve a collection of file paths to existence booleans. With ``probe=True`` (first load / manual refresh) any cache entry older than ``ttl`` is re-stat'd, and uncached paths are probed in parallel. With ``probe=False`` (background poll) the filesystem is never touched: cached values are reused at any age and unknown paths fall back to ``True`` — Resolve's own online/offline Status property still flags clips it knows are missing, so we trust it rather than paying for a network round-trip on every poll. """ distinct = {str(p) for p in paths if p} result: Dict[str, bool] = {} to_probe: List[str] = [] now = time.time() lookup_ttl = ttl if probe else float("inf") for path in distinct: cached = _cached_path_exists(path, now, lookup_ttl) if cached is not None: result[path] = cached elif probe: to_probe.append(path) else: result[path] = True if to_probe: workers = max(1, min(_PATH_PROBE_WORKERS, len(to_probe))) with ThreadPoolExecutor(max_workers=workers) as pool: for path, exists in zip(to_probe, pool.map(os.path.exists, to_probe)): result[path] = bool(exists) _store_path_exists(path, bool(exists), now) return result def _media_status(props: Dict[str, Any], file_path: Optional[str], *, file_exists: Optional[bool] = None) -> str: status_text = str(_first_prop(props, ("Status", "Media Status", "Online Status", "Offline")) or "").strip().lower() if not file_path: return "no_path" if "offline" in status_text or status_text in {"true", "yes", "1"}: return "offline" if "missing" in status_text: return "missing_file" # Pass `file_exists` in to reuse a single os.path.exists() probe — stat calls on # network source media are slow, so the caller avoids probing the same path twice. if file_exists is None: file_exists = os.path.exists(str(file_path)) if not file_exists: return "missing_file" return "online" _RESOLVE_CONTAINER_TYPE_PARTS = ( "adjustment", "compound", "fusion", "generator", "multicam", "multi cam", "sequence", "subclip", "sub clip", "timeline", "title", ) def _truthy_resolve_value(value: Any) -> bool: return str(value or "").strip().lower() in {"1", "true", "yes", "y", "on"} def _source_clip_status(record: Dict[str, Any], props: Dict[str, Any]) -> Tuple[bool, str]: media_type = str(record.get("media_type") or "").strip() media_type_lower = media_type.lower() for marker in _RESOLVE_CONTAINER_TYPE_PARTS: if marker in media_type_lower: return False, f"Resolve {media_type or 'container'} item" subclip_flag = _first_prop(props, ("Sub Clip", "SubClip", "Is Sub Clip", "IsSubClip", "Is Subclip")) if _truthy_resolve_value(subclip_flag): return False, "Resolve subclip" file_path = record.get("file_path") if not file_path: return False, "No source file path exposed" extension = os.path.splitext(str(file_path))[1].lower() if extension not in MEDIA_EXTENSIONS: return False, f"Unsupported extension: {extension or 'none'}" return True, "Source media clip" def _record_is_sequence(record: Dict[str, Any]) -> bool: media_type_lower = str(record.get("media_type") or "").strip().lower() return "sequence" in media_type_lower or "timeline" in media_type_lower def _analyzable_clip_status(record: Dict[str, Any], props: Dict[str, Any]) -> Tuple[bool, str]: source_clip, reason = _source_clip_status(record, props) if not source_clip: return False, reason if record.get("status") != "online": return False, str(record.get("status") or "offline").replace("_", " ") return True, "Online source media" def _resolve_clip_record(clip: Any, bin_path: str, selected_ids: set) -> Dict[str, Any]: props = _clip_props(clip) file_path = _first_prop(props, ("File Path", "FilePath")) media_id, _ = _safe_call(clip, "GetMediaId") clip_id = _safe_id(clip) record = { "clip_id": clip_id, "clip_name": _safe_name(clip), "bin_path": bin_path, "file_path": str(file_path) if file_path else None, "media_id": str(media_id) if media_id else None, "duration": _first_prop(props, ("Duration",)), "fps": _first_prop(props, ("FPS", "Frame Rate")), "resolution": _first_prop(props, ("Resolution",)), "media_type": _first_prop(props, ("Type", "Media Type")), "proxy": _first_prop(props, ("Proxy", "Proxy Media Path")), "resolve_status": _first_prop(props, ("Status", "Media Status", "Online Status", "Offline")), "selected": bool(clip_id and clip_id in selected_ids), } # File-existence is resolved in a single parallel batch after every clip's # Resolve properties are gathered (see resolve_media_inventory), so we stash # the props and defer existence-dependent fields to _finalize_clip_record. record["clip_key"] = stable_clip_directory(record) record["_props"] = props return record def _finalize_clip_record(record: Dict[str, Any], file_exists: bool) -> Dict[str, Any]: props = record.pop("_props", {}) or {} record["file_exists"] = file_exists record["status"] = _media_status(props, record["file_path"], file_exists=file_exists) record["source_clip"], record["source_clip_reason"] = _source_clip_status(record, props) record["analyzable"], record["analyzable_reason"] = _analyzable_clip_status(record, props) return record def _append_folder_media( folder: Any, *, bin_path: str, recursive: bool, selected_ids: set, records: List[Dict[str, Any]], warnings: List[str], limit: int, exclude_bins: Optional[set] = None, ) -> bool: clips, clip_err = _safe_call(folder, "GetClipList") if clip_err: warnings.append(f"GetClipList failed for {bin_path}: {clip_err}") clips = [] for clip in clips or []: if len(records) >= limit: return True records.append(_resolve_clip_record(clip, bin_path, selected_ids)) if not recursive: return False subfolders, folder_err = _safe_call(folder, "GetSubFolderList") if folder_err: warnings.append(f"GetSubFolderList failed for {bin_path}: {folder_err}") return False for subfolder in subfolders or []: if len(records) >= limit: return True child_name = _safe_name(subfolder, "Unnamed") if exclude_bins and child_name in exclude_bins: continue truncated = _append_folder_media( subfolder, bin_path=f"{bin_path}/{child_name}", recursive=recursive, selected_ids=selected_ids, records=records, warnings=warnings, limit=limit, exclude_bins=exclude_bins, ) if truncated: return True return False def _empty_media_counts() -> Dict[str, int]: return { "total": 0, "source_clips": 0, "sequences": 0, "analyzable": 0, "not_analyzable": 0, "online": 0, "offline": 0, "missing_file": 0, "no_path": 0, "analyzed": 0, "selected": 0, } def _analysis_status_by_clip(project_root: str, records: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: keys = [record.get("clip_key") for record in records if record.get("clip_key")] status_by_key: Dict[str, Dict[str, Any]] = {} if not keys: return status_by_key # Resolve each clip to its report via the persisted clip index, which maps # every stable id found in a report (normalized + raw file path, clip_id, # media_id) to its folder. This survives a Media Pool rename, a legacy hash # basis, AND an offline clip that no longer reports a file path but still # carries its clip_id — none of which a folder-name scan can match. See #51. clips_root = os.path.join(project_root, "clips") hash_to_folder = load_clip_index(project_root).get("hash_to_folder") or {} for record in records: clip_key = record.get("clip_key") if not clip_key: continue report_path = os.path.join(project_root, "clips", str(clip_key), "analysis.json") if not os.path.isfile(report_path): # Renamed/legacy/offline clip: the recomputed clip_key no longer # matches the folder on disk. Fall back to any of the clip's stable # hashes via the index. for clip_hash in stable_clip_match_hashes(record): folder = hash_to_folder.get(clip_hash) if not folder: continue candidate = os.path.join(clips_root, folder, "analysis.json") if os.path.isfile(candidate): report_path = candidate break if os.path.isfile(report_path): status_by_key[str(clip_key)] = { "analysis_status": "analyzed", "analysis_report_path": report_path, } db_path = os.path.join(project_root, "jobs.sqlite") if not os.path.isfile(db_path): return status_by_key # The jobs DB stores each clip under the clip_key it had when analyzed. A # clip renamed afterwards produces a new clip_key, so an exact-key match # misses its job row. Index unresolved records by their rename-stable hash # so a DB row recorded under the old name (e.g. a reused batch report living # outside the local clips/ dir) still maps back to the current clip. #51. key_set = {str(k) for k in keys} pending_hash_to_key: Dict[str, str] = {} for record in records: clip_key = record.get("clip_key") if not clip_key or str(clip_key) in status_by_key: continue for folder_hash in stable_clip_match_hashes(record): pending_hash_to_key.setdefault(folder_hash, str(clip_key)) def _apply_row(row: sqlite3.Row, target_key: str) -> None: if status_by_key.get(target_key, {}).get("analysis_status") == "analyzed": return db_status = row["status"] report_path = row["report_path"] # In media_analysis_jobs, 'succeeded' = fresh analysis written this run, # 'skipped' = an existing analysis report was reused. Both indicate the # clip has been analyzed (a report exists on disk). Normalize them to # "analyzed" for callers that just want to know "is there a report?", # but preserve the raw job state under job_status for diagnostics. report_resolves = False if isinstance(report_path, str) and report_path: try: resolved = os.path.realpath(os.path.abspath(os.path.expanduser(report_path))) report_resolves = os.path.isfile(resolved) except Exception: report_resolves = False normalized = db_status if db_status in ("succeeded", "skipped") and report_resolves: normalized = "analyzed" status_by_key[target_key] = { "analysis_status": normalized, "job_status": db_status, "cache_status": row["cache_status"], "analysis_report_path": report_path, "analysis_error": row["error"], "job_id": row["job_id"], "job_name": row["job_name"], "job_updated_at": row["updated_at"], } select_cols = ( "SELECT jc.clip_key, jc.status, jc.cache_status, jc.report_path, jc.error, " "j.job_id, j.name AS job_name, j.updated_at " "FROM job_clips jc JOIN jobs j ON j.job_id = jc.job_id" ) placeholders = ",".join("?" for _ in keys) try: conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row rows = conn.execute( f"{select_cols} WHERE jc.clip_key IN ({placeholders}) ORDER BY jc.updated_at DESC", keys, ).fetchall() for row in rows: _apply_row(row, str(row["clip_key"])) # Only pay for the unfiltered scan when a rename actually left a record # unresolved by the disk pass and exact-key match above. unresolved = { h: k for h, k in pending_hash_to_key.items() if k not in status_by_key } if unresolved: for row in conn.execute( f"{select_cols} ORDER BY jc.updated_at DESC" ).fetchall(): raw_key = str(row["clip_key"]) if raw_key in key_set: continue row_hash = clip_directory_hash(raw_key) target_key = unresolved.get(row_hash) if row_hash else None if target_key: _apply_row(row, target_key) except Exception: return status_by_key finally: try: conn.close() except Exception: pass return status_by_key # Last full Resolve walk per project_root, kept overlay-free so the analysis # status can be re-applied cheaply on every background poll without re-walking # the Media Pool (the expensive, non-parallelizable GetClipProperty pass). _INVENTORY_CACHE: Dict[str, Dict[str, Any]] = {} _INVENTORY_LOCK = threading.Lock() def _get_cached_inventory(project_root: str) -> Optional[Dict[str, Any]]: with _INVENTORY_LOCK: return _INVENTORY_CACHE.get(project_root) def _store_cached_inventory(project_root: str, entry: Dict[str, Any]) -> None: with _INVENTORY_LOCK: _INVENTORY_CACHE[project_root] = entry def _assemble_inventory_payload(project_root: str, entry: Dict[str, Any]) -> Dict[str, Any]: """Apply the (local, cheap) analysis-status overlay onto cached base records. Base records hold the Resolve-derived fields plus file existence; the analysis overlay (queued/running/analyzed, report paths, job ids) is re-read from disk every call so a background poll reflects job progress without touching Resolve. Records are copied so the cached base stays overlay-free across polls. """ records = [dict(record) for record in entry["base_records"]] status_by_key = _analysis_status_by_clip(project_root, records) counts = _empty_media_counts() counts["total"] = len(records) counts["selected"] = entry.get("selected_count", sum(1 for r in records if r.get("selected"))) for record in records: status = record.get("status") or "unknown" if status in counts: counts[status] += 1 if _record_is_sequence(record): counts["sequences"] += 1 if record.get("source_clip"): counts["source_clips"] += 1 if record.get("analyzable"): counts["analyzable"] += 1 else: counts["not_analyzable"] += 1 analysis = status_by_key.get(str(record.get("clip_key") or ""), {}) record.update(analysis) record.setdefault("analysis_status", "not analyzed") if record["analysis_status"] in {"analyzed", "succeeded", "skipped"}: counts["analyzed"] += 1 return { "success": True, "resolve_available": True, "status": "Resolve connected", "project": entry["project"], "project_root": project_root, "clips": records, "counts": counts, "truncated": bool(entry.get("truncated")), "limit": entry.get("limit"), "warnings": entry.get("warnings", []), } def resolve_media_inventory( project_root: str, *, limit: Any = 500, exclude_bins: Optional[set] = None, recursive: bool = True, probe_paths: bool = True, reuse_cached: bool = False, ) -> Dict[str, Any]: try: max_items = max(1, min(int(limit), 10000)) except (TypeError, ValueError): max_items = 500 # Background polls only need to surface analysis progress (a local, disk-backed # signal), so they reuse the last Resolve walk instead of paying for ~N serial # GetClipProperty round-trips again. A cheap project-id check still catches a # project switch made directly in Resolve (a handful of API calls vs a full # walk); we rebuild only on a confirmed mismatch. If the current project can't # be determined (Resolve down / no project open), we keep serving the cache — # a transient blip shouldn't trigger an expensive rebuild on every poll. if reuse_cached: cached = _get_cached_inventory(project_root) if cached is not None: current_id, id_error = _current_resolve_project_id() cached_id = (cached.get("project") or {}).get("id") project_changed = ( id_error is None and current_id is not None and str(current_id) != str(cached_id) ) if not project_changed: return _assemble_inventory_payload(project_root, cached) # Everything that touches the Resolve scripting API stays under the lock; the # parallel path probe and the disk overlay run outside it. with _RESOLVE_API_LOCK: resolve, resolve_error = _connect_resolve_read_only() if resolve_error: return { "success": True, "resolve_available": False, "status": "Resolve unavailable", "error": resolve_error, "clips": [], "counts": _empty_media_counts(), } pm, pm_error = _safe_call(resolve, "GetProjectManager") project = None if pm and not pm_error: project, _ = _safe_call(pm, "GetCurrentProject") if not project: return { "success": True, "resolve_available": False, "status": "No Resolve project", "error": "DaVinci Resolve is connected, but no project is open.", "clips": [], "counts": _empty_media_counts(), } media_pool, mp_error = _safe_call(project, "GetMediaPool") if not media_pool or mp_error: return { "success": True, "resolve_available": False, "status": "Media Pool unavailable", "error": mp_error or "Failed to get Resolve Media Pool", "clips": [], "counts": _empty_media_counts(), } root_folder, root_error = _safe_call(media_pool, "GetRootFolder") if not root_folder or root_error: return { "success": True, "resolve_available": False, "status": "Root folder unavailable", "error": root_error or "Failed to get Resolve root folder", "clips": [], "counts": _empty_media_counts(), } selected_ids = set() selected_clips, _ = _safe_call(media_pool, "GetSelectedClips") for clip in selected_clips or []: clip_id = _safe_id(clip) if clip_id: selected_ids.add(clip_id) warnings: List[str] = [] records: List[Dict[str, Any]] = [] truncated = _append_folder_media( root_folder, bin_path="Master", recursive=recursive, selected_ids=selected_ids, records=records, warnings=warnings, limit=max_items, exclude_bins=exclude_bins, ) project_info = { "name": _safe_name(project, "Resolve Project"), "id": _safe_id(project), } selected_count = len(selected_ids) # Resolve every clip's file path in one parallel, cache-backed batch, then # finalize existence-dependent fields (status / analyzable). existence = _probe_paths_exist( (record.get("file_path") for record in records), probe=probe_paths, ) for record in records: file_path = record.get("file_path") _finalize_clip_record(record, bool(file_path) and existence.get(str(file_path), False)) entry = { "base_records": records, "project": project_info, "selected_count": selected_count, "truncated": bool(truncated), "limit": max_items, "warnings": warnings, } _store_cached_inventory(project_root, entry) return _assemble_inventory_payload(project_root, entry) _PROJECT_CONTEXT_RE = re.compile(r"^(?P.+)-(?P[0-9a-f]{10})$") def _project_context_family_slug(project_directory: Any) -> str: name = os.path.basename(str(project_directory or "")).strip() match = _PROJECT_CONTEXT_RE.match(name) return match.group("slug") if match else name def _project_context_label(project_directory: Any) -> str: family = _project_context_family_slug(project_directory) return family.replace("-", " ").strip().title() or "Project" def _context_payload(project_name: Any, project_id: Any, output_root: Dict[str, Any], *, source: str = "dashboard") -> Dict[str, Any]: project_directory = output_root.get("project_directory") or os.path.basename(str(output_root.get("project_root") or "")) return { "project_name": str(project_name or _project_context_label(project_directory)), "project_id": str(project_id) if project_id not in (None, "") else None, "project_root": output_root.get("project_root"), "base_root": output_root.get("base_root"), "project_directory": project_directory, "family_slug": _project_context_family_slug(project_directory), "source": source, } def _context_from_project_root(base_root: str, project_root: str, *, source: str = "analysis_root") -> Optional[Dict[str, Any]]: root = os.path.realpath(os.path.abspath(os.path.expanduser(str(project_root)))) base = os.path.realpath(os.path.abspath(os.path.expanduser(str(base_root)))) try: if os.path.commonpath([root, base]) != base: return None except ValueError: return None if not os.path.isdir(root): return None project_directory = os.path.basename(root) return { "project_name": _project_context_label(project_directory), "project_id": None, "project_root": root, "base_root": base, "project_directory": project_directory, "family_slug": _project_context_family_slug(project_directory), "source": source, } @_serialize_resolve def _current_resolve_project_context(base_root: str) -> Optional[Dict[str, Any]]: resolve, resolve_error = _connect_resolve_read_only() if resolve_error: return None pm, pm_error = _safe_call(resolve, "GetProjectManager") if not pm or pm_error: return None project, _ = _safe_call(pm, "GetCurrentProject") if not project: return None project_name = _safe_name(project, "Resolve Project") project_id = _safe_id(project) root = project_root_for_dashboard( project_name=project_name, project_id=project_id, analysis_root=base_root, ) if not root.get("success"): return None return _context_payload(project_name, project_id, root, source="resolve") def _project_folder_name(folder: Any) -> str: if isinstance(folder, str): return folder name, _ = _safe_call(folder, "GetName") return str(name or folder or "").strip() def _normalize_project_folder_path(folder_path: Any) -> List[str]: if folder_path is None: return [] if isinstance(folder_path, (list, tuple)): raw_parts = [str(part or "").strip() for part in folder_path] else: raw_parts = re.split(r"[\\/]+", str(folder_path or "")) parts = [part for part in raw_parts if part and part not in {".", "/"}] if parts and parts[0].lower() in {"root", "master"}: parts = parts[1:] return parts def _goto_project_folder(pm: Any, folder_path: Any) -> Tuple[bool, Optional[str]]: parts = _normalize_project_folder_path(folder_path) _, root_error = _safe_call(pm, "GotoRootFolder") if root_error: return False, root_error for part in parts: opened, open_error = _safe_call(pm, "OpenFolder", part) if open_error: return False, open_error if not opened: return False, f"Resolve project folder not found: {part}" return True, None def _project_folder_label(folder_path: List[str]) -> str: return " / ".join(folder_path) if folder_path else "Root" @_serialize_resolve def _resolve_all_project_contexts(base_root: str, *, max_depth: int = 12, max_projects: int = 2000) -> Dict[str, Any]: resolve, resolve_error = _connect_resolve_read_only() if resolve_error: return { "success": True, "available": False, "error": resolve_error, "projects": [], "database": None, "current_folder": None, } pm, pm_error = _safe_call(resolve, "GetProjectManager") if not pm or pm_error: return { "success": True, "available": False, "error": pm_error or "ProjectManager unavailable", "projects": [], "database": None, "current_folder": None, } current_project, _ = _safe_call(pm, "GetCurrentProject") current_name = _safe_name(current_project, "") if current_project else None current_id = _safe_id(current_project) if current_project else None current_folder, _ = _safe_call(pm, "GetCurrentFolder") database, _ = _safe_call(pm, "GetCurrentDatabase") projects: List[Dict[str, Any]] = [] errors: List[str] = [] active_folder_path: Optional[List[str]] = None seen_locations = set() ok, root_error = _goto_project_folder(pm, []) if not ok: return { "success": True, "available": False, "error": root_error or "Failed to open Resolve project root folder", "projects": [], "database": database if isinstance(database, dict) else None, "current_folder": current_folder, } def visit(folder_path: List[str], depth: int = 0) -> None: nonlocal active_folder_path if depth > max_depth or len(projects) >= max_projects: return names, names_error = _safe_call(pm, "GetProjectListInCurrentFolder") if names_error: errors.append(f"{_project_folder_label(folder_path)}: {names_error}") names = [] for raw_name in names or []: if len(projects) >= max_projects: break project_name = str(raw_name or "").strip() if not project_name: continue location_key = (tuple(folder_path), project_name) if location_key in seen_locations: continue seen_locations.add(location_key) project_id = current_id if current_name and project_name == current_name else None root = resolve_output_root( project_name=project_name, project_id=project_id, analysis_root=base_root, create=False, ) project_directory = root.get("project_directory") if root.get("success") else "" is_active = bool(current_name and project_name == current_name) if is_active: active_folder_path = list(folder_path) projects.append({ "project_name": project_name, "project_id": project_id, "project_directory": project_directory, "folder_path": list(folder_path), "folder_label": _project_folder_label(folder_path), "database_label": (database or {}).get("DbName") if isinstance(database, dict) else None, "active": is_active, "can_load_resolve": True, }) folders, folders_error = _safe_call(pm, "GetFolderListInCurrentFolder") if folders_error: errors.append(f"{_project_folder_label(folder_path)} folders: {folders_error}") return for folder in folders or []: if len(projects) >= max_projects: break folder_name = _project_folder_name(folder) if not folder_name: continue opened, open_error = _safe_call(pm, "OpenFolder", folder_name) if open_error or not opened: errors.append(open_error or f"Failed to open folder {folder_name}") continue visit([*folder_path, folder_name], depth + 1) _, parent_error = _safe_call(pm, "GotoParentFolder") if parent_error: errors.append(f"Failed to return from {folder_name}: {parent_error}") _goto_project_folder(pm, folder_path) visit([]) restore_path = active_folder_path if active_folder_path is not None else [] _goto_project_folder(pm, restore_path) projects.sort(key=lambda row: (str(row.get("folder_label") or "").lower(), str(row.get("project_name") or "").lower())) return { "success": True, "available": True, "projects": projects, "count": len(projects), "database": database if isinstance(database, dict) else None, "current_folder": current_folder, "active_project": current_name, "active_folder_path": active_folder_path, "truncated": len(projects) >= max_projects, "errors": errors, "warning": "Project list truncated" if len(projects) >= max_projects else None, } @_serialize_resolve def _resolve_project_contexts(base_root: str) -> Dict[str, Any]: resolve, resolve_error = _connect_resolve_read_only() if resolve_error: return { "available": False, "error": resolve_error, "contexts": [], "current": None, "database": None, "folder": None, } pm, pm_error = _safe_call(resolve, "GetProjectManager") if not pm or pm_error: return { "available": False, "error": pm_error or "ProjectManager unavailable", "contexts": [], "current": None, "database": None, "folder": None, } current_project, _ = _safe_call(pm, "GetCurrentProject") current_name = _safe_name(current_project, "") if current_project else None current_id = _safe_id(current_project) if current_project else None names, names_error = _safe_call(pm, "GetProjectListInCurrentFolder") if names_error: names = [] database, _ = _safe_call(pm, "GetCurrentDatabase") folder, _ = _safe_call(pm, "GetCurrentFolder") contexts: List[Dict[str, Any]] = [] seen_names = set() for raw_name in names or []: project_name = str(raw_name or "").strip() if not project_name or project_name in seen_names: continue seen_names.add(project_name) project_id = current_id if current_name and project_name == current_name else None root = resolve_output_root( project_name=project_name, project_id=project_id, analysis_root=base_root, create=False, ) if not root.get("success"): continue context = _context_payload(project_name, project_id, root, source="resolve") context["resolve_project_name"] = project_name context["can_load_resolve"] = True context["resolve_current"] = bool(current_name and project_name == current_name) contexts.append(context) if current_project and current_name and current_name not in seen_names: root = resolve_output_root( project_name=current_name, project_id=current_id, analysis_root=base_root, create=False, ) if root.get("success"): context = _context_payload(current_name, current_id, root, source="resolve") context["resolve_project_name"] = current_name context["can_load_resolve"] = True context["resolve_current"] = True contexts.insert(0, context) return { "available": True, "error": names_error, "contexts": contexts, "current": next((context for context in contexts if context.get("resolve_current")), None), "database": database if isinstance(database, dict) else None, "folder": folder, } @_serialize_resolve def _load_resolve_project_context(base_root: str, project_name: Any, folder_path: Any = None) -> Dict[str, Any]: target_name = str(project_name or "").strip() if not target_name: return {"success": False, "error": "Resolve project name is required"} resolve, resolve_error = _connect_resolve_read_only() if resolve_error: return {"success": False, "error": resolve_error} pm, pm_error = _safe_call(resolve, "GetProjectManager") if not pm or pm_error: return {"success": False, "error": pm_error or "ProjectManager unavailable"} current_project, _ = _safe_call(pm, "GetCurrentProject") current_name = _safe_name(current_project, "") if current_project else None loaded_project = current_project target_folder = _normalize_project_folder_path(folder_path) if target_folder: ok, folder_error = _goto_project_folder(pm, target_folder) if not ok: return {"success": False, "error": folder_error or "Failed to open Resolve project folder"} if current_name != target_name or target_folder: loaded_project, load_error = _safe_call(pm, "LoadProject", target_name) if load_error: return {"success": False, "error": f"LoadProject failed: {load_error}"} if not loaded_project: folder_label = _project_folder_label(target_folder) if target_folder else "current project folder" return {"success": False, "error": f"Resolve project not found in {folder_label}: {target_name}"} project, _ = _safe_call(pm, "GetCurrentProject") project = project or loaded_project loaded_name = _safe_name(project, target_name) project_id = _safe_id(project) root = project_root_for_dashboard( project_name=loaded_name, project_id=project_id, analysis_root=base_root, ) if not root.get("success"): return root context = _context_payload(loaded_name, project_id, root, source="resolve") context["resolve_project_name"] = loaded_name context["can_load_resolve"] = True context["resolve_current"] = True return {"success": True, "active": context, "output_root": root} def discover_project_contexts(base_root: str, active: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: base = os.path.realpath(os.path.abspath(os.path.expanduser(str(base_root)))) contexts: List[Dict[str, Any]] = [] resolve_projects = _resolve_project_contexts(base) current = resolve_projects.get("current") contexts.extend(resolve_projects.get("contexts") or []) if os.path.isdir(base): for name in sorted(os.listdir(base)): path = os.path.join(base, name) context = _context_from_project_root(base, path) if context: context["can_load_resolve"] = False contexts.append(context) if active: contexts.append(dict(active, source=active.get("source") or "active")) deduped: List[Dict[str, Any]] = [] seen = set() for context in contexts: root = context.get("project_root") if not root or root in seen: continue seen.add(root) context["active"] = bool(active and root == active.get("project_root")) deduped.append(context) active_family = (active or {}).get("family_slug") related_roots = [ context["project_root"] for context in deduped if active_family and context.get("family_slug") == active_family and (context["project_root"] == (active or {}).get("project_root") or os.path.isdir(context["project_root"])) ] return { "success": True, "base_root": base, "active": active, "current_resolve_project": current, "resolve_projects": resolve_projects, "contexts": deduped, "related_project_roots": related_roots, } # ─── V2 Review API: clip / shot read endpoints + frame serving ────────────── # # These helpers power the bin grid, clip detail, and shot detail views. # All read directly from disk artifacts (analysis.json + corrections.json # sidecar + sampled_NNNN.jpg). When C1 (DB-as-truth) lands, swap these to # query the V2 DB — the HTTP API contract does not change. def _v2_iter_clip_dirs(project_root: str) -> List[Tuple[str, str]]: """Return (clip_slug, clip_dir_path) for each analysis.json under {project_root}/clips/.""" clips_root = os.path.join(project_root, "clips") if not os.path.isdir(clips_root): return [] rows: List[Tuple[str, str]] = [] for entry in sorted(os.listdir(clips_root)): candidate = os.path.join(clips_root, entry) if not os.path.isdir(candidate): continue if not os.path.isfile(os.path.join(candidate, "analysis.json")): continue rows.append((entry, candidate)) return rows def _v2_iter_job_report_dirs(project_root: str) -> List[Tuple[str, str]]: """Return reusable analysis report dirs referenced by this project's jobs DB.""" db_path = os.path.join(project_root, "jobs.sqlite") if not os.path.isfile(db_path): return [] base_root = os.path.dirname(os.path.realpath(project_root)) rows: List[Tuple[str, str]] = [] try: conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row job_rows = conn.execute( """ SELECT clip_key, report_path, status FROM job_clips WHERE report_path IS NOT NULL AND status IN ('succeeded', 'skipped', 'analyzed') ORDER BY updated_at DESC """ ).fetchall() except Exception: return [] finally: try: conn.close() except Exception: pass for row in job_rows: report_path = str(row["report_path"] or "") if not report_path: continue real_report = os.path.realpath(os.path.abspath(os.path.expanduser(report_path))) if os.path.basename(real_report) != "analysis.json" or not os.path.isfile(real_report): continue try: if os.path.commonpath([real_report, base_root]) != base_root: continue except ValueError: continue clip_dir = os.path.dirname(real_report) slug = str(row["clip_key"] or os.path.basename(clip_dir)) rows.append((slug, clip_dir)) return rows def _v2_iter_analysis_dirs(project_root: str) -> List[Tuple[str, str]]: """Return local reports plus reusable report dirs linked from batch jobs.""" rows: List[Tuple[str, str]] = [] seen: set = set() for slug, clip_dir in _v2_iter_clip_dirs(project_root) + _v2_iter_job_report_dirs(project_root): real_dir = os.path.realpath(clip_dir) if real_dir in seen: continue seen.add(real_dir) rows.append((slug, clip_dir)) return rows def _v2_load_analysis(clip_dir: str) -> Optional[Dict[str, Any]]: path = os.path.join(clip_dir, "analysis.json") try: with open(path, "r", encoding="utf-8") as handle: return json.load(handle) except (OSError, json.JSONDecodeError): return None def _v2_load_analysis_db_first(project_root: str, clip_dir: str) -> Optional[Dict[str, Any]]: """C1 — DB-canonical reader with JSON fallback. Falls back to analysis.json for reports that predate schema v9 and for job-linked report dirs whose rows live under another project's DB. """ try: from src.utils import analysis_store report = analysis_store.load_db_report( project_root, clip_dir=os.path.basename(clip_dir.rstrip("/\\")) ) if report is not None: return report except Exception: pass return _v2_load_analysis(clip_dir) def _v2_semantic_search(project_root: str, q: str, *, limit: int = 20) -> Dict[str, Any]: """Semantic search over the embeddings index, shaped like /api/index/query rows so the existing search-card renderer works unchanged.""" text = (q or "").strip() if not text: return {"success": True, "results": []} try: from src.utils import embeddings, timeline_brain_db as tbd found = embeddings.find_similar(project_root, text=text, kind="text", limit=limit) if not found.get("success"): return found conn = tbd.connect(project_root) resolve_ids: Dict[str, Optional[str]] = {} def resolve_clip_id(clip_uuid: Optional[str]) -> Optional[str]: if not clip_uuid: return None if clip_uuid not in resolve_ids: row = conn.execute( "SELECT resolve_clip_id, clip_dir FROM clips WHERE clip_uuid = ?", (clip_uuid,), ).fetchone() resolve_ids[clip_uuid] = (row["resolve_clip_id"] or row["clip_dir"]) if row else None return resolve_ids[clip_uuid] rows: List[Dict[str, Any]] = [] for hit in found.get("results") or []: entity_type = hit.get("entity_type") clip_uuid = hit.get("clip_uuid") or (hit.get("entity_uuid") if entity_type == "clip" else None) row: Dict[str, Any] = { "result_type": "transcript" if entity_type == "segment" else "semantic", "score": hit.get("score"), "clip_id": resolve_clip_id(clip_uuid), "clip_name": hit.get("clip_name"), } if entity_type == "shot": row["start_seconds"] = hit.get("time_seconds_start") row["summary"] = hit.get("description") elif entity_type == "segment": row["start_seconds"] = hit.get("start_seconds") row["summary"] = hit.get("text") else: row["summary"] = hit.get("summary") rows.append(row) rows = _v2_enrich_search_results(project_root, rows) return {"success": True, "query": text, "model": found.get("model"), "results": rows} except Exception as exc: # noqa: BLE001 — search must fail soft in the panel return {"success": False, "error": f"{type(exc).__name__}: {exc}"} def _v2_clip_duration(report: Dict[str, Any]) -> Optional[float]: marker_plan = report.get("clip_analysis_markers") if isinstance(report.get("clip_analysis_markers"), dict) else {} duration = marker_plan.get("duration_seconds") if isinstance(duration, (int, float)): return float(duration) technical = report.get("technical") if isinstance(report.get("technical"), dict) else {} fmt = technical.get("format") if isinstance(technical.get("format"), dict) else {} if isinstance(fmt.get("duration_seconds"), (int, float)): return float(fmt["duration_seconds"]) videos = technical.get("video") if isinstance(technical.get("video"), list) else [] first = videos[0] if videos and isinstance(videos[0], dict) else {} if isinstance(first.get("duration_seconds"), (int, float)): return float(first["duration_seconds"]) return None def _v2_pick_representative_frame_index(report: Dict[str, Any]) -> Optional[int]: """Pick the frame index (1-based, matching sampled_NNNN.jpg) for a clip thumbnail. Strategy: middle shot's first frame_index_used, falling back to the middle analysis_keyframe, falling back to frame 1. """ visual = report.get("visual") if isinstance(report.get("visual"), dict) else {} shots = visual.get("shot_descriptions") if isinstance(visual.get("shot_descriptions"), list) else [] if shots: mid_shot = shots[len(shots) // 2] if isinstance(mid_shot, dict): indices = mid_shot.get("frame_indices_used") or mid_shot.get("frame_indices") if isinstance(indices, list) and indices: first = indices[0] if isinstance(first, (int, float)): return int(first) motion = report.get("motion") if isinstance(report.get("motion"), dict) else {} keyframes = motion.get("analysis_keyframes") if isinstance(motion.get("analysis_keyframes"), list) else [] if keyframes: mid_kf = keyframes[len(keyframes) // 2] if isinstance(mid_kf, dict) and isinstance(mid_kf.get("index"), (int, float)): return int(mid_kf["index"]) return 1 def _v2_clip_summary_card(clip_slug: str, clip_dir: str, report: Dict[str, Any]) -> Dict[str, Any]: clip_block = report.get("clip") if isinstance(report.get("clip"), dict) else {} visual = report.get("visual") if isinstance(report.get("visual"), dict) else {} classification = visual.get("editorial_classification") if isinstance(visual.get("editorial_classification"), dict) else {} shots = visual.get("shot_descriptions") if isinstance(visual.get("shot_descriptions"), list) else [] editing_notes = visual.get("editing_notes") if isinstance(visual.get("editing_notes"), dict) else {} summary = visual.get("clip_summary") if not isinstance(summary, str): summary = "" oneliner = visual.get("clip_summary_oneliner") if not isinstance(oneliner, str) or not oneliner: oneliner = summary[:140] + ("…" if len(summary) > 140 else "") rep_index = _v2_pick_representative_frame_index(report) return { "clip_id": clip_block.get("clip_id"), "clip_slug": clip_slug, "clip_dir": clip_dir, "clip_name": clip_block.get("clip_name") or clip_block.get("file_path") or clip_slug, "file_path": clip_block.get("file_path"), "bin_path": clip_block.get("bin_path"), "fps": clip_block.get("fps"), "duration_seconds": _v2_clip_duration(report), "shot_count": len(shots), "analyzed_at": report.get("analyzed_at"), "vision_committed_at": report.get("vision_committed_at"), "primary_use": classification.get("primary_use"), "select_potential": classification.get("select_potential"), "clip_summary": summary, "clip_summary_oneliner": oneliner, "search_tags": list(editing_notes.get("search_tags") or []) if isinstance(editing_notes.get("search_tags"), list) else [], "representative_frame_index": rep_index, "vision_status": report.get("vision_status") or visual.get("status"), } def list_analyzed_clips(project_root: str) -> Dict[str, Any]: """List analyzed clips for the bin grid. One row per analysis.json found.""" rows: List[Dict[str, Any]] = [] for slug, clip_dir in _v2_iter_analysis_dirs(project_root): report = _v2_load_analysis_db_first(project_root, clip_dir) if report is None: continue rows.append(_v2_clip_summary_card(slug, clip_dir, report)) rows.sort(key=lambda r: (r.get("clip_name") or "").lower()) return { "success": True, "project_root": project_root, "clips": rows, "count": len(rows), } def _v2_find_clip_dir(project_root: str, clip_id: str) -> Optional[str]: """Locate the clip directory for a given clip_id under {project_root}/clips/.""" for slug, clip_dir in _v2_iter_analysis_dirs(project_root): if slug == clip_id: return clip_dir report = _v2_load_analysis(clip_dir) if not report: continue clip_block = report.get("clip") if isinstance(report.get("clip"), dict) else {} if str(clip_block.get("clip_id") or "") == clip_id: return clip_dir return None def get_analyzed_clip(project_root: str, clip_id: str) -> Dict[str, Any]: clip_dir = _v2_find_clip_dir(project_root, clip_id) if not clip_dir: return {"success": False, "error": f"No analyzed clip found for id={clip_id}"} report = _v2_load_analysis_db_first(project_root, clip_dir) if report is None: return {"success": False, "error": "analysis.json unreadable"} visual = report.get("visual") if isinstance(report.get("visual"), dict) else {} shots = visual.get("shot_descriptions") if isinstance(visual.get("shot_descriptions"), list) else [] corrections = _v2_read_corrections_for_dir(clip_dir) return { "success": True, "card": _v2_clip_summary_card(os.path.basename(clip_dir), clip_dir, report), "clip": report.get("clip") or {}, "clip_summary": visual.get("clip_summary"), "clip_summary_oneliner": visual.get("clip_summary_oneliner"), "editorial_classification": visual.get("editorial_classification") or {}, "shot_and_style": visual.get("shot_and_style") or {}, "content": visual.get("content") or {}, "slate": visual.get("slate") or {}, "motion": visual.get("motion") or {}, "cut_understanding": visual.get("cut_understanding") or {}, "editing_notes": visual.get("editing_notes") or {}, "cross_shot": visual.get("cross_shot") or {}, "coverage_groups": visual.get("coverage_groups") or [], "continuity_chains": visual.get("continuity_chains") or [], "qc": visual.get("qc") or {}, "confidence": visual.get("confidence") or {}, "shots": shots, "shot_count": len(shots), "corrections": corrections, "analyzed_at": report.get("analyzed_at"), "vision_committed_at": report.get("vision_committed_at"), } def get_analyzed_clip_shots(project_root: str, clip_id: str) -> Dict[str, Any]: """Lighter endpoint: just the shots array.""" clip_dir = _v2_find_clip_dir(project_root, clip_id) if not clip_dir: return {"success": False, "error": f"No analyzed clip found for id={clip_id}"} report = _v2_load_analysis_db_first(project_root, clip_dir) if report is None: return {"success": False, "error": "analysis.json unreadable"} visual = report.get("visual") if isinstance(report.get("visual"), dict) else {} shots = visual.get("shot_descriptions") if isinstance(visual.get("shot_descriptions"), list) else [] return { "success": True, "clip_id": clip_id, "shots": shots, "shot_count": len(shots), } def get_analyzed_clip_shot(project_root: str, clip_id: str, shot_index: int) -> Dict[str, Any]: clip_dir = _v2_find_clip_dir(project_root, clip_id) if not clip_dir: return {"success": False, "error": f"No analyzed clip found for id={clip_id}"} report = _v2_load_analysis_db_first(project_root, clip_dir) if report is None: return {"success": False, "error": "analysis.json unreadable"} visual = report.get("visual") if isinstance(report.get("visual"), dict) else {} shots = visual.get("shot_descriptions") if isinstance(visual.get("shot_descriptions"), list) else [] motion = report.get("motion") if isinstance(report.get("motion"), dict) else {} keyframes = motion.get("analysis_keyframes") if isinstance(motion.get("analysis_keyframes"), list) else [] matched: Optional[Dict[str, Any]] = None for entry in shots: if isinstance(entry, dict): try: if int(entry.get("shot_index")) == int(shot_index): matched = entry break except (TypeError, ValueError): continue if matched is None: return {"success": False, "error": f"shot_index={shot_index} not found in clip {clip_id}"} frame_indices = matched.get("frame_indices_used") or matched.get("frame_indices") or [] if not isinstance(frame_indices, list): frame_indices = [] # Resolve each frame index back to its source keyframe (time_seconds, selection_reason, file path) kf_by_index: Dict[int, Dict[str, Any]] = {} for kf in keyframes: if not isinstance(kf, dict): continue try: kf_by_index[int(kf.get("index"))] = kf except (TypeError, ValueError): continue frame_rows: List[Dict[str, Any]] = [] for raw_index in frame_indices: try: idx = int(raw_index) except (TypeError, ValueError): continue kf = kf_by_index.get(idx, {}) frame_rows.append({ "frame_index": idx, "time_seconds": kf.get("time_seconds"), "selection_reason": kf.get("selection_reason"), "delta_from_previous": kf.get("delta_from_previous"), "motion_peak": bool(kf.get("motion_peak")), }) corrections = _v2_read_corrections_for_dir(clip_dir) shot_corrections = _v2_filter_corrections_for_shot(corrections, matched.get("shot_uuid"), shot_index) # Cross-shot relationships (spec §4) come from the DB, not the report — # fill the shot page's Relationships group when confirmed rows exist. # Exported reports don't carry shot_uuid, so derive it from the DB by # clip + shot_index. try: from src.utils import analysis_store as _analysis_store from src.utils import shot_relationships as _shot_rel conn = _timeline_brain_db.connect(project_root) shot_uuid = matched.get("shot_uuid") if not shot_uuid: clip_uuid = _analysis_store.resolve_clip_uuid(conn, clip_id) if clip_uuid: hit = conn.execute( "SELECT shot_uuid FROM shots WHERE clip_uuid = ? AND shot_index = ?", (clip_uuid, int(shot_index)), ).fetchone() shot_uuid = hit["shot_uuid"] if hit else None if shot_uuid: relationships = _shot_rel.relationships_for_shot(conn, str(shot_uuid)) if relationships: matched = dict(matched) matched["relationships"] = relationships except Exception: # noqa: BLE001 — panel reads fail soft pass return { "success": True, "clip_id": clip_id, "shot_index": shot_index, "shot": matched, "frames": frame_rows, "corrections": shot_corrections, } def regenerate_clip_transcript( project_root: str, clip_id: str, *, with_words: bool = True, backend: Optional[str] = None, language: Optional[str] = None, model: Optional[str] = None, ) -> Dict[str, Any]: """Re-run transcription for a single clip, writing transcript.json (with word_timestamps when supported) and merging the result into analysis.json. Does not touch the rest of the analysis layers (visual, motion, etc.). """ clip_dir = _v2_find_clip_dir(project_root, clip_id) if not clip_dir: return {"success": False, "error": f"No analyzed clip found for id={clip_id}"} report = _v2_load_analysis(clip_dir) if report is None: return {"success": False, "error": "analysis.json unreadable"} source_file = report.get("source_file") or (report.get("clip") or {}).get("file_path") if not source_file or not os.path.isfile(str(source_file)): return {"success": False, "error": f"source file not found on disk: {source_file!r}"} try: from src.utils.media_analysis import ( _transcribe, detect_capabilities, ) except Exception as exc: return {"success": False, "error": f"transcription helpers unavailable: {exc}"} artifacts = { "clip_dir": clip_dir, "analysis_json": os.path.join(clip_dir, "analysis.json"), "transcript_json": os.path.join(clip_dir, "transcript.json"), "transcript_srt": os.path.join(clip_dir, "transcript.srt"), "transcript_vtt": os.path.join(clip_dir, "transcript.vtt"), } capabilities = detect_capabilities() transcription_opts: Dict[str, Any] = { "enabled": True, "word_timestamps": bool(with_words), # Allow model download because this is a user-initiated re-transcribe; # without this flag _transcribe will skip with a guard error. "allow_model_download": True, } if backend: transcription_opts["backend"] = backend if language: transcription_opts["language"] = language if model: transcription_opts["model"] = model payload = _transcribe(source_file, artifacts, {"transcription": transcription_opts}, capabilities) if not payload.get("success"): return {"success": False, "error": payload.get("reason") or payload.get("error") or "transcription failed", "backend": payload.get("backend")} # Patch analysis.json so the in-memory snapshot stays in sync with the new # transcript.json artifact. We only touch the `transcription` block. try: with open(artifacts["analysis_json"], "r", encoding="utf-8") as handle: updated_report = json.load(handle) except Exception: updated_report = report updated_report["transcription"] = { "success": True, "backend": payload.get("backend"), "language": payload.get("language"), "text": payload.get("text"), "segments": payload.get("segments") or [], } if payload.get("words"): updated_report["transcription"]["words"] = payload["words"] try: _atomic_write_json(artifacts["analysis_json"], updated_report) except Exception as exc: return {"success": False, "error": f"transcript written but analysis.json patch failed: {exc}"} word_segment_count = sum(1 for seg in (payload.get("segments") or []) if isinstance(seg, dict) and seg.get("words")) return { "success": True, "clip_id": clip_id, "backend": payload.get("backend"), "language": payload.get("language"), "segment_count": len(payload.get("segments") or []), "word_segment_count": word_segment_count, "wrote_words": bool(payload.get("words") or word_segment_count), } _TRANSCRIPT_CORRECTIONS_FILENAME = "transcript-corrections.json" def _transcript_corrections_path(clip_dir: str) -> str: return os.path.join(clip_dir, _TRANSCRIPT_CORRECTIONS_FILENAME) def _read_transcript_corrections(clip_dir: str) -> Optional[Dict[str, Any]]: path = _transcript_corrections_path(clip_dir) if not os.path.isfile(path): return None try: with open(path, "r", encoding="utf-8") as handle: data = json.load(handle) except Exception: return None if not isinstance(data, dict): return None return data def _normalize_transcript_segments(raw_segments: Any) -> List[Dict[str, Any]]: """Coerce a list of raw segment dicts (from transcript.json, analysis.json, or transcript-corrections.json) into the dashboard's canonical shape. Canonical: {index, start_seconds, end_seconds, text, words?: [...]}. """ def _to_float(v: Any) -> Optional[float]: try: return float(v) if v is not None else None except (TypeError, ValueError): return None out: List[Dict[str, Any]] = [] if not isinstance(raw_segments, list): return out for index, seg in enumerate(raw_segments): if not isinstance(seg, dict): continue if seg.get("deleted"): continue start = seg.get("start_seconds") if seg.get("start_seconds") is not None else seg.get("start") end = seg.get("end_seconds") if seg.get("end_seconds") is not None else seg.get("end") text = seg.get("text") if text in (None, ""): text = seg.get("content") if text in (None, ""): continue normalized: Dict[str, Any] = { "index": int(seg.get("index", index)), "start_seconds": _to_float(start), "end_seconds": _to_float(end), "text": str(text).strip(), } words = seg.get("words") if isinstance(words, list) and words: word_rows: List[Dict[str, Any]] = [] for w_index, w in enumerate(words): if not isinstance(w, dict): continue w_text = w.get("word") if w.get("word") not in (None, "") else w.get("text") if w_text in (None, ""): continue word_rows.append({ "index": int(w.get("index", w_index)), "word": str(w_text), "start_seconds": _to_float(w.get("start_seconds") if w.get("start_seconds") is not None else w.get("start")), "end_seconds": _to_float(w.get("end_seconds") if w.get("end_seconds") is not None else w.get("end")), }) if word_rows: normalized["words"] = word_rows out.append(normalized) return out def get_analyzed_clip_transcript(project_root: str, clip_id: str) -> Dict[str, Any]: clip_dir = _v2_find_clip_dir(project_root, clip_id) if not clip_dir: return {"success": False, "error": f"No analyzed clip found for id={clip_id}"} report = _v2_load_analysis(clip_dir) if report is None: return {"success": False, "error": "analysis.json unreadable"} transcription = report.get("transcription") if isinstance(report.get("transcription"), dict) else {} corrections = _read_transcript_corrections(clip_dir) corrected_segments: Optional[List[Dict[str, Any]]] = None edited_count = 0 deleted_indices: List[int] = [] if corrections and isinstance(corrections.get("segments"), list): corrected_segments = _normalize_transcript_segments(corrections["segments"]) edited_count = int(corrections.get("edited_count") or 0) deleted_indices = corrections.get("deleted_indices") or [] if corrected_segments is not None: segments = corrected_segments else: segments = _normalize_transcript_segments(transcription.get("segments")) clip_meta = report.get("clip") if isinstance(report.get("clip"), dict) else {} return { "success": True, "clip_id": clip_id, "clip_name": clip_meta.get("clip_name"), "backend": transcription.get("backend"), "language": transcription.get("language"), "text": transcription.get("text"), "segments": segments, "segment_count": len(segments), "available": bool(segments or transcription.get("text")), "has_corrections": corrections is not None, "corrections_meta": (corrections or {}).get("metadata") or { "edited_count": edited_count, "deleted_count": len(deleted_indices), "updated_at": (corrections or {}).get("updated_at"), } if corrections is not None else None, } def save_clip_transcript_corrections(project_root: str, clip_id: str, body: Dict[str, Any]) -> Dict[str, Any]: """Write the shadow segments array to /transcript-corrections.json. Expected body shape: { "segments": [ { "index": n, "start_seconds": s, "end_seconds": e, "text": "...", "words": [...] }, ... ], "edited_count": N, "deleted_indices": [int, ...] } Or "clear": true to remove the corrections file entirely. """ clip_dir = _v2_find_clip_dir(project_root, clip_id) if not clip_dir: return {"success": False, "error": f"No analyzed clip found for id={clip_id}"} path = _transcript_corrections_path(clip_dir) if bool(body.get("clear")): try: if os.path.isfile(path): os.remove(path) except Exception as exc: return {"success": False, "error": str(exc)} return {"success": True, "cleared": True, "path": path} raw_segments = body.get("segments") if not isinstance(raw_segments, list): return {"success": False, "error": "body.segments must be a list"} normalized = _normalize_transcript_segments(raw_segments) payload = { "schema_version": "1.0", "updated_at": _now_iso(), "segments": normalized, "edited_count": int(body.get("edited_count") or 0), "deleted_indices": list(body.get("deleted_indices") or []), "metadata": { "edited_count": int(body.get("edited_count") or 0), "deleted_count": len(body.get("deleted_indices") or []), "updated_at": _now_iso(), }, } try: _atomic_write_json(path, payload) except Exception as exc: return {"success": False, "error": str(exc)} return { "success": True, "path": path, "segment_count": len(normalized), "edited_count": payload["edited_count"], "deleted_count": len(payload["deleted_indices"]), } def _now_iso() -> str: import datetime as _dt return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def get_clip_frame_path(project_root: str, clip_id: str, frame_index: int) -> Optional[str]: clip_dir = _v2_find_clip_dir(project_root, clip_id) if not clip_dir: return None candidate = os.path.join(clip_dir, "frames", f"sampled_{int(frame_index):04d}.jpg") if os.path.isfile(candidate): return candidate return None def _v2_read_corrections_for_dir(clip_dir: str) -> Dict[str, Any]: path = os.path.join(clip_dir, "corrections.json") if not os.path.isfile(path): return {"schema_version": "2.0", "current": {}, "changelog": []} try: with open(path, "r", encoding="utf-8") as handle: data = json.load(handle) if not isinstance(data, dict): return {"schema_version": "2.0", "current": {}, "changelog": []} data.setdefault("schema_version", "2.0") data.setdefault("current", {}) data.setdefault("changelog", []) return data except (OSError, json.JSONDecodeError): return {"schema_version": "2.0", "current": {}, "changelog": []} def _v2_filter_corrections_for_shot( corrections: Dict[str, Any], shot_uuid: Any, shot_index: Any ) -> Dict[str, Any]: current = corrections.get("current") if isinstance(corrections.get("current"), dict) else {} changelog = corrections.get("changelog") if isinstance(corrections.get("changelog"), list) else [] keep_keys: set = set() if shot_uuid: keep_keys.add(str(shot_uuid)) if shot_index is not None: keep_keys.add(str(shot_index)) filtered_current = { key: entry for key, entry in current.items() if key.startswith("shot:") and key.split(":", 2)[1] in keep_keys } filtered_changelog = [ row for row in changelog if row.get("entity_type") == "shot" and str(row.get("entity_uuid")) in keep_keys ] return {"current": filtered_current, "changelog": filtered_changelog} def apply_clip_correction(project_root: str, clip_id: str, body: Dict[str, Any]) -> Dict[str, Any]: """Proxy POST /api/clips//corrections → media_analysis update_*_field helpers.""" clip_dir = _v2_find_clip_dir(project_root, clip_id) if not clip_dir: return {"success": False, "error": f"No analyzed clip found for id={clip_id}"} from src.server import _v2_update_field entity_type = body.get("entity_type") or body.get("entityType") or "shot" params: Dict[str, Any] = dict(body) params["clip_id"] = clip_id params["clip_dir"] = clip_dir return _v2_update_field(project_root, params, entity_type=entity_type) def export_clip_selection(project_root: str, clip_ids: List[str], fmt: str) -> Tuple[bytes, str, str]: """Build the export bytes for a selection. Returns (bytes, content_type, filename).""" fmt = (fmt or "json").strip().lower() payloads: List[Dict[str, Any]] = [] for clip_id in clip_ids: data = get_analyzed_clip(project_root, clip_id) if data.get("success"): payloads.append(data) timestamp = _now_iso().replace(":", "").replace("-", "")[:13] if fmt == "csv": import csv as _csv import io as _io buf = _io.StringIO() writer = _csv.writer(buf) writer.writerow([ "clip_id", "clip_name", "bin_path", "duration_seconds", "shot_count", "primary_use", "select_potential", "energy_arc", "style", "search_tags", "qc_warnings", "clip_summary_oneliner", "clip_summary", ]) for clip in payloads: card = clip.get("card") or {} classification = clip.get("editorial_classification") or {} editing_notes = clip.get("editing_notes") or {} qc = clip.get("qc") or {} writer.writerow([ card.get("clip_id") or "", card.get("clip_name") or "", card.get("bin_path") or "", card.get("duration_seconds") if card.get("duration_seconds") is not None else "", clip.get("shot_count") if clip.get("shot_count") is not None else "", classification.get("primary_use") or "", classification.get("select_potential") or "", classification.get("energy_arc") or "", classification.get("style") or "", "|".join(str(t) for t in (editing_notes.get("search_tags") or [])), "|".join(str(w) for w in (qc.get("warnings") or [])), clip.get("clip_summary_oneliner") or "", clip.get("clip_summary") or "", ]) return buf.getvalue().encode("utf-8"), "text/csv; charset=utf-8", f"selection-{timestamp}.csv" # JSON: full payload array text = json.dumps({"clip_count": len(payloads), "clips": payloads}, indent=2) return text.encode("utf-8"), "application/json", f"selection-{timestamp}.json" def combined_clip_analysis(project_root: str, body: Dict[str, Any]) -> Dict[str, Any]: """Synthesize a multi-clip review payload. Body: {clip_ids: [str, ...]}. Returns a unified payload with: clip_count, sources (per-clip card), clip_summaries[] (one per clip), editorial_classification (union), shot_and_style (per-clip), shots[] (all shots from all clips, source-tagged and time-offset by accumulated clip duration so they read like one strip), transcript (concatenated segments with the same time offset), tags (union), qc (union). """ clip_ids = body.get("clip_ids") if not isinstance(clip_ids, list) or not clip_ids: return {"success": False, "error": "clip_ids must be a non-empty list"} sources: List[Dict[str, Any]] = [] shot_summaries: List[Dict[str, Any]] = [] transcript_segments: List[Dict[str, Any]] = [] clip_summaries: List[Dict[str, Any]] = [] union_tags: List[str] = [] union_qc: List[str] = [] classifications: List[Dict[str, Any]] = [] shot_and_style_blocks: List[Dict[str, Any]] = [] cursor = 0.0 total_duration = 0.0 for clip_id in clip_ids: data = get_analyzed_clip(project_root, clip_id) if not data.get("success"): continue card = data.get("card") or {} duration = float(card.get("duration_seconds") or 0.0) sources.append({ "clip_id": card.get("clip_id") or clip_id, "clip_name": card.get("clip_name"), "bin_path": card.get("bin_path"), "duration_seconds": duration, "shot_count": data.get("shot_count"), "thumbnail_frame_index": card.get("thumbnail_frame_index"), "offset_seconds": cursor, }) if data.get("clip_summary"): clip_summaries.append({ "clip_id": card.get("clip_id") or clip_id, "clip_name": card.get("clip_name"), "oneliner": data.get("clip_summary_oneliner"), "summary": data.get("clip_summary"), }) if isinstance(data.get("editorial_classification"), dict): classifications.append({"clip_name": card.get("clip_name"), **data["editorial_classification"]}) if isinstance(data.get("shot_and_style"), dict): shot_and_style_blocks.append({"clip_name": card.get("clip_name"), **data["shot_and_style"]}) for shot in (data.get("shots") or []): if not isinstance(shot, dict): continue shot_summaries.append({ "source_clip_id": card.get("clip_id") or clip_id, "source_clip_name": card.get("clip_name"), "source_offset_seconds": cursor, "shot_index": shot.get("shot_index"), "time_seconds_start": (float(shot.get("time_seconds_start") or 0.0) + cursor) if shot.get("time_seconds_start") is not None else None, "time_seconds_end": (float(shot.get("time_seconds_end") or 0.0) + cursor) if shot.get("time_seconds_end") is not None else None, "frame_indices_used": shot.get("frame_indices_used") or [], "description": shot.get("description"), "qc_flags": shot.get("qc_flags") or [], }) # Transcript merge clip_dir = _v2_find_clip_dir(project_root, clip_id) if clip_dir: t = get_analyzed_clip_transcript(project_root, clip_id) if t.get("success"): for seg in (t.get("segments") or []): if not isinstance(seg, dict): continue transcript_segments.append({ "source_clip_id": card.get("clip_id") or clip_id, "source_clip_name": card.get("clip_name"), "source_offset_seconds": cursor, "start_seconds": (float(seg.get("start_seconds") or 0.0) + cursor) if seg.get("start_seconds") is not None else None, "end_seconds": (float(seg.get("end_seconds") or 0.0) + cursor) if seg.get("end_seconds") is not None else None, "text": seg.get("text"), }) for tag in ((data.get("editing_notes") or {}).get("search_tags") or []): text = str(tag).strip() if text and text not in union_tags: union_tags.append(text) for warn in ((data.get("qc") or {}).get("warnings") or []): text = str(warn).strip() if text and text not in union_qc: union_qc.append(text) cursor += duration total_duration += duration if not sources: return {"success": False, "error": "none of the requested clip_ids resolved to an analyzed clip"} return { "success": True, "clip_count": len(sources), "total_duration_seconds": total_duration, "sources": sources, "clip_summaries": clip_summaries, "editorial_classifications": classifications, "shot_and_style_blocks": shot_and_style_blocks, "shots": shot_summaries, "transcript_segments": transcript_segments, "search_tags": union_tags, "qc_warnings": union_qc, } def _v2_create_timeline_from_clips(body: Dict[str, Any]) -> Dict[str, Any]: """POST /api/resolve/create_timeline_from_clips → proxies media_pool action="create_timeline_from_clips". Body: {name?, clip_ids: [str, ...]}. """ clip_ids = body.get("clip_ids") if not isinstance(clip_ids, list) or not clip_ids: return {"success": False, "error": "clip_ids must be a non-empty list"} name = str(body.get("name") or "Review Selection").strip() or "Review Selection" try: from src.server import media_pool return media_pool("create_timeline_from_clips", params={"name": name, "clip_ids": list(clip_ids)}) except Exception as exc: # noqa: BLE001 return {"success": False, "error": f"{type(exc).__name__}: {exc}"} def _v2_open_clip_in_resolve(body: Dict[str, Any]) -> Dict[str, Any]: """Proxy POST /api/resolve/open_clip → media_pool_item(open_in_viewer) / resolve_control(save_state|restore_state). Used by the shot-detail 'Open in Resolve' button and the save-before-preview workflow. Body shape: {action: "open_in_viewer" (default), clip_id, mark_in_seconds?, mark_out_seconds?, clear_marks?, mark_type?, page?} {action: "save_state"} {action: "restore_state", state_token} """ requested_action = (body.get("action") or "open_in_viewer").strip().lower() try: if requested_action in {"save_state", "restore_state"}: from src.server import resolve_control return resolve_control(requested_action, params=body) if requested_action == "open_in_viewer": from src.server import media_pool_item params = dict(body) params.pop("action", None) params.setdefault("page", "media") return media_pool_item("open_in_viewer", params=params) return {"success": False, "error": f"Unsupported action {requested_action!r}"} except Exception as exc: # noqa: BLE001 — dashboard surface should never crash return {"success": False, "error": f"{type(exc).__name__}: {exc}"} # ─── C6: timeline version chain + brain-edit history helpers ───────────────── def list_timelines_with_versions(project_root: str) -> Dict[str, Any]: """Every timeline that has at least one archived version, with counts.""" try: conn = _timeline_brain_db.connect(project_root) except Exception as exc: return {"success": False, "error": f"{type(exc).__name__}: {exc}", "timelines": []} rows = conn.execute( """ SELECT timeline_name, COUNT(*) AS version_count, MAX(version) AS latest_version, MAX(created_at) AS most_recent FROM timeline_versions GROUP BY timeline_name ORDER BY most_recent DESC NULLS LAST """ ).fetchall() return { "success": True, "timelines": [dict(r) for r in rows], } def get_timeline_history_payload( project_root: str, timeline_name: str, *, history_limit: int = 200, ) -> Dict[str, Any]: """Combined payload: version chain + brain edits for a single timeline.""" versions = _timeline_versioning.list_timeline_versions( project_root=project_root, timeline_name=timeline_name, ) edits = _brain_edits.get_brain_edit_history( project_root=project_root, timeline_name=timeline_name, limit=history_limit, ) return { "success": True, "timeline_name": timeline_name, "versions": versions, "edits": edits, } def list_edit_plans_payload(project_root: str) -> Dict[str, Any]: """Edit-engine plan list for the panel browser (DB/file only, no Resolve). Fingerprint-corrupt plans surface as {"plan_id", "corrupt": True} warning rows rather than being silently hidden. """ try: from src.utils import edit_engine as _edit_engine return _edit_engine.list_plans(project_root, limit=50, include_corrupt=True) except Exception as exc: # noqa: BLE001 — panel reads fail soft return {"success": False, "error": f"{type(exc).__name__}: {exc}", "plans": []} def get_edit_plan_payload(project_root: str, plan_id: str) -> Dict[str, Any]: """Full plan detail for the panel, enriched for rendering: selects decisions and swap alternates gain a `thumb_frame_index` (the shot's first sampled frame, for the existing /api/clips//frames/ route) and a `resolve_clip_id` fallback mapped from clip_uuid. Enrichment is best-effort — the plan still renders without thumbnails when the DB is unavailable. """ try: from src.utils import edit_engine as _edit_engine plan = _edit_engine.load_plan(project_root, plan_id) except Exception as exc: # noqa: BLE001 return {"success": False, "error": f"{type(exc).__name__}: {exc}"} if plan is None: return {"success": False, "error": f"Plan {plan_id} not found"} if plan.get("_corrupt"): return {"success": True, "plan_id": plan_id, "corrupt": True} plan = json.loads(json.dumps(plan, default=str)) # detach a plain copy try: conn = _timeline_brain_db.connect(project_root) clip_id_cache: Dict[str, Any] = {} def _enrich(row: Dict[str, Any]) -> None: clip_uuid = str(row.get("clip_uuid") or "") if not row.get("resolve_clip_id") and clip_uuid: if clip_uuid not in clip_id_cache: hit = conn.execute( "SELECT resolve_clip_id FROM clips WHERE clip_uuid = ?", (clip_uuid,), ).fetchone() clip_id_cache[clip_uuid] = hit["resolve_clip_id"] if hit else None if clip_id_cache[clip_uuid]: row["resolve_clip_id"] = clip_id_cache[clip_uuid] shot_uuid = row.get("shot_uuid") if shot_uuid and row.get("thumb_frame_index") is None: hit = conn.execute( "SELECT MIN(frame_index) AS frame_index FROM frames WHERE shot_uuid = ?", (str(shot_uuid),), ).fetchone() if hit and hit["frame_index"] is not None: row["thumb_frame_index"] = int(hit["frame_index"]) for decision in plan.get("decisions") or []: if isinstance(decision, dict): _enrich(decision) for alternate in plan.get("alternates") or []: if isinstance(alternate, dict): _enrich(alternate) except Exception: # noqa: BLE001 — thumbnails are progressive enhancement pass return {"success": True, "corrupt": False, "plan": plan} def proxy_timeline_versioning_action(body: Dict[str, Any]) -> Dict[str, Any]: """Bridge dashboard → MCP server timeline_versioning tool. Body shape: {action, ...params}. Used for write actions (archive, rollback, prune) that need a live Resolve connection. """ action = (body.get("action") or "").strip() if not action: return {"success": False, "error": "action required"} try: from src.server import timeline_versioning as _tv_tool params = {k: v for k, v in body.items() if k != "action"} return _tv_tool(action, params=params) except Exception as exc: # noqa: BLE001 return {"success": False, "error": f"{type(exc).__name__}: {exc}"} def _v2_enrich_search_results(project_root: str, results: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Augment /api/index/query results with fps, shot_index, and a thumbnail frame. These come from the clip's analysis.json on disk so the UI can render SMPTE timecode (HH:MM:SS:FF) and a clickable card that opens the deep-link review page for the matching shot. """ if not results: return results analyses_by_clip: Dict[str, Optional[Dict[str, Any]]] = {} for row in results: clip_id = row.get("clip_id") or row.get("clip_key") if not clip_id: continue if clip_id not in analyses_by_clip: clip_dir = _v2_find_clip_dir(project_root, clip_id) analyses_by_clip[clip_id] = _v2_load_analysis(clip_dir) if clip_dir else None report = analyses_by_clip[clip_id] if not report: continue # fps from technical block technical = report.get("technical") if isinstance(report.get("technical"), dict) else {} videos = technical.get("video") if isinstance(technical.get("video"), list) else [] fps = None if videos and isinstance(videos[0], dict): raw_fps = videos[0].get("frame_rate") try: fps = float(str(raw_fps).split("/")[0]) / (float(str(raw_fps).split("/")[1]) if "/" in str(raw_fps) else 1.0) if raw_fps else None except (TypeError, ValueError, ZeroDivisionError): fps = None if not fps: marker_plan = report.get("clip_analysis_markers") if isinstance(report.get("clip_analysis_markers"), dict) else {} try: fps = float(marker_plan.get("fps")) if marker_plan.get("fps") else None except (TypeError, ValueError): fps = None row["fps"] = fps # Resolve shot_index from start_seconds against shot_descriptions visual = report.get("visual") if isinstance(report.get("visual"), dict) else {} shots = visual.get("shot_descriptions") if isinstance(visual.get("shot_descriptions"), list) else [] start_seconds = row.get("start_seconds") matched_shot: Optional[Dict[str, Any]] = None if start_seconds is not None: try: ts = float(start_seconds) for shot in shots: if not isinstance(shot, dict): continue s = shot.get("time_seconds_start") e = shot.get("time_seconds_end") if s is None or e is None: continue if float(s) <= ts < float(e): matched_shot = shot break except (TypeError, ValueError): matched_shot = None if matched_shot is None and shots: # Clip-level result or no time anchor — point at the middle shot. matched_shot = shots[len(shots) // 2] if isinstance(shots[len(shots) // 2], dict) else None if matched_shot is not None: row["shot_index"] = matched_shot.get("shot_index") frame_indices = matched_shot.get("frame_indices_used") or matched_shot.get("frame_indices") or [] if isinstance(frame_indices, list) and frame_indices: row["thumbnail_frame_index"] = frame_indices[0] if "thumbnail_frame_index" not in row: row["thumbnail_frame_index"] = _v2_pick_representative_frame_index(report) return results def read_clip_corrections(project_root: str, clip_id: str) -> Dict[str, Any]: clip_dir = _v2_find_clip_dir(project_root, clip_id) if not clip_dir: return {"success": False, "error": f"No analyzed clip found for id={clip_id}"} data = _v2_read_corrections_for_dir(clip_dir) return { "success": True, "clip_id": clip_id, "corrections_path": os.path.join(clip_dir, "corrections.json"), "current": data.get("current", {}), "changelog": data.get("changelog", []), "current_field_count": len(data.get("current", {})), "changelog_count": len(data.get("changelog", [])), } class DashboardState: def __init__(self, project_name: str, project_id: str, analysis_root: str): self.base_analysis_root = os.path.realpath(os.path.abspath(os.path.expanduser(str(analysis_root)))) if project_name == "Dashboard Analysis" and project_id == "dashboard": current = _current_resolve_project_context(self.base_analysis_root) if current: project_name = current["project_name"] project_id = current.get("project_id") self.project_name = project_name self.project_id = project_id root = project_root_for_dashboard( project_name=project_name, project_id=project_id, analysis_root=self.base_analysis_root, ) if not root.get("success"): raise RuntimeError(root.get("error") or "Invalid analysis root") self.output_root = root self.project_root = root["project_root"] self.lock = threading.Lock() def context(self) -> Dict[str, Any]: return _context_payload(self.project_name, self.project_id, self.output_root, source="active") def projects(self) -> Dict[str, Any]: return discover_project_contexts(self.base_analysis_root, self.context()) def related_project_roots(self) -> List[str]: return list(self.projects().get("related_project_roots") or []) def set_context(self, payload: Dict[str, Any]) -> Dict[str, Any]: project_root = payload.get("project_root") or payload.get("projectRoot") project_name = payload.get("project_name") or payload.get("projectName") project_id = payload.get("project_id") or payload.get("projectId") load_resolve_project = bool(payload.get("load_resolve_project") or payload.get("loadResolveProject")) if load_resolve_project: resolve_project_name = payload.get("resolve_project_name") or payload.get("resolveProjectName") or project_name resolve_project_folder_path = payload.get("resolve_project_folder_path") or payload.get("resolveProjectFolderPath") loaded = _load_resolve_project_context(self.base_analysis_root, resolve_project_name, resolve_project_folder_path) if not loaded.get("success"): return loaded active = loaded["active"] self.project_name = active["project_name"] self.project_id = active.get("project_id") self.output_root = loaded["output_root"] self.project_root = active["project_root"] return {"success": True, "active": self.context(), "projects": self.projects()} if project_root: context = _context_from_project_root(self.base_analysis_root, str(project_root), source="selected") if not context: candidate_root = os.path.realpath(os.path.abspath(os.path.expanduser(str(project_root)))) try: under_base = os.path.commonpath([candidate_root, self.base_analysis_root]) == self.base_analysis_root except ValueError: under_base = False if not under_base or not project_name: return {"success": False, "error": "Project context must be under the analysis base root"} project_directory = os.path.basename(candidate_root) context = { "project_name": project_name, "project_id": project_id, "project_root": candidate_root, "project_directory": project_directory, } project_name = project_name or context["project_name"] project_id = project_id if project_id not in (None, "") else context.get("project_id") output_root = { "success": True, "analysis_version": None, "base_root": self.base_analysis_root, "project_root": context["project_root"], "project_directory": context["project_directory"], "project_name": project_name, "project_id": project_id, "errors": [], } else: if not project_name: return {"success": False, "error": "project_name or project_root is required"} output_root = project_root_for_dashboard( project_name=project_name, project_id=project_id, analysis_root=self.base_analysis_root, ) if not output_root.get("success"): return output_root self.project_name = str(project_name or output_root.get("project_name") or "Project") self.project_id = str(project_id) if project_id not in (None, "") else None self.output_root = output_root self.project_root = output_root["project_root"] return {"success": True, "active": self.context(), "projects": self.projects()} DOC_SOURCES = { "readme": {"title": "README", "path": "README.md"}, "analysis-guide": {"title": "Media Analysis Guide", "path": "docs/guides/media-analysis-guide.md"}, "agent-skill": {"title": "Agent Skill", "path": "docs/SKILL.md"}, "release-notes": {"title": "Release Notes", "path": "CHANGELOG.md"}, } def _repo_root() -> str: return os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) def _mcp_version() -> str: try: from src.server import VERSION as _server_version # type: ignore return str(_server_version) except Exception: try: pkg_path = os.path.join(_repo_root(), "package.json") with open(pkg_path, "r", encoding="utf-8") as handle: return str(json.load(handle).get("version") or "unknown") except Exception: return "unknown" def _request_is_loopback(handler: BaseHTTPRequestHandler) -> bool: """Defensive guard: only allow privileged routes from loopback clients.""" try: addr = (handler.client_address or ("",))[0] except Exception: return False return addr in {"127.0.0.1", "::1", "localhost"} def _launch_claude_code_terminal() -> Dict[str, Any]: """Open a Terminal/iTerm window at the MCP server's project root running the ``claude`` CLI. macOS only — other platforms return a clipboard-only hint so the dashboard can show a sensible message instead of silently no-op. """ repo_root = _repo_root() if sys.platform != "darwin": return { "success": False, "error": "Terminal launch is macOS-only. Open your terminal at the project root and run `claude`, then paste the prompt.", } import shlex import shutil import subprocess claude_bin = shutil.which("claude") or "claude" cmd = f"cd {shlex.quote(repo_root)} && {shlex.quote(claude_bin)}" iterm_running = False try: check = subprocess.run( ["osascript", "-e", 'application "iTerm" is running'], capture_output=True, text=True, timeout=8, ) iterm_running = (check.stdout or "").strip().lower() == "true" except Exception: iterm_running = False if iterm_running: script = ( 'tell application "iTerm"\n' ' activate\n' f' create window with default profile command "{cmd}"\n' 'end tell' ) else: escaped = cmd.replace("\\", "\\\\").replace('"', '\\"') script = ( 'tell application "Terminal"\n' ' activate\n' f' do script "{escaped}"\n' 'end tell' ) try: proc = subprocess.run( ["osascript", "-e", script], capture_output=True, text=True, timeout=15, ) if proc.returncode != 0: return {"success": False, "error": (proc.stderr or "").strip() or "osascript failed"} return { "success": True, "terminal": "iterm" if iterm_running else "terminal", "cwd": repo_root, } except Exception as exc: return {"success": False, "error": str(exc)} def _native_directory_picker(initial: Optional[str] = None) -> Dict[str, Any]: """Open a native OS folder picker on the *server* machine and return the absolute path. The dashboard runs on localhost so the picker pops on the user's machine — works in every browser because the browser is never asked to expose a filesystem path. """ initial_dir = initial if initial and os.path.isdir(initial) else os.path.expanduser("~") if sys.platform == "darwin": # AppleScript picker — works without a Python Tk binding installed. script = ( 'tell application "System Events" to activate\n' f'set chosenFolder to choose folder with prompt "Select a folder" default location POSIX file "{initial_dir}"\n' 'POSIX path of chosenFolder' ) try: import subprocess proc = subprocess.run( ["osascript", "-e", script], capture_output=True, text=True, timeout=120, ) if proc.returncode != 0: stderr = (proc.stderr or "").strip() if "User canceled" in stderr or "cancelled" in stderr.lower(): return {"success": True, "canceled": True} return {"success": False, "error": stderr or "AppleScript picker failed"} path = (proc.stdout or "").strip().rstrip("/") if not path: return {"success": True, "canceled": True} return {"success": True, "path": path} except Exception as exc: return {"success": False, "error": str(exc)} # Fallback: tkinter on Linux/Windows. Requires a display. try: import tkinter from tkinter import filedialog except Exception as exc: return {"success": False, "error": f"native picker unavailable: {exc}"} try: root = tkinter.Tk() root.withdraw() root.attributes("-topmost", True) path = filedialog.askdirectory(initialdir=initial_dir, title="Select a folder") root.destroy() except Exception as exc: return {"success": False, "error": str(exc)} if not path: return {"success": True, "canceled": True} return {"success": True, "path": path} def _load_installer_module(): """Lazy-import install.py from the repo root. Returns (module, error).""" try: sys.path.insert(0, _repo_root()) import importlib if "install" in sys.modules: return importlib.reload(sys.modules["install"]), None import install # type: ignore return install, None except Exception as exc: return None, f"installer module unavailable: {exc}" def _resolve_mcp_paths() -> Dict[str, Any]: """Return the paths needed to write an MCP client config: python_path, server_path, api_path, lib_path. Each may be None. """ installer, error = _load_installer_module() repo = _repo_root() # Server entrypoint — same one the installer wires up. server_path = os.path.join(repo, "src", "resolve_mcp_server.py") # Prefer the repo venv python so the configured client launches the # MCP server with all dependencies available. candidates = [ os.path.join(repo, "venv", "bin", "python"), os.path.join(repo, "venv", "Scripts", "python.exe"), os.path.join(repo, ".venv", "bin", "python"), ] python_path = next((c for c in candidates if os.path.isfile(c)), None) if not python_path: python_path = sys.executable api_path = lib_path = None if installer is not None: try: api_path, lib_path = installer.find_resolve_paths() except Exception: pass return { "python_path": str(python_path) if python_path else None, "server_path": str(server_path) if os.path.isfile(server_path) else None, "api_path": str(api_path) if api_path else None, "lib_path": str(lib_path) if lib_path else None, "installer_error": error, } def _mcp_status_payload() -> Dict[str, Any]: """Return MCP server identity + per-client install status.""" installer, error = _load_installer_module() paths = _resolve_mcp_paths() clients_out: List[Dict[str, Any]] = [] if installer is None: return { "success": False, "error": error or "installer unavailable", "server": { "version": _mcp_version(), "python_path": paths.get("python_path"), "server_path": paths.get("server_path"), }, "clients": [], } for client in getattr(installer, "MCP_CLIENTS", []): try: config_path = client["get_path"]() except Exception: config_path = None config_key = client.get("config_key", "mcpServers") available = config_path is not None installed = False entry: Any = None if available and config_path and os.path.isfile(str(config_path)): try: existing = installer.read_json(str(config_path)) entry = (existing.get(config_key) or {}).get("davinci-resolve") installed = bool(entry) except Exception: installed = False clients_out.append({ "id": client["id"], "name": client["name"], "notes": client.get("notes", ""), "config_path": str(config_path) if config_path else None, "config_key": config_key, "available": bool(available), "installed": bool(installed), }) return { "success": True, "server": { "version": _mcp_version(), "python_path": paths.get("python_path"), "server_path": paths.get("server_path"), "resolve_api_path": paths.get("api_path"), "resolve_lib_path": paths.get("lib_path"), "resolve_api_detected": bool(paths.get("api_path")), "resolve_lib_detected": bool(paths.get("lib_path")), }, "clients": clients_out, "transport": _transport_status(), } def _transport_status() -> Dict[str, Any]: """Live networked-transport status (or local-only) for the MCP diagnostics card.""" try: from src.utils.mcp_transport import read_transport_state except Exception: return {"networked": False, "mode": "stdio (local)"} state = read_transport_state() if not state: return {"networked": False, "mode": "stdio (local)"} return { "networked": True, "mode": state.get("transport"), "url": state.get("url"), "loopback": state.get("loopback", True), "has_token": bool(state.get("token")), "token": state.get("token"), "pid": state.get("pid"), } def _transport_start() -> Dict[str, Any]: """Spawn a networked MCP instance (streamable-http, loopback + token).""" import subprocess as _sp from src.utils.mcp_transport import read_transport_state if read_transport_state(): return {"success": False, "error": "A networked transport instance is already running."} paths = _resolve_mcp_paths() py, script = paths.get("python_path"), paths.get("server_path") if not py or not script: return {"success": False, "error": "Could not resolve the Python interpreter or server script path."} try: _sp.Popen( [py, script, "--transport", "streamable-http"], stdin=_sp.DEVNULL, stdout=_sp.DEVNULL, stderr=_sp.DEVNULL, start_new_session=True, ) except OSError as exc: return {"success": False, "error": f"Failed to launch: {exc}"} import time as _t for _ in range(15): _t.sleep(0.2) if read_transport_state(): return {"success": True, "transport": _transport_status()} return {"success": True, "note": "Launch initiated; status will appear shortly."} def _transport_stop() -> Dict[str, Any]: """Stop the running networked MCP instance via its state-file PID.""" import signal as _sig from src.utils.mcp_transport import read_transport_state, clear_transport_state state = read_transport_state() if not state: return {"success": True, "note": "No networked transport running."} pid = state.get("pid") if isinstance(pid, int): try: os.kill(pid, _sig.SIGTERM) except (OSError, ProcessLookupError): pass clear_transport_state() return {"success": True} def _mcp_install_payload(client_id: str) -> Dict[str, Any]: """Write the MCP entry for one client by delegating to install.write_client_config.""" installer, error = _load_installer_module() if installer is None: return {"success": False, "error": error or "installer unavailable"} target = next((c for c in installer.MCP_CLIENTS if c["id"] == client_id), None) if not target: return {"success": False, "error": f"unknown client: {client_id}"} paths = _resolve_mcp_paths() if not paths.get("server_path"): return {"success": False, "error": "MCP server script not found in repo"} if not paths.get("python_path"): return {"success": False, "error": "Python interpreter not found"} if not paths.get("api_path") or not paths.get("lib_path"): return { "success": False, "error": "Resolve scripting API / library paths could not be auto-detected. Open Resolve Studio at least once or install via install.py.", } try: ok, message = installer.write_client_config( target, paths["python_path"], paths["server_path"], paths["api_path"], paths["lib_path"], dry_run=False, ) except Exception as exc: return {"success": False, "error": str(exc)} return {"success": bool(ok), "message": message, "client_id": client_id} def _mcp_uninstall_payload(client_id: str) -> Dict[str, Any]: """Remove the davinci-resolve entry from a client's config file.""" installer, error = _load_installer_module() if installer is None: return {"success": False, "error": error or "installer unavailable"} target = next((c for c in installer.MCP_CLIENTS if c["id"] == client_id), None) if not target: return {"success": False, "error": f"unknown client: {client_id}"} try: config_path = target["get_path"]() except Exception as exc: return {"success": False, "error": str(exc)} if not config_path or not os.path.isfile(str(config_path)): return {"success": True, "message": "Nothing to remove (config file does not exist).", "client_id": client_id} config_key = target.get("config_key", "mcpServers") try: existing = installer.read_json(str(config_path)) servers = existing.get(config_key) or {} if "davinci-resolve" not in servers: return {"success": True, "message": "Nothing to remove (entry not present).", "client_id": client_id} del servers["davinci-resolve"] if servers: existing[config_key] = servers else: existing.pop(config_key, None) installer.write_json(str(config_path), existing) except Exception as exc: return {"success": False, "error": str(exc)} return {"success": True, "message": f"Removed davinci-resolve from {config_path}", "client_id": client_id} def _list_active_batch_jobs(project_root: str) -> List[Dict[str, Any]]: """Across the analysis base root, find batch jobs with status='running'. Used by the update apply path: refuse to update mid-job because it would corrupt in-flight clip analysis state (the new build's schema may not match what the running batch was started against). """ out: List[Dict[str, Any]] = [] if not project_root: return out base = os.path.dirname(os.path.normpath(project_root)) if not os.path.isdir(base): return out for entry in sorted(os.listdir(base)): candidate = os.path.join(base, entry) if not os.path.isdir(candidate): continue try: payload = list_batch_jobs(candidate, limit=200) except Exception: continue for job in payload.get("jobs") or []: if (job.get("status") or "").lower() == "running": out.append({ "project_root": candidate, "job_id": job.get("job_id"), "status": job.get("status"), "started_at": job.get("started_at"), }) return out def _update_apply_payload(*, strategy: str = "refuse_on_dirty", force_active_jobs: bool = False, project_root: Optional[str] = None) -> Dict[str, Any]: """Apply a guarded git fast-forward update by delegating to install.py's existing apply_safe_self_update. Returns a structured result for the UI. """ # Active-job lock — refuse to update if a batch analysis job is mid-flight. # Pass force_active_jobs=true to override (the user explicitly accepts the # risk of in-flight state being inconsistent with the new build). if not force_active_jobs and project_root: active = _list_active_batch_jobs(project_root) if active: return { "success": False, "reason": "active_jobs", "message": f"{len(active)} batch analysis job(s) are currently running. Cancel them or pass force=true to override.", "active_jobs": active, } try: sys.path.insert(0, _repo_root()) from install import apply_safe_self_update # type: ignore except Exception as exc: return {"success": False, "error": f"update helper unavailable: {exc}"} try: result = apply_safe_self_update(_repo_root(), dry_run=False, initiator="dashboard", strategy=strategy) except Exception as exc: return {"success": False, "error": str(exc)} out: Dict[str, Any] = { "success": bool(result.get("success")), "changed": bool(result.get("changed")), "reason": result.get("reason"), "message": result.get("message"), "current_version": _mcp_version(), "from_version": result.get("from_version"), "to_version": result.get("to_version"), "from_sha": result.get("from_sha"), "to_sha": result.get("to_sha"), } if result.get("success") and result.get("changed"): out["restart_required"] = True # Eagerly migrate per-project DBs so schema bumps in the new build # surface immediately instead of waiting for the next analysis call. out["db_migrations"] = _eager_migrate_after_update(project_root) # Drop a restart-needed marker the host / dashboard can poll for. _write_restart_marker(_repo_root(), result) # Surface stash status on the result. for k in ("stash_ref", "stash_pop_conflict", "remediation"): if k in result and result[k] is not None: out[k] = result[k] return out def _write_restart_marker(repo_root: str, update_result: Dict[str, Any]) -> None: """Drop a `.mcp_restart_needed` marker file with update metadata. The MCP server is a child process of the host (Claude Code, etc.) so we can't restart it ourselves. The marker is a hint the host can poll via `/api/restart_needed` or by reading the file directly. """ log_dir = os.path.join(repo_root, "logs") try: os.makedirs(log_dir, exist_ok=True) marker = { "needed": True, "from_version": update_result.get("from_version"), "to_version": update_result.get("to_version"), "from_sha": update_result.get("from_sha"), "to_sha": update_result.get("to_sha"), "applied_at": _now_iso_safe(), } with open(os.path.join(log_dir, ".mcp_restart_needed"), "w", encoding="utf-8") as fh: json.dump(marker, fh, indent=2) except OSError: pass def _now_iso_safe() -> str: import time as _time return _time.strftime("%Y-%m-%dT%H:%M:%SZ", _time.gmtime()) def _read_restart_marker(repo_root: str) -> Dict[str, Any]: path = os.path.join(repo_root, "logs", ".mcp_restart_needed") if not os.path.isfile(path): return {"needed": False} try: with open(path, "r", encoding="utf-8") as fh: payload = json.load(fh) if isinstance(payload, dict): payload.setdefault("needed", True) return payload except (OSError, json.JSONDecodeError): pass return {"needed": True, "marker_path": path} def _clear_restart_marker(repo_root: str) -> Dict[str, Any]: path = os.path.join(repo_root, "logs", ".mcp_restart_needed") try: os.remove(path) except OSError: return {"success": False, "error": "marker not present or unreadable"} return {"success": True} def _eager_migrate_after_update(project_root: Optional[str] = None) -> Dict[str, Any]: """Walk every project under the analysis base root and open + migrate its timeline_brain.sqlite. Surfaces schema bumps from the new build right after `git pull` instead of on next per-project work.""" if project_root: base = os.path.dirname(os.path.normpath(project_root)) else: base = os.path.expanduser("~/Documents/davinci-resolve-mcp-analysis") migrated: List[Dict[str, Any]] = [] if not os.path.isdir(base): return {"success": True, "migrated": migrated, "note": "no base root found"} for entry in sorted(os.listdir(base)): candidate = os.path.join(base, entry) if not os.path.isdir(os.path.join(candidate, "_soul")): continue try: _timeline_brain_db.connect(candidate) migrated.append({"project_root": candidate, "ok": True}) except Exception as exc: migrated.append({"project_root": candidate, "ok": False, "error": str(exc)}) return {"success": True, "migrated": migrated} def _update_rollback_payload() -> Dict[str, Any]: try: sys.path.insert(0, _repo_root()) from install import rollback_to_previous_build # type: ignore except Exception as exc: return {"success": False, "error": f"rollback helper unavailable: {exc}"} try: result = rollback_to_previous_build(_repo_root(), initiator="dashboard") except Exception as exc: return {"success": False, "error": str(exc)} out: Dict[str, Any] = dict(result) out["current_version"] = _mcp_version() if result.get("success"): out["restart_required"] = True return out def _update_history_payload(limit: int = 20) -> Dict[str, Any]: try: sys.path.insert(0, _repo_root()) from install import read_update_history # type: ignore except Exception as exc: return {"success": False, "error": f"history helper unavailable: {exc}", "entries": []} try: return read_update_history(_repo_root(), limit=limit) except Exception as exc: return {"success": False, "error": str(exc), "entries": []} def _update_preview_payload() -> Dict[str, Any]: """Render the about-to-apply update for user confirmation. Returns release notes, flagged breaking changes, channel, prerelease flag, and the target SHA so the dashboard can show a meaningful modal before `git pull` actually runs. """ try: sys.path.insert(0, _repo_root()) from install import preview_update # type: ignore except Exception as exc: return {"success": False, "error": f"preview helper unavailable: {exc}"} try: return preview_update(_repo_root()) except Exception as exc: return {"success": False, "error": str(exc)} def _update_status_payload(project_root: Optional[str], *, force: bool = False) -> Dict[str, Any]: current = _mcp_version() base = { "success": True, "current_version": current, "update_available": False, "status": "unknown", } # Always use the repo root, not the analysis project root — that's where the # MCP server's startup check writes its state file, so dashboard and server # share cache instead of running independent checks. update_project_dir = _repo_root() try: from src.utils.update_check import ( check_for_updates, get_cached_update_status, ) except Exception as exc: base["error"] = f"update check unavailable: {exc}" return base try: if force: payload = check_for_updates(current, update_project_dir, force=True) else: payload = get_cached_update_status(update_project_dir, current_version=current) # Cache miss → run a real check now so the UI gets a useful answer # on first load instead of "hasn't run yet". if isinstance(payload, dict) and payload.get("status") == "unknown": payload = check_for_updates(current, update_project_dir) except Exception as exc: base["error"] = str(exc) return base if not isinstance(payload, dict): return base status = str(payload.get("status") or "unknown") latest = payload.get("latest_version") or payload.get("latest") update_available = status == "update_available" return { "success": True, "current_version": current, "latest_version": str(latest) if latest else None, "status": status, "update_available": bool(update_available), "checked_at": payload.get("checked_at_iso") or payload.get("checked_at"), "snooze_until": payload.get("snooze_until_iso") or payload.get("snooze_until"), "update_mode": payload.get("update_mode"), "release_url": payload.get("release_url") or payload.get("html_url"), "release_notes": payload.get("release_notes") or payload.get("body"), } def _dashboard_doc(doc_id: Any) -> Dict[str, Any]: key = str(doc_id or "readme") source = DOC_SOURCES.get(key) if not source: return {"success": False, "error": "Unknown document"} repo_root = _repo_root() rel_path = str(source["path"]) path = os.path.abspath(os.path.join(repo_root, rel_path)) if not path.startswith(repo_root + os.sep): return {"success": False, "error": "Document path escaped repository root"} try: with open(path, "r", encoding="utf-8") as handle: content = handle.read() except OSError as exc: return {"success": False, "error": str(exc)} return { "success": True, "doc": key, "title": source["title"], "path": rel_path, "content": content, } def _setup_defaults(action: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: from src.server import setup as server_setup return server_setup(action, params or {}) def _inventory_prefs() -> Tuple[int, Optional[set]]: """Return (limit, exclude_bins) for the inventory walk from media-analysis preferences. ``exclude_bins`` is None when nothing is configured, so the walk indexes every folder by default.""" try: from src.server import _media_analysis_effective_preferences prefs = _media_analysis_effective_preferences() except Exception: # noqa: BLE001 — fall back to built-in defaults prefs = {} try: limit = max(1, min(int(prefs.get("inventory_limit", 500)), 10000)) except (TypeError, ValueError): limit = 500 raw = prefs.get("inventory_exclude_bins") exclude = {part.strip() for part in str(raw).split(",") if part.strip()} if raw else None return limit, (exclude or None) class Handler(BaseHTTPRequestHandler): state: DashboardState def log_message(self, fmt: str, *args: Any) -> None: return def _json(self, payload: Dict[str, Any], status: int = 200) -> None: raw = json.dumps(payload, ensure_ascii=False, default=str).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) def _json_etag(self, payload: Dict[str, Any]) -> None: """JSON response with an ETag so unchanged polls short-circuit to 304. The Resolve media inventory is re-fetched every few seconds; when the serialized payload is byte-identical to what the client already holds we skip both the body transfer and the client-side re-render of the table. """ raw = json.dumps(payload, ensure_ascii=False, default=str).encode("utf-8") etag = '"' + hashlib.md5(raw).hexdigest() + '"' if self.headers.get("If-None-Match") == etag: tiny = json.dumps({"success": True, "unchanged": True, "etag": etag}).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(tiny))) self.send_header("ETag", etag) self.end_headers() self.wfile.write(tiny) return self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self.send_header("ETag", etag) self.end_headers() self.wfile.write(raw) def _serve_file(self, path: str, content_type: str = "application/octet-stream") -> None: try: with open(path, "rb") as handle: raw = handle.read() except OSError: self._json({"success": False, "error": "File not found"}, HTTPStatus.NOT_FOUND) return self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(raw))) self.send_header("Cache-Control", "private, max-age=300") self.end_headers() self.wfile.write(raw) def _html(self) -> None: raw = HTML.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) def _serve_clip_export(self, body: Dict[str, Any]) -> None: clip_ids = body.get("clip_ids") fmt = body.get("format") or "json" if not isinstance(clip_ids, list) or not clip_ids: self._json({"success": False, "error": "clip_ids must be a non-empty list"}, HTTPStatus.BAD_REQUEST) return try: raw, content_type, filename = export_clip_selection(self.state.project_root, list(clip_ids), str(fmt)) except Exception as exc: # noqa: BLE001 self._json({"success": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) return self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(raw))) self.send_header("Content-Disposition", f'attachment; filename="{filename}"') self.end_headers() self.wfile.write(raw) def _body(self) -> Dict[str, Any]: length = int(self.headers.get("Content-Length") or 0) if length <= 0: return {} raw = self.rfile.read(length).decode("utf-8") try: payload = json.loads(raw) except json.JSONDecodeError: return {} return payload if isinstance(payload, dict) else {} def do_GET(self) -> None: try: self._route_get() except Exception as exc: # pragma: no cover - runtime safety for dashboard users self._json({"success": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) def do_POST(self) -> None: try: self._route_post() except Exception as exc: # pragma: no cover self._json({"success": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) def _route_get(self) -> None: parsed = urlparse(self.path) path = parsed.path query = parse_qs(parsed.query) if path == "/": self._html() return if path == "/api/boot": self._json( { "success": True, "project_name": self.state.project_name, "project_id": self.state.project_id, "project_root": self.state.project_root, "repo_root": _repo_root(), "codex_workspace": _repo_root(), "output_root": self.state.output_root, "active_context": self.state.context(), "related_project_roots": self.state.related_project_roots(), "capabilities": detect_capabilities(), "resolve": _resolve_identity(), "mcp_version": _mcp_version(), } ) return if path == "/api/projects": self._json(self.state.projects()) return if path == "/api/update/status": force = (query.get("force") or ["0"])[0].lower() in {"1", "true", "yes"} self._json(_update_status_payload(self.state.project_root, force=force)) return if path == "/api/update/history": try: limit = int((query.get("limit") or ["20"])[0]) except (TypeError, ValueError): limit = 20 self._json(_update_history_payload(limit=limit)) return if path == "/api/restart_needed": self._json(_read_restart_marker(_repo_root())) return if path == "/api/update/preview": self._json(_update_preview_payload()) return if path == "/api/mcp/status": self._json(_mcp_status_payload()) return if path == "/api/projects/all": self._json(_resolve_all_project_contexts(self.state.base_analysis_root)) return if path == "/api/jobs": self._json(list_batch_jobs(self.state.project_root)) return if path == "/api/docs": doc = (query.get("doc") or ["readme"])[0] payload = _dashboard_doc(doc) self._json(payload, 200 if payload.get("success") else 404) return if path.startswith("/api/doc_asset/"): rel = unquote(path[len("/api/doc_asset/"):]) base = os.path.realpath(os.path.join(_repo_root(), "docs", "images")) full = os.path.realpath(os.path.join(base, rel)) if not (full.startswith(base + os.sep) or full == base): self._json({"success": False, "error": "path escape"}, HTTPStatus.FORBIDDEN) return content_types = {".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".gif": "image/gif", ".svg": "image/svg+xml", ".webp": "image/webp"} ext = os.path.splitext(full)[1].lower() if ext not in content_types or not os.path.isfile(full): self._json({"success": False, "error": "not found"}, HTTPStatus.NOT_FOUND) return self._serve_file(full, content_types[ext]) return if path == "/api/setup/schema": self._json(_setup_defaults("schema")) return if path == "/api/setup/defaults": self._json(_setup_defaults("get_defaults")) return if path == "/api/resolve/media": pref_limit, exclude_bins = _inventory_prefs() self._json_etag( resolve_media_inventory( self.state.project_root, limit=(query.get("limit") or [pref_limit])[0], exclude_bins=exclude_bins, recursive=(query.get("recursive") or ["true"])[0].lower() not in {"0", "false", "no"}, probe_paths=(query.get("probe") or ["1"])[0].lower() not in {"0", "false", "no"}, reuse_cached=(query.get("reuse") or ["0"])[0].lower() in {"1", "true", "yes"}, ) ) return if path.startswith("/api/jobs/"): job_id = path.split("/")[3] self._json(batch_job_status(self.state.project_root, job_id)) return if path == "/api/index/status": self._json(analysis_index_status(self.state.project_root)) return if path == "/api/coverage": # Standalone readiness rollup — no live Resolve required. The # `coverage_report` action gives target-vs-records detail; this # endpoint summarizes what the analysis directory already knows. self._json(analysis_root_coverage(self.state.project_root)) return if path == "/api/index/query": q = (query.get("q") or [""])[0] payload = query_analysis_index(self.state.project_root, q, limit=(query.get("limit") or [20])[0]) if payload.get("success") and payload.get("results"): payload["results"] = _v2_enrich_search_results(self.state.project_root, payload["results"]) self._json(payload) return if path == "/api/entities": try: from src.utils import entities as _entities self._json(_entities.list_entities(self.state.project_root)) except Exception as exc: # noqa: BLE001 — panel reads fail soft self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/search/semantic": q = (query.get("q") or [""])[0] try: limit = int((query.get("limit") or ["20"])[0]) except (TypeError, ValueError): limit = 20 self._json(_v2_semantic_search(self.state.project_root, q, limit=limit)) return # ─── C6 timeline-history surface ─────────────────────────────── if path == "/api/timeline_versions": self._json(list_timelines_with_versions(self.state.project_root)) return if path == "/api/timeline_versions/diff": timeline_name = (query.get("timeline_name") or [""])[0] try: from_version = int((query.get("from_version") or [""])[0]) to_version = int((query.get("to_version") or [""])[0]) except (ValueError, TypeError): self._json({"success": False, "error": "from_version and to_version (ints) required"}, HTTPStatus.BAD_REQUEST) return if not timeline_name: self._json({"success": False, "error": "timeline_name required"}, HTTPStatus.BAD_REQUEST) return self._json({"success": True, **_timeline_versioning.diff_versions( project_root=self.state.project_root, timeline_name=timeline_name, from_version=from_version, to_version=to_version, )}) return if path.startswith("/api/timeline_versions/"): timeline_name = unquote(path[len("/api/timeline_versions/"):]) if not timeline_name: self._json({"success": False, "error": "timeline_name required"}, HTTPStatus.BAD_REQUEST) return self._json(get_timeline_history_payload(self.state.project_root, timeline_name)) return # ─── Edit-engine plan browser (DB/file only — no Resolve) ─────── if path == "/api/edit_plans": self._json(list_edit_plans_payload(self.state.project_root)) return if path.startswith("/api/edit_plans/"): plan_id = unquote(path[len("/api/edit_plans/"):]) if not plan_id: self._json({"success": False, "error": "plan_id required"}, HTTPStatus.BAD_REQUEST) return payload = get_edit_plan_payload(self.state.project_root, plan_id) if not payload.get("success") and "not found" in str(payload.get("error", "")): self._json(payload, HTTPStatus.NOT_FOUND) return self._json(payload) return if path == "/api/brain_edits/registry": self._json({"success": True, **_brain_edits.read_brain_edits_registry(self.state.project_root)}) return if path == "/api/caps/history": try: from src.utils import analysis_caps as _ac days = int((query.get("days") or ["30"])[0]) self._json({ "success": True, "history": _ac.get_usage_history(project_root=self.state.project_root, days=days), }) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/caps/refusals": try: from src.utils import analysis_caps as _ac limit = int((query.get("limit") or ["20"])[0]) self._json({ "success": True, "events": _ac.get_caps_events( project_root=self.state.project_root, event_type=(query.get("event_type") or ["refusal"])[0] or "refusal", limit=limit, ), }) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/media_pool_changes": try: from src.utils import media_pool_changes as _mpc limit = int((query.get("limit") or ["50"])[0]) self._json({ "success": True, "changes": _mpc.get_media_pool_change_history( project_root=self.state.project_root, limit=limit, ), }) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/runs": try: from src.utils import analysis_runs as _ar limit = int((query.get("limit") or ["50"])[0]) self._json({ "success": True, "runs": _ar.list_runs(project_root=self.state.project_root, limit=limit), "current_run_id": _ar.current_run_id(), }) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/caps": # Effective caps + per-project usage rollup. Proxies into the # media_analysis tool's get_caps action which already does the # preference lookup + DB rollup. try: from src.server import media_analysis as _ma_tool import asyncio result = asyncio.run(_ma_tool("get_caps", params={})) self._json(result) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/resolve_ai_usage": # Ledger of Resolve-local 21.0 AI ops (read straight from this # project's brain DB — no Resolve round-trip needed). try: from src.utils import resolve_ai_ledger as _ledger root = self.state.project_root self._json({ "success": True, "summary": _ledger.get_summary(project_root=root), "recent": _ledger.get_usage(project_root=root, limit=50), }) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/resolve_ai/governance": # Effective governance tier + this project's render usage. Proxies # into the media_analysis get_ai_governance action. try: from src.server import media_analysis as _ma_tool import asyncio self._json(asyncio.run(_ma_tool("get_ai_governance", params={}))) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path.startswith("/api/timeline_thumbnail/"): rel = unquote(path[len("/api/timeline_thumbnail/"):]) # Path is /; constrain it to live under _soul/timeline_versions base = os.path.join(self.state.project_root, "_soul", "timeline_versions") full = os.path.realpath(os.path.join(base, rel)) if not full.startswith(os.path.realpath(base) + os.sep) and full != os.path.realpath(base): self._json({"success": False, "error": "path escape"}, HTTPStatus.FORBIDDEN) return if not os.path.isfile(full): self._json({"success": False, "error": "not found"}, HTTPStatus.NOT_FOUND) return try: with open(full, "rb") as fh: data = fh.read() self.send_response(200) self.send_header("Content-Type", "image/png") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) except OSError as exc: self._json({"success": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) return # ─── V2 Review API ────────────────────────────────────────────── if path == "/api/clips": self._json(list_analyzed_clips(self.state.project_root)) return if path == "/api/panel_state": state_payload = read_panel_state(self.state.project_root) or {} self._json({"success": True, "state": state_payload}) return if path.startswith("/api/clips/"): parts = path.split("/") # /api/clips/ → parts = ["", "api", "clips", ""] if len(parts) >= 4: clip_id = parts[3] tail = parts[4:] if not tail: self._json(get_analyzed_clip(self.state.project_root, clip_id)) return if tail == ["shots"]: self._json(get_analyzed_clip_shots(self.state.project_root, clip_id)) return if tail == ["transcript"]: self._json(get_analyzed_clip_transcript(self.state.project_root, clip_id)) return if len(tail) == 2 and tail[0] == "shots": try: shot_index = int(tail[1]) except ValueError: self._json({"success": False, "error": "shot_index must be an integer"}, HTTPStatus.BAD_REQUEST) return self._json(get_analyzed_clip_shot(self.state.project_root, clip_id, shot_index)) return if len(tail) == 2 and tail[0] == "frames": try: frame_index = int(tail[1]) except ValueError: self._json({"success": False, "error": "frame_index must be an integer"}, HTTPStatus.BAD_REQUEST) return frame_path = get_clip_frame_path(self.state.project_root, clip_id, frame_index) if not frame_path: self._json({"success": False, "error": f"Frame {frame_index} not found for clip {clip_id}"}, HTTPStatus.NOT_FOUND) return self._serve_file(frame_path, content_type="image/jpeg") return if tail == ["corrections"]: self._json(read_clip_corrections(self.state.project_root, clip_id)) return self._json({"success": False, "error": "Not found"}, HTTPStatus.NOT_FOUND) def _route_post(self) -> None: parsed = urlparse(self.path) path = parsed.path body = self._body() if path == "/api/browse/directory": if not _request_is_loopback(self): self._json({"success": False, "error": "Native folder picker is only available to loopback clients."}, HTTPStatus.FORBIDDEN) return initial = body.get("initial") or body.get("current") or None self._json(_native_directory_picker(initial=str(initial) if initial else None)) return if path == "/api/launch/claude-code": if not _request_is_loopback(self): self._json({"success": False, "error": "Terminal launch is loopback-only."}, HTTPStatus.FORBIDDEN) return self._json(_launch_claude_code_terminal()) return if path == "/api/update/apply": if not _request_is_loopback(self): self._json({"success": False, "error": "Self-update is only available to loopback clients."}, HTTPStatus.FORBIDDEN) return strategy = (body.get("strategy") or "refuse_on_dirty").strip().lower() if strategy not in {"refuse_on_dirty", "stash_if_needed"}: strategy = "refuse_on_dirty" force_active_jobs = bool(body.get("force_active_jobs") or body.get("force")) self._json(_update_apply_payload( strategy=strategy, force_active_jobs=force_active_jobs, project_root=self.state.project_root, )) return if path == "/api/restart_needed/clear": if not _request_is_loopback(self): self._json({"success": False, "error": "Loopback only."}, HTTPStatus.FORBIDDEN) return self._json(_clear_restart_marker(_repo_root())) return if path == "/api/update/rollback": if not _request_is_loopback(self): self._json({"success": False, "error": "Rollback is only available to loopback clients."}, HTTPStatus.FORBIDDEN) return self._json(_update_rollback_payload()) return if path.startswith("/api/clips/") and path.endswith("/transcript/regenerate"): if not _request_is_loopback(self): self._json({"success": False, "error": "Transcript regeneration is loopback-only."}, HTTPStatus.FORBIDDEN) return clip_id = path.split("/")[3] # Serialize the analysis.json read-modify-write: this server is # threaded, so concurrent regenerations (or a regen racing a batch # job's report write) would interleave and the last writer would drop # the other's updates (PS9). with self.state.lock: result = regenerate_clip_transcript( self.state.project_root, clip_id, with_words=bool(body.get("with_words", True)), backend=body.get("backend") or None, language=body.get("language") or None, model=body.get("model") or None, ) self._json(result) return if path.startswith("/api/clips/") and path.endswith("/transcript/corrections"): if not _request_is_loopback(self): self._json({"success": False, "error": "Transcript edits are loopback-only."}, HTTPStatus.FORBIDDEN) return clip_id = path.split("/")[3] self._json(save_clip_transcript_corrections(self.state.project_root, clip_id, body)) return if path == "/api/mcp/install": if not _request_is_loopback(self): self._json({"success": False, "error": "MCP install routes are loopback-only."}, HTTPStatus.FORBIDDEN) return client_id = str(body.get("client_id") or "").strip() if not client_id: self._json({"success": False, "error": "client_id is required"}, HTTPStatus.BAD_REQUEST) return self._json(_mcp_install_payload(client_id)) return if path == "/api/mcp/uninstall": if not _request_is_loopback(self): self._json({"success": False, "error": "MCP install routes are loopback-only."}, HTTPStatus.FORBIDDEN) return client_id = str(body.get("client_id") or "").strip() if not client_id: self._json({"success": False, "error": "client_id is required"}, HTTPStatus.BAD_REQUEST) return self._json(_mcp_uninstall_payload(client_id)) return if path == "/api/mcp/transport/start": if not _request_is_loopback(self): self._json({"success": False, "error": "Transport management is loopback-only."}, HTTPStatus.FORBIDDEN) return self._json(_transport_start()) return if path == "/api/mcp/transport/stop": if not _request_is_loopback(self): self._json({"success": False, "error": "Transport management is loopback-only."}, HTTPStatus.FORBIDDEN) return self._json(_transport_stop()) return if path == "/api/jobs": paths = body.get("paths") or [] if isinstance(paths, str): paths = [line.strip() for line in paths.splitlines() if line.strip()] params = { "depth": body.get("depth") or "standard", "max_analysis_frames": body.get("max_analysis_frames", 8), "vision": body.get("vision") or {"enabled": False}, "transcription": body.get("transcription") or {"enabled": True, "allow_model_download": True}, "cleanup_frames": True, "reuse_project_roots": self.state.related_project_roots(), } # Honor the saved frame-sampling mode (or an explicit per-job override) # so batch runs match the user's chosen coverage/cost. Falls back to the # recommended mode when the user hasn't set a default yet (batch jobs # shouldn't block on the first-run prompt). try: from src.server import ( _media_analysis_effective_preferences as _ma_eff_prefs, ) from src.utils import media_analysis as _ma_mod _ma_prefs = _ma_eff_prefs() params["sampling_mode"] = ( body.get("sampling_mode") or _ma_prefs.get("sampling_mode_default") or _ma_mod.RECOMMENDED_SAMPLING_MODE ) params["frames_per_minute"] = body.get("frames_per_minute") or _ma_prefs.get("sampling_frames_per_minute") params["frame_floor"] = body.get("frame_floor") or _ma_prefs.get("sampling_frame_floor") params["frame_ceiling"] = body.get("frame_ceiling") or _ma_prefs.get("sampling_frame_ceiling") except Exception: # Best-effort; the engine still applies its own defaults. pass with self.state.lock: created = create_batch_job_from_paths( project_name=self.state.project_name, project_id=self.state.project_id, paths=paths, analysis_root=self.state.output_root["base_root"], recursive=bool(body.get("recursive", True)), params=params, name=body.get("name"), ) self._json(created, 200 if created.get("success") else 400) return if path.startswith("/api/jobs/") and path.endswith("/run"): job_id = path.split("/")[3] with self.state.lock: result = run_batch_job_slice( self.state.project_root, job_id, max_clips=body.get("max_clips", 1), max_seconds=body.get("max_seconds"), ) self._json(result, 200 if result.get("success") else 400) return if path.startswith("/api/jobs/") and path.endswith("/cancel"): job_id = path.split("/")[3] self._json(cancel_batch_job(self.state.project_root, job_id)) return if path.startswith("/api/jobs/") and path.endswith("/resume"): job_id = path.split("/")[3] self._json(resume_batch_job(self.state.project_root, job_id)) return if path == "/api/index/build": with self.state.lock: self._json(build_analysis_index(self.state.project_root)) return if path == "/api/setup/defaults": payload = _setup_defaults("set_defaults", body) self._json(payload, 200 if payload.get("success") else 400) return if path == "/api/setup/clear": payload = _setup_defaults("clear_defaults", body) self._json(payload, 200 if payload.get("success") else 400) return if path == "/api/context": with self.state.lock: payload = self.state.set_context(body) self._json(payload, 200 if payload.get("success") else 400) return # ─── C6 timeline-history write actions (loopback only) ───────── if path == "/api/timeline_versions/action": if not _request_is_loopback(self): self._json({"success": False, "error": "Timeline versioning writes are loopback-only."}, HTTPStatus.FORBIDDEN) return self._json(proxy_timeline_versioning_action(body)) return if path == "/api/caps": if not _request_is_loopback(self): self._json({"success": False, "error": "Caps writes are loopback-only."}, HTTPStatus.FORBIDDEN) return try: from src.server import media_analysis as _ma_tool import asyncio result = asyncio.run(_ma_tool("set_caps_preset", params=body)) self._json(result, 200 if result.get("success") else 400) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/resolve_ai/run": # Run a Resolve 21 AI op from the panel. Loopback-only because it # mutates Resolve (and the media-creators write new files). The # confirm-token two-step is handled by the consolidated tool; the # 'confirmation_required' shape is relayed to the panel as 200. if not _request_is_loopback(self): self._json({"success": False, "error": "Loopback only."}, HTTPStatus.FORBIDDEN) return try: result = _run_resolve_ai_op(body) ok = bool(result.get("success")) or result.get("status") == "confirmation_required" self._json(result, 200 if ok else 400) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/resolve_ai/governance": if not _request_is_loopback(self): self._json({"success": False, "error": "Loopback only."}, HTTPStatus.FORBIDDEN) return try: from src.server import media_analysis as _ma_tool import asyncio result = asyncio.run(_ma_tool("set_ai_governance", params=body)) self._json(result, 200 if result.get("success") else 400) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/caps/reset_day": if not _request_is_loopback(self): self._json({"success": False, "error": "Loopback only."}, HTTPStatus.FORBIDDEN) return try: from src.utils import analysis_caps as _ac result = _ac.reset_day_usage( project_root=self.state.project_root, day_bucket=body.get("day_bucket"), ) self._json(result) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/runs/begin": if not _request_is_loopback(self): self._json({"success": False, "error": "Loopback only."}, HTTPStatus.FORBIDDEN) return try: from src.utils import analysis_runs as _ar result = _ar.begin_run( project_root=self.state.project_root, label=body.get("label"), initiator=body.get("initiator") or "dashboard", ) self._json(result) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/runs/end": if not _request_is_loopback(self): self._json({"success": False, "error": "Loopback only."}, HTTPStatus.FORBIDDEN) return try: from src.utils import analysis_runs as _ar result = _ar.end_run( project_root=self.state.project_root, analysis_run_id=body.get("analysis_run_id"), ) self._json(result) except Exception as exc: self._json({"success": False, "error": f"{type(exc).__name__}: {exc}"}) return if path == "/api/update/channel": if not _request_is_loopback(self): self._json({"success": False, "error": "Loopback only."}, HTTPStatus.FORBIDDEN) return channel = (body.get("channel") or "stable").strip().lower() if channel not in {"stable", "beta", "dev"}: self._json({"success": False, "error": "channel must be stable | beta | dev"}, HTTPStatus.BAD_REQUEST) return os.environ["DAVINCI_RESOLVE_MCP_UPDATE_CHANNEL"] = channel self._json({"success": True, "channel": channel, "note": "Set for this process; persist via env var to survive restart."}) return # ─── V2 Review API (writes) ───────────────────────────────────── if path == "/api/panel_state": merge = body.pop("__merge__", True) written_by = body.pop("__written_by__", "control_panel") result = write_panel_state( self.state.project_root, {k: v for k, v in body.items() if not k.startswith("__")}, written_by=written_by, merge=bool(merge), ) self._json(result, 200 if result.get("success") else 400) return if path == "/api/resolve/open_clip": result = _v2_open_clip_in_resolve(body) self._json(result, 200 if result.get("success") else 400) return if path == "/api/resolve/create_timeline_from_clips": if not _request_is_loopback(self): self._json({"success": False, "error": "Timeline creation is loopback-only."}, HTTPStatus.FORBIDDEN) return result = _v2_create_timeline_from_clips(body) self._json(result, 200 if result.get("success") else 400) return if path == "/api/clips/combined": result = combined_clip_analysis(self.state.project_root, body) self._json(result, 200 if result.get("success") else 400) return if path == "/api/clips/export": return self._serve_clip_export(body) if path.startswith("/api/clips/"): parts = path.split("/") if len(parts) >= 5 and parts[4] == "corrections": clip_id = parts[3] result = apply_clip_correction(self.state.project_root, clip_id, body) self._json(result, 200 if result.get("success") else 400) return self._json({"success": False, "error": "Not found"}, HTTPStatus.NOT_FOUND) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run the local Resolve MCP control panel.") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8765) parser.add_argument("--project-name", default="Dashboard Analysis") parser.add_argument("--project-id", default="dashboard") parser.add_argument("--analysis-root", default=os.path.expanduser("~/Documents/davinci-resolve-mcp-analysis")) open_group = parser.add_mutually_exclusive_group() open_group.add_argument("--open", dest="open", action="store_true", help="Open the control panel in the default browser.") open_group.add_argument("--no-open", dest="open", action="store_false", help="Run the control panel server without opening a browser.") parser.set_defaults(open=False) return parser.parse_args() def _warm_inventory_cache(project_root: str) -> None: """Build the first Resolve inventory in the background at startup. Populates the inventory + path-existence caches before the browser connects so the first dashboard open paints live data immediately instead of waiting on a cold Media Pool walk. Best-effort: if Resolve isn't up yet this no-ops and the first real request builds normally. """ try: pref_limit, exclude_bins = _inventory_prefs() resolve_media_inventory(project_root, limit=pref_limit, exclude_bins=exclude_bins) except Exception: # noqa: BLE001 — warm-up must never crash startup pass def main() -> None: from src.utils import actor_identity actor_identity.set_instance("control-panel") args = parse_args() state = DashboardState(args.project_name, args.project_id, args.analysis_root) Handler.state = state server = ThreadingHTTPServer((args.host, args.port), Handler) url = f"http://{args.host}:{args.port}" print(f"DaVinci Resolve MCP: {url}") print(f"Project analysis root: {state.project_root}") threading.Thread(target=_warm_inventory_cache, args=(state.project_root,), daemon=True).start() if args.open: webbrowser.open(url) try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() if __name__ == "__main__": main()