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

Aggregiert deep_processed der letzten 24h, sortiert Top-3 nach Prio+Confidence,
formatiert Telegram-MarkdownV2-Message mit Brain-Note-Pfaden + 1 Entscheidungs-
frage des Tages, sendet via mcp tool (oder Bot-API direkt fuer Cron-Run).

Anti-Spam: bei 0 promoted -> kurze Status-Message statt voller Brief.
"""
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-briefing")
from aria_audit import audit as _audit  # KAR-220 P2
import json, os, re, sqlite3, sys, urllib.parse, urllib.request, yaml
from datetime import datetime, timezone, timedelta
from pathlib import Path

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

MD2_SPECIAL = r"_*[]()~`>#+-=|{}.!"

# KAR-116: Skill-Existenz-Check vor „Skill bauen?"-Frage
# Heuristik: Substring-Match-Ratio von Title-Tokens gegen Skill-Frontmatter (name+description)
# "evals" in Title matched "regression-evals" + "llm-evaluation" + "evals-skill" usw.
SKILL_MATCH_THRESHOLD = 0.40  # Anteil der Title-Tokens die in Skill-Tokens als Substring vorkommen
SKILL_STOPWORDS = frozenset((
    "the a an of and or for to in on at by with from as is are was were be been being "
    "how why what when where who which this that these those it its their them "
    "your you i we our us my mine yours his her them they "
    "ai llm gpt claude code agent agents using build builds built make makes made "
    "skill skills tool tools how-to howto guide tutorial demo"
).split())


def _tokenize(text: str) -> set[str]:
    tokens = re.findall(r"[a-z0-9][a-z0-9\-_]{2,}", (text or "").lower())
    return {t for t in tokens if t not in SKILL_STOPWORDS}


def _skill_frontmatter(skill_md: Path) -> dict:
    """Parse minimal frontmatter (name, description) from a SKILL.md."""
    try:
        text = skill_md.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return {}
    fm = {}
    in_fm = False
    for line in text.splitlines():
        if line.strip() == "---":
            if in_fm:
                break
            in_fm = True
            continue
        if in_fm:
            m = re.match(r"^(name|description|trigger):\s*(.*?)\s*$", line)
            if m:
                fm[m.group(1)] = m.group(2).strip("\"'")
    return fm


def find_similar_skill(title: str, threshold: float = SKILL_MATCH_THRESHOLD) -> dict | None:
    """KAR-116: Sucht installed Skill mit Word-Overlap zum Video-Title.

    Returns dict mit `name`, `description`, `score`, `path` oder None.
    """
    if not SKILLS_DIR.exists():
        return None
    title_tokens = _tokenize(title)
    if not title_tokens:
        return None

    best = None
    best_score = 0.0
    for skill_md in SKILLS_DIR.glob("*/SKILL.md"):
        fm = _skill_frontmatter(skill_md)
        if not fm:
            continue
        skill_blob = f"{fm.get('name','')} {fm.get('description','')}".lower()
        skill_tokens = _tokenize(skill_blob)
        if not skill_tokens:
            continue
        # Substring-Match: jeder Title-Token zählt wenn er Teil eines Skill-Tokens ist (oder umgekehrt)
        hits = 0
        for tt in title_tokens:
            if len(tt) < 4:
                continue
            if tt in skill_blob or any(tt in s or s in tt for s in skill_tokens):
                hits += 1
        score = hits / len(title_tokens) if title_tokens else 0
        if score > best_score:
            best_score = score
            best = {
                "name": fm.get("name", skill_md.parent.name),
                "description": fm.get("description", ""),
                "score": score,
                "path": str(skill_md),
            }
    if best and best_score >= threshold:
        return best
    return None


def md2_escape(text: str) -> str:
    out = []
    for ch in text:
        if ch in MD2_SPECIAL or ch == "\\":
            out.append("\\" + ch)
        else:
            out.append(ch)
    return "".join(out)


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 fetch_window(con: sqlite3.Connection, hours: int = 28) -> dict:
    since = (datetime.now(timezone.utc) - timedelta(hours=hours)).strftime("%Y-%m-%d %H:%M:%S")
    ingested = con.execute(
        "SELECT COUNT(*) FROM ingested WHERE ingested_at >= ?", (since,)
    ).fetchone()[0]
    triaged = con.execute(
        "SELECT verdict, COUNT(*) FROM triaged WHERE triaged_at >= ? GROUP BY verdict",
        (since,),
    ).fetchall()
    triage_breakdown = {r[0]: r[1] for r in triaged}
    deeps = con.execute(
        """SELECT d.video_id, d.classification, d.priority, d.confidence,
                   d.brain_note_path, d.kar_issue_id, d.cost_usd,
                   t.avg_score,
                   i.title, i.channel, i.url, i.duration_seconds
            FROM deep_processed d
            JOIN ingested i ON i.video_id = d.video_id
            LEFT JOIN triaged t ON t.video_id = d.video_id
            WHERE d.processed_at >= ?
            ORDER BY
              CASE d.priority WHEN 'P0' THEN 0 WHEN 'P1' THEN 1 ELSE 2 END,
              d.confidence DESC,
              t.avg_score DESC
            LIMIT 50""",
        (since,),
    ).fetchall()
    return {
        "ingested": ingested,
        "triage_breakdown": triage_breakdown,
        "deeps": deeps,
        "since": since,
    }


def todays_costs(con: sqlite3.Connection) -> dict:
    today = datetime.now(timezone.utc).date().isoformat()
    row = con.execute("SELECT triage_usd, deep_usd, total_usd FROM daily_costs WHERE date=?", (today,)).fetchone()
    if not row:
        return {"triage": 0.0, "deep": 0.0, "total": 0.0}
    return {"triage": row[0] or 0.0, "deep": row[1] or 0.0, "total": row[2] or 0.0}


def fetch_todays_autocreated_kars(hours_back: int = 28) -> list[dict]:
    """KAR-563: Scan Brain-Notes for linear:-Field with mtime within window.

    aria-akp-to-linear.service runs daily ~04:00 UTC and writes linear: KAR-NNN
    into the matching Brain-Note. We surface that work in the briefing instead
    of asking Kais a redundant approval question.
    """
    inbox = Path("/root/aria/brain/00-Inbox/Videos")
    if not inbox.exists():
        return []
    since_ts = (datetime.now(timezone.utc) - timedelta(hours=hours_back)).timestamp()
    out = []
    for note in inbox.glob("*.md"):
        try:
            if note.stat().st_mtime < since_ts:
                continue
        except OSError:
            continue
        # Parse a minimal frontmatter block for linear:, title:, klassifikation:, prioritaet:
        try:
            text = note.read_text(encoding="utf-8", errors="replace")
        except OSError:
            continue
        if not text.startswith("---"):
            continue
        end = text.find("\n---", 4)
        if end == -1:
            continue
        fm = {}
        for line in text[4:end].splitlines():
            m = re.match(r"^(\w+):\s*(.*)$", line)
            if m:
                fm[m.group(1)] = m.group(2).strip().strip("\"'")
        kar_id = fm.get("linear", "")
        if not kar_id or not kar_id.startswith("KAR-"):
            continue
        out.append({
            "kar_id": kar_id,
            "title": fm.get("title", note.stem)[:80],
            "klassifikation": fm.get("klassifikation", ""),
            "prioritaet": fm.get("prioritaet", ""),
            "note_filename": note.name,
            "mtime": note.stat().st_mtime,
        })
    out.sort(key=lambda d: d["mtime"], reverse=True)
    return out


def build_message(window: dict, costs: dict, top_n: int) -> tuple[str, bool]:
    """Returns (markdownv2_text, is_full_briefing)."""
    deeps = window["deeps"]
    promoted_count = len(deeps)
    triage = window["triage_breakdown"]
    triage_summary = (
        f'{triage.get("promote",0)}p / {triage.get("summary_only",0)}s / {triage.get("skip",0)}sk'
    )

    if promoted_count == 0:
        # Anti-spam short briefing
        lines = [
            "*AKP — Tages\\-Briefing*",
            "",
            f"• ingested: {window['ingested']}",
            f"• triaged: {md2_escape(triage_summary)}",
            f"• promoted/deep\\-processed: 0",
            "",
            f"Heute keine Top\\-Tier\\-Funde\\. Kosten: ${md2_escape(format(costs['total'],'.4f'))}",
        ]
        # KAR-563: Status-Section auch im Anti-Spam-Pfad — aria-akp-to-linear
        # kann gestern's promotes heute autonom als KAR anlegen, das sehen wir hier.
        autocreated = fetch_todays_autocreated_kars()
        if autocreated:
            lines.append("")
            lines.append("*Heute autonom als KAR angelegt*")
            for k in autocreated[:8]:
                klass_short = (k["klassifikation"] or "")[:1] or "?"
                prio_short = k["prioritaet"] or ""
                tag = md2_escape(f"{klass_short}/{prio_short}") if (klass_short != "?" or prio_short) else ""
                tag_part = f" \\({tag}\\)" if tag else ""
                lines.append(f"• `{md2_escape(k['kar_id'])}` {md2_escape(k['title'])}{tag_part}")
            if len(autocreated) > 8:
                lines.append(f"• \\.\\.\\.und {len(autocreated)-8} weitere")
        return "\n".join(lines), False

    top = deeps[:top_n]
    lines = ["*AKP — Tages\\-Briefing*", ""]
    lines.append(
        f"• ingested {window['ingested']} \\| triaged {md2_escape(triage_summary)} \\| deep {promoted_count}"
    )
    lines.append(
        f"• Kosten heute: ${md2_escape(format(costs['total'],'.4f'))} \\(triage ${md2_escape(format(costs['triage'],'.4f'))}, deep ${md2_escape(format(costs['deep'],'.4f'))}\\)"
    )
    lines.append("")
    lines.append(f"*Top {len(top)}*")
    for i, d in enumerate(top, 1):
        title = (d["title"] or "")[:80]
        channel = (d["channel"] or "")[:30]
        klass = d["classification"]
        prio = d["priority"]
        conf = d["confidence"] or 0
        path_short = Path(d["brain_note_path"]).name if d["brain_note_path"] else "(no-note)"
        lines.append(f"")
        lines.append(f"{i}\\. *{md2_escape(title)}*")
        lines.append(f"   {md2_escape(channel)} \\| {md2_escape(klass)} \\| {prio} \\| conf {md2_escape(format(conf,'.2f'))}")
        lines.append(f"   `{md2_escape(path_short)}`")

    # KAR-563: Replace "Entscheidungsfrage" with autonomous-creation status.
    # Standing Order (feedback_always_kar_and_autonomous, 2026-05-21): Aria legt
    # P0/P1-Issues autonom an via aria-akp-to-linear.service — Briefing meldet
    # nur, fragt nicht.
    autocreated = fetch_todays_autocreated_kars()
    if autocreated:
        lines.append("")
        lines.append("*Heute autonom als KAR angelegt*")
        for k in autocreated[:8]:
            klass_short = (k["klassifikation"] or "")[:1] or "?"
            prio_short = k["prioritaet"] or ""
            tag = md2_escape(f"{klass_short}/{prio_short}") if (klass_short != "?" or prio_short) else ""
            tag_part = f" \\({tag}\\)" if tag else ""
            lines.append(f"• `{md2_escape(k['kar_id'])}` {md2_escape(k['title'])}{tag_part}")
        if len(autocreated) > 8:
            lines.append(f"• \\.\\.\\.und {len(autocreated)-8} weitere")
    else:
        # 2026-05-24: aria-akp-to-linear.timer DEAKTIVIERT (Backlog-Policy). Videos
        # werden nicht mehr auto als KAR promotet — daher ist "0 autocreated" der
        # Normalzustand, KEINE Drift-Warning mehr. Videos bleiben Brain-Notes.
        pass

    return "\n".join(lines), True


def send_telegram(token: str, chat_id: str, text: str, parse_mode: str = "MarkdownV2") -> bool:
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    data = urllib.parse.urlencode({
        "chat_id": chat_id,
        "text": text,
        "parse_mode": parse_mode,
        "disable_web_page_preview": "true",
    }).encode()
    req = urllib.request.Request(url, data=data, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=15) as resp:
            return resp.status == 200
    except Exception as e:
        print(f"[akp-briefing] telegram error: {e}", file=sys.stderr)
        return False


def get_telegram_token() -> str | None:
    candidates = [
        "/root/.claude/channels/telegram/.env",
        "/root/aria/.env",
        "/root/.env",
    ]
    keys = ("TG_TOKEN", "TELEGRAM_BOT_TOKEN", "TELEGRAM_TOKEN")
    for path in candidates:
        if not Path(path).exists():
            continue
        for line in Path(path).read_text().splitlines():
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, _, v = line.partition("=")
            if k.strip() in keys and v.strip():
                return v.strip().strip('"').strip("'")
    return os.environ.get("TG_TOKEN") or os.environ.get("TELEGRAM_BOT_TOKEN")


def log_run(con: sqlite3.Connection, processed: int, notes: str) -> None:
    con.execute(
        "INSERT INTO run_log (stage, finished_at, status, items_processed, notes) VALUES ('briefing', datetime('now'), 'completed', ?, ?)",
        (processed, notes),
    )
    con.commit()


COOKIES_PATH = "/root/.config/yt-dlp/cookies.txt"
COOKIE_STALE_DAYS = 10


def cookie_freshness_warning() -> str:
    """KAR-692: warnt im Briefing, wenn yt-dlp-Cookies stale werden — sonst stoppt der AKP-Download
    still (YouTube blockt den VPS, Transkriptor-URL-Fallback ist für YouTube kaputt, KAR-571).
    MarkdownV2-escaped. Leerer String wenn alles frisch."""
    try:
        age_days = int((datetime.now(timezone.utc).timestamp() - os.path.getmtime(COOKIES_PATH)) / 86400)
    except OSError:
        return "⚠️ *yt\\-dlp Cookies fehlen* \\- AKP\\-Download blockt\\. Bitte `cookies.txt` erneuern\\."
    if age_days >= COOKIE_STALE_DAYS:
        return (f"⚠️ *yt\\-dlp Cookies {age_days} Tage alt* \\- AKP\\-Download läuft bald tot\\. "
                "Bitte `cookies.txt` erneuern \\(Chrome\\-Export, dann SCP\\)\\.")
    return ""


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

    window = fetch_window(con)
    costs = todays_costs(con)
    text, is_full = build_message(window, costs, top_n)

    if "--dry-run" in argv:
        print(text)
        log_run(con, len(window["deeps"]), "dry-run")
        return 0

    token = get_telegram_token()
    if not token:
        print("[akp-briefing] no telegram token found, dry-run mode")
        print(text)
        log_run(con, len(window["deeps"]), "no_token_fallback_print")
        return 0

    ok = send_telegram(token, chat_id, text)
    _log.event("briefing_sent", items=len(window["deeps"]), ok=ok, full_briefing=is_full)
    _audit("aria-akp-briefing", "briefing_sent", target_kind="briefing", target_id=datetime.now(timezone.utc).strftime("%Y-%m-%d"), payload={"items": len(window["deeps"]), "full_briefing": is_full}, status="ok" if ok else "error")
    log_run(con, len(window["deeps"]), f"sent={ok} full={is_full}")
    print(f"[akp-briefing] sent={ok} items={len(window['deeps'])} full_briefing={is_full}")

    # KAR-692: Cookie-Freshness-Reminder als ISOLIERTE Nachricht (bricht das Briefing nie).
    warn = cookie_freshness_warning()
    if warn:
        send_telegram(token, chat_id, warn)
        print("[akp-briefing] cookie-freshness reminder sent (cookies stale/missing)")

    # Briefing-Archive (KAR-379): persist daily briefing as Brain-Note
    try:
        archive_dir = Path("/root/aria/brain/05-Referenzen/akp-briefings")
        archive_dir.mkdir(parents=True, exist_ok=True)
        date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
        archive_path = archive_dir / f"{date_str}.md"
        # Demote MD2-escapes for archive (readable text)
        unescaped = re.sub(r"\\([_*\[\]()~`>#+\-=|{}.!])", r"\1", text)
        frontmatter = f"""---
