#!/usr/bin/env python3
"""
Aria Knowledge Pipeline (AKP) — ArXiv-Connector Stage 1 (KAR-462)

Holt aktuelle ArXiv-Papers via Atom-API fuer konfigurierte Kategorien,
filtert + scored sie (Saravia-Pattern: title/abstract-keyword + author-
follow-Liste), picked Top-N, schreibt JSON nach
brain/akp/raw/arxiv/<cat>/<arxiv-id>.json und trackt State in
shared brain/youtube/.state.sqlite (source='arxiv').

Reuse: Stage 2 Triage + Stage 3 Deep + Stage 4 Briefing — alles
source-agnostic via raw_path-Lookup. Gleicher Konsumenten-Vertrag
wie aria-akp-rss-ingest.py.

Quelle: KAR-462. Saravia-Pattern aus
brain/02-Wissen/video-elvis-saravia-llm-artifacts-2026-05-21.md.
"""
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-arxiv-ingest")

import hashlib
import json
import os
import re
import sqlite3
import sys
import time
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from datetime import datetime, timedelta, timezone
from pathlib import Path

import yaml

CONFIG_PATH = Path("/root/aria/brain/youtube/.config.yaml")
ARXIV_CONFIG_PATH = Path("/root/aria/brain/05-Referenzen/arxiv-subscriptions.md")
ARXIV_RAW_DIR = Path("/root/aria/brain/akp/raw/arxiv")
ARXIV_API_BASE = "https://export.arxiv.org/api/query"
DEFAULT_CATEGORIES = ["cs.AI", "cs.CL", "cs.LG"]
DEFAULT_LOOKBACK_HOURS = 24
DEFAULT_MAX_PER_CATEGORY = 50
DEFAULT_PICK_TOP_N = 10
USER_AGENT = "aria-akp-arxiv-ingest/1.0 (Aria-Pipeline; +https://kadicon)"
HTTP_TIMEOUT_S = 30

NS = {
    "atom": "http://www.w3.org/2005/Atom",
    "arxiv": "http://arxiv.org/schemas/atom",
    "opensearch": "http://a9.com/-/spec/opensearch/1.1/",
}


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


def load_arxiv_config() -> dict:
    """Parse arxiv-subscriptions.md for categories + scoring-keywords + author-follow."""
    cfg = {
        "categories": list(DEFAULT_CATEGORIES),
        "keyword_boost": {},
        "author_follow": set(),
        "lookback_hours": DEFAULT_LOOKBACK_HOURS,
        "max_per_category": DEFAULT_MAX_PER_CATEGORY,
        "pick_top_n": DEFAULT_PICK_TOP_N,
    }
    if not ARXIV_CONFIG_PATH.exists():
        return cfg
    # Exact prefix-match: more specific names first to avoid substring collision
    # (e.g. "## Pick (Top-N pro Category)" also contains "categor").
    section_key = None
    for line in ARXIV_CONFIG_PATH.read_text().splitlines():
        line = line.strip()
        if line.startswith("## "):
            header = line[3:].strip().lower()
            if header.startswith("categor"):
                section_key = "categories"
            elif header.startswith("keyword"):
                section_key = "keywords"
            elif header.startswith("author"):
                section_key = "authors"
            elif header.startswith("lookback"):
                section_key = "lookback"
            elif header.startswith("pick"):
                section_key = "pick"
            else:
                section_key = None
            continue
        if not line.startswith("- "):
            continue
        item = line[2:].split("#", 1)[0].strip()
        if not item:
            continue
        if section_key == "categories":
            cfg["categories"] = [c.strip() for c in item.split(",") if c.strip()]
        elif section_key == "keywords":
            kw, _, weight = item.partition("=")
            try:
                cfg["keyword_boost"][kw.strip().lower()] = float(weight) if weight else 1.0
            except ValueError:
                cfg["keyword_boost"][kw.strip().lower()] = 1.0
        elif section_key == "authors":
            cfg["author_follow"].add(item.lower())
        elif section_key == "lookback":
            try:
                cfg["lookback_hours"] = int(item)
            except ValueError:
                pass
        elif section_key == "pick":
            try:
                cfg["pick_top_n"] = int(item)
            except ValueError:
                pass
    return cfg


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


