Cookbook

export TRANSCRIPTAPI_KEY="ta_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

# /health needs no key and costs nothing — a safe smoke test before a big job
curl -s "https://api.transcriptapi.io/health"
# {"status":"ok"}

curl -s "https://api.transcriptapi.io/transcript?video_id=dQw4w9WgXcQ" \
  -H "Authorization: Bearer $TRANSCRIPTAPI_KEY" | head -c 200

# ta.py — pip install httpx
import os
import time

import httpx

BASE = "https://api.transcriptapi.io"

client = httpx.Client(
    base_url=BASE,
    headers={"Authorization": f"Bearer {os.environ['TRANSCRIPTAPI_KEY']}"},
    timeout=httpx.Timeout(120.0),   # a long transcript with translation is slow
)


class OutOfCredits(RuntimeError):
    """402 — stop the whole job, no amount of retrying makes credits appear."""


class PermanentUpstream(RuntimeError):
    """502 for a reason that will never change: no captions, private, deleted."""


# The 502 detail carries the upstream exception name. These describe the video
# itself, so a retry only burns wall-clock time.
PERMANENT = (
    "TranscriptsDisabled",
    "NoTranscriptFound",
    "VideoUnavailable",
    "InvalidVideoId",
    "AgeRestricted",
)


def call(path: str, attempts: int = 4, **params):
    """One GET against the API, with retries only where they can help."""
    for attempt in range(attempts):
        r = client.get(path, params=params)

        if r.status_code == 200:
            return r.json()

        try:
            detail = r.json().get("detail", r.text)
        except ValueError:
            detail = r.text

        if r.status_code == 401:
            raise RuntimeError(f"bad or missing API key: {detail}")
        if r.status_code == 402:
            raise OutOfCredits(detail)
        if r.status_code == 422:
            raise ValueError(f"bad parameters for {path}: {detail}")
        if r.status_code == 502:
            if any(name in detail for name in PERMANENT):
                raise PermanentUpstream(detail)
            # Transient: YouTube blocked an exit node or timed out. The credits
            # were already refunded, so the retry is the only thing it costs.
            time.sleep(1.5 * (attempt + 1))
            continue

        r.raise_for_status()

    raise RuntimeError(f"{path} still failing after {attempts} attempts")

# corpus.py — python corpus.py PLZHQObOWTQDPD3MizzM2xVFitgF8hE_ab course.txt
import re
import sys
from concurrent.futures import ThreadPoolExecutor

from ta import call, OutOfCredits, PermanentUpstream


def flatten(segments) -> str:
    """Segments are 2-6 words each. Joined and de-spaced they read as prose."""
    return re.sub(r"\s+", " ", " ".join(s["text"].strip() for s in segments)).strip()


def fetch(video):
    """Returns (video, text, error). Only 402 is allowed to escape and kill the
    pool — everything else is a per-video problem, not a job-level one."""
    try:
        data = call("/transcript", video_id=video["id"], language="en")
        return video, flatten(data["transcript"]), None
    except OutOfCredits:
        raise
    except (PermanentUpstream, RuntimeError, ValueError) as exc:
        return video, None, str(exc)


def build(playlist_id: str, out_path: str, workers: int = 6):
    playlist = call("/playlist", playlist_id=playlist_id, limit=500)
    videos = playlist["videos"]
    print(f"{playlist['total']} videos, up to {1 + len(videos)} credits on a cold run")

    with ThreadPoolExecutor(max_workers=workers) as pool:
        # map keeps playlist order, which for a course is chapter order
        rows = list(pool.map(fetch, videos))

    skipped = []
    with open(out_path, "w", encoding="utf-8") as fh:
        for video, text, error in rows:
            if error:
                skipped.append((video["id"], video["title"], error))
                continue
            fh.write(f"\n\n### {video['title']}\n")
            fh.write(f"https://www.youtube.com/watch?v={video['id']}\n\n")
            fh.write(text + "\n")

    words = sum(len(t.split()) for _, t, e in rows if not e)
    print(f"wrote {out_path}: {len(rows) - len(skipped)} videos, ~{words:,} words")
    for video_id, title, error in skipped:
        print(f"  skipped {video_id} ({title}): {error}")


