YouTube transcripts in LangChain, without the blocked loader

# transcriptapi.py — one function, shared by every loader on this page
import os, time, requests

API = "https://api.transcriptapi.io"
HEADERS = {"Authorization": f"Bearer {os.environ['TRANSCRIPTAPI_KEY']}"}


def fetch_chunks(video_id: str, language: str = "en", window: int = 60) -> list[dict]:
    """The transcript of one video, grouped into `window`-second chunks.

    Returns [] when the video has no captions (404, refunded, retryable: false).
    Retries once when YouTube was unreachable (503, refunded, retryable: true).
    """
    for attempt in range(2):
        r = requests.get(
            f"{API}/transcript",
            params={"video_id": video_id, "language": language},
            headers=HEADERS,
            timeout=60,
        )
        if r.ok:
            break
        body = r.json()
        if body.get("retryable") is False:
            return []                 # captions off, private, deleted — credit refunded
        if body.get("retryable") is not True:
            r.raise_for_status()      # 401 / 402 / 422 — fix the request, not the video
        time.sleep(2 * (attempt + 1))
    r.raise_for_status()

    buckets: dict[int, list[dict]] = {}
    for seg in r.json()["transcript"]:            # [{"start": 0.0, "duration": 3.52, "text": "..."}]
        buckets.setdefault(int(seg["start"] // window), []).append(seg)

    chunks = []
    for segs in buckets.values():                 # insertion order == playback order
        start = segs[0]["start"]
        end = round(segs[-1]["start"] + segs[-1]["duration"], 2)
        chunks.append({
            "text": " ".join(s["text"] for s in segs),
            "video_id": video_id,
            "language": language,
            "start": start,
            "end": end,
            "source": f"https://www.youtube.com/watch?v={video_id}&t={int(start)}s",
        })
    return chunks

from typing import Iterator
from langchain_core.document_loaders import BaseLoader
from langchain_core.documents import Document
from transcriptapi import fetch_chunks


class TranscriptAPILoader(BaseLoader):
    """Same shape as YoutubeLoader in CHUNKS mode, minus the datacenter block."""

    def __init__(self, video_id: str, language: str = "en", window: int = 60):
        self.video_id, self.language, self.window = video_id, language, window

    def lazy_load(self) -> Iterator[Document]:
        for c in fetch_chunks(self.video_id, self.language, self.window):
            yield Document(page_content=c.pop("text"), metadata=c)


# before:
#   YoutubeLoader.from_youtube_url(url, transcript_format=TranscriptFormat.CHUNKS,
#                                  chunk_size_seconds=60).load()
# after:
docs = TranscriptAPILoader("dQw4w9WgXcQ").load()

docs[3].metadata
# {'video_id': 'dQw4w9WgXcQ', 'language': 'en', 'start': 180.16, 'end': 239.8,
#  'source': 'https://www.youtube.com/watch?v=dQw4w9WgXcQ&t=180s'}

import requests
from transcriptapi import API, HEADERS
from loader import TranscriptAPILoader


def channel_video_ids(channel: str, limit: int = 200) -> list[str]:
    """Newest uploads of a channel. channel is a UC… id or an @handle. 1 credit."""
    r = requests.get(f"{API}/channel/videos",
                     params={"channel_id": channel, "limit": limit},
                     headers=HEADERS, timeout=30)
    r.raise_for_status()
    return [v["id"] for v in r.json()["videos"]]


def search_video_ids(q: str, limit: int = 50) -> list[str]:
    """Top YouTube results for a query. 1 credit."""
    r = requests.get(f"{API}/search", params={"q": q, "limit": limit},
                     headers=HEADERS, timeout=30)
    r.raise_for_status()
    return [v["id"] for v in r.json()["results"]]


docs = []
for vid in channel_video_ids("@3blue1brown"):        # 1 credit
    docs.extend(TranscriptAPILoader(vid).load())    # 1 credit each; 0 on a cache hit

# any embeddings, any vector store — this is ordinary LangChain from here on
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

store = FAISS.from_documents(docs, OpenAIEmbeddings())
for hit in store.similarity_search("why is e to the i pi minus one", k=3):
    print(hit.metadata["source"], "—", hit.page_content[:80])

import { BaseDocumentLoader } from "@langchain/core/document_loaders/base";
import { Document } from "@langchain/core/documents";

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

export class TranscriptAPILoader extends BaseDocumentLoader {
  constructor(private videoId: string, private window = 60) {
    super();
  }

  async load(): Promise<Document[]> {
    const res = await fetch(`https://api.transcriptapi.io/transcript?video_id=${this.videoId}`, {
      headers: { Authorization: `Bearer ${process.env.TRANSCRIPTAPI_KEY}` },
    });
    if (!res.ok) {
      const err = await res.json();
      if (err.retryable === false) return [];   // no captions — refunded, skip the video
      throw new Error(err.detail);              // 503 — let your job queue retry it
    }
    const { transcript } = (await res.json()) as { transcript: Segment[] };

    const buckets = new Map<number, Segment[]>();
    for (const s of transcript) {
      const k = Math.floor(s.start / this.window);
      buckets.set(k, [...(buckets.get(k) ?? []), s]);
    }
    return [...buckets.values()].map((segs) => {
      const last = segs[segs.length - 1];
      return new Document({
        pageContent: segs.map((s) => s.text).join(" "),
        metadata: {
          video_id: this.videoId,
          start: segs[0].start,
          end: last.start + last.duration,
          source: `https://www.youtube.com/watch?v=${this.videoId}&t=${Math.floor(segs[0].start)}s`,
        },
      });
    });
  }
}

// before: YoutubeLoader.createFromUrl(url, { language: "en" })
const docs = await new TranscriptAPILoader("dQw4w9WgXcQ").load();

from llama_index.core import Document, VectorStoreIndex
from llama_index.core.readers.base import BaseReader
from transcriptapi import fetch_chunks


class TranscriptAPIReader(BaseReader):
    """Replaces YoutubeTranscriptReader. One Document per time window, not per video."""

    def load_data(self, video_ids: list[str], language: str = "en") -> list[Document]:
        docs = []
        for vid in video_ids:
            for c in fetch_chunks(vid, language):
                text = c.pop("text")
                docs.append(Document(text=text, metadata=c, id_=f"{vid}:{int(c['start'])}"))
        return docs


index = VectorStoreIndex.from_documents(TranscriptAPIReader().load_data(["dQw4w9WgXcQ"]))
response = index.as_query_engine().query("What is promised in the chorus?")
for node in response.source_nodes:
    print(node.metadata["source"], node.score)

HTTP 404
{
  "detail": "This video has captions turned off, so there is no transcript to fetch. Your credits have been refunded.",
  "error": "TranscriptsDisabled",
  "hint": "Retrying will not help — the uploader has disabled captions for this video.",
  "retryable": false,
  "credits_refunded": true
}