title: AKP Briefing {date_str}
type: reference
tags: [akp, briefing, daily]
date: {date_str}
status: aktiv
linear: KAR-74
deep_items: {len(window['deeps'])}
full_briefing: {is_full}
telegram_sent: {ok}
---

"""
        archive_path.write_text(frontmatter + unescaped)
        _log.event("briefing_archived", path=str(archive_path), deep_items=len(window["deeps"]))
        print(f"[akp-briefing] archived to {archive_path}")
    except Exception as e:
        print(f"[akp-briefing] archive error (non-fatal): {e}", file=sys.stderr)

    # Additive Web Push channel (KAR-303 phase 5b kickoff). Best-effort:
    # any failure here must not block the Telegram pipeline above.
    try:
        sys.path.insert(0, "/root/projekte/aria-control")
        from app.push import send_push  # type: ignore[import-not-found]

        first_lines = text.split("\n", 4)
        summary = first_lines[0] if first_lines else "AKP-Briefing"
        body = "\n".join(first_lines[1:4]).strip() or "Neue Erkenntnisse im Brain"
        push_result = send_push(
            title=summary[:100],
            body=body[:200],
            url="/?t=" + os.environ.get("ARIA_CONTROL_TOKEN", "")[:0] + "&tab=home",
        )
        print(f"[akp-briefing] push sent={push_result['sent']} removed={push_result['removed']}")
    except Exception as e:
        print(f"[akp-briefing] push skipped: {e}")

    return 0 if ok else 1


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