#!/usr/bin/env python3
"""
Aria Knowledge Pipeline (AKP) — Stage 3: Deep Process
KAR-74 · 2026-05-12

Liest alle triaged-promote Videos ohne deep_processed-Eintrag, schickt vollen
Transcript an Opus 4.7 mit kombiniertem Prompt:
  - 7-Punkt Praxisanalyse (Standing Order feedback_video_analysis_protocol)
  - Cross-Reference zu Brain via aria-brain-search.py
  - Klassifikation Brain-Entry/Issue/Skill/Spike/Ignore + Priority P0-P2
  - Inline Multi-Level-Challenge (Devils-Advocate -> Steel-Man -> Synthese)
  - Confidence-Score (Opus self-rated)

Output: Brain-Note in 00-Inbox/Videos/<date>-<channel>-<slug>.md (Standing-Order Frontmatter)
Auto-Issue gated auf P0 + confidence>=0.85 + Cross-Ref dedupe.

Gated auf ANTHROPIC_API_KEY.
"""
from __future__ import annotations
import sys as _sys
_sys.path.insert(0, "/root/aria/lib")
from aria_logging import get_logger as _get_logger
_log = _get_logger("aria-akp-deep")
from aria_audit import audit as _audit  # KAR-220 P2
from aria_schemas import DeepProcessedItem as _DeepItem  # KAR-220 P2
from aria_llm_trace import trace_llm_call  # KAR-744 LLM-Tracing

import json, os, re, sqlite3, subprocess, sys, time, yaml
from datetime import datetime, timezone
from pathlib import Path

CONFIG_PATH = Path("/root/aria/brain/youtube/.config.yaml")

DEEP_PROMPT_TEMPLATE = """Du bist Aria im Deep-Process-Modus fuer KAR-74 AKP.

Auftrag: dieses Video tief analysieren nach dem Standing Order "feedback_video_analysis_protocol" und das Ergebnis als JSON liefern.

Video-Metadata:
- Titel: {title}
- Channel: {channel}
- Dauer: {duration_min} min
- Upload: {upload_date}
- URL: {url}

Cross-Reference (top 3 verwandte bestehende Brain-Notes, falls Match):
{cross_ref_block}

Transcript:
{transcript}

---

Liefere AUSSCHLIESSLICH ein JSON-Objekt mit dieser Struktur (keine Markdown-Codefences drumherum, kein Vor- oder Nachwort):

{{
  "kernaussage": "1-2 Saetze",
  "praxisanalyse": {{
    "1_was_lernen": "...",
    "2_wie_einsetzen": "...",
    "3_vorteil_aria_kadi": "...",
    "4_besser_machen": "...",
    "5_lohnt_warum": "...",
    "6_naechste_schritte": "..."
  }},
  "kritik": "Pflichtfeld — was ist schwach, redundant, schon-bekannt, ueberholt? Nicht schoenfaerben.",
  "klassifikation": "Brain-Entry|Issue|Skill|Spike|Ignore",
  "prioritaet": "P0|P1|P2",
  "confidence": 0.0,
  "thema": "Schlagwort fuer Frontmatter",
  "nutzen": "Ein-Satz-Nutzen fuer Aria/KADi/MRR",
  "umsetzungsidee": "konkrete Action wenn klassifikation=Issue/Skill/Spike, sonst null",
  "tags": ["array", "von", "tags"],
  "issue_proposal": {{
    "title": "Wenn klassifikation=Issue: KAR-Issue-Vorschlag-Title, sonst null",
    "body_md": "Wenn klassifikation=Issue: kurzer Body inkl. why+how+effort, sonst null",
    "estimate": "S|M|L|XL"
  }},
  "challenge": {{
    "devils_advocate": "Warum sollte das NICHT umgesetzt werden? 1-2 Saetze.",
    "steel_man": "Bestes Argument FUER Umsetzung. 1-2 Saetze.",
    "synthese": "Aria-Verdict: Promote / Refine / Drop — mit kurzer Begruendung."
  }}
}}

Klassifikations-Regeln:
- Brain-Entry: Wissen aufschreiben, kein Action-Item
- Issue: konkretes KAR-Backlog-Item das gebaut werden sollte
- Skill: neuer .claude/skills/-Eintrag oder bestehender Skill-Update
- Spike: Architektur-Untersuchung noetig, mehrere Optionen offen
- Ignore: trotz Triage-Promote bei Deep-Inspect doch nicht relevant (Override)

Confidence: ehrlich self-rated 0.0-1.0. Bei Klassifikation=Issue: nur >=0.85 erlaubt Auto-Issue.
Prioritaet: P0=jetzt, P1=diese Woche, P2=Backlog.
"""