def fetch_category(category: str, max_results: int) -> list[dict]:
    """Query arxiv API for one category, return list of paper dicts."""
    params = {
        "search_query": f"cat:{category}",
        "sortBy": "submittedDate",
        "sortOrder": "descending",
        "max_results": max_results,
    }
    url = f"{ARXIV_API_BASE}?{urllib.parse.urlencode(params)}"
    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT_S) as resp:
        body = resp.read()
    root = ET.fromstring(body)
    out = []
    for entry in root.findall("atom:entry", NS):
        arxiv_id_full = (entry.findtext("atom:id", default="", namespaces=NS) or "").strip()
        # http://arxiv.org/abs/2603.12345v1 -> 2603.12345
        m = re.search(r"abs/([^v\s]+)", arxiv_id_full)
        arxiv_id = m.group(1) if m else hashlib.sha1(arxiv_id_full.encode()).hexdigest()[:16]
        title = (entry.findtext("atom:title", default="", namespaces=NS) or "").strip()
        title = re.sub(r"\s+", " ", title)
        summary = (entry.findtext("atom:summary", default="", namespaces=NS) or "").strip()
        summary = re.sub(r"\s+", " ", summary)
        pub_at = (entry.findtext("atom:published", default="", namespaces=NS) or "").strip()
        authors = [
            (a.findtext("atom:name", default="", namespaces=NS) or "").strip()
            for a in entry.findall("atom:author", NS)
        ]
        pdf_link = ""
        html_link = ""
        for lnk in entry.findall("atom:link", NS):
            href = lnk.get("href", "")
            if lnk.get("title") == "pdf":
                pdf_link = href
            elif lnk.get("rel") == "alternate":
                html_link = href
        primary_cat = ""
        pc = entry.find("arxiv:primary_category", NS)
        if pc is not None:
            primary_cat = pc.get("term", "")
        out.append({
            "arxiv_id": arxiv_id,
            "title": title,
            "summary": summary,
            "published_at": pub_at,
            "authors": authors,
            "pdf_url": pdf_link,
            "html_url": html_link or f"http://arxiv.org/abs/{arxiv_id}",
            "primary_category": primary_cat,
            "category_queried": category,
        })
    return out


def parse_arxiv_published(s: str) -> datetime | None:
    """Parse 2026-05-22T17:30:00Z -> datetime UTC."""
    try:
        return datetime.fromisoformat(s.replace("Z", "+00:00"))
    except (TypeError, ValueError):
        return None


def cheap_filter(paper: dict, arxiv_cfg: dict, cfg: dict) -> str | None:
    """Return skip-reason or None if pass."""
    pub = parse_arxiv_published(paper["published_at"])
    if pub:
        age_h = (datetime.now(timezone.utc) - pub).total_seconds() / 3600
        if age_h > arxiv_cfg["lookback_hours"]:
            return "too_old"
    if len(paper["summary"]) < 200:
        return "too_short_abstract"
    title = paper["title"]
    blocklist = cfg.get("ingest", {}).get("title_blocklist_regex", "")
    if blocklist and re.search(blocklist, title, re.IGNORECASE):
        return "title_blocklist"
    return None


def score_paper(paper: dict, arxiv_cfg: dict) -> float:
    """Saravia-Pattern: weight by title/abstract keyword + author-follow."""
    score = 0.0
    text_lower = (paper["title"] + " " + paper["summary"]).lower()
    for kw, weight in arxiv_cfg["keyword_boost"].items():
        if kw in text_lower:
            score += weight
    if arxiv_cfg["author_follow"]:
        for a in paper["authors"]:
            if a.lower() in arxiv_cfg["author_follow"]:
                score += 5.0
    pub = parse_arxiv_published(paper["published_at"])
    if pub:
        age_h = (datetime.now(timezone.utc) - pub).total_seconds() / 3600
        if age_h < 12:
            score += 0.5
    return score


def already_ingested(con: sqlite3.Connection, eid: str) -> bool:
    cur = con.execute("SELECT 1 FROM ingested WHERE video_id=?", (eid,))
    return cur.fetchone() is not None


def store_skipped(con: sqlite3.Connection, eid: str, cat: str, title: str, reason: str, url: str) -> None:
    con.execute(
        "INSERT OR IGNORE INTO ingested (video_id, channel, title, url, skipped_reason, source)"
        " VALUES (?,?,?,?,?,'arxiv')",
        (eid, cat, title[:200], url, reason),
    )
    con.commit()


def store_ingested(con: sqlite3.Connection, eid: str, cat: str, title: str,
                   url: str, raw_path: str, content_chars: int, published_at: str) -> None:
    con.execute(
        "INSERT OR REPLACE INTO ingested (video_id, channel, channel_handle, title, "
        "duration_seconds, published_at, url, transcript_chars, raw_path, source) "
        "VALUES (?,?,?,?,?,?,?,?,?,'arxiv')",
        (eid, cat, cat, title[:300], None, published_at, url,
         content_chars, raw_path),
    )
    con.commit()


