open_notebook/scripts/add_video_source.py

#!/usr/bin/env python3
"""Fuegt ein Video (YouTube & alles, was yt-dlp kennt) als Quelle in ein Notebook ein —
auch ohne Untertitel.

Zwei Wege, automatisch gewaehlt:

  Untertitel vorhanden  -> die URL wird direkt als Link-Quelle uebergeben. Open Notebook
                           holt das Transkript selbst (schnell, kein Download).
  keine Untertitel      -> yt-dlp laedt nur die Tonspur, die wird als Audio-Quelle
                           hochgeladen. Open Notebook transkribiert sie mit dem
                           eingestellten Speech-to-Text-Modell (hier: der lokale
                           faster-whisper-Server auf :8902) und bettet sie ein.

Der zweite Weg ist noetig, weil Open Notebook fuer Videos nur die vorhandenen Untertitel
liest — gibt es keine, bleibt die Quelle leer. yt-dlp laeuft nur auf dem Host (nicht im
Container), deshalb diese Bruecke.

Beispiele:
    ./scripts/add_video_source.py https://www.youtube.com/watch?v=...
    ./scripts/add_video_source.py <url> --notebook notebook:abc --language de
    ./scripts/add_video_source.py <url> --force-audio     # Untertitel ignorieren
"""
from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

API = "http://127.0.0.1:5055"
REPO = Path(__file__).resolve().parent.parent

# Sprachen, deren Untertitel wir akzeptieren. Deckungsgleich mit
# config/content_core.yaml (youtube_transcripts.preferred_languages) — sonst wuerde
# dieses Skript den Link-Weg waehlen, obwohl Open Notebook die Sprache gar nicht sucht.
CAPTION_LANGS = ["de", "en", "es", "pt"]

# Whisper braucht nicht mehr: mono, 16 kHz. Haelt den Upload klein (~15 MB/Stunde).
AUDIO_POSTPROC = "-ac 1 -ar 16000 -b:a 32k"


def die(msg: str) -> int:
    print(f"FEHLER: {msg}", file=sys.stderr)
    return 1


def run(cmd: list[str], **kw) -> subprocess.CompletedProcess:
    return subprocess.run(cmd, check=True, capture_output=True, text=True, **kw)


def api_get(path: str) -> dict | list:
    with urllib.request.urlopen(f"{API}{path}", timeout=30) as r:
        return json.load(r)


def probe(url: str) -> dict:
    """Metadaten holen, ohne etwas herunterzuladen."""
    try:
        out = run(["yt-dlp", "--dump-single-json", "--no-warnings", "--skip-download", url])
    except subprocess.CalledProcessError as e:
        raise SystemExit(die(f"yt-dlp kann die URL nicht lesen:\n{e.stderr.strip()[:500]}"))
    return json.loads(out.stdout)


YOUTUBE_ID = re.compile(
    r"(?:youtu\.be/|youtube\.com/(?:embed/|v/|watch\?v=|watch\?.+&v=))([\w-]{11})"
)


def caption_langs(url: str) -> list[str]:
    """Untertitelsprachen, die Open Notebook tatsaechlich verwenden kann.

    Bewusst NICHT ueber yt-dlp: dessen "automatic_captions" listet auch YouTubes
    Auto-Uebersetzungen (>100 Sprachen, u.a. immer "de"), die von
    youtube-transcript-api gar nicht als Transkript gefunden werden. Wer sich darauf
    verlaesst, waehlt bei einem z.B. rein japanischen Video den Link-Weg — und Open
    Notebook legt wieder eine leere Quelle an. Also fragen wir genau die Bibliothek,
    die im Container auch die Entscheidung trifft.

    Nicht-YouTube-Quellen: leere Liste — fuer die kennt content-core keinen
    Untertitel-Weg, da fuehrt ohnehin nur die Tonspur zum Ziel.
    """
    m = YOUTUBE_ID.search(url)
    if not m:
        return []
    probe_code = (
        "from youtube_transcript_api import YouTubeTranscriptApi as A;"
        f"print(' '.join(t.language_code for t in A().list('{m.group(1)}')))"
    )
    try:
        out = run(["docker", "compose", "exec", "-T", "open_notebook",
                   "/app/.venv/bin/python", "-c", probe_code], cwd=REPO)
    except subprocess.CalledProcessError:
        return []  # kein Transkript vorhanden (oder Abruf blockiert) -> Tonspur
    have = out.stdout.split()
    return [lang for lang in CAPTION_LANGS
            if any(h == lang or h.startswith(f"{lang}-") for h in have)]


def download_audio(url: str, dest_dir: Path, stem: str) -> Path:
    """Tonspur laden. `stem` macht den Dateinamen eindeutig — Open Notebook legt den
    Upload unter seinem Originalnamen in data/uploads/ ab, ein fixer Name wie
    "audio.mp3" wuerde beim naechsten Video kollidieren."""
    print("Lade Tonspur (yt-dlp)...")
    run([
        "yt-dlp", "-x", "--audio-format", "mp3", "--audio-quality", "5",
        "--postprocessor-args", f"ExtractAudio:{AUDIO_POSTPROC}",
        "--no-warnings", "--no-playlist",
        "-o", str(dest_dir / f"{stem}.%(ext)s"), url,
    ])
    files = list(dest_dir.glob(f"{stem}.mp3"))
    if not files:
        raise SystemExit(die("yt-dlp hat keine Audiodatei erzeugt."))
    f = files[0]
    print(f"  {f.name}: {f.stat().st_size / 1e6:.1f} MB")
    return f