def load_config() -> dict:
    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 get_pending_promote(con: sqlite3.Connection, limit: int = 100) -> list[sqlite3.Row]:
    return list(con.execute(
        """SELECT i.video_id, i.channel, i.title, i.duration_seconds, i.url,
                   i.published_at AS upload_date, i.raw_path,
                   t.avg_score, t.aria_relevance, t.one_line_reason
            FROM ingested i
            JOIN triaged t ON t.video_id = i.video_id
            LEFT JOIN deep_processed d ON d.video_id = i.video_id
            WHERE t.verdict = 'promote'
              AND d.video_id IS NULL
            ORDER BY t.avg_score DESC, t.triaged_at ASC
            LIMIT ?""",
        (limit,),
    ))


def opus_pricing_usd(input_tokens: int, output_tokens: int) -> float:
    # Opus 4.7 pricing per Anthropic page (Stand 2026): $15/MTok input, $75/MTok output
    return (input_tokens / 1_000_000) * 15.0 + (output_tokens / 1_000_000) * 75.0


def sonnet_pricing_usd(input_tokens: int, output_tokens: int) -> float:
    # Sonnet 4.6 pricing (Stand 2026): ~$3/MTok input, ~$15/MTok output (Critic-Modell, KAR-747)
    return (input_tokens / 1_000_000) * 3.0 + (output_tokens / 1_000_000) * 15.0


def todays_cost(con: sqlite3.Connection, key: str) -> float:
    today = datetime.now(timezone.utc).date().isoformat()
    row = con.execute(f"SELECT {key} FROM daily_costs WHERE date=?", (today,)).fetchone()
    return float(row[0]) if row else 0.0


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


def cross_ref(query: str, brain_search_path: str, top_n: int) -> list[dict]:
    """Call aria-brain-search.py for top-N matches. Best-effort, no crash on failure."""
    if not Path(brain_search_path).exists():
        return []
    try:
        r = subprocess.run(
            ["python3", brain_search_path, query, "--top", str(top_n), "--json"],
            capture_output=True, text=True, timeout=15,
        )
    except subprocess.TimeoutExpired:
        return []
    if r.returncode != 0:
        # Try without --json (older script versions)
        try:
            r = subprocess.run(
                ["python3", brain_search_path, query, "--top", str(top_n)],
                capture_output=True, text=True, timeout=15,
            )
        except subprocess.TimeoutExpired:
            return []
        if r.returncode != 0:
            return []
        out = []
        for line in r.stdout.splitlines():
            line = line.strip()
            if not line or line.startswith(("==", "Top", "#", "-" * 3)):
                continue
            m = re.match(r"^\s*[\d.]+\s+(.+\.md)\s*$", line)
            if m:
                out.append({"path": m.group(1), "score": 0})
            elif line.endswith(".md"):
                out.append({"path": line, "score": 0})
        return out[:top_n]
    try:
        data = json.loads(r.stdout)
        if isinstance(data, list):
            return data[:top_n]
    except json.JSONDecodeError:
        pass
    return []


