Add YouTube ingestion: captions-first, faster-whisper GPU fallback

Videos become vault documents (transcript .txt + .json sidecar,
source="youtube") so they flow through embed/summarize/search/ask
alongside the papers. yt-dlp fetches captions when present; faster-whisper
transcribes audio on the GPU when they're missing.

- clawlibrary/youtube.py: fetch metadata, captions via yt-dlp (json3),
  whisper transcription with CUDA/CPU auto-detect, ingest_video + queue loader.
  Network/GPU work is injected for unit-testability.
- extract.iter_documents: discover any .txt+.json doc (papers AND transcripts);
  embed/summarize now iterate documents, not just PDFs.
- CLI `youtube` command (--queue/--url/--mode/--whisper-*), youtube.yaml queue.
- [youtube] optional extra (yt-dlp, faster-whisper). 10 new tests (72 total).

Co-Authored-By: Claude Opus 4.8 <[email protected]>

README.md

@@ -94,6 +94,36 @@ of the vault — no re-downloads.
 provides a [requester-pays S3 bucket](https://info.arxiv.org/help/bulk_data_s3.html)
 (`s3://arxiv/`) and a Kaggle dataset for that — no throttling, by design.
 
+## YouTube ingestion
+
+`clawlibrary youtube` pulls video knowledge into the **same vault** as the
+papers. Each video becomes a document — a `{video_id}.txt` transcript plus a
+`{video_id}.json` sidecar (`source="youtube"`) — so it flows straight through
+`embed` / `summarize` / `search` / `ask` next to the literature.
+
+Transcription is **captions-first**: existing YouTube captions (manual or
+auto) are fetched via yt-dlp (fast, free, no GPU). When a video has none,
+**faster-whisper** transcribes the audio on the GPU.
+
+```bash
+pip install -e ".[youtube]"          # yt-dlp + faster-whisper (CUDA-heavy)
+
+clawlibrary youtube --dry-run         # list what the queue would ingest
+clawlibrary youtube                   # process youtube.yaml
+clawlibrary youtube --url "https://www.youtube.com/watch?v=..." \
+                    --topic "quantum error correction" --max 5
+clawlibrary youtube --mode whisper --whisper-model large-v3   # force GPU transcription
+
+# then fold the new transcripts into semantic search / RAG:
+clawlibrary embed
+clawlibrary ask "What did the talks say about agentic planning?"
+```
+
+`youtube.yaml` is the queue (channels, playlists, or single videos), mirroring
+`topics.yaml`. Modes: `auto` (captions → whisper fallback, default), `captions`
+(only), `whisper` (force). On a GPU node, `--whisper-device auto` picks CUDA;
+it falls back to CPU/int8 when no GPU is present.
+
 ## Subscription content (TDM)
 
 For papers with no open-access copy, the legitimate path is your institution's

clawlibrary/__main__.py

@@ -4,6 +4,12 @@ Commands:
   run [--topic Q] [--dry-run] [--max N]    harvest the topic queue
   stats                                    show vault statistics
   export [--format csv|json] [--out P]     export the vault index
+  extract [--limit N] [--vlm-fallback]     extract full text from vault PDFs
+  summarize [--limit N] [--model M]        LLM structured summaries (gemma4)
+  embed [--limit N]                        embed documents for semantic search
+  search QUERY [--k N]                     semantic search across the vault
+  ask QUESTION [--k N]                     RAG answer from retrieved passages
+  youtube [--queue F] [--url U --topic T]  ingest YouTube videos as transcripts
 """
 
 from __future__ import annotations
@@ -181,7 +187,7 @@ def cmd_extract(args: argparse.Namespace) -> int:
 def cmd_summarize(args: argparse.Namespace) -> int:
     import json as _json
 
-    from .extract import iter_pdfs
+    from .extract import iter_documents
     from .ollama_client import OllamaClient
     from .summarize import append_insight, has_summary, summarize_one
 
@@ -191,22 +197,22 @@ def cmd_summarize(args: argparse.Namespace) -> int:
         console.print("[red]Ollama not reachable.[/red]")
         return 2
     insights = Path(config.vault.root) / "insights.md"
-    pdfs = [p for p in iter_pdfs(Path(config.vault.root)) if p.with_suffix(".txt").exists()]
+    docs = list(iter_documents(Path(config.vault.root)))
     if args.limit:
-        pdfs = pdfs[: args.limit]
+        docs = docs[: args.limit]
 
     done = skipped = failed = 0
-    total = len(pdfs)
+    total = len(docs)
-    for i, pdf in enumerate(pdfs, 1):
+    for i, doc in enumerate(docs, 1):
-        side = pdf.with_suffix(".json")
+        side = doc.with_suffix(".json")
         if not args.force and has_summary(side):
             skipped += 1
             continue
         try:
-            summary = summarize_one(pdf, oc)
+            summary = summarize_one(doc, oc)
         except Exception as e:  # noqa: BLE001 — one paper must not abort the batch
             failed += 1
-            console.print(f"[red]✗[/red] {pdf.name}: {e}")
+            console.print(f"[red]✗[/red] {doc.name}: {e}")
             continue
         if summary is not None:
             append_insight(insights, _json.loads(side.read_text()))
@@ -222,7 +228,7 @@ def cmd_summarize(args: argparse.Namespace) -> int:
 
 def cmd_embed(args: argparse.Namespace) -> int:
     from .embedstore import EmbedStore, add_document, merge
-    from .extract import iter_pdfs
+    from .extract import iter_documents
     from .ollama_client import OllamaClient
 
     config, _ = _load(args.config, args.topics)
@@ -234,26 +240,26 @@ def cmd_embed(args: argparse.Namespace) -> int:
 
     store = EmbedStore.load(root)
     seen = store.dois
-    pdfs = [p for p in iter_pdfs(root) if p.with_suffix(".txt").exists()]
+    docs = list(iter_documents(root))
     if args.limit:
-        pdfs = pdfs[: args.limit]
+        docs = docs[: args.limit]
 
     added = skipped = 0
-    total = len(pdfs)
+    total = len(docs)
-    for i, pdf in enumerate(pdfs, 1):
+    for i, doc in enumerate(docs, 1):
-        data = json.loads(pdf.with_suffix(".json").read_text())
+        data = json.loads(doc.with_suffix(".json").read_text())
-        doi = data.get("doi") or data.get("id") or pdf.stem
+        doi = data.get("doi") or data.get("id") or doc.stem
         if doi in seen:
             skipped += 1
             continue
-        text = pdf.with_suffix(".txt").read_text()
+        text = doc.read_text()
         try:
             vecs, rows = add_document(
                 oc, text=text, doi=doi, title=data.get("title", ""),
-                topic=data.get("topic", ""), text_path=str(pdf.with_suffix(".txt")),
+                topic=data.get("topic", ""), text_path=str(doc),
             )
         except Exception as e:  # noqa: BLE001 — checkpoint already saved; skip & continue
-            console.print(f"[red]✗[/red] {pdf.name}: {e}")
+            console.print(f"[red]✗[/red] {doc.name}: {e}")
             continue
         store = merge(store, vecs, rows)
         seen.add(doi)
@@ -322,6 +328,100 @@ def cmd_ask(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_youtube(args: argparse.Namespace) -> int:
+    from . import youtube as yt
+
+    config, _ = _load(args.config, args.topics)
+    root = Path(config.vault.root)
+    run_id = f"yt-{args.mode}"
+
+    # Build the work list: either an ad-hoc --url, or the youtube.yaml queue.
+    if args.url:
+        if not args.topic:
+            console.print("[red]--url requires --topic[/red]")
+            return 1
+        work = [(args.url, args.topic, args.max, args.mode)]
+    else:
+        try:
+            _defaults, sources = yt.load_queue(args.queue)
+        except (FileNotFoundError, ValueError) as e:
+            console.print(f"[red]{e}[/red]")
+            return 2
+        work = [
+            (s.url, s.topic, args.max or s.max_videos, args.mode or s.mode or "auto")
+            for s in sources
+        ]
+
+    transcribers: dict[str, object] = {}
+
+    def transcriber_for(mode: str):
+        if mode not in transcribers:
+            transcribers[mode] = yt.make_transcriber(
+                mode=mode,
+                whisper_model=args.whisper_model,
+                whisper_device=args.whisper_device,
+                whisper_compute=args.whisper_compute,
+            )
+        return transcribers[mode]
+
+    ingested = skipped = empty = failed = 0
+    seen_total = 0
+    for src_url, topic, cap, mode in work:
+        console.print(f"[bold cyan]→ {topic}[/bold cyan]  [dim]{src_url}[/dim]")
+        try:
+            videos = yt.list_source_videos(src_url, max_videos=cap)
+        except Exception as e:  # noqa: BLE001
+            console.print(f"  [red]✗ could not list videos:[/red] {e}")
+            failed += 1
+            continue
+        for v in videos:
+            if args.limit and seen_total >= args.limit:
+                break
+            seen_total += 1
+            try:
+                res = yt.ingest_video(
+                    v["url"], topic, root, run_id=run_id,
+                    transcribe_fn=transcriber_for(mode),
+                    dry_run=args.dry_run, force=args.force,
+                )
+            except Exception as e:  # noqa: BLE001 — one video must not abort the batch
+                failed += 1
+                console.print(f"  [red]✗[/red] {v.get('title', v['id'])[:70]}  [dim]{e}[/dim]")
+                continue
+            status = res["status"]
+            title = (res.get("title") or v.get("title") or v["id"])[:70]
+            if status == "ingested":
+                ingested += 1
+                console.print(
+                    f"  [green]✓[/green] {title}  "
+                    f"[dim]({res['method']}, {res['words']} words)[/dim]"
+                )
+            elif status == "planned":
+                console.print(f"  [yellow]·[/yellow] {title}  [dim](would ingest)[/dim]")
+            elif status == "dedup":
+                skipped += 1
+            elif status == "no_transcript":
+                empty += 1
+                console.print(f"  [yellow]∅[/yellow] {title}  [dim]no transcript[/dim]")
+            else:  # fetch_failed
+                failed += 1
+                console.print(f"  [red]✗[/red] {title}  [dim]{status}[/dim]")
+        if args.limit and seen_total >= args.limit:
+            break
+
+    table = Table(title="YouTube ingest", header_style="bold")
+    table.add_column("Metric")
+    table.add_column("Count", justify="right")
+    table.add_row("Ingested", str(ingested))
+    table.add_row("Skipped (dedup)", str(skipped))
+    table.add_row("No transcript", str(empty))
+    table.add_row("Failed", str(failed))
+    console.print(table)
+    if ingested:
+        console.print("[dim]Next: `clawlibrary embed` then `clawlibrary ask ...`[/dim]")
+    return 0
+
+
 def build_parser() -> argparse.ArgumentParser:
     p = argparse.ArgumentParser(prog="clawlibrary", description=__doc__)
     p.add_argument("--config", default="config.yaml", help="path to config.yaml")
@@ -372,6 +472,22 @@ def build_parser() -> argparse.ArgumentParser:
     pa.add_argument("--k", type=int, default=6, help="passages to retrieve")
     pa.add_argument("--model", default="gemma4:E4B", help="Ollama model for the answer")
     pa.set_defaults(func=cmd_ask)
+
+    py = sub.add_parser("youtube", help="ingest YouTube videos as transcripts into the vault")
+    py.add_argument("--queue", default="youtube.yaml", help="path to the youtube.yaml queue")
+    py.add_argument("--url", help="ad-hoc channel/playlist/video URL (requires --topic)")
+    py.add_argument("--topic", help="topic label for --url (vault folder + relevance)")
+    py.add_argument("--mode", choices=["auto", "captions", "whisper"], default="auto",
+                    help="auto=captions then whisper fallback; captions=only; whisper=force")
+    py.add_argument("--max", type=int, default=None, help="cap videos per source")
+    py.add_argument("--limit", type=int, default=None, help="cap total videos this run")
+    py.add_argument("--force", action="store_true", help="re-ingest even if already present")
+    py.add_argument("--dry-run", action="store_true", help="list what would be ingested")
+    py.add_argument("--whisper-model", default="large-v3", help="faster-whisper model size")
+    py.add_argument("--whisper-device", default="auto", choices=["auto", "cuda", "cpu"])
+    py.add_argument("--whisper-compute", default=None,
+                    help="ctranslate2 compute type (e.g. float16, int8); default by device")
+    py.set_defaults(func=cmd_youtube)
     return p
 
 

clawlibrary/extract.py

@@ -111,3 +111,17 @@ def extract_one(pdf_path: Path, *, min_chars: int = MIN_CHARS_PER_PAGE, vlm=None
 
 def iter_pdfs(vault_root: Path):
     yield from sorted(vault_root.glob("*/*.pdf"))
+
+
+def iter_documents(vault_root: Path):
+    """Yield every text-ready document's .txt path (extracted papers AND
+    transcripts), i.e. any ``*/*.txt`` with a sibling ``.json`` sidecar.
+
+    This is the discovery surface for embed/summarize: PDFs gain a .txt after
+    `extract`, YouTube transcripts have one from the start — both look the same
+    here, so video knowledge is embedded and summarized alongside the papers.
+    Callers use ``txt.with_suffix(".json")`` for the sidecar.
+    """
+    for txt in sorted(vault_root.glob("*/*.txt")):
+        if txt.with_suffix(".json").exists():
+            yield txt

