"""Stage 4: fetch a timestamped transcript (no video download). 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 json import os import time import urllib.error import urllib.parse import urllib.request 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 # --- 1.x instance API --- api = None if proxy: try: from youtube_transcript_api.proxies import GenericProxyConfig api = YouTubeTranscriptApi( proxy_config=GenericProxyConfig(http_url=proxy, https_url=proxy)) except Exception: api = YouTubeTranscriptApi() else: api = YouTubeTranscriptApi() if hasattr(api, "fetch"): fetched = api.fetch(video_id, languages=langs) return [{"text": s.text, "start": s.start, "duration": s.duration} for s in fetched] # --- 0.6.x classmethod API --- kwargs = {"languages": langs} if proxy: kwargs["proxies"] = {"http": proxy, "https": proxy} data = YouTubeTranscriptApi.get_transcript(video_id, **kwargs) return [{"text": d["text"], "start": d["start"], "duration": d.get("duration", 0)} for d in data] def _is_transient(exc: Exception) -> bool: s = (type(exc).__name__ + " " + str(exc)).lower() return any(k in s for k in ("ssl", "eof", "timed out", "timeout", "reset", "connection", "max retries", "temporarily")) 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``. 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, 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_transcriptapi(video_id, langs) except Exception as 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: 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: text = (r.get("text") or "").strip() if not text: continue start = float(r.get("start") or 0.0) dur = float(r.get("duration") or 0.0) segs.append({"start": start, "end": start + dur, "text": text}) if not segs: raise TranscriptError("The transcript came back empty for this video.") return segs def transcript_text(segs: list[dict]) -> str: """Render segments as ``[mm:ss] text`` lines for the LLM and the UI preview.""" lines = [] for s in segs: m, sec = divmod(int(s["start"]), 60) lines.append(f"[{m:02d}:{sec:02d}] {s['text']}") return "\n".join(lines) def _hint(exc: Exception) -> str: name = type(exc).__name__ msg = str(exc) low = (name + " " + msg).lower() # Check specific transcript states first (note: "transcripts" contains "ip"). if "disabled" in low or "transcriptsdisabled" in low: return "This video has transcripts/captions disabled — pick another video." if "notranscript" in low or "no transcript" in low: return "No transcript is available for this video in the requested languages." if "unavailable" in low or "videounavailable" in low: return "The video is unavailable (private/removed/region-locked)." if any(k in low for k in ("ssl", "eof", "reset", "connection", "max retries", "timed out", "blocked", "forbidden", "too many", "429")): return ("YouTube blocked the transcript request from this Space's IP " "(datacenter IPs are commonly blocked — seen here as a TLS/SSL reset). " "Set a residential proxy as the YT_PROXY Space secret and retry. " f"[{name}]") return f"Could not fetch transcript: {name}: {msg[:200]}"