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

Liest alle ingested Videos ohne triaged-Eintrag, schickt Transcript-Excerpt an
Haiku 4.5 mit JSON-Rubric (depth/novelty/aria_relevance/production_ready 0-3),
schreibt Verdict (promote/summary_only/skip) nach triage/<video-id>.json,
trackt Cost gegen Daily-Cap.

Gated auf ANTHROPIC_API_KEY — fehlt der, wird sauber gequeued (kein Crash).
"""
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-triage")
from aria_audit import audit as _audit  # KAR-220 P2

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

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

TRIAGE_PROMPT = """Du bist Arias Triage-Filter fuer Wissens-Videos.

Bewerte das Video auf 4 Achsen mit Skala 0=schwach, 1=mittelmaessig, 2=gut, 3=stark.

Achsen:
1. depth: Tiefe der Insights vs. Oberflaeche (Skript-Tiefe, Argumentations-Schritte)
2. novelty: Neuigkeit vs. wiedergekaute Best-Practices (Datum/Stand-Bezug, neue Konzepte)
3. aria_relevance: Relevanz fuer Aria-Ziele (Multi-Agent-Patterns, Eval-Setup, Memory-Design, LLM-Distribution, Engineering-Principles, Brain-Architektur, Observability, Cost-Optimierung)
4. production_ready: Direkt umsetzbar vs. theoretisch (Code/Workflow/Architektur in der Praxis nutzbar)

Verdict-Regel (Threshold gesenkt 2026-05-23, KAR-573 — vorher 2.0/1.5):
- promote: avg(scores) >= 1.5 ODER aria_relevance >= 2
- summary_only: avg(scores) >= 1.0
- skip: alles andere

Begruendung: 319 summary_only-Videos waren ein "oberflaechlich behandelter"
Backlog der Wert haben koennte. Lockere Schwelle bringt mehr Videos in den
Deep-Process. Sollte das Brain mit Noise zumuellen: Threshold wieder hoch.

Output AUSSCHLIESSLICH ein JSON-Objekt mit dieser Struktur (keine Erklaerung drumherum):
{{"depth": int, "novelty": int, "aria_relevance": int, "production_ready": int, "verdict": "promote"|"summary_only"|"skip", "one_line_reason": str}}

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

Transcript-Excerpt:
{transcript_excerpt}
"""


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(con: sqlite3.Connection, limit: int = 250) -> 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
            FROM ingested i
            LEFT JOIN triaged t ON t.video_id = i.video_id
            WHERE i.skipped_reason IS NULL
              AND i.raw_path IS NOT NULL
              AND t.video_id IS NULL
            ORDER BY i.ingested_at ASC
            LIMIT ?""",
        (limit,),
    ))


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, triage_delta: float = 0.0, deep_delta: float = 0.0) -> None:
    today = datetime.now(timezone.utc).date().isoformat()
    con.execute(
        "INSERT INTO daily_costs (date, triage_usd, deep_usd, total_usd) VALUES (?,?,?,?) "
        "ON CONFLICT(date) DO UPDATE SET triage_usd=triage_usd+?, deep_usd=deep_usd+?, total_usd=total_usd+?",
        (today, triage_delta, deep_delta, triage_delta + deep_delta,
         triage_delta, deep_delta, triage_delta + deep_delta),
    )
    con.commit()


def haiku_pricing_usd(input_tokens: int, output_tokens: int) -> float:
    # Haiku 4.5 pricing (per Anthropic Pricing Page, Stand 2026): $1/MTok input, $5/MTok output
    return (input_tokens / 1_000_000) * 1.0 + (output_tokens / 1_000_000) * 5.0


def call_haiku(client, model: str, prompt: str) -> tuple[dict | None, dict]:
    """Call Haiku, expect JSON. Returns (parsed_json or None, usage_dict)."""
    try:
        resp = client.messages.create(
            model=model,
            max_tokens=400,
            messages=[{"role": "user", "content": prompt}],
        )
    except Exception as e:
        return None, {"error": str(e), "input_tokens": 0, "output_tokens": 0}
    text = "".join(b.text for b in resp.content if hasattr(b, "text"))
    parsed = None
    try:
        if "```" in text:
            text = text.split("```")[1]
            if text.startswith("json"):
                text = text[4:]
        text = text.strip()
        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,
    }
    return parsed, usage


def store_triage(con: sqlite3.Connection, video_id: str, scores: dict, verdict: str, reason: str, path: str, cost: float) -> None:
    avg = (scores["depth"] + scores["novelty"] + scores["aria_relevance"] + scores["production_ready"]) / 4.0
    con.execute(
        "INSERT OR REPLACE INTO triaged (video_id, depth, novelty, aria_relevance, production_ready, avg_score, verdict, one_line_reason, triage_path, cost_usd) VALUES (?,?,?,?,?,?,?,?,?,?)",
        (video_id, scores["depth"], scores["novelty"], scores["aria_relevance"], scores["production_ready"], avg, verdict, reason, path, cost),
    )
    con.commit()