clawlibrary/youtube.py

@@ -0,0 +1,423 @@
+"""Ingest YouTube videos into the vault as transcripts.
+
+Captions-first, GPU-fallback. Two transcript paths:
+  1. Existing captions (manual or auto-generated) fetched via yt-dlp — fast,
+     free, no GPU.
+  2. faster-whisper transcription of the audio — the GPU fallback used when a
+     video has no usable captions (or when ``mode="whisper"`` forces it).
+
+A video becomes a vault "document" exactly like an extracted paper: a
+``{video_id}.txt`` transcript plus a ``{video_id}.json`` sidecar
+(``source="youtube"``), so it flows straight into ``embed`` / ``summarize`` /
+``search`` / ``ask`` alongside the papers — no special-casing downstream.
+
+Network/GPU work is injected (``info_fetcher`` / ``transcribe_fn``) so the
+ingestion logic is unit-testable without yt-dlp or a GPU.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import tempfile
+from datetime import UTC, datetime
+from pathlib import Path
+
+import yaml
+
+from .models import VaultEntry, url_hash
+from .vault import VaultManager, slugify, topic_slug
+
+WATCH_URL = "https://www.youtube.com/watch?v={id}"
+DEFAULT_CAPTION_LANGS = ["en", "en-US", "en-GB", "en-orig", "en.*"]
+_VIDEO_ID_RE = re.compile(r"(?:v=|/shorts/|/embed/|youtu\.be/)([A-Za-z0-9_-]{11})")
+
+
+# --------------------------------------------------------------------------- #
+# Lazy optional deps (yt-dlp, faster-whisper, ctranslate2)
+# --------------------------------------------------------------------------- #
+def _require_ytdlp():
+    try:
+        from yt_dlp import YoutubeDL  # noqa: PLC0415
+    except ImportError as e:  # pragma: no cover - import-guard
+        raise RuntimeError(
+            "yt-dlp is required for YouTube ingestion. Install with:\n"
+            '    pip install -e ".[youtube]"'
+        ) from e
+    return YoutubeDL
+
+
+def _require_whisper():
+    try:
+        from faster_whisper import WhisperModel  # noqa: PLC0415
+    except ImportError as e:  # pragma: no cover - import-guard
+        raise RuntimeError(
+            "faster-whisper is required for audio transcription. Install with:\n"
+            '    pip install -e ".[youtube]"'
+        ) from e
+    return WhisperModel
+
+
+# --------------------------------------------------------------------------- #
+# Helpers
+# --------------------------------------------------------------------------- #
+def video_id_of(url_or_id: str) -> str:
+    """Extract the 11-char video id from a URL, or return it if already an id."""
+    s = url_or_id.strip()
+    m = _VIDEO_ID_RE.search(s)
+    if m:
+        return m.group(1)
+    if re.fullmatch(r"[A-Za-z0-9_-]{11}", s):
+        return s
+    # Fall back to the last path segment (best effort).
+    return s.rstrip("/").rsplit("/", 1)[-1]
+
+
+def parse_json3(raw: str) -> str:
+    """Parse a YouTube json3 caption file into plain text.
+
+    json3 is ``{"events": [{"segs": [{"utf8": "word"}, ...]}, ...]}``; we
+    concatenate the segment text and collapse runs of blank lines.
+    """
+    data = json.loads(raw)
+    parts: list[str] = []
+    for ev in data.get("events", []) or []:
+        for seg in ev.get("segs", []) or []:
+            t = seg.get("utf8")
+            if t:
+                parts.append(t)
+    text = "".join(parts)
+    return re.sub(r"\n{3,}", "\n\n", text).strip()
+
+
+# --------------------------------------------------------------------------- #
+# Real network/GPU implementations (skipped by tests via injection)
+# --------------------------------------------------------------------------- #
+def _ydl_base_opts() -> dict:
+    return {"quiet": True, "no_warnings": True, "noprogress": True, "ignoreerrors": True}
+
+
+def fetch_video_info(url: str) -> dict | None:
+    """Return yt-dlp's metadata dict for a single video (no download)."""
+    YoutubeDL = _require_ytdlp()
+    opts = {**_ydl_base_opts(), "skip_download": True}
+    with YoutubeDL(opts) as ydl:
+        return ydl.extract_info(url, download=False)
+
+
+def list_source_videos(url: str, *, max_videos: int | None = None) -> list[dict]:
+    """Expand a channel/playlist/video URL into a flat list of {id, title, url}.
+
+    Uses extract_flat so we don't fetch full metadata for every entry up front;
+    full metadata is fetched per-video at ingest time.
+    """
+    YoutubeDL = _require_ytdlp()
+    opts = {**_ydl_base_opts(), "extract_flat": "in_playlist", "skip_download": True}
+    if max_videos:
+        opts["playlistend"] = max_videos
+    with YoutubeDL(opts) as ydl:
+        info = ydl.extract_info(url, download=False)
+    if not info:
+        return []
+    entries = info.get("entries")
+    if entries is None:  # a single video, not a playlist/channel
+        vid = info.get("id") or video_id_of(url)
+        return [{"id": vid, "title": info.get("title", ""), "url": WATCH_URL.format(id=vid)}]
+    out: list[dict] = []
+    for e in entries:
+        if not e:
+            continue
+        vid = e.get("id")
+        if not vid:
+            continue
+        out.append(
+            {
+                "id": vid,
+                "title": e.get("title", ""),
+                "url": e.get("url") or WATCH_URL.format(id=vid),
+            }
+        )
+        if max_videos and len(out) >= max_videos:
+            break
+    return out
+
+
+def _captions_via_ytdlp(url: str, langs: list[str]) -> tuple[str | None, str]:
+    """Download the best available caption track via yt-dlp; return (text, method)."""
+    YoutubeDL = _require_ytdlp()
+    with tempfile.TemporaryDirectory() as tmp:
+        opts = {
+            **_ydl_base_opts(),
+            "skip_download": True,
+            "writesubtitles": True,
+            "writeautomaticsub": True,
+            "subtitleslangs": langs,
+            "subtitlesformat": "json3",
+            "outtmpl": str(Path(tmp) / "%(id)s.%(ext)s"),
+        }
+        with YoutubeDL(opts) as ydl:
+            info = ydl.extract_info(url, download=True) or {}
+        files = sorted(Path(tmp).glob("*.json3"))
+        if not files:
+            return None, "captions:none"
+        text = parse_json3(files[0].read_text())
+        if not text:
+            return None, "captions:none"
+        # Manual subs are higher quality than auto-generated; label which we got.
+        manual = set(info.get("subtitles") or {})
+        lang = files[0].name.split(".")[-2] if "." in files[0].name else ""
+        method = "captions:manual" if lang in manual else "captions:auto"
+        return text, method
+
+
+def _detect_whisper_device(prefer: str = "auto") -> tuple[str, str]:
+    """Return (device, compute_type): CUDA fp16 when available, else CPU int8."""
+    if prefer == "cpu":
+        return "cpu", "int8"
+    if prefer == "cuda":
+        return "cuda", "float16"
+    try:
+        import ctranslate2  # noqa: PLC0415
+
+        if ctranslate2.get_cuda_device_count() > 0:
+            return "cuda", "float16"
+    except Exception:  # noqa: BLE001 - any probe failure → CPU
+        pass
+    return "cpu", "int8"
+
+
+_WHISPER_CACHE: dict = {}
+
+
+def whisper_transcribe(
+    audio_path: str | Path,
+    *,
+    model_size: str = "large-v3",
+    device: str = "auto",
+    compute_type: str | None = None,
+    beam_size: int = 5,
+) -> str:
+    """Transcribe an audio file with faster-whisper; returns the joined text."""
+    WhisperModel = _require_whisper()
+    dev, ct = _detect_whisper_device(device)
+    if compute_type:
+        ct = compute_type
+    key = (model_size, dev, ct)
+    if key not in _WHISPER_CACHE:
+        _WHISPER_CACHE[key] = WhisperModel(model_size, device=dev, compute_type=ct)
+    model = _WHISPER_CACHE[key]
+    segments, _info = model.transcribe(str(audio_path), beam_size=beam_size, vad_filter=True)
+    return " ".join(seg.text.strip() for seg in segments).strip()
+
+
+def _download_audio(url: str, dest_dir: str) -> Path:
+    """Download bestaudio for a video into dest_dir; return the file path."""
+    YoutubeDL = _require_ytdlp()
+    opts = {
+        **_ydl_base_opts(),
+        "format": "bestaudio/best",
+        "outtmpl": str(Path(dest_dir) / "%(id)s.%(ext)s"),
+    }
+    with YoutubeDL(opts) as ydl:
+        info = ydl.extract_info(url, download=True)
+        path = Path(ydl.prepare_filename(info))
+    if not path.exists():
+        files = [p for p in Path(dest_dir).iterdir() if p.is_file()]
+        if files:
+            path = files[0]
+    return path
+
+
+def make_transcriber(
+    *,
+    mode: str = "auto",
+    langs: list[str] | None = None,
+    whisper_model: str = "large-v3",
+    whisper_device: str = "auto",
+    whisper_compute: str | None = None,
+):
+    """Build a ``transcribe_fn(info) -> (text, method)`` from runtime settings.
+
+    mode:
+      - "auto"     : captions first, whisper fallback (default)
+      - "captions" : captions only (skip if none)
+      - "whisper"  : always transcribe audio with faster-whisper
+    """
+    langs = langs or DEFAULT_CAPTION_LANGS
+
+    def transcribe(info: dict) -> tuple[str | None, str]:
+        url = info.get("webpage_url") or info.get("original_url") or WATCH_URL.format(
+            id=info.get("id", "")
+        )
+        if mode in ("auto", "captions"):
+            text, method = _captions_via_ytdlp(url, langs)
+            if text:
+                return text, method
+            if mode == "captions":
+                return None, "captions:none"
+        # whisper path (forced, or auto-fallback when captions missing)
+        with tempfile.TemporaryDirectory() as tmp:
+            audio = _download_audio(url, tmp)
+            if not audio.exists():
+                return None, "whisper:no_audio"
+            text = whisper_transcribe(
+                audio,
+                model_size=whisper_model,
+                device=whisper_device,
+                compute_type=whisper_compute,
+            )
+        return (text or None), f"whisper:{whisper_model}"
+
+    return transcribe
+
+
+# --------------------------------------------------------------------------- #
+# Ingestion
+# --------------------------------------------------------------------------- #
+def _sidecar_path(vault_root: Path, topic: str, vid: str) -> Path:
+    return Path(vault_root) / topic_slug(topic) / f"{vid}.json"
+
+
+def already_ingested(vault_root: Path, topic: str, vid: str) -> bool:
+    side = _sidecar_path(vault_root, topic, vid)
+    txt = side.with_suffix(".txt")
+    return side.exists() and txt.exists() and txt.stat().st_size > 0
+
+
+def ingest_video(
+    url_or_id: str,
+    topic: str,
+    vault_root: str | Path,
+    *,
+    run_id: str = "youtube",
+    transcribe_fn,
+    info_fetcher=fetch_video_info,
+    dry_run: bool = False,
+    force: bool = False,
+    now: datetime | None = None,
+) -> dict:
+    """Ingest one video into the vault. Returns a status dict.
+
+    status ∈ {"dedup", "planned", "no_transcript", "fetch_failed", "ingested"}.
+    """
+    vault_root = Path(vault_root)
+    vid = video_id_of(url_or_id)
+    url = WATCH_URL.format(id=vid)
+
+    if not force and already_ingested(vault_root, topic, vid):
+        return {"status": "dedup", "video_id": vid}
+
+    info = info_fetcher(url)
+    if not info:
+        return {"status": "fetch_failed", "video_id": vid}
+
+    title = (info.get("title") or "").strip() or "(untitled video)"
+    channel = info.get("uploader") or info.get("channel") or ""
+    description = (info.get("description") or "").strip()
+    duration = info.get("duration")
+    upload_date = info.get("upload_date") or ""  # "YYYYMMDD"
+    year = int(upload_date[:4]) if upload_date[:4].isdigit() else None
+
+    if dry_run:
+        return {"status": "planned", "video_id": vid, "title": title, "channel": channel}
+
+    text, method = transcribe_fn(info)
+    if not text or not text.strip():
+        return {"status": "no_transcript", "video_id": vid, "title": title, "method": method}
+
+    side = _sidecar_path(vault_root, topic, vid)
+    txt_path = side.with_suffix(".txt")
+    txt_path.parent.mkdir(parents=True, exist_ok=True)
+    txt_path.write_text(text)
+
+    stamp = (now or datetime.now(UTC)).isoformat()
+    entry = VaultEntry(
+        id=f"youtube:{vid}",
+        doi=f"youtube:{vid}",
+        doi_slug=vid,
+        title=title,
+        authors=[channel] if channel else [],
+        journal=channel or "YouTube",
+        year=year,
+        abstract=(description[:2000] or None),
+        topic=topic,
+        record_url=url,
+        resolved_pdf_url=url,
+        url_hash=url_hash(url),
+        source="youtube",
+        vault_path=str(txt_path),
+        file_size_bytes=len(text.encode("utf-8")),
+        downloaded_at=stamp,
+        run_id=run_id,
+    )
+    data = entry.model_dump()
+    data.update(
+        {
+            "text_path": str(txt_path),
+            "word_count": len(text.split()),
+            "char_count": len(text),
+            "transcript_method": method,
+            "channel": channel,
+            "duration_seconds": duration,
+            "upload_date": upload_date,
+            "video_id": vid,
+            "video_url": url,
+        }
+    )
+    side.write_text(json.dumps(data, indent=2))
+
+    # Register in the vault index so `stats`/`export` see it.
+    VaultManager(vault_root).append_to_index(entry)
+    return {
+        "status": "ingested",
+        "video_id": vid,
+        "title": title,
+        "method": method,
+        "words": data["word_count"],
+    }
+
+
+# --------------------------------------------------------------------------- #
+# Queue config (youtube.yaml)
+# --------------------------------------------------------------------------- #
+class YouTubeSource:
+    """One entry in youtube.yaml: a channel, playlist, or single video."""
+
+    def __init__(self, url: str, topic: str, *, name: str = "", max_videos: int | None = None,
+                 mode: str | None = None):
+        self.url = url
+        self.topic = topic or slugify(name or url)
+        self.name = name
+        self.max_videos = max_videos
+        self.mode = mode
+
+
+def load_queue(path: str | Path) -> tuple[dict, list[YouTubeSource]]:
+    """Load youtube.yaml → (defaults, sources). Mirrors topics.yaml's shape."""
+    p = Path(path)
+    if not p.exists():
+        raise FileNotFoundError(f"youtube queue not found: {p}")
+    data = yaml.safe_load(p.read_text()) or {}
+    if not isinstance(data, dict):
+        raise ValueError(f"{p} must be a YAML mapping")
+    defaults = data.get("defaults", {}) or {}
+    raw = data.get("sources", []) or []
+    if not raw:
+        raise ValueError("youtube.yaml must define at least one entry under 'sources'")
+    sources: list[YouTubeSource] = []
+    for item in raw:
+        if not isinstance(item, dict):
+            raise ValueError(f"each source must be a mapping, got: {item!r}")
+        url = item.get("url")
+        if not url:
+            raise ValueError(f"source missing 'url': {item!r}")
+        sources.append(
+            YouTubeSource(
+                url=url,
+                topic=item.get("topic", ""),
+                name=item.get("name", ""),
+                max_videos=item.get("max_videos", defaults.get("max_videos")),
+                mode=item.get("mode", defaults.get("mode")),
+            )
+        )
+    return defaults, sources

