#!/usr/bin/env python3
"""videngineer · MCP connector

Gives any agent (Claude Desktop / Claude Code / Cursor / any MCP client) the ability
to reverse-engineer a video — WITHOUT exposing videngineer's raw API surface. The
agent only ever sees these clean tools; the endpoints, auth flow, and topology stay
inside this connector.

Tools (all read-only except analyze_video, which starts an analysis on YOUR account):
  • analyze_video(url, label?)              → start an analysis, returns a job_id
  • get_status(job_id)                      → queued | running | complete | error
  • get_report(job_id, preset?, fields?)    → one analysis; lean by default, full or
                                              per-field on request
  • list_analyses()                         → your analyses, newest first
  • get_context_bundle(analysis_ids, preset)→ 1..10 of YOUR analyses packaged as ONE
                                              generation-ready payload (remix /
                                              synthesis / hooks_only) for an external
                                              image/video tool

Ownership: every id is authorized server-side against YOUR key, per id. An analysis
you don't own comes back redacted (owned:false — title/thumbnail/hook-type only)
with the option to analyze the source URL yourself. Nothing here ever spends your
credits automatically.

Setup (the customer does this once):
  export VIDENGINEER_API_KEY="<their key>"   # Account → API access → Generate key
  # optional: export VIDENGINEER_BASE_URL="https://videngineer.com"
  python videngineer_mcp.py        # or via uvx / the Claude config (see README.md)

Auth: the API key is sent as a Bearer token to videngineer; a valid key acts as the
customer's account, on their plan. One key per customer (see README for issuing keys).
"""
from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor

try:
    from mcp.server.fastmcp import FastMCP
except ImportError:
    raise SystemExit(
        "Missing dependency. Install with:  pip install 'mcp[cli]'  (or: uv add mcp)"
    )

BASE = os.environ.get("VIDENGINEER_BASE_URL", "https://videngineer.com").rstrip("/")
KEY = os.environ.get("VIDENGINEER_API_KEY", "").strip()

MAX_BUNDLE = 10          # hard cap on ids per bundle
PRESETS = ("remix", "synthesis", "hooks_only")
REPORT_PRESETS = ("lean", "full")

mcp = FastMCP("videngineer")