if __name__ == "__main__":
    try:
        build(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else "corpus.txt")
    except OutOfCredits as exc:
        # Everything fetched so far is cached, so topping up and re-running
        # resumes for free rather than starting over.
        sys.exit(f"out of credits: {exc}")

# index.py — a transcript search engine in one file, no services required
import re
import sqlite3

from ta import call, PermanentUpstream

DB = sqlite3.connect("transcripts.db")
DB.executescript("""
CREATE VIRTUAL TABLE IF NOT EXISTS snippets USING fts5(
    video_id UNINDEXED,
    title,
    start UNINDEXED,
    body
);
""")


def windows(segments, span: float = 30.0):
    """Group segments into ~30 second windows.

    A single segment is a fragment of a sentence, so indexing per segment means
    a phrase query never matches and every hit is unreadable. 30 seconds is
    long enough to hold a thought and short enough to jump to.
    """
    buf, t0 = [], None
    for seg in segments:
        if t0 is None:
            t0 = seg["start"]
        buf.append(seg["text"].strip())
        if seg["start"] + seg["duration"] - t0 >= span:
            yield t0, " ".join(buf)
            buf, t0 = [], None
    if buf:
        yield t0, " ".join(buf)


def index_channel(channel_id: str, limit: int = 50):
    listing = call("/channel/videos", channel_id=channel_id, limit=limit)

    for video in listing["videos"]:
        done = DB.execute(
            "SELECT 1 FROM snippets WHERE video_id = ? LIMIT 1", (video["id"],)
        ).fetchone()
        if done:
            continue

        try:
            data = call("/transcript", video_id=video["id"], language="en")
        except PermanentUpstream:
            continue   # captions disabled or the video is gone; nothing to index

        DB.executemany(
            "INSERT INTO snippets (video_id, title, start, body) VALUES (?, ?, ?, ?)",
            [
                (video["id"], video["title"], start, text)
                for start, text in windows(data["transcript"])
            ],
        )
        DB.commit()
        print(f"indexed {video['title']}")


def search(query: str, limit: int = 10):
    rows = DB.execute(
        """
        SELECT video_id, title, start,
               snippet(snippets, 3, '>>', '<<', '...', 14)
        FROM snippets
        WHERE snippets MATCH ?
        ORDER BY rank
        LIMIT ?
        """,
        (query, limit),
    ).fetchall()

    for video_id, title, start, hit in rows:
        at = max(0, int(start) - 2)   # land just before the phrase, not on top of it
        print(f"{title}  [{at // 60}:{at % 60:02d}]")
        print(f"  https://www.youtube.com/watch?v={video_id}&t={at}s")
        print(f"  {hit}\n")


if __name__ == "__main__":
    index_channel("@Fireship", limit=60)
    search("react server components")

// Rendering a hit: the timestamp is the whole point, so make it clickable.
export type Hit = { videoId: string; title: string; start: number; body: string };

export function deepLink(videoId: string, start: number, lead = 2): string {
  const at = Math.max(0, Math.floor(start) - lead);
  return `https://www.youtube.com/watch?v=${videoId}&t=${at}s`;
}

export function stamp(start: number): string {
  const total = Math.floor(start);
  const pad = (n: number) => String(n).padStart(2, "0");
  const h = Math.floor(total / 3600);
  const m = Math.floor((total % 3600) / 60);
  const s = total % 60;
  return h > 0 ? `${h}:${pad(m)}:${pad(s)}` : `${m}:${pad(s)}`;
}

// <a href={deepLink(hit.videoId, hit.start)}>{stamp(hit.start)}</a>

