Ingesting YouTube transcripts: yt-dlp for subs, Whisper when subs don't exist

YOUTUBE_API_KEY=AIza...
YOUTUBE_CHANNEL_HANDLE=@TheDoersglobal
DATABASE_URL=postgresql+psycopg://localhost:5432/clipdex
 
# yt-dlp pulls cookies from your signed-in browser to bypass YouTube's
# bot check. Set to chrome / safari / firefox / brave, or empty to disable.
YTDLP_COOKIES_FROM_BROWSER=brave
 
# Bound first runs (a 1000-episode channel takes a while). 0 = unbounded.
MAX_VIDEOS_PER_RUN=2

# Handle -> channel id + uploads playlist id
curl -s "https://www.googleapis.com/youtube/v3/channels?part=contentDetails,snippet&forHandle=TheDoersglobal&key=${YOUTUBE_API_KEY}" \
  | jq '.items[0] | {title: .snippet.title, id: .id, uploads: .contentDetails.relatedPlaylists.uploads}'
# -> { "title": "The Doers", "id": "UCZbKwSM00cPCNCdJ-ClfqFQ", "uploads": "UUZbKwSM00cPCNCdJ-ClfqFQ" }

ingest/
├── pyproject.toml
└── src/
    └── clipdex_ingest/
        ├── __init__.py
        ├── __main__.py       # python -m clipdex_ingest
        ├── client.py         # YouTube Data API wrapper + QuotaExceeded
        ├── captions.py       # track picker + WebVTT parser
        ├── fallback.py       # yt-dlp + Whisper (optional extra)
        ├── store.py          # Postgres writes
        ├── run.py            # entrypoint behind `task ingest`
        └── settings.py       # Pydantic Settings

[project]
name = "clipdex-ingest"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "httpx>=0.27",
    "webvtt-py>=0.5",
    "psycopg[binary]>=3.2",
    "sqlalchemy>=2.0",
    "pydantic-settings>=2.5",
    "clipdex-schema",     # workspace dependency
]
 
[project.optional-dependencies]
fallback = [
    "yt-dlp>=2024.10",
    "openai-whisper>=20240930",
]

import httpx
from datetime import datetime
from clipdex_ingest.settings import settings
 
YT = "https://www.googleapis.com/youtube/v3"
 
async def list_uploads(playlist_id: str, since: datetime | None = None) -> list[dict]:
    """Return all videos in the uploads playlist newer than `since`."""
    out: list[dict] = []
    page_token: str | None = None
    async with httpx.AsyncClient(timeout=15) as http:
        while True:
            params = {
                "part": "snippet,contentDetails",
                "playlistId": playlist_id,
                "maxResults": 50,
                "key": settings.youtube_api_key,
            }
            if page_token:
                params["pageToken"] = page_token
            r = await http.get(f"{YT}/playlistItems", params=params)
            r.raise_for_status()
            data = r.json()
 
            stop = False
            for item in data["items"]:
                published = datetime.fromisoformat(
                    item["contentDetails"]["videoPublishedAt"].replace("Z", "+00:00")
                )
                if since and published <= since:
                    stop = True
                    break
                out.append({
                    "video_id": item["contentDetails"]["videoId"],
                    "title": item["snippet"]["title"],
                    "published_at": published,
                })
            page_token = data.get("nextPageToken")
            if stop or not page_token:
                break
    return out

import asyncio
from pathlib import Path
 
import webvtt
 
from clipdex_ingest.settings import settings
from clipdex_schema import TranscriptSegment
 
 
def _cache_dir() -> Path:
    return Path(settings.cache_dir) / "captions"
 
 
def _fetch_vtt_sync(video_id: str) -> Path | None:
    """Download the auto-generated English subs as WebVTT. No media downloaded."""
    import yt_dlp
 
    out_dir = _cache_dir()
    out_dir.mkdir(parents=True, exist_ok=True)
    template = str(out_dir / f"{video_id}.%(ext)s")
 
    ydl_opts: dict = {
        "skip_download": True,
        "writesubtitles": True,
        "writeautomaticsub": True,
        "subtitleslangs": ["en", "en-US", "en-GB"],
        "subtitlesformat": "vtt",
        "outtmpl": template,
        "quiet": True,
        "no_warnings": True,
        # Subs-only: shorts, premieres, and bot-flagged videos sometimes
        # have no resolvable media formats. We don't care — we're not
        # downloading media — so swallow that specific error.
        "ignore_no_formats_error": True,
    }
    if settings.ytdlp_cookies_from_browser:
        ydl_opts["cookiesfrombrowser"] = (settings.ytdlp_cookies_from_browser,)
 
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        ydl.download([f"https://www.youtube.com/watch?v={video_id}"])
 
    for lang in ("en", "en-US", "en-GB"):
        candidate = out_dir / f"{video_id}.{lang}.vtt"
        if candidate.exists():
            return candidate
    return None
 
 