def write_triage_file(cfg: dict, video_id: str, payload: dict) -> str:
    out_dir = Path(cfg["paths"]["triage_dir"])
    out_dir.mkdir(parents=True, exist_ok=True)
    p = out_dir / f"{video_id}.json"
    p.write_text(json.dumps(payload, indent=2, ensure_ascii=False))
    return str(p)


def log_run_start(con: sqlite3.Connection) -> int:
    cur = con.execute("INSERT INTO run_log (stage, status) VALUES ('triage', '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(con, limit=10000)
        print(f"[akp-triage] ANTHROPIC_API_KEY not set. {len(pending)} videos queued.")
        log_run_start(con)
        log_run_finish(con, con.execute("SELECT MAX(id) FROM run_log").fetchone()[0], 0, len(pending), 0, "no_api_key")
        return 0

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

    cost_cap = cfg["cost_caps"]["triage_max_usd"]
    todays = todays_cost(con, "triage_usd")
    if todays >= cost_cap:
        print(f"[akp-triage] daily cost-cap reached ({todays:.4f}/{cost_cap} USD). Skipping run.")
        return 0

    pending = get_pending(con)
    if not pending:
        print("[akp-triage] no pending videos.")
        return 0

    print(f"[akp-triage] {len(pending)} pending. cost-budget left: ${cost_cap - todays:.4f}")
    run_id = log_run_start(con)

    excerpt_max = cfg["triage"]["transcript_max_chars"]
    promote_avg = cfg["triage"]["promote_threshold_avg"]
    promote_aria = cfg["triage"]["promote_threshold_aria"]
    summary_avg = cfg["triage"]["summary_threshold_avg"]
    model = cfg["models"]["triage_model"]

    processed = 0
    failed = 0
    skipped = 0

    for row in pending:
        if (todays_cost(con, "triage_usd") >= cost_cap):
            print("[akp-triage] cost-cap reached mid-run, stopping.")
            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]
        if not transcript.strip():
            store_triage(con, row["video_id"], {"depth":0,"novelty":0,"aria_relevance":0,"production_ready":0}, "skip", "empty_transcript", "", 0.0)
            skipped += 1
            continue
        prompt = TRIAGE_PROMPT.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",
            transcript_excerpt=transcript,
        )
        parsed, usage = call_haiku(client, model, prompt)
        cost = haiku_pricing_usd(usage["input_tokens"], usage["output_tokens"])
        add_cost(con, triage_delta=cost)

        if not parsed or "verdict" not in parsed:
            store_triage(con, row["video_id"], {"depth":0,"novelty":0,"aria_relevance":0,"production_ready":0}, "skip", f"parse_fail: {usage.get('error','')[:100]}", "", cost)
            failed += 1
            continue

        scores = {
            "depth": int(parsed.get("depth", 0)),
            "novelty": int(parsed.get("novelty", 0)),
            "aria_relevance": int(parsed.get("aria_relevance", 0)),
            "production_ready": int(parsed.get("production_ready", 0)),
        }
        avg = sum(scores.values()) / 4.0
        verdict_raw = parsed.get("verdict", "skip")
        if avg >= promote_avg and scores["aria_relevance"] >= promote_aria:
            verdict = "promote"
        elif avg >= summary_avg or scores["aria_relevance"] >= promote_aria:
            verdict = "summary_only"
        else:
            verdict = "skip"
        if verdict_raw in ("promote", "summary_only", "skip") and verdict_raw != verdict:
            verdict = verdict_raw if verdict_raw == "skip" else verdict

        reason = parsed.get("one_line_reason", "")[:300]
        payload = {
            "video_id": row["video_id"],
            "channel": row["channel"],
            "title": row["title"],
            "scores": scores,
            "avg_score": avg,
            "verdict": verdict,
            "one_line_reason": reason,
            "model": model,
            "cost_usd": cost,
            "triaged_at": datetime.now(timezone.utc).isoformat(),
        }
        path = write_triage_file(cfg, row["video_id"], payload)
        store_triage(con, row["video_id"], scores, verdict, reason, path, cost)
        processed += 1
        time.sleep(0.2)

    log_run_finish(con, run_id, processed, skipped, failed, f"todays_cost=${todays_cost(con,'triage_usd'):.4f}")
    print(f"[akp-triage] processed={processed} skipped={skipped} failed={failed} cost_today=${todays_cost(con,'triage_usd'):.4f}")
    con.close()
    return 0


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