""" services/transcript_service.py [v2 — cascade + diagnostics] "No transcript extracted" is never one bug. It's a chain, and you need to know WHICH link broke. So this service tries six sources in order and RECORDS what happened at each one: 1. client_override caption segments the browser already fetched (best: the browser holds the OAuth token AND has a residential IP) 2. srt_upload creator pasted / uploaded an .srt or .vtt 3. captions_api official captions.download with the owner's OAuth token ← IP-agnostic. THIS is the one that works on HF Spaces. 4. scrape youtube-transcript-api (+ optional proxy). Works on your laptop, fails on AWS/HF: YouTube blocks datacenter IPs. 5. whisper yt-dlp → ffmpeg → Groq whisper-large-v3 (verbose_json, segment timestamps). ENABLE_WHISPER=1. Note yt-dlp hits the SAME IP block, so it needs a proxy/cookies on a Space. 6. chapters description chapters — not speech, but a real time→text map. Flagged kind="chapters" so nothing ever quotes it as words the creator said. Every attempt lands in `Transcript.attempts` and is surfaced by GET /api/coach/transcript-check/{video_id} — so you see the exact reason in one call instead of guessing. env: TRANSCRIPT_PROXY http(s) proxy for tiers 4/5 TRANSCRIPT_LANGS comma list (default "en,en-US,en-GB,hi,es,pt,fr,de,id") ENABLE_WHISPER "1" to allow the audio-transcription fallback """ from __future__ import annotations import asyncio import os import re import subprocess import tempfile import time from dataclasses import dataclass from typing import Optional import httpx from services import captions_service as caps TRANSCRIPT_PROXY = os.getenv("TRANSCRIPT_PROXY", "").strip() PREFERRED_LANGS = [l.strip() for l in os.getenv( "TRANSCRIPT_LANGS", "en,en-US,en-GB,hi,es,pt,fr,de,id").split(",") if l.strip()] ENABLE_WHISPER = os.getenv("ENABLE_WHISPER", "").strip() in ("1", "true", "yes") GROQ_API_KEY = os.getenv("GROQ_API_KEY", "") WHISPER_MODEL = os.getenv("WHISPER_MODEL", "whisper-large-v3") _CACHE: dict[str, tuple[float, "Transcript"]] = {} _CACHE_TTL = 60 * 60 * 6 # ── Model ─────────────────────────────────────────────────────────────────── @dataclass class Segment: start: float duration: float text: str @property def end(self) -> float: return self.start + self.duration class Transcript: def __init__(self, segments: list[Segment], language: str = "", generated: bool = True, source: str = "none", kind: str = "speech"): self.segments = sorted(segments, key=lambda s: s.start) self.language = language self.generated = generated self.source = source self.kind = kind # "speech" | "chapters" | "none" self.attempts: list[dict] = [] @property def available(self) -> bool: return len(self.segments) > 0 @property def is_speech(self) -> bool: return self.available and self.kind == "speech" @property def word_count(self) -> int: return sum(len(s.text.split()) for s in self.segments) @property def duration(self) -> float: return self.segments[-1].end if self.segments else 0.0 def window(self, start: float, end: float, max_chars: int = 420) -> str: if not self.segments: return "" hits = [s.text for s in self.segments if s.end > start and s.start < end] text = _clean(" ".join(hits)) return text[:max_chars].rstrip() + ("…" if len(text) > max_chars else "") def at(self, t: float, before: float = 6.0, after: float = 3.0) -> str: return self.window(max(0.0, t - before), t + after) def section_at(self, t: float) -> str: """Chapter mode: which section was playing at second t.""" cur = "" for s in self.segments: if s.start <= t: cur = s.text return cur def timeline(self, step: float = 30.0, max_chars_per_row: int = 150, limit: int = 40) -> list[dict]: if not self.segments: return [] out, t = [], 0.0 while t < self.duration and len(out) < limit: txt = self.window(t, t + step, max_chars=max_chars_per_row) if txt: out.append({"t": int(t), "timestamp": fmt_ts(t), "text": txt}) t += step return out def to_dict(self) -> dict: return { "available": self.available, "kind": self.kind, "source": self.source, "language": self.language, "auto_generated": self.generated, "word_count": self.word_count, "segments": len(self.segments), "attempts": self.attempts, } EMPTY = Transcript([], source="none", kind="none") def fmt_ts(seconds: float) -> str: s = int(max(0, seconds)) return f"{s // 60}:{s % 60:02d}" def _clean(text: str) -> str: text = re.sub(r"\[[^\]]{0,30}\]", " ", text) text = (text.replace("\n", " ").replace("'", "'") .replace("&", "&").replace(""", '"')) return re.sub(r"\s+", " ", text).strip() def _segs(raw: list[dict]) -> list[Segment]: return [Segment(float(r.get("start", 0)), float(r.get("duration") or 0), _clean(r.get("text", ""))) for r in raw if _clean(r.get("text", ""))] # ── Tier 4: scrape ────────────────────────────────────────────────────────── def _proxies() -> Optional[dict]: if not TRANSCRIPT_PROXY: return None return {"http": TRANSCRIPT_PROXY, "https": TRANSCRIPT_PROXY} def _scrape_sync(video_id: str) -> list[dict]: from youtube_transcript_api import YouTubeTranscriptApi def pick(listing): for finder in ("find_manually_created_transcript", "find_generated_transcript"): try: return getattr(listing, finder)(PREFERRED_LANGS) except Exception: pass for t in listing: return t raise ValueError("no caption tracks") try: # new instance API (>= 1.0) api = YouTubeTranscriptApi(proxies=_proxies()) if _proxies() else YouTubeTranscriptApi() tr = pick(api.list(video_id)) return [{"start": f.start, "duration": f.duration, "text": f.text} for f in tr.fetch()] except (AttributeError, TypeError): pass # old static API (<= 0.6) tr = pick(YouTubeTranscriptApi.list_transcripts(video_id, proxies=_proxies())) return tr.fetch() # ── Tier 5: audio → Groq Whisper ──────────────────────────────────────────── async def _whisper(video_id: str) -> list[dict]: """ yt-dlp pulls the audio, ffmpeg downsamples to 16kHz mono, Groq whisper-large-v3 returns segment-level timestamps. Chunked at 20 minutes to stay under the upload limit; offsets stitched back on. Needs: pip install yt-dlp + ffmpeg on PATH (packages.txt: ffmpeg) Caveat: yt-dlp hits the SAME YouTube IP block on a Space — give it TRANSCRIPT_PROXY, or run this tier somewhere with a clean IP. """ if not GROQ_API_KEY: raise RuntimeError("GROQ_API_KEY not set") with tempfile.TemporaryDirectory() as tmp: audio = os.path.join(tmp, "a.m4a") cmd = ["yt-dlp", "-f", "bestaudio", "-o", audio, f"https://www.youtube.com/watch?v={video_id}", "--quiet", "--no-warnings"] if TRANSCRIPT_PROXY: cmd += ["--proxy", TRANSCRIPT_PROXY] p = await asyncio.to_thread(subprocess.run, cmd, capture_output=True, text=True, timeout=300) if p.returncode != 0 or not os.path.exists(audio): raise RuntimeError(f"yt-dlp failed: {(p.stderr or '')[:160]}") pattern = os.path.join(tmp, "c%03d.wav") await asyncio.to_thread( subprocess.run, ["ffmpeg", "-i", audio, "-ac", "1", "-ar", "16000", "-f", "segment", "-segment_time", "1200", pattern, "-y", "-loglevel", "error"], capture_output=True, timeout=600) chunks = sorted(f for f in os.listdir(tmp) if f.startswith("c") and f.endswith(".wav")) if not chunks: raise RuntimeError("ffmpeg produced no audio") out: list[dict] = [] async with httpx.AsyncClient(timeout=180.0) as client: for i, name in enumerate(chunks): offset = i * 1200 with open(os.path.join(tmp, name), "rb") as fh: r = await client.post( "https://api.groq.com/openai/v1/audio/transcriptions", headers={"Authorization": f"Bearer {GROQ_API_KEY}"}, files={"file": (name, fh, "audio/wav")}, data={"model": WHISPER_MODEL, "response_format": "verbose_json", "timestamp_granularities[]": "segment"}, ) if r.status_code != 200: raise RuntimeError(f"Groq whisper {r.status_code}: {r.text[:140]}") for s in r.json().get("segments", []): out.append({"start": float(s["start"]) + offset, "duration": float(s["end"]) - float(s["start"]), "text": s.get("text", "")}) return out # ── The cascade ───────────────────────────────────────────────────────────── async def get_transcript(video_id: str, *, override: Optional[list[dict]] = None, srt_text: Optional[str] = None, access_token: Optional[str] = None, description: str = "") -> Transcript: """Never raises. Always explains itself through .attempts.""" if not (override or srt_text): hit = _CACHE.get(video_id) if hit and time.time() - hit[0] < _CACHE_TTL and hit[1].is_speech: return hit[1] attempts: list[dict] = [] def ok(t: Transcript, tier: str, note: str = "") -> Transcript: attempts.append({"tier": tier, "ok": True, "note": note or f"{len(t.segments)} segments"}) t.attempts = attempts _CACHE[video_id] = (time.time(), t) return t def fail(tier: str, why: str) -> None: attempts.append({"tier": tier, "ok": False, "note": why[:220]}) print(f"[transcript] {video_id} · {tier}: {why[:220]}") # 1 ── client-supplied segments (browser already has the token + a clean IP) if override: segs = _segs(override) if segs: return ok(Transcript(segs, "supplied", False, "client_override"), "client_override") fail("client_override", "supplied but empty") # 2 ── pasted / uploaded SRT or VTT if srt_text and srt_text.strip(): try: segs = _segs(caps.parse_srt(srt_text)) if segs: return ok(Transcript(segs, "supplied", False, "srt_upload"), "srt_upload") fail("srt_upload", "parsed to 0 cues — is the file really SRT/VTT?") except Exception as e: fail("srt_upload", f"{type(e).__name__}: {e}") # 3 ── OFFICIAL captions API — the route that works on Hugging Face if access_token: try: srt, snip = await caps.download(video_id, access_token, PREFERRED_LANGS) segs = _segs(caps.parse_srt(srt)) if segs: t = Transcript(segs, snip.get("language", ""), snip.get("trackKind") == "ASR", "captions_api") return ok(t, "captions_api", f"{len(segs)} cues · {snip.get('trackKind')}") fail("captions_api", "downloaded file had no cues") except caps.CaptionsError as e: fail("captions_api", str(e)) except Exception as e: fail("captions_api", f"{type(e).__name__}: {e}") else: fail("captions_api", "no OAuth access_token supplied — the official, IP-proof route is unusable") # 4 ── scrape (expected to fail on AWS/HF without a proxy) try: segs = _segs(await asyncio.to_thread(_scrape_sync, video_id)) if segs: return ok(Transcript(segs, "", True, "scrape"), "scrape") fail("scrape", "no cues returned") except Exception as e: msg = f"{type(e).__name__}: {e}" if any(k in msg.lower() for k in ("block", "ip", "too many", "429", "captcha", "bot", "consent")): msg += " ← YouTube is blocking this server's IP. Expected on Hugging Face/AWS. Use the captions API or set TRANSCRIPT_PROXY." fail("scrape", msg) # 5 ── whisper on the audio if ENABLE_WHISPER: try: segs = _segs(await _whisper(video_id)) if segs: return ok(Transcript(segs, "", True, "whisper"), "whisper") fail("whisper", "no segments") except Exception as e: fail("whisper", f"{type(e).__name__}: {e}") else: fail("whisper", "disabled (ENABLE_WHISPER=1; needs yt-dlp + ffmpeg + a clean IP/proxy)") # 6 ── chapters: not speech, but a real time→text map try: ch = _segs(caps.parse_chapters(description)) if ch: t = Transcript(ch, "", False, "chapters", kind="chapters") return ok(t, "chapters", f"{len(ch)} chapters — LOW FIDELITY, not speech") fail("chapters", "no timestamped chapters in the description") except Exception as e: fail("chapters", str(e)) empty = Transcript([], source="none", kind="none") empty.attempts = attempts return empty def unavailable_reason(t: Transcript) -> str: if t.is_speech: return "" if t.kind == "chapters": return ("No captions available — the Coach is using your description chapters as a coarse time " "map. It can say WHICH SECTION viewers left in, but not the exact words. Turn on " "auto-captions in YouTube Studio for word-level diagnosis.") return ("No transcript could be extracted. Most likely: the 'youtube.force-ssl' scope isn't granted " "(so the official captions API is closed to us) AND YouTube is blocking this server's IP for " "scraping. Reconnecting the channel to grant that scope fixes it with no proxy needed.")