pyproject.toml

@@ -28,6 +28,12 @@ dev = [
     "pytest-cov>=5.0",
     "ruff>=0.5",
 ]
+# YouTube ingestion: yt-dlp for metadata+captions, faster-whisper for the
+# GPU audio-transcription fallback. Heavy (CUDA libs) — installed on demand.
+youtube = [
+    "yt-dlp>=2024.1",
+    "faster-whisper>=1.0",
+]
 
 [project.scripts]
 clawlibrary = "clawlibrary.__main__:main"

tests/test_youtube.py

@@ -0,0 +1,152 @@
+import json
+from datetime import UTC, datetime
+
+import pytest
+
+from clawlibrary import youtube as yt
+from clawlibrary.extract import iter_documents
+
+FIXED_NOW = datetime(2026, 6, 4, tzinfo=UTC)
+
+
+def _info(vid="dQw4w9WgXcQ", title="Agentic Planning Explained"):
+    return {
+        "id": vid,
+        "title": title,
+        "uploader": "AI Channel",
+        "description": "A talk about LLM agents and planning.",
+        "duration": 600,
+        "upload_date": "20250115",
+        "webpage_url": yt.WATCH_URL.format(id=vid),
+    }
+
+
+def _fetcher(info):
+    return lambda url: info
+
+
+def _transcriber(text, method="captions:auto"):
+    return lambda info: (text, method)
+
+
+# ---- pure helpers ----------------------------------------------------------
+def test_video_id_of_handles_url_forms():
+    vid = "dQw4w9WgXcQ"
+    assert yt.video_id_of(f"https://www.youtube.com/watch?v={vid}") == vid
+    assert yt.video_id_of(f"https://youtu.be/{vid}") == vid
+    assert yt.video_id_of(f"https://www.youtube.com/shorts/{vid}") == vid
+    assert yt.video_id_of(vid) == vid
+
+
+def test_parse_json3():
+    raw = json.dumps({"events": [
+        {"segs": [{"utf8": "hello "}, {"utf8": "world"}]},
+        {"segs": [{"utf8": "\n"}]},
+        {"segs": [{"utf8": "again"}]},
+    ]})
+    assert yt.parse_json3(raw) == "hello world\nagain"
+
+
+# ---- ingest_video ----------------------------------------------------------
+def test_ingest_video_writes_txt_and_sidecar(tmp_path):
+    res = yt.ingest_video(
+        "https://youtu.be/dQw4w9WgXcQ", "ai agents", tmp_path,
+        transcribe_fn=_transcriber("LLM agents plan and use tools. " * 20),
+        info_fetcher=_fetcher(_info()),
+        now=FIXED_NOW,
+    )
+    assert res["status"] == "ingested"
+    side = tmp_path / "ai-agents" / "dQw4w9WgXcQ.json"
+    txt = side.with_suffix(".txt")
+    assert txt.exists() and txt.read_text().startswith("LLM agents")
+    data = json.loads(side.read_text())
+    assert data["source"] == "youtube"
+    assert data["doi"] == "youtube:dQw4w9WgXcQ"
+    assert data["transcript_method"] == "captions:auto"
+    assert data["channel"] == "AI Channel"
+    assert data["year"] == 2025
+    assert data["word_count"] > 50
+    # registered in the vault index
+    idx = json.loads((tmp_path / "index.json").read_text())
+    assert any(e["id"] == "youtube:dQw4w9WgXcQ" for e in idx)
+
+
+def test_ingest_video_dedup(tmp_path):
+    args = dict(transcribe_fn=_transcriber("some transcript text here"),
+                info_fetcher=_fetcher(_info()), now=FIXED_NOW)
+    first = yt.ingest_video("dQw4w9WgXcQ", "t", tmp_path, **args)
+    second = yt.ingest_video("dQw4w9WgXcQ", "t", tmp_path, **args)
+    assert first["status"] == "ingested"
+    assert second["status"] == "dedup"
+    # force re-ingests
+    forced = yt.ingest_video("dQw4w9WgXcQ", "t", tmp_path, force=True, **args)
+    assert forced["status"] == "ingested"
+
+
+def test_ingest_video_dry_run_writes_nothing(tmp_path):
+    res = yt.ingest_video(
+        "dQw4w9WgXcQ", "t", tmp_path, dry_run=True,
+        transcribe_fn=_transcriber("should not be used"),
+        info_fetcher=_fetcher(_info()),
+    )
+    assert res["status"] == "planned"
+    assert not list(tmp_path.glob("**/*.txt"))
+
+
+def test_ingest_video_no_transcript(tmp_path):
+    res = yt.ingest_video(
+        "dQw4w9WgXcQ", "t", tmp_path,
+        transcribe_fn=_transcriber("", method="captions:none"),
+        info_fetcher=_fetcher(_info()),
+    )
+    assert res["status"] == "no_transcript"
+    assert not list(tmp_path.glob("**/*.txt"))
+
+
+def test_ingest_video_fetch_failed(tmp_path):
+    res = yt.ingest_video(
+        "dQw4w9WgXcQ", "t", tmp_path,
+        transcribe_fn=_transcriber("x"),
+        info_fetcher=lambda url: None,
+    )
+    assert res["status"] == "fetch_failed"
+
+
+def test_video_doc_discovered_by_iter_documents(tmp_path):
+    yt.ingest_video(
+        "dQw4w9WgXcQ", "quantum", tmp_path,
+        transcribe_fn=_transcriber("surface code error correction threshold"),
+        info_fetcher=_fetcher(_info()),
+        now=FIXED_NOW,
+    )
+    docs = list(iter_documents(tmp_path))
+    assert len(docs) == 1
+    assert docs[0].name == "dQw4w9WgXcQ.txt"
+    # the embed/summarize contract: sibling .json with a usable dedup key
+    data = json.loads(docs[0].with_suffix(".json").read_text())
+    assert (data.get("doi") or data.get("id")) == "youtube:dQw4w9WgXcQ"
+
+
+# ---- queue config ----------------------------------------------------------
+def test_load_queue(tmp_path):
+    q = tmp_path / "youtube.yaml"
+    q.write_text(
+        "defaults:\n  max_videos: 10\n  mode: auto\n"
+        "sources:\n"
+        "  - name: Chan\n    url: https://youtube.com/@chan/videos\n    topic: ai\n"
+        "  - url: https://youtu.be/dQw4w9WgXcQ\n    topic: quantum\n    max_videos: 3\n"
+    )
+    defaults, sources = yt.load_queue(q)
+    assert defaults["max_videos"] == 10
+    assert len(sources) == 2
+    assert sources[0].topic == "ai"
+    assert sources[1].max_videos == 3  # per-source override
+
+
+def test_load_queue_errors(tmp_path):
+    with pytest.raises(FileNotFoundError):
+        yt.load_queue(tmp_path / "missing.yaml")
+    bad = tmp_path / "bad.yaml"
+    bad.write_text("sources: []\n")
+    with pytest.raises(ValueError):
+        yt.load_queue(bad)