async def fetch_captions(video_id: str) -> list[TranscriptSegment] | None:
    """Return parsed segments, or None if no caption track is available."""
    out_dir = _cache_dir()
    for lang in ("en", "en-US", "en-GB"):
        candidate = out_dir / f"{video_id}.{lang}.vtt"
        if candidate.exists():
            return list(_parse_vtt(candidate, video_id)) or None
    cached = await asyncio.to_thread(_fetch_vtt_sync, video_id)
    if cached is None:
        return None
    return list(_parse_vtt(cached, video_id)) or None

from collections.abc import Iterator
 
 
def _parse_vtt(path: Path, video_id: str) -> Iterator[TranscriptSegment]:
    """Parse YouTube auto-caption WebVTT, collapsing the rolling-cue format."""
    seq = 0
    prev_text: str | None = None
    for cue in webvtt.read(str(path)):
        start_ms = int(cue.start_in_seconds * 1000)
        end_ms = int(cue.end_in_seconds * 1000)
        # Skip the 10ms rolling transition cues.
        if end_ms - start_ms < 200:
            continue
        lines = [line.strip() for line in cue.text.splitlines() if line.strip()]
        if not lines:
            continue
        # Multi-line cues are "prev line + new line" — keep only the new one.
        text = lines[-1]
        if text == prev_text:
            continue
        prev_text = text
        yield TranscriptSegment(
            video_id=video_id,
            seq=seq,
            start_ms=start_ms,
            end_ms=end_ms,
            text=text,
            source="youtube-captions",
        )
        seq += 1

from typing import Literal
from pydantic import BaseModel
 
CaptionSource = Literal["youtube-captions", "whisper-fallback"]
 
class TranscriptSegment(BaseModel):
    video_id: str
    seq: int
    start_ms: int
    end_ms: int
    text: str
    source: CaptionSource

from pathlib import Path
 
from clipdex_ingest.settings import settings
from clipdex_schema import TranscriptSegment
 
 
def transcribe_with_whisper(video_id: str) -> list[TranscriptSegment]:
    """Download audio with yt-dlp, transcribe with local Whisper, delete the media.
 
    The `fallback` extra must be installed:
        uv sync --package clipdex-ingest --extra fallback
    """
    try:
        import whisper
        import yt_dlp
    except ImportError as e:
        raise RuntimeError(
            "Whisper fallback requested but the `fallback` extra is not installed.\n"
            "Run: uv sync --package clipdex-ingest --extra fallback"
        ) from e
 
    audio_dir = Path(settings.cache_dir) / "audio"
    audio_dir.mkdir(parents=True, exist_ok=True)
    audio_path = audio_dir / f"{video_id}.m4a"
 
    ydl_opts: dict = {
        "format": "bestaudio[ext=m4a]/bestaudio",
        "outtmpl": str(audio_path.with_suffix(".%(ext)s")),
        "quiet": True,
        "no_warnings": True,
    }
    if settings.ytdlp_cookies_from_browser:
        ydl_opts["cookiesfrombrowser"] = (settings.ytdlp_cookies_from_browser,)
 
    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
        ydl.download([f"https://www.youtube.com/watch?v={video_id}"])
 
    model = whisper.load_model(settings.whisper_model)  # "base" by default
    result = model.transcribe(str(audio_path), word_timestamps=False)
 
    # Delete the audio file once transcribed — we keep the text, not the media.
    audio_path.unlink(missing_ok=True)
 
    return [
        TranscriptSegment(
            video_id=video_id,
            seq=i,
            start_ms=int(seg["start"] * 1000),
            end_ms=int(seg["end"] * 1000),
            text=seg["text"].strip(),
            source="whisper-fallback",
        )
        for i, seg in enumerate(result["segments"])
    ]