# subtitles.py — write .srt and .vtt for a video, plus a translated track
from math import ceil

from ta import call


def cues(segments):
    """(start, end, text) with overlaps removed.

    YouTube timings routinely overlap by a few hundred milliseconds. Players
    handle that as a flicker or a dropped line, so each cue is clamped to the
    start of the next one.
    """
    out = []
    for i, seg in enumerate(segments):
        start = float(seg["start"])
        end = start + float(seg["duration"])
        if i + 1 < len(segments):
            end = min(end, float(segments[i + 1]["start"]))
        if end <= start:
            end = start + 0.2
        text = " ".join(seg["text"].split())
        if text:
            out.append((start, end, text))
    return out


def clock(seconds: float, sep: str) -> str:
    """SRT wants 00:00:01,500 — VTT wants 00:00:01.500. Same digits, one comma."""
    ms = int(round(seconds * 1000))
    h, ms = divmod(ms, 3_600_000)
    m, ms = divmod(ms, 60_000)
    s, ms = divmod(ms, 1000)
    return f"{h:02d}:{m:02d}:{s:02d}{sep}{ms:03d}"


def to_srt(segments) -> str:
    blocks = [
        f"{i}\n{clock(a, ',')} --> {clock(b, ',')}\n{t}\n"
        for i, (a, b, t) in enumerate(cues(segments), start=1)
    ]
    return "\n".join(blocks)


def to_vtt(segments) -> str:
    blocks = ["WEBVTT\n"] + [
        f"{clock(a, '.')} --> {clock(b, '.')}\n{t}\n" for a, b, t in cues(segments)
    ]
    return "\n".join(blocks)


def translation_credits(segment_count: int) -> int:
    """What a translate_to call costs: 1 for the call, plus 1 per 40 segments
    with a floor of 2."""
    return 1 + max(2, ceil(segment_count / 40))


def write_tracks(video_id: str, translate_to: str | None = None):
    original = call("/transcript", video_id=video_id, language="en")
    segments = original["transcript"]

    for ext, body in (("srt", to_srt(segments)), ("vtt", to_vtt(segments))):
        with open(f"{video_id}.en.{ext}", "w", encoding="utf-8") as fh:
            fh.write(body)

    if not translate_to:
        return

    # The segment count is only known after the first fetch — and that fetch is
    # now cached, so it is not billed again when the translated call repeats it.
    print(f"{len(segments)} segments -> {translation_credits(len(segments))} credits")

    other = call(
        "/transcript", video_id=video_id, language="en", translate_to=translate_to
    )
    lang = other.get("translated_to", translate_to)
    with open(f"{video_id}.{lang}.vtt", "w", encoding="utf-8") as fh:
        fh.write(to_vtt(other["transcript"]))


if __name__ == "__main__":
    write_tracks("dQw4w9WgXcQ", translate_to="de")

<video controls src="/clips/dQw4w9WgXcQ.mp4">
  <track kind="subtitles" srclang="en" label="English"
         src="/subs/dQw4w9WgXcQ.en.vtt" default>
  <track kind="subtitles" srclang="de" label="Deutsch"
         src="/subs/dQw4w9WgXcQ.de.vtt">
</video>

# Subscribing and unsubscribing are free. Only the transcript fetch costs.
curl -X POST "https://api.transcriptapi.io/channel/subscribe" \
  -H "Authorization: Bearer $TRANSCRIPTAPI_KEY" \
  -H "Content-Type: application/json" \
  -d '{"channel_id":"UC_x5XG1OV2P6uZZ5FSM9Ttw","webhook_url":"https://you.example/hooks/yt/s3cr3t"}'

# {"status":"subscribed","channel_id":"UC_x5XG1OV2P6uZZ5FSM9Ttw", ...}

# Stop watching — channel_id goes in the query string for the DELETE
curl -X DELETE "https://api.transcriptapi.io/channel/subscribe?channel_id=UC_x5XG1OV2P6uZZ5FSM9Ttw" \
  -H "Authorization: Bearer $TRANSCRIPTAPI_KEY"