def post_source(fields: dict[str, str], file: Path | None = None) -> dict:
    """POST /api/sources als multipart/form-data (das erwartet die API)."""
    cmd = ["curl", "-sS", "-X", "POST", f"{API}/api/sources"]
    for k, v in fields.items():
        cmd += ["-F", f"{k}={v}"]
    if file is not None:
        cmd += ["-F", f"file=@{file}"]
    out = run(cmd).stdout
    try:
        return json.loads(out)
    except json.JSONDecodeError:
        raise SystemExit(die(f"Unerwartete Antwort der API: {out[:300]}"))


def cleanup_upload(filename: str) -> None:
    """Die hochgeladene MP3 nach dem Transkribieren wegraeumen.

    Eigentlich ist dafuer das API-Flag delete_source=true gedacht — das ist in dieser
    content-core-Version aber wirkungslos: extract_content verliert das Feld beim
    Aufbau des Graph-State (der zurueckgegebene State zeigt delete_source=None), der
    delete_file-Knoten sieht es nie und ueberspringt das Loeschen. Ohne diesen Schritt
    blieben pro Video ein paar MB in notebook_data/uploads/ liegen, die niemand mehr
    braucht — der Text steckt ja laengst in der Quelle.

    Loeschen im Container, weil die Datei dort als root angelegt wird.
    """
    path = f"/app/data/uploads/{filename}"
    try:
        run(["docker", "compose", "exec", "-T", "open_notebook", "rm", "-f", "--", path],
            cwd=REPO)
        print(f"Aufgeraeumt: uploads/{filename}")
    except subprocess.CalledProcessError as e:
        print(f"Hinweis: uploads/{filename} konnte nicht geloescht werden "
              f"({e.stderr.strip()[:100]}) — stoert nicht, belegt nur Platz.",
              file=sys.stderr)


def wait_for_content(source_id: str, timeout_s: int) -> dict | None:
    """Auf die asynchrone Verarbeitung warten (Transkription kann Minuten dauern)."""
    quoted = urllib.parse.quote(source_id, safe="")
    deadline = time.time() + timeout_s
    last = None
    while time.time() < deadline:
        try:
            last = api_get(f"/api/sources/{quoted}")
        except urllib.error.HTTPError:
            pass
        if last:
            text = last.get("full_text") or ""
            if text.strip():
                return last
        print("  ... wird verarbeitet", end="\r", flush=True)
        time.sleep(10)
    return None


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("url")
    ap.add_argument("--notebook", help="Notebook-ID (Default: das einzige vorhandene)")
    ap.add_argument("--language", help="Sprache fuers Transkribieren erzwingen, z.B. de")
    ap.add_argument("--force-audio", action="store_true",
                    help="Untertitel ignorieren und immer die Tonspur transkribieren")
    ap.add_argument("--timeout", type=int, default=1800, metavar="S",
                    help="Wartezeit auf die Verarbeitung (Default: 1800s)")
    args = ap.parse_args()

    try:
        notebooks = api_get("/api/notebooks")
    except (urllib.error.URLError, TimeoutError) as e:
        return die(f"Open-Notebook-API unter {API} nicht erreichbar ({e}).")

    notebook_id = args.notebook
    if not notebook_id:
        if len(notebooks) != 1:
            return die("Mehrere (oder keine) Notebooks vorhanden — bitte --notebook angeben:\n"
                       + "\n".join(f"  {n['id']}  {n.get('name')}" for n in notebooks))
        notebook_id = notebooks[0]["id"]
        print(f"Notebook: {notebooks[0].get('name')} ({notebook_id})")

    meta = probe(args.url)
    title = meta.get("title") or args.url
    dur = meta.get("duration")
    langs = [] if args.force_audio else caption_langs(args.url)
    print(f"Video   : {title}" + (f"  ({dur // 60}:{dur % 60:02d} min)" if dur else ""))

    use_audio = args.force_audio or not langs
    if args.force_audio:
        print("Untertitel werden ignoriert (--force-audio) — Tonspur wird transkribiert.")
    elif langs:
        print(f"Untertitel vorhanden ({', '.join(langs)}) — nutze den direkten Weg.")
    else:
        print("Keine brauchbaren Untertitel — Tonspur wird transkribiert.")

    common = {
        "notebook_id": notebook_id,
        "embed": "true",
        "async_processing": "true",
        "title": title,
    }

    uploaded_name: str | None = None
    if not use_audio:
        src = post_source({**common, "type": "link", "url": args.url})
    else:
        stem = meta.get("id") or "video"
        with tempfile.TemporaryDirectory() as td:
            audio = download_audio(args.url, Path(td), stem)
            print("Lade zu Open Notebook hoch; die Transkription laeuft dort auf dem "
                  "lokalen Whisper-Server...")
            src = post_source({**common, "type": "upload"}, file=audio)
            uploaded_name = audio.name

    source_id = src.get("id")
    if not source_id:
        return die(f"Quelle wurde nicht angelegt: {json.dumps(src)[:300]}")
    print(f"Quelle angelegt: {source_id}")

    done = wait_for_content(source_id, args.timeout)
    if not done:
        return die(f"Nach {args.timeout}s kein Inhalt. Status pruefen mit:\n"
                   f"  docker compose logs --tail 50 open_notebook")

    text = done.get("full_text") or ""
    print(f"\nFertig: {len(text)} Zeichen Inhalt.")
    print(f"Anfang: {text[:160].strip()}...")

    if uploaded_name:
        cleanup_upload(uploaded_name)
    return 0


if __name__ == "__main__":
    sys.exit(main())