refactor(video_analysis): extract_transcript.py uses yt-dlp VTT directly (skip youtube-transcript-api which consistently fails for these videos)

youtube-transcript-api v1.2.4 returns XML parse error on empty response for ALL videos in this campaign. yt-dlp's --write-auto-subs reliably returns 1000s of segments per video. Switched to yt-dlp as the primary path.

Tests updated to mock _fetch_via_ytdlp instead of _fetch_raw_transcript. 8/8 tests passing.

scripts/video_analysis/extract_transcript.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import json
 import re
+import subprocess
 import time
 from dataclasses import dataclass
 from datetime import datetime, timezone
@@ -9,8 +10,6 @@ from pathlib import Path
 from typing import Any
 from urllib.parse import parse_qs, urlparse
 
-from youtube_transcript_api import YouTubeTranscriptApi
-
 from scripts.video_analysis.error_types import ErrorInfo, make_error
 
 
@@ -69,12 +68,35 @@ def format_transcript_json(video_id: str, segments: list[dict[str, Any]]) -> dic
  }
 
 
-def _fetch_raw_transcript(video_id: str) -> list[dict[str, Any]]:
+def _parse_vtt_segments(vtt_path: Path) -> list[dict[str, Any]]:
- fetched = YouTubeTranscriptApi.get_transcript(video_id)
+ text = vtt_path.read_text(encoding="utf-8")
- return [
+ segments: list[dict[str, Any]] = []
-  {"start": float(s["start"]), "duration": float(s["duration"]), "text": str(s["text"])}
+ pattern = re.compile(r"(\d{2}):(\d{2}):(\d{2})\.(\d{3})\s+-->", re.MULTILINE)
-  for s in fetched
+ blocks = re.split(r"\n\n+", text)
- ]
+ for block in blocks:
+  match = pattern.search(block)
+  if not match:
+   continue
+  h, m, s, ms = match.groups()
+  start = int(h) * 3600 + int(m) * 60 + int(s) + int(ms) / 1000.0
+  lines = [ln.strip() for ln in block.split("\n") if ln.strip() and not pattern.match(ln) and "-->" not in ln]
+  text_content = " ".join(lines)
+  if text_content:
+   segments.append({"start": start, "duration": 0.0, "text": text_content})
+ return segments
+
+
+def _fetch_via_ytdlp(video_id: str, working_dir: Path) -> list[dict[str, Any]]:
+ completed = subprocess.run(
+  ["yt-dlp", "--write-auto-subs", "--sub-langs", "en", "--sub-format", "vtt",
+   "--skip-download", "--output", str(working_dir / video_id),
+   f"https://youtu.be/{video_id}"],
+  capture_output=True, text=True,
+ )
+ candidates = list(working_dir.glob(f"{video_id}*.vtt"))
+ if not candidates:
+  raise RuntimeError(f"yt-dlp VTT fetch failed: {completed.stderr[:300]}")
+ return _parse_vtt_segments(candidates[0])
 
 
 def extract_transcript(url_or_id: str, output: Path, retries: int = 3) -> _Ok | _Err:
@@ -82,19 +104,16 @@ def extract_transcript(url_or_id: str, output: Path, retries: int = 3) -> _Ok |
  if parsed.is_err():
   return parsed
  video_id = parsed.value
+ output.parent.mkdir(parents=True, exist_ok=True)
  last_exc: Exception | None = None
- segments: list[dict[str, Any]] = []
  for attempt in range(retries):
   try:
-   segments = _fetch_raw_transcript(video_id)
+   segments = _fetch_via_ytdlp(video_id, output.parent)
-   break
+   data = format_transcript_json(video_id, segments)
+   output.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
+   return ok(data)
   except Exception as e:
    last_exc = e
    if attempt < retries - 1:
     time.sleep(2 ** attempt)
- if not segments:
+ return err(make_error("TranscriptFetchError", "fetch", str(last_exc) if last_exc else "no segments"))
-  return err(make_error("NetworkError", "fetch", str(last_exc) if last_exc else "no segments"))
- data = format_transcript_json(video_id, segments)
- output.parent.mkdir(parents=True, exist_ok=True)
- output.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
- return ok(data)

tests/test_video_analysis_extract_transcript.py

@@ -43,7 +43,7 @@ def test_extract_transcript_success(tmp_path: Path) -> None:
   {"start": 0.0, "duration": 5.0, "text": "Hello world"},
   {"start": 5.0, "duration": 3.0, "text": "Goodbye world"},
  ]
- with patch("scripts.video_analysis.extract_transcript._fetch_raw_transcript") as mock_fetch:
+ with patch("scripts.video_analysis.extract_transcript._fetch_via_ytdlp") as mock_fetch:
   mock_fetch.return_value = fake_segments
   result = extract_transcript("https://youtu.be/ABCDEFGHIJK", tmp_path / "transcript.json")
   assert result.is_ok()
@@ -54,14 +54,14 @@ def test_extract_transcript_success(tmp_path: Path) -> None:
 
 
 def test_extract_transcript_network_error(tmp_path: Path) -> None:
- with patch("scripts.video_analysis.extract_transcript._fetch_raw_transcript") as mock_fetch:
+ with patch("scripts.video_analysis.extract_transcript._fetch_via_ytdlp") as mock_fetch:
   mock_fetch.side_effect = Exception("network unreachable")
   result = extract_transcript("https://youtu.be/ABCDEFGHIJK", tmp_path / "transcript.json")
   assert result.is_err()
 
 
 def test_extract_transcript_retries_then_fails(tmp_path: Path) -> None:
- with patch("scripts.video_analysis.extract_transcript._fetch_raw_transcript") as mock_fetch:
+ with patch("scripts.video_analysis.extract_transcript._fetch_via_ytdlp") as mock_fetch:
   mock_fetch.side_effect = Exception("transient")
   result = extract_transcript("https://youtu.be/ABCDEFGHIJK", tmp_path / "transcript.json", retries=2)
   assert result.is_err()