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

Liest Podcast-RSS-Feed-Liste, pullt neue Episoden, downloaded Audio temporaer,
transcribed via groq-whisper, schreibt JSON nach brain/akp/raw/podcast/.

Cheap-Filter (Audio-spezifisch):
- duration 20-180 min (skip kurze Trailer + extra-lange Marathons)
- age <14 days
- title-blocklist + Werbung-Patterns
- skip wenn Show in `youtube-channel-subscriptions.md` parallel abgedeckt
  (z.B. Latent Space hat YouTube-Version, vermeidet Doppel-Ingest)

Cost-Cap (separat von Triage/Deep):
- groq-whisper-large-v3-turbo: $0.04/h Audio
- Bei 5 Episoden/Tag á 60 min = $0.20/Tag
- Daily-Cap konfigurierbar in .config.yaml
"""
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-podcast-ingest")

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

import feedparser

CONFIG_PATH = Path("/root/aria/brain/youtube/.config.yaml")
PODCAST_LIST_PATH = Path("/root/aria/brain/05-Referenzen/podcast-subscriptions.md")
PODCAST_RAW_DIR = Path("/root/aria/brain/akp/raw/podcast")
# Hybrid-Router: Groq für ≤60s, Transkriptor für >60s (Pro-Plan flat-rate, KAR-127)
WHISPER_SCRIPT = Path("/root/aria/scripts/aria-transcribe-route.py")

# Default-Cost-Cap (USD/Tag) — analog .config.yaml pattern
DEFAULT_WHISPER_DAILY_CAP_USD = 1.0  # erlaubt ~25h Audio/Tag

AD_PATTERNS_TITLE = [
    r"\bsponsor\b",
    r"\bsponsored by\b",
    r"\bbrought to you by\b",
]


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_feeds(memory_path: Path) -> list[str]:
    out = []
    url_re = re.compile(r"^-\s+(https?://\S+)")
    if not memory_path.exists():
        return []
    for line in memory_path.read_text().splitlines():
        m = url_re.match(line.strip())
        if m:
            out.append(m.group(1).rstrip(",;)"))
    seen = set(); deduped = []
    for u in out:
        if u in seen: continue
        seen.add(u); deduped.append(u)
    return deduped


def feed_slug(url: str) -> str:
    s = re.sub(r"https?://", "", url)
    s = re.sub(r"[^\w]+", "-", s).strip("-").lower()
    return s[:60]


def episode_id(entry) -> str:
    for key in ("id", "guid"):
        v = getattr(entry, key, None)
        if v:
            return hashlib.sha1(v.encode()).hexdigest()[:16]
    link = getattr(entry, "link", "") or ""
    title = getattr(entry, "title", "") or ""
    return hashlib.sha1((link + title).encode()).hexdigest()[:16]


def find_audio_url(entry) -> tuple[str | None, int | None]:
    """Return (audio_url, duration_seconds_estimate or None)."""
    for enc in entry.get("enclosures", []) or []:
        if isinstance(enc, dict):
            t = enc.get("type") or ""
            href = enc.get("href") or enc.get("url")
            if t.startswith("audio/") and href:
                return href, None
    # itunes:duration
    dur_str = entry.get("itunes_duration", "")
    duration = None
    if dur_str:
        parts = dur_str.split(":")
        try:
            parts = [int(p) for p in parts]
            if len(parts) == 3:
                duration = parts[0] * 3600 + parts[1] * 60 + parts[2]
            elif len(parts) == 2:
                duration = parts[0] * 60 + parts[1]
            elif len(parts) == 1:
                duration = parts[0]
        except ValueError:
            pass
    # no audio url found
    return None, duration


def episode_published_at(entry):
    for key in ("published_parsed", "updated_parsed"):
        v = entry.get(key)
        if v:
            try:
                return datetime(*v[:6], tzinfo=timezone.utc)
            except (TypeError, ValueError):
                pass
    return None


def cheap_filter(entry, duration: int | None) -> str | None:
    title = entry.get("title", "") or ""
    pub = episode_published_at(entry)
    if pub and (datetime.now(timezone.utc) - pub) > timedelta(days=14):
        return "too_old"
    if duration is not None:
        if duration < 20 * 60:
            return "too_short_lt20min"
        if duration > 180 * 60:
            return "too_long_gt3h"
    for pat in AD_PATTERNS_TITLE:
        if re.search(pat, title, re.IGNORECASE):
            return "ad_title"
    return None


def download_audio(audio_url: str, tmp_dir: Path, proxy: str | None) -> Path | None:
    """Use yt-dlp to download audio (handles redirects, partial fetch, formats)."""
    out_path = tmp_dir / "episode.mp3"
    cmd = ["yt-dlp", "-o", str(out_path), "--no-warnings",
           "--no-playlist", "--quiet", "--socket-timeout", "120"]
    if proxy:
        cmd += ["--proxy", proxy]
    cmd.append(audio_url)
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    except subprocess.TimeoutExpired:
        return None
    candidates = list(tmp_dir.glob("episode.*"))
    if candidates:
        return candidates[0]
    if r.returncode == 0 and out_path.exists():
        return out_path
    return None


def transcribe(audio_path: Path) -> str | None:
    """Call aria-transcribe-route.py — duration-based router (Groq ≤60s, Transkriptor >60s).

    Podcasts sind typischerweise 20-180 min → routet immer zu Transkriptor (Pro-Plan flat-rate,
    bessere Quality + Speaker-Diarization). Falls Transkriptor failed → automatic Fallback
    auf Groq (im Router selbst implementiert).
    """
    try:
        r = subprocess.run(
            ["python3", str(WHISPER_SCRIPT), str(audio_path)],
            capture_output=True, text=True, timeout=1800,  # 30 min — Transkriptor poll-cap 15 min + buffer
        )
    except subprocess.TimeoutExpired:
        return None
    if r.returncode != 0:
        return None
    text = r.stdout.strip()
    return text if text else None


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


def ensure_whisper_column(con) -> None:
    """Fix 09.07.2026: whisper_daily_usd verglich gegen total_usd (= triage+deep+
    whisper gemischt) — ein teurer Deep-Tag (>3$) hat damit den Podcast-Ingest
    ausgehungert, obwohl 0$ Whisper verbraucht waren. Eigene Spalte trennt das."""
    cols = [r[1] for r in con.execute("PRAGMA table_info(daily_costs)")]
    if "whisper_usd" not in cols:
        con.execute("ALTER TABLE daily_costs ADD COLUMN whisper_usd REAL DEFAULT 0")
        con.commit()


def todays_whisper_cost(con) -> float:
    ensure_whisper_column(con)
    today = datetime.now(timezone.utc).date().isoformat()
    row = con.execute(
        "SELECT whisper_usd FROM daily_costs WHERE date=?", (today,)
    ).fetchone()
    return float(row[0]) if row and row[0] is not None else 0.0


def add_whisper_cost(con, delta: float) -> None:
    ensure_whisper_column(con)
    today = datetime.now(timezone.utc).date().isoformat()
    # total_usd weiter mitzaehlen — der globale total_max_usd-Cap soll Whisper sehen.
    con.execute(
        "INSERT INTO daily_costs (date, triage_usd, deep_usd, total_usd, whisper_usd) VALUES (?,0,0,?,?) "
        "ON CONFLICT(date) DO UPDATE SET total_usd=total_usd+?, whisper_usd=whisper_usd+?",
        (today, delta, delta, delta, delta),
    )
    con.commit()


def write_raw(eid, slug, entry, transcript, duration, audio_url):
    out_dir = PODCAST_RAW_DIR / slug
    out_dir.mkdir(parents=True, exist_ok=True)
    out_path = out_dir / f"{eid}.json"
    pub = episode_published_at(entry)
    payload = {
        "video_id": f"pod_{eid}",
        "channel_slug": slug,
        "channel_handle": slug,
        "title": entry.get("title", ""),
        "url": entry.get("link", "") or audio_url,
        "audio_url": audio_url,
        "duration_seconds": duration,
        "upload_date": pub.strftime("%Y%m%d") if pub else "",
        "published_at": pub.isoformat() if pub else "",
        "transcript": transcript,
        "transcript_chars": len(transcript),
        "ingested_at": datetime.now(timezone.utc).isoformat(),
        "source": "podcast",
    }
    out_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False))
    return str(out_path)


def store_ingested(con, eid, slug, title, url, raw_path, content_chars, duration, pub_iso):
    con.execute(
        "INSERT OR REPLACE INTO ingested (video_id, channel, channel_handle, title, "
        "duration_seconds, published_at, url, transcript_chars, raw_path, source) "
        "VALUES (?,?,?,?,?,?,?,?,?,'podcast')",
        (f"pod_{eid}", slug, slug, title[:300], duration, pub_iso, url,
         content_chars, raw_path),
    )
    con.commit()


def store_skipped(con, eid, slug, title, url, reason):
    con.execute(
        "INSERT OR IGNORE INTO ingested (video_id, channel, title, url, skipped_reason, source) VALUES (?,?,?,?,?,'podcast')",
        (f"pod_{eid}", slug, title[:200], url, reason),
    )
    con.commit()


def log_run_start(con):
    cur = con.execute("INSERT INTO run_log (stage, status) VALUES ('ingest-podcast', 'running')")
    con.commit()
    return cur.lastrowid


def log_run_finish(con, run_id, processed, skipped, failed, notes=""):
    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):
    cfg = load_config()
    con = db_conn(cfg["paths"]["state_db"])
    feeds = parse_feeds(PODCAST_LIST_PATH)
    if not feeds:
        print("[akp-podcast] no feeds")
        return 0

    if not os.environ.get("GROQ_API_KEY"):
        print("[akp-podcast] GROQ_API_KEY not set — transcribe will fail.", file=sys.stderr)
        return 0

    # KAR-742 Nachtrag 09.07.2026: Podcast-Enclosures liegen auf offenen CDNs
    # (Libsyn/Simplecast/Changelog) — die blocken Datacenter-IPs nicht. Der
    # Residential-Proxy ist hier unnoetig und hat den 3-GB-Monats-Cap in Tagen
    # verbrannt (30-150 MB/Episode). Proxy nur noch per explizitem Opt-in.
    proxy = os.environ.get("YT_PROXY") if os.environ.get("AKP_PODCAST_USE_PROXY") == "1" else None

    cap = float(cfg.get("cost_caps", {}).get("whisper_daily_usd", DEFAULT_WHISPER_DAILY_CAP_USD))

    max_per_feed = 5
    if "--backfill" in argv:
        max_per_feed = 10

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

    for url in feeds:
        slug = feed_slug(url)
        try:
            parsed = feedparser.parse(url)
        except Exception as e:
            notes_lines.append(f"{slug}: parse_err {str(e)[:60]}")
            failed += 1
            continue
        if parsed.bozo and not parsed.entries:
            notes_lines.append(f"{slug}: bozo_no_entries")
            failed += 1
            continue
        for entry in parsed.entries[:max_per_feed]:
            eid = episode_id(entry)
            if already_ingested(con, f"pod_{eid}"):
                continue
            audio_url, duration = find_audio_url(entry)
            if not audio_url:
                store_skipped(con, eid, slug, entry.get("title", ""), entry.get("link", ""), "no_audio_enclosure")
                skipped += 1
                continue
            reason = cheap_filter(entry, duration)
            if reason:
                store_skipped(con, eid, slug, entry.get("title", ""), entry.get("link", ""), reason)
                skipped += 1
                continue
            # Cost-check vor Download
            est_cost = (duration or 3600) / 3600 * 0.04  # estimated
            if todays_whisper_cost(con) + est_cost > cap:
                notes_lines.append(f"{slug}/{eid}: cost-cap reached")
                break

            # Download + transcribe in tempdir
            with tempfile.TemporaryDirectory(prefix="akp-pod-") as tmp:
                tmp_dir = Path(tmp)
                audio_path = download_audio(audio_url, tmp_dir, proxy)
                if not audio_path or audio_path.stat().st_size < 1000:
                    notes_lines.append(f"{slug}/{eid}: download_fail")
                    failed += 1
                    continue
                transcript = transcribe(audio_path)
            if not transcript:
                notes_lines.append(f"{slug}/{eid}: transcribe_fail")
                failed += 1
                continue

            actual_cost = (duration or 3600) / 3600 * 0.04
            add_whisper_cost(con, actual_cost)
            raw_path = write_raw(eid, slug, entry, transcript, duration, audio_url)
            pub = episode_published_at(entry)
            store_ingested(con, eid, slug, entry.get("title", ""), entry.get("link", "") or audio_url,
                           raw_path, len(transcript), duration, pub.isoformat() if pub else "")
            processed += 1
            time.sleep(0.5)

        # Inter-feed cost-cap check
        if todays_whisper_cost(con) > cap:
            notes_lines.append("cost-cap reached, stopping all feeds")
            break

    log_run_finish(con, run_id, processed, skipped, failed, "; ".join(notes_lines)[:500])
    print(f"[akp-podcast] processed={processed} skipped={skipped} failed={failed} cost=${todays_whisper_cost(con):.3f}")
    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:]))