# ─── internal API plumbing (never surfaced to the agent) ───────────────
def _request(method: str, path: str, body: dict | None = None, timeout: int = 90) -> dict:
    if not KEY:
        return {"error": "VIDENGINEER_API_KEY is not set. Generate a key at videngineer.com: Account → API access → Generate key."}
    data = json.dumps(body).encode("utf-8") if body is not None else None
    req = urllib.request.Request(
        f"{BASE}{path}", data=data, method=method,
        headers={"Authorization": f"Bearer {KEY}", "Content-Type": "application/json",
                 "User-Agent": "videngineer-mcp/1.5"},
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            raw = r.read()
            return json.loads(raw) if raw else {}
    except urllib.error.HTTPError as e:
        body_txt = e.read().decode("utf-8", errors="replace")
        try:
            return {"error": json.loads(body_txt).get("error", f"HTTP {e.code}"),
                    "detail": body_txt[:200], "status_code": e.code}
        except Exception:
            return {"error": f"HTTP {e.code}", "detail": body_txt[:200], "status_code": e.code}
    except Exception as e:
        return {"error": f"request failed: {e}"}


def _fetch_context(analysis_id: str) -> dict:
    """One analysis via the ownership-checked context endpoint. The server decides
    owned vs redacted per id against THIS key — the connector never fetches an
    analysis body any other way."""
    return _request("GET", f"/api/v1/project/{analysis_id}/context")


# ─── distill layer: compact, labeled, token-lean (never the raw envelope) ──
def _beats(bp: dict, cap: int = 30) -> list[str]:
    """scene_timeline → compact beat lines a generation tool can follow."""
    out = []
    for s in (bp.get("scene_timeline") or [])[:cap]:
        ts = s.get("timestamp") or ""
        kind = s.get("scene_type") or "scene"
        subj = (s.get("subject") or "").strip()
        txt = (s.get("text_on_screen") or "").strip()
        line = f"{ts} · {kind} — {subj}" if subj else f"{ts} · {kind}"
        if txt:
            line += f' · text:"{txt[:60]}"'
        out.append(line[:180])
    return out


def _voice_line(va: dict) -> str:
    """voice_analysis → one castable line."""
    if not isinstance(va, dict) or not va:
        return "no narration detected"
    parts = [str(va[k]) for k in
             ("gender", "age_range", "tone", "pace", "voice_quality", "energy_level")
             if va.get(k)]
    if va.get("is_ai_voice"):
        parts.append("AI-generated voice")
    line = " · ".join(parts) or (va.get("voice_notes") or "").strip()
    return line[:200] or "no narration detected"


def _top_moves(sc: dict, n: int = 3) -> list[str]:
    """scorecard → its n strongest dimensions, with the one-line read."""
    dims = sc.get("dimensions") or []
    dims = sorted((d for d in dims if isinstance(d, dict)),
                  key=lambda d: d.get("score") or 0, reverse=True)[:n]
    return [f"{d.get('label') or d.get('key')} {d.get('score')}/10 — {d.get('read', '')}"[:200]
            for d in dims]


def _scorecard_slice(sc: dict) -> dict:
    return {"overall": sc.get("overall"), "verdict": sc.get("verdict"),
            "top_moves": _top_moves(sc)}


def _hook_text(bp: dict) -> str:
    return ((bp.get("opening_hook") or bp.get("hook") or "").strip())[:240]


def _pacing(bp: dict) -> dict:
    dur = bp.get("video_duration_seconds") or 0
    scenes = bp.get("distinct_scenes") or len(bp.get("scene_timeline") or [])
    return {"duration_seconds": dur, "scene_count": scenes,
            "avg_scene_seconds": round(dur / scenes, 1) if scenes else None,
            "hook_score": bp.get("hook_score")}


def _style(bp: dict) -> dict:
    rec = bp.get("production_recommendation") or {}
    return {"dominant_style": rec.get("dominant_style"),
            "dominant_angle": rec.get("dominant_angle"),
            "animated_frames_pct": bp.get("animated_frames_pct")}


def _assets(ctx: dict, keyframes: int) -> dict:
    frames = ((ctx.get("context") or {}).get("frame_urls") or [])[:keyframes]
    return {"thumb_url": ctx.get("thumbnail_url"), "keyframe_urls": frames}


def _clusters_slice(bp: dict, cap: int = 8) -> list[dict]:
    groups = ((bp.get("deep_clusters") or {}).get("technique_groups") or [])[:cap]
    out = []
    for g in groups:
        if not isinstance(g, dict):
            continue
        out.append({"technique": g.get("technique") or g.get("name"),
                    "count": g.get("count") or len(g.get("frame_urls") or []) or None,
                    "example_frame": (g.get("frame_urls") or [None])[0]})
    return out


def _not_owned(ctx: dict) -> dict:
    """The redacted shape for an analysis this key does NOT own. Built ONLY from
    the server's already-redacted payload — preview + how to add it yourself.
    Never auto-analyzes; the credit cost is stated so the caller decides."""
    cost = ctx.get("analyze_cost_credits")
    return {
        "id": ctx.get("job_id"),
        "owned": False,
        "preview": {"title": ctx.get("title"), "thumbnail_url": ctx.get("thumbnail_url"),
                    "hook_type": ctx.get("hook_type"),
                    "duration_seconds": ctx.get("duration_seconds")},
        "message": "This analysis isn't in your library — preview only.",
        "add_to_library": {
            "tool": "analyze_video",
            "url": ctx.get("source_url"),
            "cost_credits": cost,
            "note": ("Analyzing this URL adds the full teardown to your library"
                     + (f" for {cost} credits" if cost else "")
                     + ". It is never triggered automatically — call analyze_video yourself."),
        },
    }


def _compact_item(ctx: dict, preset: str) -> dict:
    """One OWNED, complete analysis → the compact labeled object for a bundle."""
    c = ctx.get("context") or {}
    bp = c.get("blueprint") or {}
    base = {"id": ctx.get("job_id"), "owned": True, "title": ctx.get("title"),
            "hook_type": ctx.get("hook_type"), "hook": _hook_text(bp)}
    if preset == "hooks_only":
        base.update({
            "opening_beats": _beats(bp, cap=3),
            "scorecard": _scorecard_slice(c.get("scorecard") or {}),
            "assets": _assets(ctx, keyframes=3),
        })
        return base
    if preset == "synthesis":  # thin per-video ref; the pattern layer carries the merge
        sc = c.get("scorecard") or {}
        base.update({
            "duration_seconds": ctx.get("duration_seconds"),
            "scene_count": bp.get("distinct_scenes"),
            "overall_score": sc.get("overall"),
            "assets": {"thumb_url": ctx.get("thumbnail_url"), "keyframe_urls": []},
        })
        return base
    # remix — deep, generation-ready
    base.update({
        "blueprint": {"beats": _beats(bp, cap=30), "pacing": _pacing(bp),
                      "style": _style(bp)},
        "voice": _voice_line(c.get("voice_analysis") or {}),
        "transcript": (c.get("transcript") or "")[:8000],
        "recreation_prompts": c.get("recreation_prompts") or [],
        "scorecard": _scorecard_slice(c.get("scorecard") or {}),
        "assets": _assets(ctx, keyframes=12),
    })
    return base


def _synthesis_layer(items: list[dict], contexts: list[dict]) -> dict:
    """The merged pattern layer across N owned analyses — computed here, honestly,
    from structure (no model call): what the videos share, where the hooks diverge."""
    n = len(contexts)
    # scene_type presence + average normalized position across videos
    seen: dict[str, list[float]] = {}
    for ctx in contexts:
        bp = (ctx.get("context") or {}).get("blueprint") or {}
        tl = bp.get("scene_timeline") or []
        total = max(len(tl), 1)
        counted = set()
        for i, s in enumerate(tl):
            kind = s.get("scene_type")
            if not kind or kind in counted:
                continue
            counted.add(kind)
            seen.setdefault(kind, []).append(i / total)
    common_beats = [
        {"scene_type": k, "in_videos": f"{len(v)}/{n}",
         "avg_position_pct": round(100 * sum(v) / len(v))}
        for k, v in sorted(seen.items(), key=lambda kv: sum(kv[1]) / len(kv[1]))
        if len(v) * 2 >= n
    ]
    divergent_hooks = [
        {"id": ctx.get("job_id"), "title": ctx.get("title"),
         "hook_type": ctx.get("hook_type"),
         "hook": _hook_text((ctx.get("context") or {}).get("blueprint") or {}),
         "hook_score": (((ctx.get("context") or {}).get("blueprint")) or {}).get("hook_score")}
        for ctx in contexts
    ]
    # voice traits identical across every narrated video
    traits: dict[str, str] | None = None
    for ctx in contexts:
        va = (ctx.get("context") or {}).get("voice_analysis") or {}
        cur = {k: str(va[k]) for k in
               ("gender", "tone", "pace", "voice_quality", "energy_level") if va.get(k)}
        if not cur:
            continue
        traits = cur if traits is None else {k: v for k, v in traits.items() if cur.get(k) == v}
    return {"videos_merged": n, "common_beats": common_beats,
            "divergent_hooks": divergent_hooks,
            "shared_voice_traits": traits or {}}


# ─── the only surface the agent sees ───────────────────────────────────
@mcp.tool()
def analyze_video(url: str, label: str = "") -> dict:
    """Start reverse-engineering a video. Accepts a YouTube, TikTok, Instagram, Vimeo,
    X/Twitter, Google Drive, or direct-MP4 URL. Returns a job_id — pass it to
    get_report() to fetch the full breakdown once it's done (~90 seconds)."""
    r = _request("POST", "/api/v1/analyze", {"url": url, "name": label, "interval": "auto"})
    if r.get("error"):
        return r
    return {"job_id": r.get("job_id"), "status": r.get("status", "queued"),
            "note": "Call get_report(job_id) to retrieve the breakdown (usually ready in ~90s)."}


@mcp.tool()
def get_status(job_id: str) -> dict:
    """Check whether an analysis is queued, running, complete, or errored."""
    r = _fetch_context(job_id)
    if r.get("error"):
        return r
    if not r.get("owned"):
        return _not_owned(r)
    return {"job_id": job_id, "status": r.get("status"), "progress": r.get("progress")}


@mcp.tool()
def get_report(job_id: str, preset: str = "lean", fields: list[str] | None = None,
               wait: bool = True) -> dict:
    """Read one analysis from your library. preset="lean" (default) is the compact
    read: hook, scorecard verdict + top moves, beat preview, voice, counts.
    preset="full" adds the complete blueprint, transcript, voice profile,
    recreation prompts, technique clusters, cast, and ROI. Or pass fields=[...]
    to pull exactly what you need (see `available_fields` in the lean response).
    If wait=True (default), polls until the analysis completes (up to ~5 minutes)."""
    if preset not in REPORT_PRESETS:
        return {"error": f"Unknown preset '{preset}'. Use one of: {', '.join(REPORT_PRESETS)}."}
    deadline = time.time() + 300
    while True:
        r = _fetch_context(job_id)
        if r.get("error"):
            return r
        if not r.get("owned"):
            return _not_owned(r)
        status = r.get("status")
        if status == "complete":
            break
        if status == "error":
            return {"job_id": job_id, "status": "error", "error": "analysis failed"}
        if not wait or time.time() > deadline:
            return {"job_id": job_id, "status": status, "progress": r.get("progress"),
                    "note": "Not finished yet — call get_report again in a bit."}
        time.sleep(5)

    c = r.get("context") or {}
    bp = c.get("blueprint") or {}
    full_map = {
        "blueprint": {k: v for k, v in bp.items()
                      if k not in ("scene_timeline", "deep_clusters")},
        "beats": _beats(bp, cap=60),
        "transcript": c.get("transcript") or "",
        "voice_profile": c.get("voice_analysis") or {},
        "classification": c.get("classification") or {},
        "scorecard": c.get("scorecard") or {},
        "recreation_prompts": c.get("recreation_prompts") or [],
        "scene_clusters": _clusters_slice(bp, cap=20),
        "cast": c.get("entity_analysis") or {},
        "roi": c.get("roi") or {},
        "assets": _assets(r, keyframes=12),
    }
    head = {"job_id": job_id, "owned": True, "status": "complete",
            "video": r.get("title") or r.get("source_url")}
    if fields:
        unknown = [f for f in fields if f not in full_map]
        if unknown:
            return {"error": f"Unknown fields: {unknown}. Available: {sorted(full_map)}"}
        return {**head, **{f: full_map[f] for f in fields}}
    if preset == "full":
        return {**head, **full_map}
    return {**head,
            "hook_type": r.get("hook_type"),
            "hook": _hook_text(bp),
            "scorecard": _scorecard_slice(c.get("scorecard") or {}),
            "beats_preview": _beats(bp, cap=8),
            "voice": _voice_line(c.get("voice_analysis") or {}),
            "counts": {"scenes": bp.get("distinct_scenes"),
                       "duration_seconds": r.get("duration_seconds"),
                       "recreation_prompts": len(c.get("recreation_prompts") or []),
                       "keyframes": len(c.get("frame_urls") or [])},
            "assets": {"thumb_url": r.get("thumbnail_url")},
            "available_fields": sorted(full_map),
            "note": 'Lean view. get_report(job_id, preset="full") or fields=[...] for more.'}


@mcp.tool()
def get_context_bundle(analysis_ids: list[str], preset: str = "remix") -> dict:
    """Package 1..10 of YOUR analyses into ONE generation-ready payload for an
    external image/video tool. Presets:
      • "remix" — deep dive per video: beats, pacing, style, voice, transcript,
        recreation prompts (image+video, [YOUR PRODUCT] placeholders), keyframe URLs.
        Best with 1 video.
      • "synthesis" — N videos merged into a pattern layer (common_beats,
        divergent_hooks, shared_voice_traits) + thin per-video refs. No transcripts.
      • "hooks_only" — openings + scorecard only, across N videos.
    Each id is authorized against YOUR key; anything you don't own comes back
    redacted (owned:false) with the analyze-it-yourself option — never full data."""
    if preset not in PRESETS:
        return {"error": f"Unknown preset '{preset}'. Use one of: {', '.join(PRESETS)}."}
    ids, seen = [], set()
    for i in [str(x).strip() for x in (analysis_ids or []) if str(x).strip()]:
        if i not in seen:
            seen.add(i); ids.append(i)
    if not ids:
        return {"error": "analysis_ids is empty — pass at least one job_id (see list_analyses)."}
    truncated = len(ids) > MAX_BUNDLE
    kept = ids[:MAX_BUNDLE]

    # fan-out: one ownership-checked fetch PER id — the server redacts foreign ids
    with ThreadPoolExecutor(max_workers=min(5, len(kept))) as pool:
        fetched = list(pool.map(_fetch_context, kept))

    items, not_owned, unavailable, owned_ctx = [], [], [], []
    for an_id, ctx in zip(kept, fetched):
        if ctx.get("error"):
            unavailable.append({"id": an_id, "owned": False,
                                "reason": "not found" if ctx.get("status_code") == 404
                                          else ctx.get("error")})
        elif not ctx.get("owned"):
            not_owned.append(_not_owned(ctx))
        elif ctx.get("status") != "complete":
            unavailable.append({"id": an_id, "owned": True, "reason": f"analysis is {ctx.get('status')}",
                                "note": "Wait for it to complete, then re-bundle."})
        else:
            owned_ctx.append(ctx)
            items.append(_compact_item(ctx, preset))

    out = {"preset": preset, "requested": len(ids), "included": len(items),
           "truncated": truncated}
    if truncated:
        out["note"] = f"Bundles cap at {MAX_BUNDLE} analyses — {len(ids) - MAX_BUNDLE} id(s) dropped."
    if preset == "synthesis" and owned_ctx:
        out["pattern_layer"] = _synthesis_layer(items, owned_ctx)
    out["videos" if preset == "synthesis" else "items"] = items
    if not_owned:
        out["not_owned"] = not_owned
    if unavailable:
        out["unavailable"] = unavailable
    if items:
        out["generation_note"] = ("recreation_prompts are image-first (I2V): generate the still "
                                  "from image_prompt, animate with video_prompt, keep on-screen "
                                  "text in your editor. Swap [YOUR PRODUCT]/[YOUR TEXT] placeholders.")
    return out


@mcp.tool()
def list_analyses() -> dict:
    """List the analyses on your videngineer account, newest first — job_id, title,
    status, and created date. Use it to track what's been analyzed, then pass any
    job_id to get_report() to read that breakdown (one or many)."""
    r = _request("GET", "/api/v1/projects")
    if r.get("error"):
        return r
    items = r.get("projects") or []
    items = sorted(items, key=lambda p: p.get("created_at") or "", reverse=True)
    return {
        "count": len(items),
        "analyses": [
            {"job_id": p.get("id"), "title": p.get("name"),
             "status": p.get("status"), "created_at": p.get("created_at")}
            for p in items
        ],
        "note": "Pass a job_id to get_report(), or several to get_context_bundle().",
    }


if __name__ == "__main__":
    mcp.run()