CREATE TABLE processed_videos (
    video_id        TEXT PRIMARY KEY,
    title           TEXT NOT NULL,
    published_at    TIMESTAMPTZ NOT NULL,
    status          TEXT NOT NULL CHECK (status IN ('pending', 'done', 'failed')),
    source          TEXT,
    segment_count   INT,
    ingested_at     TIMESTAMPTZ,
    error           TEXT
);
 
CREATE TABLE transcript_segments (
    video_id   TEXT NOT NULL REFERENCES processed_videos(video_id) ON DELETE CASCADE,
    seq        INT NOT NULL,
    start_ms   INT NOT NULL,
    end_ms     INT NOT NULL,
    text       TEXT NOT NULL,
    source     TEXT NOT NULL,
    PRIMARY KEY (video_id, seq)
);
 
CREATE INDEX transcript_segments_video_idx ON transcript_segments (video_id, start_ms);

from datetime import datetime
 
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
 
from clipdex_schema import TranscriptSegment
 
 
async def latest_published(session: AsyncSession) -> datetime | None:
    """Watermark: ignore failed rows so a failure's now() timestamp doesn't
    mask real new uploads on the next run."""
    r = await session.execute(
        text("SELECT MAX(published_at) FROM processed_videos WHERE status = 'done'")
    )
    return r.scalar()
 
 
async def already_done(session: AsyncSession, video_id: str) -> bool:
    r = await session.execute(
        text("SELECT status FROM processed_videos WHERE video_id = :v"),
        {"v": video_id},
    )
    row = r.first()
    return row is not None and row.status == "done"
 
 
async def save_video(
    session: AsyncSession,
    *,
    video_id: str,
    title: str,
    published_at: datetime,
    segments: list[TranscriptSegment],
    source: str,
) -> None:
    """Replace any prior data for `video_id` and mark it done."""
    try:
        await session.execute(
            text(
                """
                INSERT INTO processed_videos
                  (video_id, title, published_at, status, source, segment_count, ingested_at)
                VALUES
                  (:vid, :title, :pub, 'done', :src, :n, now())
                ON CONFLICT (video_id) DO UPDATE
                SET status = 'done',
                    source = EXCLUDED.source,
                    segment_count = EXCLUDED.segment_count,
                    ingested_at = EXCLUDED.ingested_at,
                    error = NULL
                """
            ),
            {"vid": video_id, "title": title, "pub": published_at,
             "src": source, "n": len(segments)},
        )
        await session.execute(
            text("DELETE FROM transcript_segments WHERE video_id = :v"),
            {"v": video_id},
        )
        if segments:
            await session.execute(
                text(
                    """
                    INSERT INTO transcript_segments
                      (video_id, seq, start_ms, end_ms, text, source)
                    VALUES
                      (:video_id, :seq, :start_ms, :end_ms, :text, :source)
                    """
                ),
                [s.model_dump() for s in segments],
            )
        await session.commit()
    except Exception:
        await session.rollback()
        raise
 
 
async def mark_failed(session: AsyncSession, video_id: str, error: str) -> None:
    """Record a failed video so we know not to retry it on every run."""
    try:
        await session.execute(
            text(
                """
                INSERT INTO processed_videos
                  (video_id, title, published_at, status, error, ingested_at)
                VALUES (:vid, :vid, now(), 'failed', :err, now())
                ON CONFLICT (video_id) DO UPDATE
                SET status = 'failed', error = EXCLUDED.error, ingested_at = now()
                """
            ),
            {"vid": video_id, "err": error[:1000]},
        )
        await session.commit()
    except Exception:
        await session.rollback()
        raise

import asyncio
import logging
import os
 
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 
from clipdex_ingest.captions import fetch_captions
from clipdex_ingest.client import QuotaExceeded, list_uploads, resolve_uploads_playlist
from clipdex_ingest.settings import settings
from clipdex_ingest.store import already_done, latest_published, mark_failed, save_video
 
