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

Liest Channel-Liste aus brain/05-Referenzen/youtube-channel-subscriptions.md,
holt neue Videos via yt-dlp --flat-playlist (kein YouTube-Data-API-Key noetig),
filtert per Cheap-Filter (title-blocklist + duration), pullt Transcripts via
youtube_transcript_api (timedtext-Endpoint, ~KB/Video — KAR-742), mit yt-dlp
Auto-Subs nur noch als gecappten Fallback, schreibt JSON nach
brain/youtube/raw/<channel>/<video-id>.json, trackt State in brain/youtube/.state.sqlite.

Discovery-Bypass: liest *.url / *.txt aus brain/youtube/discover/.
"""
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-ingest")
from aria_audit import audit as _audit  # KAR-220 P2

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")


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 parse_channels(memory_path: str) -> list[dict]:
    """Parse channel list from markdown memory file. Returns [{handle/id, slug}]."""
    out: list[dict] = []
    line_re = re.compile(r"^-\s+(@\w[\w.-]*|UC[\w-]{20,24})\b")
    with open(memory_path) as f:
        for line in f:
            m = line_re.match(line.strip())
            if not m:
                continue
            ident = m.group(1)
            slug = ident.lstrip("@").lower()
            out.append({"ident": ident, "slug": slug})
    seen = set()
    deduped = []
    for c in out:
        if c["ident"] in seen:
            continue
        seen.add(c["ident"])
        deduped.append(c)
    return deduped


def channel_url(ident: str) -> str:
    if ident.startswith("@"):
        return f"https://www.youtube.com/{ident}/videos"
    if ident.startswith("UC"):
        return f"https://www.youtube.com/channel/{ident}/videos"
    return ident


def yt_dlp_flat(url: str, max_items: int, cookies: str | None, proxy: str | None) -> list[dict]:
    """Fetch flat-playlist of channel — only metadata, no downloads."""
    cmd = [
        "yt-dlp",
        "--flat-playlist",
        "-J",
        "--playlist-end", str(max_items),
        "--ignore-errors",
        "--no-warnings",
        "--socket-timeout", "30",
    ]
    if cookies and Path(cookies).exists():
        cmd += ["--cookies", cookies]
    if proxy:
        cmd += ["--proxy", proxy]
    cmd.append(url)
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
    except subprocess.TimeoutExpired:
        return []
    if r.returncode != 0 and not r.stdout.strip():
        return []
    try:
        data = json.loads(r.stdout)
    except json.JSONDecodeError:
        return []
    if not isinstance(data, dict):
        return []
    entries = data.get("entries") or []
    out = []
    for e in entries:
        if not e or e.get("_type") not in (None, "url"):
            continue
        vid = e.get("id")
        if not vid:
            continue
        out.append({
            "video_id": vid,
            "title": e.get("title") or "",
            "duration": e.get("duration"),
            "url": e.get("url") or f"https://www.youtube.com/watch?v={vid}",
            "uploader": e.get("uploader") or e.get("channel") or "",
            "upload_date": e.get("upload_date") or "",
        })
    return out


def yt_dlp_transcript(url: str, cookies: str | None, proxy: str | None) -> tuple[str | None, dict | None]:
    """Pull auto-subs + metadata. Returns (transcript_text, metadata_dict).

    Two-call pattern: first metadata via --dump-json (lightweight), then sub-only
    download. Combining --write-sub with -J is unreliable because yt-dlp tries
    to extract video formats too.
    """
    import tempfile
    tmp = tempfile.mkdtemp(prefix="akp-")
    base_args = ["--no-warnings", "--socket-timeout", "60"]
    if proxy:
        base_args += ["--proxy", proxy]
    if cookies and Path(cookies).exists():
        base_args += ["--cookies", cookies]
    try:
        meta_cmd = ["yt-dlp", *base_args, "--skip-download", "--dump-json", url]
        try:
            mr = subprocess.run(meta_cmd, capture_output=True, text=True, timeout=60)
            meta = json.loads(mr.stdout) if mr.stdout.strip() else None
        except (subprocess.TimeoutExpired, json.JSONDecodeError):
            meta = None
        sub_cmd = [
            "yt-dlp", *base_args,
            "--skip-download",
            "--write-auto-sub",
            "--sub-lang", "en.*,en,de",
            "--sub-format", "vtt",
            "-o", f"{tmp}/%(id)s.%(ext)s",
            url,
        ]
        try:
            sr = subprocess.run(sub_cmd, capture_output=True, text=True, timeout=120)
        except subprocess.TimeoutExpired:
            return None, meta
        vtt_files = sorted(Path(tmp).glob("*.vtt"))
        if not vtt_files:
            err = (sr.stderr or "").strip().splitlines()[-1:] or [""]
            print(f"  [no_transcript] {url}: {err[0][:120]}", file=sys.stderr)
            return None, meta
        text = vtt_to_text(vtt_files[0].read_text(errors="replace"))
        return text, meta
    finally:
        for p in Path(tmp).iterdir():
            try:
                p.unlink()
            except OSError:
                pass
        try:
            Path(tmp).rmdir()
        except OSError:
            pass


def vtt_to_text(vtt: str) -> str:
    """Strip VTT timestamps + cue numbers, dedupe consecutive lines."""
    out_lines: list[str] = []
    last = ""
    for raw in vtt.splitlines():
        line = raw.strip()
        if not line or line.startswith(("WEBVTT", "NOTE", "Kind:", "Language:")):
            continue
        if "-->" in line or re.match(r"^\d+$", line):
            continue
        cleaned = re.sub(r"<[^>]+>", "", line)
        cleaned = re.sub(r"&\w+;", " ", cleaned).strip()
        if cleaned and cleaned != last:
            out_lines.append(cleaned)
            last = cleaned
    return "\n".join(out_lines)


def ytapi_transcript(video_id: str, proxy: str | None, languages: list[str]) -> tuple[str | None, str]:
    """KAR-742: primary transcript path via youtube_transcript_api.

    Trifft nur den `timedtext`-Endpoint (~KB statt MB/Video via yt-dlp Watch-Page).
    Proxy als GenericProxyConfig (unser Webshare-Setup nutzt IP-Auth ohne User/Pass,
    deshalb NICHT WebshareProxyConfig).

    Returns (text_or_None, status) mit status in:
      "ok"      transcript geholt
      "none"    Video hat nachweislich kein Transcript -> skip, kein yt-dlp-Fallback
      "blocked" Proxy quota/IP-Block -> Lauf abbrechen (Bandbreite schonen)
      "error"   transienter Fehler -> optionaler, gecappter yt-dlp-Fallback erlaubt
    """
    try:
        from youtube_transcript_api import (
            YouTubeTranscriptApi, RequestBlocked, IpBlocked, TranscriptsDisabled,
            NoTranscriptFound, VideoUnavailable, VideoUnplayable, AgeRestricted,
            InvalidVideoId,
        )
        from youtube_transcript_api.proxies import GenericProxyConfig
    except ImportError as e:
        print(f"  [ytapi_import_fail] {e}", file=sys.stderr)
        return None, "error"

    pc = GenericProxyConfig(http_url=proxy, https_url=proxy) if proxy else None
    api = YouTubeTranscriptApi(proxy_config=pc)
    try:
        fetched = api.fetch(video_id, languages=tuple(languages))
    except (RequestBlocked, IpBlocked) as e:
        print(f"  [ytapi_blocked] {video_id}: {type(e).__name__}", file=sys.stderr)
        return None, "blocked"
    except NoTranscriptFound:
        # bevorzugte Sprachen fehlen -> irgendein verfuegbares Transcript nehmen
        try:
            first = next(iter(api.list(video_id)), None)
            if first is None:
                return None, "none"
            fetched = first.fetch()
        except (RequestBlocked, IpBlocked):
            return None, "blocked"
        except Exception as e:
            print(f"  [ytapi_none] {video_id}: {type(e).__name__}", file=sys.stderr)
            return None, "none"
    except (TranscriptsDisabled, VideoUnavailable, VideoUnplayable, AgeRestricted, InvalidVideoId) as e:
        print(f"  [ytapi_none] {video_id}: {type(e).__name__}", file=sys.stderr)
        return None, "none"
    except Exception as e:
        print(f"  [ytapi_error] {video_id}: {type(e).__name__}: {str(e)[:120]}", file=sys.stderr)
        return None, "error"

    text = "\n".join(s.text.strip() for s in fetched if getattr(s, "text", "").strip())
    return (text, "ok") if text else (None, "none")


def yt_dlp_meta(url: str, cookies: str | None, proxy: str | None) -> dict | None:
    """Metadata-only fetch (--dump-json) fuer Discovery-Items (selten, manuell).

    Channel-Videos brauchen das NICHT — deren Metadaten kommen aus --flat-playlist.
    """
    cmd = ["yt-dlp", "--no-warnings", "--socket-timeout", "60", "--skip-download", "--dump-json"]
    if proxy:
        cmd += ["--proxy", proxy]
    if cookies and Path(cookies).exists():
        cmd += ["--cookies", cookies]
    cmd.append(url)
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
        return json.loads(r.stdout) if r.stdout.strip() else None
    except (subprocess.TimeoutExpired, json.JSONDecodeError):
        return None


def cheap_filter(video: dict, cfg: dict) -> str | None:
    """Return skip-reason or None if pass."""
    title = video.get("title") or ""
    duration = video.get("duration")
    blocklist = cfg["ingest"]["title_blocklist_regex"]
    if blocklist and re.search(blocklist, title, re.IGNORECASE):
        return "title_blocklist"
    if duration is not None:
        if duration < cfg["ingest"]["duration_min_seconds"]:
            return "too_short"
        if duration > cfg["ingest"]["duration_max_seconds"]:
            return "too_long"
    return None


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


def store_skipped(con: sqlite3.Connection, video: dict, channel_slug: str, reason: str) -> None:
    con.execute(
        "INSERT OR IGNORE INTO ingested (video_id, channel, title, duration_seconds, url, skipped_reason) VALUES (?,?,?,?,?,?)",
        (video["video_id"], channel_slug, video.get("title", ""), video.get("duration"), video.get("url", ""), reason),
    )
    con.commit()


def store_ingested(con: sqlite3.Connection, video: dict, channel_slug: str, raw_path: str, transcript_chars: int) -> 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 (?,?,?,?,?,?,?,?,?,?)",
        (
            video["video_id"],
            channel_slug,
            video.get("uploader", channel_slug),
            video.get("title", ""),
            video.get("duration"),
            video.get("upload_date", ""),
            video.get("url", ""),
            transcript_chars,
            raw_path,
            video.get("_source", "channel"),
        ),
    )
    con.commit()


def write_raw(video: dict, channel_slug: str, transcript: str, meta: dict | None, cfg: dict) -> str:
    out_dir = Path(cfg["paths"]["raw_dir"]) / channel_slug
    out_dir.mkdir(parents=True, exist_ok=True)
    out_path = out_dir / f"{video['video_id']}.json"
    payload = {
        "video_id": video["video_id"],
        "channel_slug": channel_slug,
        "channel_handle": video.get("uploader") or channel_slug,
        "title": video.get("title", ""),
        "duration_seconds": video.get("duration"),
        "url": video.get("url", ""),
        "upload_date": video.get("upload_date", ""),
        "transcript": transcript,
        "transcript_chars": len(transcript),
        "ingested_at": datetime.now(timezone.utc).isoformat(),
        "source": video.get("_source", "channel"),
        "yt_meta": {
            "view_count": meta.get("view_count") if meta else None,
            "like_count": meta.get("like_count") if meta else None,
            "channel_id": meta.get("channel_id") if meta else None,
            "categories": meta.get("categories") if meta else None,
            "tags": (meta.get("tags") or [])[:20] if meta else None,
        },
    }
    out_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False))
    return str(out_path)


def collect_discovery(cfg: dict) -> list[dict]:
    """Read discovery folder for *.url and *.txt files with YouTube URLs."""
    out: list[dict] = []
    discover_dir = Path(cfg["paths"]["discover_dir"])
    if not discover_dir.exists():
        return out
    url_re = re.compile(r"https?://[^\s'\"<>]+(?:youtube\.com/watch\?v=|youtu\.be/)[\w-]+")
    for f in discover_dir.glob("*.url"):
        text = f.read_text(errors="replace")
        for m in url_re.finditer(text):
            out.append({"url": m.group(0), "_source": "discover", "_drop_file": str(f)})
    for f in discover_dir.glob("*.txt"):
        text = f.read_text(errors="replace")
        for m in url_re.finditer(text):
            out.append({"url": m.group(0), "_source": "discover", "_drop_file": str(f)})
    return out


def video_id_from_url(url: str) -> str | None:
    m = re.search(r"(?:v=|youtu\.be/)([\w-]{6,})", url)
    return m.group(1) if m else None


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

    channels = parse_channels(cfg["paths"]["channel_memory"])
    discoveries = collect_discovery(cfg)

    cookies = cfg["ingest"].get("yt_dlp_cookies")
    proxy = os.environ.get("YT_PROXY") or cfg["ingest"].get("yt_dlp_proxy") or None

    # KAR-742: ytapi-Sprachpraeferenz + gecappter yt-dlp-Fallback gegen Bandbreiten-Burn
    languages = cfg["ingest"].get("transcript_languages") or ["en", "de", "en-US", "en-GB"]
    fallback_max = int(cfg["ingest"].get("yt_dlp_fallback_max", 5))
    fallback_used = 0
    blocked = False

    max_per_channel = cfg["ingest"]["videos_per_channel_per_run"]
    if "--backfill" in argv:
        max_per_channel = cfg["backfill"]["videos_per_channel_initial"]

    only_channel = None
    if "--only" in argv:
        idx = argv.index("--only")
        if idx + 1 < len(argv):
            only_channel = argv[idx + 1].lower().lstrip("@")
    if only_channel:
        channels = [c for c in channels if c["slug"] == only_channel]

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

    for ch in channels:
        if blocked:
            break
        url = channel_url(ch["ident"])
        videos = yt_dlp_flat(url, max_per_channel, cookies, proxy)
        if not videos:
            notes_lines.append(f"{ch['ident']}: 0 videos returned")
            continue
        for v in videos:
            if blocked:
                break
            if already_ingested(con, v["video_id"]):
                continue
            reason = cheap_filter(v, cfg)
            if reason:
                store_skipped(con, v, ch["slug"], reason)
                skipped += 1
                continue
            # KAR-742: ytapi primary (KB). Metadaten kommen aus --flat-playlist (v),
            # also KEIN --dump-json mehr pro Video.
            transcript, status = ytapi_transcript(v["video_id"], proxy, languages)
            meta = None
            if status == "blocked":
                blocked = True
                notes_lines.append("ytapi proxy blocked — Lauf abgebrochen (Bandbreite)")
                break
            if not transcript and status == "error" and fallback_used < fallback_max:
                fallback_used += 1
                transcript, meta = yt_dlp_transcript(v["url"], cookies, proxy)
            if not transcript:
                store_skipped(con, v, ch["slug"], "no_transcript")
                failed += 1
                continue
            raw_path = write_raw(v, ch["slug"], transcript, meta, cfg)
            store_ingested(con, v, ch["slug"], raw_path, len(transcript))
            processed += 1
            time.sleep(0.5)

    for d in discoveries:
        if blocked:
            break
        vid = video_id_from_url(d["url"])
        if not vid or already_ingested(con, vid):
            try:
                Path(d["_drop_file"]).unlink()
            except (OSError, KeyError):
                pass
            continue
        d["video_id"] = vid
        d["_source"] = "discover"
        # KAR-742: ytapi primary; Metadaten erst NACH erfolgreichem Transcript holen
        # (spart den schweren --dump-json-Call bei Videos ohne Caption).
        transcript, status = ytapi_transcript(vid, proxy, languages)
        if status == "blocked":
            blocked = True
            notes_lines.append("ytapi proxy blocked (discovery) — abgebrochen")
            break
        if not transcript and status == "error" and fallback_used < fallback_max:
            fallback_used += 1
            transcript, _ = yt_dlp_transcript(d["url"], cookies, proxy)
        if not transcript:
            store_skipped(con, {"video_id": vid, "title": "discover", "duration": None, "url": d["url"]}, "discover", "no_transcript")
            failed += 1
            continue
        meta = yt_dlp_meta(d["url"], cookies, proxy)
        ch_slug = (meta or {}).get("channel", "discover").lower().replace(" ", "-")
        v = {
            "video_id": vid,
            "title": (meta or {}).get("title", ""),
            "duration": (meta or {}).get("duration"),
            "url": d["url"],
            "uploader": (meta or {}).get("uploader") or (meta or {}).get("channel") or "discover",
            "upload_date": (meta or {}).get("upload_date", ""),
            "_source": "discover",
        }
        raw_path = write_raw(v, ch_slug, transcript, meta, cfg)
        store_ingested(con, v, ch_slug, raw_path, len(transcript))
        processed += 1
        try:
            Path(d["_drop_file"]).unlink()
        except OSError:
            pass

    log_run_finish(con, run_id, processed, skipped, failed, "; ".join(notes_lines)[:500])
    print(f"[akp-ingest] processed={processed} skipped={skipped} failed={failed} "
          f"yt_dlp_fallback={fallback_used} blocked={blocked}")
    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:]))