From 338573b1e822a1591d662916d7e4b1da53c82a7b Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sun, 21 Jun 2026 16:33:44 -0400 Subject: [PATCH] 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 | 53 +++++++++++++------ .../test_video_analysis_extract_transcript.py | 6 +-- 2 files changed, 39 insertions(+), 20 deletions(-) diff --git a/scripts/video_analysis/extract_transcript.py b/scripts/video_analysis/extract_transcript.py index 81c019ae..f25d6ef8 100644 --- a/scripts/video_analysis/extract_transcript.py +++ b/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]]: - fetched = YouTubeTranscriptApi.get_transcript(video_id) - return [ - {"start": float(s["start"]), "duration": float(s["duration"]), "text": str(s["text"])} - for s in fetched - ] +def _parse_vtt_segments(vtt_path: Path) -> list[dict[str, Any]]: + text = vtt_path.read_text(encoding="utf-8") + segments: list[dict[str, Any]] = [] + pattern = re.compile(r"(\d{2}):(\d{2}):(\d{2})\.(\d{3})\s+-->", re.MULTILINE) + 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) - break + segments = _fetch_via_ytdlp(video_id, output.parent) + 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("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) + return err(make_error("TranscriptFetchError", "fetch", str(last_exc) if last_exc else "no segments")) diff --git a/tests/test_video_analysis_extract_transcript.py b/tests/test_video_analysis_extract_transcript.py index 04812b67..1dfac3f8 100644 --- a/tests/test_video_analysis_extract_transcript.py +++ b/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()