Transcript via transcriptapi.com (no proxy) as primary; youtube-transcript-api fallback

pipeline/transcribe.py

@@ -1,29 +1,64 @@
-"""Stage 4: fetch a timestamped transcript via youtube-transcript-api (no download).
+"""Stage 4: fetch a timestamped transcript (no video download).
 
-youtube-transcript-api scrapes ``www.youtube.com``, which datacenter IPs (like a Space)
-often get blocked from — usually as a TLS reset / SSL EOF rather than a clean 4xx. We
-retry a few times for transient failures and, when it's clearly a block, point the user at
-the ``YT_PROXY`` fallback. The returned snippets already carry ``start``/``duration``
-timestamps, which we normalize into ``{start, end, text}`` segments.
+Primary source is **transcriptapi.com** (a hosted transcript API). Because it's called on
+its own domain — not ``youtube.com`` — the Space reaches it directly, so transcripts need
+**no proxy** at all. Set the ``TRANSCRIPTAPI_KEY`` Space secret to enable it.
+
+Fallback is **youtube-transcript-api**, which scrapes ``www.youtube.com`` and therefore
+does need ``YT_PROXY`` from a datacenter IP (the block usually shows up as a TLS/SSL reset).
+
+Either way the snippets carry ``start``/``duration`` timing, normalized into
+``{start, end, text}`` segments the rest of the pipeline expects.
 """
 from __future__ import annotations
 
-import re
+import json
+import os
 import time
+import urllib.error
+import urllib.parse
+import urllib.request
 
-# Strip credentials from any proxy URL that a library exception might echo, so a pasted
-# http://user:pass@host proxy never leaks into the UI/logs.
-_CRED_RE = re.compile(r"(https?://)[^/@\s]+@")
-
-
-def _redact(text: str) -> str:
-    return _CRED_RE.sub(r"\1", str(text))
+TRANSCRIPTAPI_URL = "https://transcriptapi.com/api/v2/youtube/transcript"
 
 
 class TranscriptError(RuntimeError):
     """Raised when no usable transcript can be fetched (disabled, blocked, none)."""
 
 
+def _fetch_transcriptapi(video_id: str, langs: list[str]) -> list[dict]:
+    """Fetch timestamped snippets from transcriptapi.com. Requires TRANSCRIPTAPI_KEY.
+
+    Returns ``[{text, start, duration}]``. Raises on auth/credit/HTTP errors so the
+    caller can fall back to youtube-transcript-api.
+    """
+    key = os.environ.get("TRANSCRIPTAPI_KEY", "").strip()
+    if not key:
+        raise RuntimeError("TRANSCRIPTAPI_KEY not set")
+    params = {"video_url": video_id, "format": "json", "include_timestamp": "true"}
+    if langs:
+        params["language"] = ",".join(langs)
+    url = TRANSCRIPTAPI_URL + "?" + urllib.parse.urlencode(params)
+    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {key}",
+                                               "Accept": "application/json"})
+    try:
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            data = json.load(resp)
+    except urllib.error.HTTPError as exc:
+        body = exc.read().decode("utf-8", "ignore")
+        if exc.code in (401, 403):
+            raise RuntimeError(f"transcriptapi auth failed (HTTP {exc.code}); "
+                               f"check TRANSCRIPTAPI_KEY. {body[:120]}") from exc
+        if exc.code in (402, 429):
+            raise RuntimeError(f"transcriptapi out of credits / rate-limited "
+                               f"(HTTP {exc.code}). {body[:120]}") from exc
+        raise RuntimeError(f"transcriptapi HTTP {exc.code}: {body[:160]}") from exc
+    snippets = data.get("transcript") or []
+    return [{"text": s.get("text", ""),
+             "start": s.get("start", 0.0),
+             "duration": s.get("duration", 0.0)} for s in snippets]
+
+
 def _fetch_raw(video_id: str, langs: list[str], proxy: str | None) -> list[dict]:
     from youtube_transcript_api import YouTubeTranscriptApi
 
@@ -61,23 +96,37 @@ def get_segments(video_id: str, languages=("en", "en-US", "en-GB"),
                  proxy: str | None = None, attempts: int = 3) -> list[dict]:
     """Return timestamped segments ``[{start, end, text}]`` for ``video_id``.
 
-    Retries transient connection/TLS failures; raises TranscriptError otherwise.
-    ``proxy`` (the YT_PROXY secret) routes around datacenter-IP blocks.
+    Tries transcriptapi.com first (if ``TRANSCRIPTAPI_KEY`` is set — no proxy needed),
+    then falls back to youtube-transcript-api (which uses ``proxy`` to dodge datacenter
+    blocks). Retries transient connection/TLS failures; raises TranscriptError otherwise.
     """
     langs = list(languages)
-    raw, last = None, None
-    for attempt in range(attempts):
+    raw, last, api_err = None, None, None
+
+    # Primary: transcriptapi.com (own domain -> reachable from the Space without a proxy).
+    if os.environ.get("TRANSCRIPTAPI_KEY", "").strip():
         try:
-            raw = _fetch_raw(video_id, langs, proxy)
-            break
+            raw = _fetch_transcriptapi(video_id, langs)
         except Exception as exc:
-            last = exc
-            if _is_transient(exc) and attempt < attempts - 1:
-                time.sleep(1.5 * (attempt + 1))
-                continue
-            raise TranscriptError(_hint(exc)) from exc
+            api_err = exc  # remember, but still try the fallback below
+
+    # Fallback: youtube-transcript-api (needs YT_PROXY from a datacenter IP).
+    if raw is None:
+        for attempt in range(attempts):
+            try:
+                raw = _fetch_raw(video_id, langs, proxy)
+                break
+            except Exception as exc:
+                last = exc
+                if _is_transient(exc) and attempt < attempts - 1:
+                    time.sleep(1.5 * (attempt + 1))
+                    continue
+                break
     if raw is None:
-        raise TranscriptError(_hint(last) if last else "Unknown transcript error.")
+        hint = _hint(last) if last else "Unknown transcript error."
+        if api_err is not None:
+            hint = f"transcriptapi.com: {api_err} | fallback {hint}"
+        raise TranscriptError(hint)
 
     segs = []
     for r in raw:
@@ -103,7 +152,7 @@ def transcript_text(segs: list[dict]) -> str:
 
 def _hint(exc: Exception) -> str:
     name = type(exc).__name__
-    msg = _redact(str(exc))
+    msg = str(exc)
     low = (name + " " + msg).lower()
     # Check specific transcript states first (note: "transcripts" contains "ip").
     if "disabled" in low or "transcriptsdisabled" in low: