#!/usr/bin/env python3
"""KAR-565 — AKP Stage 3.5: Visual-Frame Analyse fuer Promote-Klassifizierte Videos.

Nach Stage 3 (deep) laeuft Stage 3.5 fuer Klassifikation in (Issue, Skill, Spike).
Liefert visual_insights die NICHT im Transcript stehen (Slides, Code-Demos, Diagramme).

Pipeline:
1. SQL: deep_processed Videos mit klassifikation IN (Issue, Skill, Spike) ohne `## Visual Insights` in Brain-Note
2. Frame-Extract via `claude-watch/scripts/extract.sh` (8 evenly-spaced JPGs, gecacht)
3. Multi-Modal-Call (Sonnet 4.6 Default, Gemini Flash via VISION_FORCE=gemini)
4. Append `## Visual Insights` Section an Brain-Note
5. Cost-Tracking in `daily_costs.vision_usd` (auto-add column)

Cost-Profile:
- Sonnet 4.6: ~$0.04/Video (8 frames + transcript-excerpt input + summary output)
- Gemini Flash: ~$0.0004/Video (100x guenstiger, niedrigere Quality)

Env:
- VISION_FORCE=sonnet|gemini  — Default sonnet
- VISION_MAX=N                — max Videos pro Run (Default 5)
- ANTHROPIC_API_KEY or PRIVATE_OPS_ANTHROPIC_API_KEY  — for Sonnet
- GEMINI_API_KEY              — for Gemini fallback
"""
from __future__ import annotations

import argparse
import base64
import json
import os
import re
import sqlite3
import subprocess
import sys
import time
import urllib.request
from datetime import datetime, timezone
from pathlib import Path

sys.path.insert(0, "/root/aria/lib")
from aria_llm_trace import trace_llm_call  # KAR-744 LLM-Tracing

CONFIG_PATH = Path("/root/aria/brain/youtube/.config.yaml")
EXTRACT_SH = "/root/.claude/skills/claude-watch/scripts/extract.sh"
SONNET_MODEL = "claude-sonnet-4-6"
GEMINI_MODEL = "gemini-2.5-flash"

VISION_PROMPT = """Du bist Aria im Visual-Insights-Modus. Du siehst {n_frames} Frames aus einem AI-Engineering Video und einen Transcript-Auszug.

Ziel: Finde NEUE Erkenntnisse die VISUELL gezeigt werden aber im Transcript nicht (oder nur vage) erklaert sind.

Konkret achten auf:
- Code-Snippets in Slides (zeige sie woertlich)
- Architektur-Diagramme (skizziere die Komponenten + Pfeile)
- Charts/Numbers (gib die Daten wieder)
- API-Endpoints, Filenames, Config-Keys auf Slides
- UI-Screenshots mit relevanten Elementen
- Hand-Schrift / Whiteboard / Sketches

NICHT zu reportieren:
- Speaker-Face Closeups ohne Visual-Inhalt
- Generic Logos / Title-Slides / Outro
- Verallgemeinerungen die der Transcript schon liefert

Liefere AUSSCHLIESSLICH JSON, keine Markdown-Fences. MAX 5 Bullets visual_insights, MAX 3 slide_code_snippets, MAX 3 diagrams_described — keep responses tight:

{{
  "visual_insights": [
    "Bullet 1 — konkretes Visual-Detail (5-15 Woerter, max 5)"
  ],
  "slide_code_snippets": [
    "exakter Code aus Slide (max 3, max 200 Zeichen je)"
  ],
  "diagrams_described": [
    "Architektur: Component A -> B -> C (1-Satz, max 3)"
  ],
  "frame_count_useful": 0,
  "transcript_complement": "1-2 Saetze: was Vision ergaenzt das Transcript nicht hat"
}}

Wenn keine substantiellen Visual-Insights: alle Arrays leer, `transcript_complement: "Video ist transcript-vollstaendig, kein Visual-Mehrwert."`"""


def load_config() -> dict:
    import yaml
    with CONFIG_PATH.open() as f:
        return yaml.safe_load(f)


def db_conn(db_path: str) -> sqlite3.Connection:
    con = sqlite3.connect(db_path)
    con.row_factory = sqlite3.Row
    return con


def ensure_cost_column(con: sqlite3.Connection) -> None:
    try:
        con.execute("ALTER TABLE daily_costs ADD COLUMN vision_usd REAL DEFAULT 0")
        con.commit()
    except sqlite3.OperationalError:
        pass