def call_opus(client, model: str, prompt: str, video_id: str | None = None,
              trace_script: str = "aria-akp-deep") -> tuple[dict | None, dict]:
    t0 = time.monotonic()
    try:
        resp = client.messages.create(
            model=model,
            max_tokens=3500,
            messages=[{"role": "user", "content": prompt}],
        )
    except Exception as e:
        trace_llm_call(script=trace_script, model=model, prompt=prompt,
                       latency_ms=int((time.monotonic() - t0) * 1000),
                       video_id=video_id, error=str(e))
        return None, {"error": str(e), "input_tokens": 0, "output_tokens": 0}
    raw_text = "".join(b.text for b in resp.content if hasattr(b, "text"))
    text = raw_text
    parsed = None
    try:
        if "```" in text:
            parts = text.split("```")
            for p in parts:
                p = p.strip()
                if p.startswith("json"):
                    p = p[4:].strip()
                if p.startswith("{"):
                    text = p
                    break
        first = text.find("{"); last = text.rfind("}")
        if first != -1 and last != -1:
            parsed = json.loads(text[first:last+1])
    except (json.JSONDecodeError, ValueError):
        parsed = None
    usage = {"input_tokens": resp.usage.input_tokens, "output_tokens": resp.usage.output_tokens}
    trace_llm_call(script=trace_script, model=model, prompt=prompt,
                   response_text=raw_text, usage=usage,
                   latency_ms=int((time.monotonic() - t0) * 1000), video_id=video_id)
    return parsed, usage


CRITIC_PROMPT_TEMPLATE = """Du bist ein UNABHAENGIGER, adversarialer Critic. Ein anderes Modell hat aus einem Video-Transcript die folgende Analyse erzeugt. Pruefe sie streng — NICHT schoenfaerben, im Zweifel kritischer.

Pruefe:
1. Halluzination: Steht alles in der Analyse wirklich im Transcript, oder erfunden/uebergeneralisiert?
2. Wert: Ist der behauptete Nutzen fuer Aria/KADi real oder generisches AI-Hype-Wiederkaeuen?
3. Klassifikation: Ist die klassifikation (Brain-Entry/Issue/Skill/Spike/Ignore) angemessen, oder wurde Hype als Issue/Skill ueberbewertet?
4. Confidence: Ist die self-rated confidence gerechtfertigt?

Transcript (Auszug):
{transcript}

Analyse (JSON des Generators):
{analysis}

Liefere AUSSCHLIESSLICH ein JSON-Objekt (keine Codefences, kein Vor-/Nachwort):
{{
  "verdict": "keep|refine|drop",
  "confidence": 0.0,
  "hallucination_risk": "low|medium|high",
  "issues": ["konkrete Schwaeche 1", "..."],
  "corrected_klassifikation": "nur wenn die urspruengliche falsch ist, sonst null",
  "one_line": "1-Satz-Urteil"
}}

Verdict-Regeln:
- keep: akkurat + wertvoll, Klassifikation passt.
- refine: brauchbar, aber Schwaechen (ueberbewertete Klassifikation, schwache Belege).
- drop: halluziniert, wertlos oder reines Hype — sollte Ignore sein.
Sei im Zweifel strenger (Default refine statt keep)."""


def call_critic(client, model: str, transcript: str, parsed: dict, video_id: str | None = None) -> tuple[dict | None, dict]:
    """KAR-747: separater, unabhaengiger Critic-Call (eigener Context, Cross-Model).

    Generator != Evaluator. Gibt (critic_dict_or_None, usage) zurueck.
    """
    prompt = CRITIC_PROMPT_TEMPLATE.format(
        transcript=(transcript or "")[:8000],
        analysis=json.dumps(parsed, ensure_ascii=False)[:6000],
    )
    return call_opus(client, model, prompt, video_id=video_id, trace_script="aria-akp-deep-critic")


def slugify(s: str, maxlen: int = 50) -> str:
    s = re.sub(r"[^\w\s-]", "", s.lower())
    s = re.sub(r"[\s_-]+", "-", s).strip("-")
    return s[:maxlen]


