用于 RAG 和 LLM 流水线的视频文稿

import requests
 
API = "https://www.transcriptdock.com"
H = {"Authorization": "Bearer td_live_YOUR_KEY"}
 
def transcript(url: str) -> dict:
    job = requests.post(f"{API}/v1/jobs", headers={**H, "Idempotency-Key": f"rag-{url}"[:128]},
        json={"source": {"url": url}, "mode": "captions_only"}).json()
    while job["status"] not in ("succeeded", "failed", "cancelled"):
        job = requests.get(f"{API}/v1/jobs/{job['id']}", params={"wait": 25}, headers=H).json()
    if job["status"] != "succeeded":
        raise RuntimeError(job["error"]["code"])
    return requests.get(f"{API}/v1/transcripts/{job['result_id']}", headers=H).json()
 
def chunks(t: dict, seconds: float = 60.0):
    """Group caption segments into windows of about `seconds`, never splitting a segment."""
    url = t["source"]["canonical_url"]
    buf, start = [], None
    for seg in t["segments"]:
        if start is None:
            start = seg["start"]
        buf.append(seg["text"])
        if seg["end"] - start >= seconds:
            yield {"text": " ".join(buf), "start": start, "url": f"{url}&t={int(start)}s" if "youtube" in url else url}
            buf, start = [], None
    if buf:
        yield {"text": " ".join(buf), "start": start, "url": url}
 
t = transcript("https://www.youtube.com/watch?v=dQw4w9WgXcQ")
for c in chunks(t):
    print(c["start"], c["url"], c["text"][:80])  # embed c["text"], store start and url as metadata