# hook.py — uvicorn hook:app --port 8080
# pip install fastapi uvicorn httpx openai
import os

import httpx
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request
from openai import OpenAI

from your_app import notify_ops, publish, retry_in   # your own plumbing

app = FastAPI()
llm = OpenAI()

HOOK_SECRET = os.environ["HOOK_SECRET"]
API_KEY = os.environ["TRANSCRIPTAPI_KEY"]

seen: set[str] = set()   # use Redis or a table once you have more than one worker


@app.post("/hooks/yt/{secret}")
async def on_new_video(secret: str, request: Request, background: BackgroundTasks):
    # Deliveries are unsigned, so the shared secret lives in the path and the
    # body is treated as a nudge to go fetch — never as trusted data.
    if secret != HOOK_SECRET:
        raise HTTPException(status_code=404)

    body = await request.json()
    video_id = body.get("video_id")
    if not video_id:
        raise HTTPException(status_code=400, detail="no video_id")

    if video_id in seen:
        return {"ok": True, "duplicate": True}
    seen.add(video_id)

    # Answer immediately; a webhook sender is not waiting for your summary.
    background.add_task(summarise, body.get("channel_id"), video_id)
    return {"ok": True}


async def summarise(channel_id: str, video_id: str):
    async with httpx.AsyncClient(timeout=120.0) as http:
        r = await http.get(
            "https://api.transcriptapi.io/transcript",
            params={"video_id": video_id},
            headers={"Authorization": f"Bearer {API_KEY}"},
        )

    if r.status_code == 402:
        return notify_ops(f"out of credits, skipped {video_id}")
    if r.status_code == 502:
        # Auto-captions can trail the upload by minutes. The credits were
        # refunded, so re-queueing this is free.
        return retry_in(minutes=20, channel_id=channel_id, video_id=video_id)
    r.raise_for_status()

    text = " ".join(s["text"] for s in r.json()["transcript"])
    if len(text) < 200:
        return   # a stub or a Short; not worth a model call

    reply = llm.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Summarise this video transcript in five bullets. "
                           "Keep the speaker's claims and numbers, drop the filler.",
            },
            {"role": "user", "content": text[:120_000]},
        ],
    )
    publish(channel_id, video_id, reply.choices[0].message.content)

// server.ts — npm i express
import express from "express";

const app = express();
app.use(express.json());

const API_KEY = process.env.TRANSCRIPTAPI_KEY!;
const seen = new Set<string>();

app.post("/hooks/yt/:secret", (req, res) => {
  if (req.params.secret !== process.env.HOOK_SECRET) return res.sendStatus(404);

  const { channel_id, video_id } = req.body ?? {};
  if (!video_id) return res.sendStatus(400);
  if (seen.has(video_id)) return res.json({ ok: true, duplicate: true });
  seen.add(video_id);

  res.json({ ok: true });                  // ack first
  void summarise(channel_id, video_id);    // then do the slow part
});

type Segment = { start: number; duration: number; text: string };

async function summarise(channelId: string, videoId: string) {
  const url = new URL("https://api.transcriptapi.io/transcript");
  url.searchParams.set("video_id", videoId);

  const res = await fetch(url, {
    headers: { Authorization: `Bearer ${API_KEY}` },
  });

  if (!res.ok) {
    // 502 means it was refunded — put it back on the queue, do not drop it.
    console.error("transcript failed", res.status, await res.text());
    return;
  }

  const { transcript } = (await res.json()) as { transcript: Segment[] };
  const text = transcript.map((s) => s.text).join(" ");
  await handOffToYourModel(channelId, videoId, text);   // your own plumbing
}

app.listen(8080);

# rag.py — chunks that carry their own clock
import re

from ta import call