youtube.yaml

@@ -0,0 +1,32 @@
+# YouTube ingest queue — videos to pull into the vault as transcripts.
+#
+# Each video becomes a vault document (transcript .txt + .json sidecar,
+# source="youtube"), so it flows into `embed` / `summarize` / `search` / `ask`
+# alongside the papers. Captions are used when available (fast, no GPU);
+# faster-whisper transcribes audio on the GPU when a video has none.
+#
+# Run:  clawlibrary youtube                 # whole queue
+#       clawlibrary youtube --dry-run        # list what would be ingested
+#       clawlibrary youtube --url <URL> --topic "my topic" --max 5
+#
+# `url` may be a single video, a playlist, or a channel (e.g. .../@handle/videos).
+
+defaults:
+  max_videos: 20        # cap per source
+  mode: auto            # auto = captions first, whisper fallback
+
+sources:
+  - name: "Yannic Kilcher"
+    url: "https://www.youtube.com/@YannicKilcher/videos"
+    topic: "AI research paper walkthroughs"
+    max_videos: 15
+
+  - name: "Two Minute Papers"
+    url: "https://www.youtube.com/@TwoMinutePapers/videos"
+    topic: "AI research summaries"
+    max_videos: 15
+
+  - name: "Qiskit"
+    url: "https://www.youtube.com/@qiskit/videos"
+    topic: "quantum computing"
+    max_videos: 15