def add_vision_cost(con: sqlite3.Connection, delta: float) -> None:
    today = datetime.now(timezone.utc).date().isoformat()
    con.execute(
        "INSERT INTO daily_costs (date, triage_usd, deep_usd, vision_usd, total_usd) VALUES (?,0,0,?,?) "
        "ON CONFLICT(date) DO UPDATE SET vision_usd = COALESCE(vision_usd,0) + ?, total_usd = total_usd + ?",
        (today, delta, delta, delta, delta),
    )
    con.commit()


def get_pending(con: sqlite3.Connection, limit: int) -> list[sqlite3.Row]:
    return list(con.execute(
        """SELECT d.video_id, d.brain_note_path, d.classification, d.priority, d.confidence,
                   i.url, i.title, i.channel, i.duration_seconds
            FROM deep_processed d
            JOIN ingested i ON i.video_id = d.video_id
            WHERE d.classification IN ('Issue','Skill','Spike')
              AND d.brain_note_path IS NOT NULL
              AND i.url IS NOT NULL
              AND i.source = 'youtube'
            ORDER BY d.processed_at DESC
            LIMIT ?""",
        (limit * 3,),  # over-fetch since some may already have Visual Insights
    ))


def has_visual_insights(brain_note: Path) -> bool:
    if not brain_note.exists():
        return True  # skip — can't write to missing file
    try:
        return "## Visual Insights" in brain_note.read_text()
    except OSError:
        return True


def run_extract(url: str, work_root: str = "/tmp") -> Path | None:
    """Call claude-watch/extract.sh, return frames dir."""
    try:
        r = subprocess.run(
            ["bash", EXTRACT_SH, url],
            capture_output=True, text=True, timeout=600,
            env={**os.environ, "FRAMES": "8", "FRAME_WIDTH": "768"},
        )
    except subprocess.TimeoutExpired:
        return None
    if r.returncode != 0:
        sys.stderr.write(f"[vision] extract.sh fail: {r.stderr[-200:]}\n")
        return None
    work_dir = r.stdout.strip().splitlines()[-1] if r.stdout.strip() else None
    if not work_dir or not Path(work_dir).is_dir():
        return None
    return Path(work_dir)