def chunks(segments, target_words: int = 220, overlap_words: int = 40):
    """Split a transcript into overlapping windows, each keeping start and end.

    The overlap exists because a sentence that straddles a boundary is a
    sentence neither chunk can answer from.
    """
    out, buf, words = [], [], 0

    for seg in segments:
        buf.append(seg)
        words += len(seg["text"].split())
        if words < target_words:
            continue

        out.append(pack(buf))
        keep, kept = [], 0
        for s in reversed(buf):
            keep.insert(0, s)
            kept += len(s["text"].split())
            if kept >= overlap_words:
                break
        buf, words = keep, kept

    # Flush the tail, unless it is nothing but the overlap already emitted.
    if buf and (not out or words > overlap_words):
        out.append(pack(buf))
    return out


def pack(buf):
    last = buf[-1]
    return {
        "start": round(float(buf[0]["start"]), 2),
        "end": round(float(last["start"]) + float(last["duration"]), 2),
        "text": " ".join(s["text"].strip() for s in buf),
    }


def context(rows) -> str:
    """Each chunk is labelled with the second it starts at, so the model can
    cite it and you can turn the citation back into a link."""
    return "\n\n".join(f"[{int(c['start'])}s] {c['text']}" for c in rows)


CITATION = re.compile(r"\[(\d+)s\]")


def linkify(answer: str, video_id: str) -> str:
    return CITATION.sub(
        lambda m: f"[{m.group(1)}s](https://www.youtube.com/watch?v={video_id}&t={m.group(1)}s)",
        answer,
    )


def ask(video_id: str, question: str) -> str:
    from openai import OpenAI

    data = call("/transcript", video_id=video_id, language="en")
    rows = chunks(data["transcript"])

    reply = OpenAI().chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Answer only from the transcript below. Every claim ends "
                           "with the [NNNs] marker of the chunk it came from. If the "
                           "transcript does not answer the question, say so.\n\n"
                           + context(rows),
            },
            {"role": "user", "content": question},
        ],
    )
    return linkify(reply.choices[0].message.content, video_id)


if __name__ == "__main__":
    print(ask("dQw4w9WgXcQ", "What promises does the singer make?"))

# Across many videos, embed the same chunks and keep the clock in the metadata.
from rag import chunks
from ta import call


def channel_chunks(channel_id: str, limit: int = 50):
    listing = call("/channel/videos", channel_id=channel_id, limit=limit)

    for video in listing["videos"]:
        data = call("/transcript", video_id=video["id"], language="en")
        for c in chunks(data["transcript"]):
            yield {
                "id": f"{video['id']}:{int(c['start'])}",
                "text": c["text"],
                "metadata": {
                    "video_id": video["id"],
                    "title": video["title"],
                    "start": c["start"],
                    "end": c["end"],
                    "url": f"https://www.youtube.com/watch?v={video['id']}&t={int(c['start'])}s",
                },
            }


# rows = list(channel_chunks("@Fireship", limit=50))
# collection.add(ids=[r["id"] for r in rows],
#                documents=[r["text"] for r in rows],
#                metadatas=[r["metadata"] for r in rows])

// transcriptapi.ts — every rule about this API in one place, zero dependencies
const BASE = "https://api.transcriptapi.io";

export type Segment = { start: number; duration: number; text: string };
export type TranscriptResponse = {
  video_id: string;
  transcript: Segment[];
  translated_to?: string;
};
export type SearchHit = {
  id: string;
  title: string;
  channel: string;
  views: string;
  duration: string;
};
export type ChannelVideo = {
  id: string;
  title: string;
  views: string;
  duration: string;
  published: string;
};

export class ApiError extends Error {
  constructor(readonly status: number, readonly detail: string) {
    super(`${status}: ${detail}`);
    this.name = "ApiError";
  }
}
export class OutOfCredits extends ApiError {}
export class BadRequest extends ApiError {}

const PERMANENT_UPSTREAM = [
  "TranscriptsDisabled",
  "NoTranscriptFound",
  "VideoUnavailable",
  "InvalidVideoId",
  "AgeRestricted",
];