log = logging.getLogger("clipdex.ingest")
 
 
async def ingest_once(enable_fallback: bool | None = None) -> dict[str, int]:
    if enable_fallback is None:
        enable_fallback = os.getenv("CLIPDEX_ENABLE_FALLBACK", "").lower() in ("1", "true", "yes")
 
    if not settings.youtube_api_key:
        raise RuntimeError("YOUTUBE_API_KEY is not set; copy .env.example to .env and fill it in.")
 
    playlist_id = settings.uploads_playlist_id or await resolve_uploads_playlist(
        channel_id=settings.youtube_channel_id,
        handle=settings.youtube_channel_handle,
    )
 
    db_url = settings.database_url
    if db_url.startswith("postgresql://"):
        # SQLAlchemy maps the bare scheme to psycopg2; we ship psycopg3.
        db_url = "postgresql+psycopg://" + db_url[len("postgresql://"):]
 
    engine = create_async_engine(db_url)
    stats = {"ok": 0, "skipped": 0, "failed": 0}
 
    async with AsyncSession(engine) as session:
        since = await latest_published(session)
        uploads = await list_uploads(playlist_id, since=since)
        if settings.max_videos_per_run > 0:
            uploads = uploads[: settings.max_videos_per_run]
        log.info("ingest: %d uploads to process (since %s)", len(uploads), since)
 
        for v in uploads:
            vid = v["video_id"]
            if await already_done(session, vid):
                continue
            try:
                segments = await fetch_captions(vid)
                source: str = "youtube-captions"
                if segments is None:
                    if not enable_fallback:
                        log.info("ingest: %s no captions, fallback disabled — skipped", vid)
                        stats["skipped"] += 1
                        continue
                    log.info("ingest: %s no captions, falling back to whisper", vid)
                    from clipdex_ingest.fallback import transcribe_with_whisper
 
                    segments = transcribe_with_whisper(vid)
                    source = "whisper-fallback"
 
                await save_video(
                    session,
                    video_id=vid,
                    title=v["title"],
                    published_at=v["published_at"],
                    segments=segments,
                    source=source,
                )
                log.info("ingest: %s -> %d segments (%s)", vid, len(segments), source)
                stats["ok"] += 1
            except QuotaExceeded as e:
                log.warning("ingest: quota exceeded, stopping cleanly: %s", e)
                break
            except Exception as e:  # noqa: BLE001 — we want per-video isolation
                log.exception("ingest: %s failed", vid)
                await session.rollback()
                await mark_failed(session, vid, str(e))
                stats["failed"] += 1
 
    await engine.dispose()
    log.info("ingest: done. %d ok, %d skipped, %d failed",
             stats["ok"], stats["skipped"], stats["failed"])
    return stats

INFO clipdex.ingest: resolved uploads playlist: UUZbKwSM00cPCNCdJ-ClfqFQ
INFO clipdex.ingest: 2 uploads to process (since None)
INFO clipdex.ingest: kjKyj1ZEaf4 -> 6 segments (youtube-captions)
INFO clipdex.ingest: Agjt03cV7ac -> 2387 segments (youtube-captions)
INFO clipdex.ingest: done. 2 ok, 0 skipped, 0 failed

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
  <key>Label</key>
  <string>com.poudelprakash.clipdex.ingest</string>
  <key>ProgramArguments</key>
  <array>
    <string>/Users/prakash/.local/bin/task</string>
    <string>-d</string>
    <string>/Users/prakash/workspaces/2023/personal/clipdex</string>
    <string>ingest</string>
  </array>
  <key>StartCalendarInterval</key>
  <dict>
    <key>Hour</key><integer>3</integer>
    <key>Minute</key><integer>0</integer>
  </dict>
  <key>StandardOutPath</key>
  <string>/Users/prakash/Library/Logs/clipdex-ingest.log</string>
  <key>StandardErrorPath</key>
  <string>/Users/prakash/Library/Logs/clipdex-ingest.err</string>
</dict>
</plist>

[Unit]
Description=Daily YouTube ingest for the podcast index
 
[Timer]
OnCalendar=*-*-* 03:00:00
Persistent=true
 
[Install]
WantedBy=timers.target

from httpx import HTTPStatusError
 
class QuotaExceeded(Exception):
    pass
 
def raise_for_quota(resp):
    if resp.status_code == 403:
        body = resp.json()
        for err in body.get("error", {}).get("errors", []):
            if err.get("reason") == "quotaExceeded":
                raise QuotaExceeded(body["error"]["message"])
    resp.raise_for_status()