def write_raw(paper: dict, score: float) -> str:
    cat_slug = re.sub(r"[^\w.-]+", "-", paper.get("primary_category") or paper.get("category_queried"))
    out_dir = ARXIV_RAW_DIR / cat_slug
    out_dir.mkdir(parents=True, exist_ok=True)
    out_path = out_dir / f"{paper['arxiv_id'].replace('/', '_')}.json"
    pub = parse_arxiv_published(paper["published_at"])
    payload = {
        "video_id": paper["arxiv_id"],
        "channel_slug": cat_slug,
        "channel_handle": cat_slug,
        "title": paper["title"],
        "url": paper["html_url"],
        "upload_date": pub.strftime("%Y%m%d") if pub else "",
        "published_at": pub.isoformat() if pub else "",
        # Stage 2/3 expect "transcript" — for arxiv we use abstract as transcript
        "transcript": paper["summary"],
        "transcript_chars": len(paper["summary"]),
        "ingested_at": datetime.now(timezone.utc).isoformat(),
        "source": "arxiv",
        "arxiv_meta": {
            "authors": paper["authors"],
            "primary_category": paper["primary_category"],
            "pdf_url": paper["pdf_url"],
            "score": round(score, 3),
        },
    }
    out_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False))
    return str(out_path)


def log_run_start(con: sqlite3.Connection) -> int:
    cur = con.execute("INSERT INTO run_log (stage, status) VALUES ('ingest-arxiv', '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()
    arxiv_cfg = load_arxiv_config()
    con = db_conn(cfg["paths"]["state_db"])

    only_cat = None
    if "--only" in argv:
        idx = argv.index("--only")
        if idx + 1 < len(argv):
            only_cat = argv[idx + 1]

    dry_run = "--dry-run" in argv

    if "--backfill" in argv:
        arxiv_cfg["lookback_hours"] = max(arxiv_cfg["lookback_hours"], 7 * 24)
        arxiv_cfg["max_per_category"] = max(arxiv_cfg["max_per_category"], 200)

    run_id = log_run_start(con)
    processed = 0
    skipped = 0
    failed = 0
    notes_lines: list[str] = []

    for category in arxiv_cfg["categories"]:
        if only_cat and only_cat != category:
            continue
        try:
            papers = fetch_category(category, arxiv_cfg["max_per_category"])
        except Exception as e:
            notes_lines.append(f"{category}: fetch_error: {str(e)[:80]}")
            failed += 1
            continue

        scored: list[tuple[float, dict]] = []
        for p in papers:
            if already_ingested(con, p["arxiv_id"]):
                continue
            reason = cheap_filter(p, arxiv_cfg, cfg)
            if reason:
                if not dry_run:
                    store_skipped(con, p["arxiv_id"], category, p["title"], reason, p["html_url"])
                skipped += 1
                continue
            s = score_paper(p, arxiv_cfg)
            scored.append((s, p))

        scored.sort(key=lambda t: t[0], reverse=True)
        top_n = scored[: arxiv_cfg["pick_top_n"]]

        for score, p in top_n:
            try:
                if dry_run:
                    print(f"  [DRY] {category} score={score:.2f} {p['arxiv_id']} {p['title'][:80]}")
                else:
                    raw_path = write_raw(p, score)
                    pub = parse_arxiv_published(p["published_at"])
                    store_ingested(
                        con, p["arxiv_id"], category, p["title"],
                        p["html_url"], raw_path, len(p["summary"]),
                        pub.isoformat() if pub else "",
                    )
                processed += 1
            except Exception as e:
                failed += 1
                notes_lines.append(f"{category}/{p['arxiv_id']}: write_error: {str(e)[:80]}")
                continue

        below_cutoff = scored[arxiv_cfg["pick_top_n"]:]
        for _score, p in below_cutoff:
            if not dry_run:
                store_skipped(con, p["arxiv_id"], category, p["title"], "below_pick_cutoff", p["html_url"])
            skipped += 1

        time.sleep(3.0)  # ArXiv rate-limit: be polite

    if not dry_run:
        log_run_finish(con, run_id, processed, skipped, failed, "; ".join(notes_lines)[:500])
    print(f"[akp-arxiv] processed={processed} skipped={skipped} failed={failed} cats={len(arxiv_cfg['categories'])}")
    if notes_lines:
        for n in notes_lines[:5]:
            print(f"  NOTE: {n}")
    con.close()
    return 0


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