def call_sonnet(frames: list[Path], transcript_excerpt: str, api_key: str) -> tuple[dict | None, dict]:
    t0 = time.monotonic()
    prompt_static = VISION_PROMPT.format(n_frames=len(frames))
    transcript_block = f"Transcript-Excerpt (truncated to 6 KB):\n\n{transcript_excerpt[:6000]}"
    # KAR-744: Trace-Prompt ohne Bild-b64 (sonst Riesen-Zeilen); n_frames als Hinweis.
    trace_prompt = f"{prompt_static}\n\n{transcript_block}\n\n[+{len(frames)} frames]"
    content = [
        {"type": "text", "text": prompt_static},
        {"type": "text", "text": transcript_block},
    ]
    for f in frames:
        b64 = base64.b64encode(f.read_bytes()).decode("ascii")
        content.append({
            "type": "image",
            "source": {"type": "base64", "media_type": "image/jpeg", "data": b64},
        })
    body = {
        "model": SONNET_MODEL,
        "max_tokens": 2500,
        "messages": [{"role": "user", "content": content}],
    }
    req = urllib.request.Request(
        "https://api.anthropic.com/v1/messages",
        data=json.dumps(body).encode(),
        headers={
            "x-api-key": api_key,
            "anthropic-version": "2023-06-01",
            "content-type": "application/json",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            d = json.load(resp)
    except Exception as e:
        trace_llm_call(script="aria-akp-vision", model=SONNET_MODEL, prompt=trace_prompt,
                       latency_ms=int((time.monotonic() - t0) * 1000), error=str(e))
        return None, {"error": str(e)}
    text = "".join(b.get("text", "") for b in d.get("content", []))
    usage = d.get("usage", {})
    trace_llm_call(script="aria-akp-vision", model=SONNET_MODEL, prompt=trace_prompt,
                   response_text=text,
                   usage={"input_tokens": usage.get("input_tokens"), "output_tokens": usage.get("output_tokens")},
                   latency_ms=int((time.monotonic() - t0) * 1000))
    cost = (usage.get("input_tokens", 0) / 1e6) * 3.0 + (usage.get("output_tokens", 0) / 1e6) * 15.0
    try:
        first = text.find("{")
        last = text.rfind("}")
        parsed = json.loads(text[first:last + 1])
    except (json.JSONDecodeError, ValueError):
        parsed = None
    return parsed, {"input_tokens": usage.get("input_tokens"), "output_tokens": usage.get("output_tokens"), "cost_usd": round(cost, 5), "raw": text[:600]}


def call_gemini(frames: list[Path], transcript_excerpt: str, api_key: str) -> tuple[dict | None, dict]:
    t0 = time.monotonic()
    prompt_static = VISION_PROMPT.format(n_frames=len(frames))
    transcript_block = f"Transcript-Excerpt (truncated to 6 KB):\n\n{transcript_excerpt[:6000]}"
    trace_prompt = f"{prompt_static}\n\n{transcript_block}\n\n[+{len(frames)} frames]"
    parts = [
        {"text": prompt_static},
        {"text": transcript_block},
    ]
    for f in frames:
        b64 = base64.b64encode(f.read_bytes()).decode("ascii")
        parts.append({"inline_data": {"mime_type": "image/jpeg", "data": b64}})
    body = {
        "contents": [{"parts": parts}],
        "generationConfig": {"maxOutputTokens": 1500},
    }
    url = f"https://generativelanguage.googleapis.com/v1beta/models/{GEMINI_MODEL}:generateContent?key={api_key}"
    req = urllib.request.Request(url, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            d = json.load(resp)
    except Exception as e:
        trace_llm_call(script="aria-akp-vision", model=GEMINI_MODEL, prompt=trace_prompt,
                       latency_ms=int((time.monotonic() - t0) * 1000), error=str(e))
        return None, {"error": str(e)}
    candidate = (d.get("candidates") or [{}])[0]
    text = "".join(p.get("text", "") for p in (candidate.get("content") or {}).get("parts", []))
    usage = d.get("usageMetadata", {})
    trace_llm_call(script="aria-akp-vision", model=GEMINI_MODEL, prompt=trace_prompt,
                   response_text=text,
                   usage={"input_tokens": usage.get("promptTokenCount"), "output_tokens": usage.get("candidatesTokenCount")},
                   latency_ms=int((time.monotonic() - t0) * 1000))
    cost = (usage.get("promptTokenCount", 0) * 0.10 + usage.get("candidatesTokenCount", 0) * 0.40) / 1e6
    try:
        first = text.find("{")
        last = text.rfind("}")
        parsed = json.loads(text[first:last + 1])
    except (json.JSONDecodeError, ValueError):
        parsed = None
    return parsed, {"input_tokens": usage.get("promptTokenCount"), "output_tokens": usage.get("candidatesTokenCount"), "cost_usd": round(cost, 6), "raw": text[:600]}


def append_visual_insights(brain_note: Path, insights: dict, model: str, cost: float) -> None:
    text = brain_note.read_text()
    section = ["", "## Visual Insights", ""]
    bullets = insights.get("visual_insights") or []
    if bullets:
        for b in bullets:
            section.append(f"- {b}")
    else:
        section.append("- (keine substantiellen Visual-Insights — Video transcript-vollstaendig)")
    code = insights.get("slide_code_snippets") or []
    if code:
        section.append("")
        section.append("**Slide-Code:**")
        for c in code:
            section.append(f"```\n{c}\n```")
    diagrams = insights.get("diagrams_described") or []
    if diagrams:
        section.append("")
        section.append("**Diagramme:**")
        for d in diagrams:
            section.append(f"- {d}")
    complement = insights.get("transcript_complement")
    if complement:
        section.append("")
        section.append(f"**Transcript-Ergaenzung:** {complement}")
    section.append("")
    section.append(f"*Auto-generated by aria-akp-vision.py ({model}, cost ${cost:.4f}, {datetime.now(timezone.utc).isoformat()})*")
    section.append("")
    # Insert before the trailing auto-generated footer if present, else append
    if "---\n*Auto-generated by aria-akp-deep.py" in text:
        text = text.replace("---\n*Auto-generated by aria-akp-deep.py",
                            "\n".join(section) + "\n---\n*Auto-generated by aria-akp-deep.py")
    else:
        text = text.rstrip() + "\n\n" + "\n".join(section)
    brain_note.write_text(text)


def get_transcript_excerpt(con: sqlite3.Connection, video_id: str) -> str:
    row = con.execute("SELECT raw_path FROM ingested WHERE video_id = ?", (video_id,)).fetchone()
    if not row or not row["raw_path"]:
        return ""
    p = Path(row["raw_path"])
    if not p.exists():
        return ""
    try:
        data = json.loads(p.read_text())
        return (data.get("transcript") or "")[:8000]
    except (json.JSONDecodeError, OSError):
        return ""


def get_api_key(name: str) -> str | None:
    val = os.environ.get(name)
    if val:
        return val
    # Inline-fallback: parse .env
    for path in ("/root/aria/.env",):
        if Path(path).exists():
            for line in Path(path).read_text().splitlines():
                if line.startswith(f"{name}="):
                    return line.split("=", 1)[1].strip().strip('"').strip("'")
    return None


def get_anthropic_key() -> str | None:
    return get_api_key("ANTHROPIC_API_KEY") or get_api_key("PRIVATE_OPS_ANTHROPIC_API_KEY")


def get_gemini_key() -> str | None:
    val = os.environ.get("GEMINI_API_KEY")
    if val:
        return val
    sec = Path("/root/.aria-secrets/gemini.env")
    if sec.exists():
        for line in sec.read_text().splitlines():
            if line.startswith("GEMINI_API_KEY="):
                return line.split("=", 1)[1].strip().strip('"').strip("'")
    return None


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--max", type=int, default=int(os.environ.get("VISION_MAX", "5")))
    ap.add_argument("--video-id", help="Run for one specific video_id (debug)")
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--force-model", choices=["sonnet", "gemini"],
                    default=os.environ.get("VISION_FORCE", "sonnet"))
    args = ap.parse_args()

    cfg = load_config()
    con = db_conn(cfg["paths"]["state_db"])
    ensure_cost_column(con)

    if args.video_id:
        rows = list(con.execute(
            """SELECT d.video_id, d.brain_note_path, d.classification, d.priority, d.confidence,
                       i.url, i.title, i.channel
                FROM deep_processed d JOIN ingested i ON i.video_id = d.video_id
                WHERE d.video_id = ?""", (args.video_id,)
        ))
    else:
        rows = get_pending(con, args.max)

    # Filter to those without existing Visual Insights
    candidates = []
    for row in rows:
        bn = Path(row["brain_note_path"] or "")
        if not bn.exists():
            continue
        if not has_visual_insights(bn):
            candidates.append(row)
        if len(candidates) >= args.max:
            break
    print(f"[akp-vision] candidates: {len(candidates)} model={args.force_model}")

    if args.dry_run:
        for r in candidates[:5]:
            print(f"  DRY  {r['video_id'][:11]}  {r['classification']:5s}  {r['priority']}  {(r['title'] or '')[:60]}")
        return 0

    api_key_sonnet = get_anthropic_key()
    api_key_gemini = get_gemini_key()
    if args.force_model == "sonnet" and not api_key_sonnet:
        print("[akp-vision] no ANTHROPIC_API_KEY, falling back to gemini", file=sys.stderr)
        args.force_model = "gemini"
    if args.force_model == "gemini" and not api_key_gemini:
        print("[akp-vision] ERROR: no API key available", file=sys.stderr)
        return 2

    processed = skipped = failed = 0
    for i, row in enumerate(candidates, 1):
        vid = row["video_id"]
        url = row["url"]
        brain_note = Path(row["brain_note_path"])
        print(f"[{i}/{len(candidates)}] {vid} {(row['title'] or '')[:50]}...")

        work_dir = run_extract(url)
        if not work_dir:
            print(f"  SKIP no frames")
            skipped += 1
            continue
        frames = sorted(work_dir.glob("frames/frame_*.jpg"))[:8]
        if not frames:
            print(f"  SKIP no frame files")
            skipped += 1
            continue

        transcript = get_transcript_excerpt(con, vid)

        if args.force_model == "sonnet":
            insights, meta = call_sonnet(frames, transcript, api_key_sonnet)
            model_name = SONNET_MODEL
        else:
            insights, meta = call_gemini(frames, transcript, api_key_gemini)
            model_name = GEMINI_MODEL

        if not insights:
            print(f"  FAIL parse: {meta}")
            failed += 1
            continue

        cost = meta.get("cost_usd") or 0.0
        add_vision_cost(con, cost)
        append_visual_insights(brain_note, insights, model_name, cost)
        n_ins = len(insights.get("visual_insights") or [])
        print(f"  OK {n_ins} insights, ${cost:.4f} ({meta.get('input_tokens')}/{meta.get('output_tokens')})")
        processed += 1
        time.sleep(1.0)

    print(f"[akp-vision] processed={processed} skipped={skipped} failed={failed} model={args.force_model}")
    return 0


if __name__ == "__main__":
    sys.exit(main())