#!/usr/bin/env python3 """AKP-Retry: scannt DB nach skipped_reason='no_transcript' und versucht mit aktuellem cookies-File neu zu ingesten. KAR-23 Follow-Up nach Cookie-Fix. Usage: python3 aria-akp-retry-no-transcript.py [--limit N] [--batch N] [--dry-run] """ 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-retry-no-transcript") import argparse import json import re import sqlite3 import subprocess import sys import tempfile import time import yaml from datetime import datetime, timezone from pathlib import Path CONFIG_PATH = Path("/root/aria/brain/youtube/.config.yaml") def load_config() -> dict: return yaml.safe_load(CONFIG_PATH.read_text()) def db_conn(db_path: str) -> sqlite3.Connection: con = sqlite3.connect(db_path) con.row_factory = sqlite3.Row return con 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 yt_dlp_transcript(url: str, cookies: str | None, proxy: str | None): """Return (transcript_text_or_None, meta_dict_or_None).""" tmp = tempfile.mkdtemp() base_args = ["--no-warnings"] if proxy: base_args += ["--proxy", proxy] if cookies and Path(cookies).exists(): base_args += ["--cookies", cookies] try: # Metadata meta_cmd = ["yt-dlp", *base_args, "--skip-download", "--dump-json", url] meta = None try: mr = subprocess.run(meta_cmd, capture_output=True, text=True, timeout=60) if mr.stdout.strip(): meta = json.loads(mr.stdout) except (subprocess.TimeoutExpired, json.JSONDecodeError): meta = None # Subtitle 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: 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: 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 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", "retry-no-transcript"), "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, }, } out_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2)) return str(out_path) 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", "retry-no-transcript"), ), ) con.commit() def main(): ap = argparse.ArgumentParser() ap.add_argument("--limit", type=int, default=10000) ap.add_argument("--batch", type=int, default=50) ap.add_argument("--sleep-between-batches", type=float, default=3.0) ap.add_argument("--dry-run", action="store_true") args = ap.parse_args() cfg = load_config() con = db_conn(cfg["paths"]["state_db"]) cookies = cfg["ingest"].get("yt_dlp_cookies", "") proxy = cfg["ingest"].get("yt_dlp_proxy", "") or None duration_min = cfg["ingest"].get("duration_min_seconds", 300) duration_max = cfg["ingest"].get("duration_max_seconds", 10800) videos = list(con.execute( """SELECT video_id, channel, title, duration_seconds, url FROM ingested WHERE skipped_reason='no_transcript' ORDER BY ingested_at DESC LIMIT ?""", (args.limit,), )) print(f"[akp-retry] {len(videos)} no_transcript videos pending") if not videos: return 0 if args.dry_run: print("[akp-retry] DRY-RUN — first 5 videos:") for v in videos[:5]: print(f" {v['video_id']} {v['channel']} {(v['title'] or '')[:60]}") return 0 processed_ok = 0 still_no_transcript = 0 too_short = 0 too_long = 0 failed = 0 started = time.time() for i, v in enumerate(videos, 1): if i % args.batch == 0: elapsed = time.time() - started rate = i / elapsed eta = (len(videos) - i) / rate if rate > 0 else 0 print(f"[akp-retry] {i}/{len(videos)} ok={processed_ok} still={still_no_transcript} short={too_short} long={too_long} fail={failed} | rate={rate:.2f}/s ETA={eta/60:.1f}min", flush=True) time.sleep(args.sleep_between_batches) try: transcript, meta = yt_dlp_transcript(v["url"], cookies, proxy) except Exception: failed += 1 continue if not transcript: still_no_transcript += 1 continue duration = (meta or {}).get("duration") or v["duration_seconds"] or 0 if duration and duration < duration_min: too_short += 1 con.execute("UPDATE ingested SET skipped_reason='too_short' WHERE video_id=?", (v["video_id"],)) con.commit() continue if duration and duration > duration_max: too_long += 1 con.execute("UPDATE ingested SET skipped_reason='too_long' WHERE video_id=?", (v["video_id"],)) con.commit() continue video = { "video_id": v["video_id"], "title": (meta or {}).get("title") or v["title"], "duration": duration, "url": v["url"], "uploader": (meta or {}).get("uploader") or v["channel"], "upload_date": (meta or {}).get("upload_date", ""), "_source": "retry-no-transcript", } try: raw_path = write_raw(video, v["channel"], transcript, meta, cfg) con.execute("DELETE FROM ingested WHERE video_id=? AND skipped_reason='no_transcript'", (v["video_id"],)) store_ingested(con, video, v["channel"], raw_path, len(transcript)) processed_ok += 1 except Exception as e: failed += 1 print(f" ERR storing {v['video_id']}: {e}", file=sys.stderr) print(f"[akp-retry] DONE — ok={processed_ok} still={still_no_transcript} short={too_short} long={too_long} fail={failed} | total={len(videos)}") return 0 if __name__ == "__main__": _log.event("script_start") sys.exit(main() or 0)