const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

export class TranscriptAPI {
  constructor(
    private readonly key: string,
    private readonly retries = 3,
  ) {}

  private async get<T>(
    path: string,
    params: Record<string, string | number | undefined>,
  ): Promise<T> {
    const url = new URL(BASE + path);
    for (const [k, v] of Object.entries(params)) {
      if (v !== undefined) url.searchParams.set(k, String(v));
    }

    for (let attempt = 0; ; attempt++) {
      const res = await fetch(url, {
        headers: { Authorization: `Bearer ${this.key}` },
      });
      if (res.ok) return (await res.json()) as T;

      const detail = await res.text();

      // 401 and 422 are bugs in the caller. Retrying cannot fix a typo.
      if (res.status === 401 || res.status === 422) {
        throw new BadRequest(res.status, detail);
      }
      // 402 needs a top-up, not a backoff loop. Let it abort the whole job.
      if (res.status === 402) throw new OutOfCredits(402, detail);
      // 502 was refunded. Retry it unless the reason is about the video itself.
      if (res.status === 502 && PERMANENT_UPSTREAM.some((n) => detail.includes(n))) {
        throw new ApiError(502, detail);
      }
      if (attempt >= this.retries) throw new ApiError(res.status, detail);

      await sleep(400 * 2 ** attempt + Math.random() * 200);
    }
  }

  transcript(videoId: string, opts: { language?: string; translateTo?: string } = {}) {
    return this.get<TranscriptResponse>("/transcript", {
      video_id: videoId,
      language: opts.language ?? "en",
      translate_to: opts.translateTo,
    });
  }

  search(q: string, limit = 20) {
    return this.get<{ query: string; results: SearchHit[] }>("/search", { q, limit });
  }

  channelVideos(channelId: string, limit = 50) {
    return this.get<{ channel_id: string; title: string; total: number; videos: ChannelVideo[] }>(
      "/channel/videos",
      { channel_id: channelId, limit },
    );
  }

  channelSearch(channelId: string, q: string, limit = 20) {
    return this.get<{ channel_id: string; query: string; results: SearchHit[] }>(
      "/channel/search",
      { channel_id: channelId, q, limit },
    );
  }

  playlist(playlistId: string, limit = 100) {
    return this.get<{ playlist_id: string; total: number; videos: ChannelVideo[] }>(
      "/playlist",
      { playlist_id: playlistId, limit },
    );
  }
}

// A bounded pool that stops the moment the balance runs out.
export async function mapPool<T, R>(
  items: readonly T[],
  limit: number,
  fn: (item: T, index: number) => Promise<R>,
): Promise<R[]> {
  const out = new Array<R>(items.length);
  let cursor = 0;
  let failure: unknown = null;

  const worker = async () => {
    while (cursor < items.length && failure === null) {
      const i = cursor++;
      try {
        out[i] = await fn(items[i], i);
      } catch (err) {
        failure = err;
      }
    }
  };

  await Promise.all(
    Array.from({ length: Math.min(limit, items.length) }, worker),
  );
  if (failure !== null) throw failure;
  return out;
}

// Usage: re-running this after a crash costs nothing for the parts that
// already succeeded, because the server serves those from cache for free.
const api = new TranscriptAPI(process.env.TRANSCRIPTAPI_KEY!);
const { videos } = await api.playlist("PLZHQObOWTQDPD3MizzM2xVFitgF8hE_ab", 200);

try {
  const texts = await mapPool(videos, 6, async (v) => {
    const { transcript } = await api.transcript(v.id);
    return { id: v.id, title: v.title, text: transcript.map((s) => s.text).join(" ") };
  });
  console.log(`${texts.length} transcripts`);
} catch (err) {
  if (err instanceof OutOfCredits) {
    console.error("balance exhausted — top up and re-run, the rest is cached");
  } else {
    throw err;
  }
}