def write_brain_note(cfg: dict, video: sqlite3.Row, parsed: dict, cross_refs: list[dict], cost: float, critic: dict | None = None) -> str:
    out_dir = Path(cfg["deep"]["output_inbox_dir"])
    out_dir.mkdir(parents=True, exist_ok=True)
    today = datetime.now().strftime("%Y-%m-%d")
    slug = slugify(video["title"] or video["video_id"], 40)
    fname = f"{today}-{slugify(video['channel'] or 'akp', 20)}-{slug}.md"
    path = out_dir / fname

    fm_lines = [
        "---",
        f'title: "{(video["title"] or "").replace(chr(34),chr(39))}"',
        f"source: {video['url']}",
        "plattform: youtube",
        f"channel: {video['channel']}",
        f"duration_seconds: {video['duration_seconds']}",
        f'thema: {parsed.get("thema","")}',
        f'nutzen: "{parsed.get("nutzen","").replace(chr(34),chr(39))}"',
        f'umsetzungsidee: "{(parsed.get("umsetzungsidee") or "").replace(chr(34),chr(39))}"',
        f'prioritaet: {parsed.get("prioritaet","P2")}',
        f'klassifikation: {parsed.get("klassifikation","Brain-Entry")}',
        f'confidence: {parsed.get("confidence",0.0)}',
        "status: aktiv",
        f'tags: [{", ".join(parsed.get("tags",[]) + ["akp", "video", "auto-ingested"])}]',
        f"date: {today}",
        f"akp_run: kar-74",
        "---",
        "",
    ]
    body_lines = [
        f"# {video['title']}",
        "",
        f"**Channel:** {video['channel']} · **Dauer:** {int((video['duration_seconds'] or 0)/60)} min · **URL:** {video['url']}",
        "",
        "## Kernaussage",
        parsed.get("kernaussage", "(nicht extrahiert)"),
        "",
        "## 7-Punkt Praxisanalyse",
        f"### 1. Was koennen wir lernen?",
        parsed.get("praxisanalyse", {}).get("1_was_lernen", "-"),
        "",
        f"### 2. Wie koennen wir es einsetzen?",
        parsed.get("praxisanalyse", {}).get("2_wie_einsetzen", "-"),
        "",
        f"### 3. Konkreter Vorteil fuer Aria / KADi / Supplier Pulse / Workflow / Tools?",
        parsed.get("praxisanalyse", {}).get("3_vorteil_aria_kadi", "-"),
        "",
        f"### 4. Was koennen wir dadurch besser machen?",
        parsed.get("praxisanalyse", {}).get("4_besser_machen", "-"),
        "",
        f"### 5. Lohnt sich die Umsetzung — warum?",
        parsed.get("praxisanalyse", {}).get("5_lohnt_warum", "-"),
        "",
        f"### 6. Naechste Schritte",
        parsed.get("praxisanalyse", {}).get("6_naechste_schritte", "-"),
        "",
        "## Kritik (Pflichtfeld)",
        parsed.get("kritik", "(keine)"),
        "",
        "## Multi-Level-Challenge (inline, KAR-72)",
        f"**Devils-Advocate:** {parsed.get('challenge',{}).get('devils_advocate','-')}",
        "",
        f"**Steel-Man:** {parsed.get('challenge',{}).get('steel_man','-')}",
        "",
        f"**Synthese:** {parsed.get('challenge',{}).get('synthese','-')}",
        "",
    ]
    if critic:
        issues = critic.get("issues") or []
        issue_block = "\n".join(f"- {i}" for i in issues) if issues else "- (keine)"
        body_lines += [
            "## Adversarial-Critic (separater Call, Cross-Model, KAR-747)",
            f"**Verdict:** {critic.get('verdict','-')} · **Confidence:** {critic.get('confidence','-')} · **Halluzinations-Risiko:** {critic.get('hallucination_risk','-')}",
            "",
            f"**Urteil:** {critic.get('one_line','-')}",
            "",
            (f"**Korrigierte Klassifikation:** {critic.get('corrected_klassifikation')}" if critic.get('corrected_klassifikation') else ""),
            "**Schwaechen:**",
            issue_block,
            "",
        ]
    body_lines += [
        "## Cross-Reference (Brain-Match)",
    ]
    if cross_refs:
        for c in cross_refs:
            p = c.get("path", "")
            link = Path(p).stem
            body_lines.append(f"- [[{link}]] (score {c.get('score','-')})")
    else:
        body_lines.append("- (keine Brain-Matches)")
    body_lines += [
        "",
        f"## Klassifikation: **{parsed.get('klassifikation','Brain-Entry')}** · Prioritaet **{parsed.get('prioritaet','P2')}** · Confidence **{parsed.get('confidence',0.0)}**",
        "",
    ]
    if parsed.get("issue_proposal", {}).get("title"):
        ip = parsed["issue_proposal"]
        body_lines += [
            "## Issue-Proposal (auto-generated)",
            f"**Title:** {ip.get('title','')}",
            f"**Estimate:** {ip.get('estimate','M')}",
            "",
            ip.get("body_md", ""),
            "",
        ]
    body_lines += [
        "---",
        f"*Auto-generated by aria-akp-deep.py · cost ${cost:.4f} · {datetime.now(timezone.utc).isoformat()}*",
    ]
    path.write_text("\n".join(fm_lines + body_lines))
    return str(path)


def store_deep(con: sqlite3.Connection, video_id: str, classification: str, priority: str, confidence: float, kar_id: str | None, brain_path: str, cross_refs: list[dict], cost: float) -> None:
    con.execute(
        "INSERT OR REPLACE INTO deep_processed (video_id, classification, priority, confidence, kar_issue_id, brain_note_path, cross_refs_json, cost_usd) VALUES (?,?,?,?,?,?,?,?)",
        (video_id, classification, priority, confidence, kar_id, brain_path, json.dumps(cross_refs), cost),
    )
    con.commit()


def log_run_start(con: sqlite3.Connection) -> int:
    cur = con.execute("INSERT INTO run_log (stage, status) VALUES ('deep', 'running')")
    con.commit()
    return cur.lastrowid


def log_run_finish(con: sqlite3.Connection, run_id: int, processed: int, skipped: int, failed: int, notes: str = "") -> None:
    con.execute(
        "UPDATE run_log SET finished_at=datetime('now'), status='completed', items_processed=?, items_skipped=?, items_failed=?, notes=? WHERE id=?",
        (processed, skipped, failed, notes, run_id),
    )
    con.commit()


def main(argv: list[str]) -> int:
    cfg = load_config()
    con = db_conn(cfg["paths"]["state_db"])

    api_key = os.environ.get("ANTHROPIC_API_KEY")
    if not api_key:
        pending = get_pending_promote(con, 10000)
        print(f"[akp-deep] ANTHROPIC_API_KEY not set. {len(pending)} promoted videos queued.")
        return 0

    try:
        from anthropic import Anthropic
    except ImportError:
        print("[akp-deep] anthropic SDK missing.")
        return 2
    client = Anthropic(api_key=api_key)

    cost_cap = cfg["cost_caps"]["deep_max_usd"]
    total_cap = cfg["cost_caps"]["total_max_usd"]
    todays_total = todays_cost(con, "total_usd")
    todays_deep = todays_cost(con, "deep_usd")
    if todays_deep >= cost_cap or todays_total >= total_cap:
        print(f"[akp-deep] cost-cap reached. deep=${todays_deep:.4f}/${cost_cap}, total=${todays_total:.4f}/${total_cap}")
        return 0

    pending = get_pending_promote(con)
    if not pending:
        print("[akp-deep] no pending promoted videos.")
        return 0

    print(f"[akp-deep] {len(pending)} pending. budget left: deep=${cost_cap-todays_deep:.4f} total=${total_cap-todays_total:.4f}")
    run_id = log_run_start(con)

    excerpt_max = cfg["deep"]["transcript_max_chars"]
    cross_n = cfg["deep"]["cross_ref_top_n"]
    model = cfg["models"]["deep_model"]
    brain_search = cfg["paths"]["brain_search"]
    # KAR-747: separater Adversarial-Critic (Cross-Model, Default an)
    critic_enabled = bool(cfg["deep"].get("critic_enabled", True))
    critic_model = cfg["models"].get("critic_model", "claude-sonnet-4-6")
    verdict_counts: dict[str, int] = {}

    processed = 0
    skipped = 0
    failed = 0

    for row in pending:
        if todays_cost(con, "deep_usd") >= cost_cap or todays_cost(con, "total_usd") >= total_cap:
            print("[akp-deep] cost-cap reached mid-run.")
            break
        try:
            raw = json.loads(Path(row["raw_path"]).read_text())
        except (OSError, json.JSONDecodeError):
            failed += 1
            continue
        transcript = (raw.get("transcript") or "")[:excerpt_max]

        cross_query = f"{row['title']} {row['channel']}"[:200]
        crefs = cross_ref(cross_query, brain_search, cross_n)
        cref_block = "(keine Matches)" if not crefs else "\n".join(
            f"- {c.get('path','?')} (score {c.get('score','-')})" for c in crefs
        )

        prompt = DEEP_PROMPT_TEMPLATE.format(
            title=row["title"] or "",
            channel=row["channel"] or "",
            duration_min=int((row["duration_seconds"] or 0) / 60),
            upload_date=row["upload_date"] or "unknown",
            url=row["url"] or "",
            cross_ref_block=cref_block,
            transcript=transcript,
        )
        parsed, usage = call_opus(client, model, prompt, video_id=row["video_id"])
        cost = opus_pricing_usd(usage["input_tokens"], usage["output_tokens"])
        add_cost(con, deep_delta=cost)

        if not parsed or "klassifikation" not in parsed:
            store_deep(con, row["video_id"], "Ignore", "P2", 0.0, None, "", crefs, cost)
            failed += 1
            continue

        classification = parsed.get("klassifikation", "Brain-Entry")
        priority = parsed.get("prioritaet", "P2")
        confidence = float(parsed.get("confidence", 0.0))

        # KAR-747: separater Critic-Call (Generator != Evaluator). Erst messen, noch
        # kein Auto-Override der Klassifikation — Verdict landet in Note + Trace.
        critic = None
        if critic_enabled:
            critic, c_usage = call_critic(client, critic_model, transcript, parsed, video_id=row["video_id"])
            if c_usage and not c_usage.get("error"):
                c_cost = sonnet_pricing_usd(c_usage.get("input_tokens", 0), c_usage.get("output_tokens", 0))
                add_cost(con, deep_delta=c_cost)
                cost += c_cost
            if critic:
                v = critic.get("verdict", "?")
                verdict_counts[v] = verdict_counts.get(v, 0) + 1

        brain_path = write_brain_note(cfg, row, parsed, crefs, cost, critic=critic)
        store_deep(con, row["video_id"], classification, priority, confidence, None, brain_path, crefs, cost)
        processed += 1
        time.sleep(0.5)

    critic_note = f" critic_verdicts={verdict_counts}" if verdict_counts else ""
    log_run_finish(con, run_id, processed, skipped, failed, f"deep_today=${todays_cost(con,'deep_usd'):.4f}{critic_note}")
    print(f"[akp-deep] processed={processed} skipped={skipped} failed={failed} deep_today=${todays_cost(con,'deep_usd'):.4f}{critic_note}")
    con.close()
    return 0


if __name__ == "__main__":
    _log.event("script_start")
    sys.exit(main(sys.argv[1:]))