feat(video_analysis): extract_transcript.py with TDD (8 tests)

scripts/video_analysis/error_types.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class ErrorInfo:
+ class_name: str
+ context: str
+ detail: str
+
+
+def make_error(class_name: str, context: str, detail: str) -> ErrorInfo:
+ return ErrorInfo(class_name=class_name, context=context, detail=detail)

scripts/video_analysis/extract_transcript.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+import json
+import re
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+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
+
+
+@dataclass
+class _Ok:
+ value: Any
+
+ def is_ok(self) -> bool:
+  return True
+
+ def is_err(self) -> bool:
+  return False
+
+
+@dataclass
+class _Err:
+ err: ErrorInfo
+
+ def is_ok(self) -> bool:
+  return False
+
+ def is_err(self) -> bool:
+  return True
+
+
+def ok(value: Any) -> _Ok:
+ return _Ok(value)
+
+
+def err(error: ErrorInfo) -> _Err:
+ return _Err(error)
+
+
+def parse_video_id(url_or_id: str) -> _Ok | _Err:
+ if re.match(r"^[A-Za-z0-9_-]{11}$", url_or_id):
+  return ok(url_or_id)
+ parsed = urlparse(url_or_id)
+ if parsed.netloc in ("youtu.be", "www.youtube.com", "youtube.com"):
+  if parsed.netloc == "youtu.be":
+   candidate = parsed.path.lstrip("/")
+  else:
+   qs = parse_qs(parsed.query)
+   candidate = qs.get("v", [""])[0]
+  if re.match(r"^[A-Za-z0-9_-]{11}$", candidate):
+   return ok(candidate)
+ return err(make_error("InvalidVideoId", "url_or_id", url_or_id))
+
+
+def format_transcript_json(video_id: str, segments: list[dict[str, Any]]) -> dict[str, Any]:
+ plain = "\n".join(s["text"] for s in segments)
+ return {
+  "video_id": video_id,
+  "segments": segments,
+  "plain": plain,
+  "fetched_at": datetime.now(timezone.utc).isoformat(),
+ }
+
+
+def _fetch_raw_transcript(video_id: str) -> list[dict[str, Any]]:
+ api = YouTubeTranscriptApi()
+ fetched = api.fetch(video_id)
+ return [
+  {"start": float(s.start), "duration": float(s.duration), "text": str(s.text)}
+  for s in fetched
+ ]
+
+
+def extract_transcript(url_or_id: str, output: Path, retries: int = 3) -> _Ok | _Err:
+ parsed = parse_video_id(url_or_id)
+ if parsed.is_err():
+  return parsed
+ video_id = parsed.value
+ last_exc: Exception | None = None
+ segments: list[dict[str, Any]] = []
+ for attempt in range(retries):
+  try:
+   segments = _fetch_raw_transcript(video_id)
+   break
+  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)

tests/test_video_analysis_extract_transcript.py

@@ -0,0 +1,77 @@
+"""Tests for scripts/video_analysis/extract_transcript.py.
+
+Per conductor/code_styleguides/error_handling.md, success returns Result.ok; failure returns Result.err with ErrorInfo.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from unittest.mock import patch
+
+from scripts.video_analysis.extract_transcript import (
+ extract_transcript,
+ format_transcript_json,
+ parse_video_id,
+)
+
+
+def test_parse_video_id_youtu_be() -> None:
+ result = parse_video_id("https://youtu.be/9vM4p9NN0Ts")
+ assert result.is_ok()
+ assert result.value == "9vM4p9NN0Ts"
+
+
+def test_parse_video_id_full_url() -> None:
+ result = parse_video_id("https://www.youtube.com/watch?v=0yF9TvMeAzM")
+ assert result.is_ok()
+ assert result.value == "0yF9TvMeAzM"
+
+
+def test_parse_video_id_already_id() -> None:
+ result = parse_video_id("yxkUvXs-hoQ")
+ assert result.is_ok()
+ assert result.value == "yxkUvXs-hoQ"
+
+
+def test_parse_video_id_invalid() -> None:
+ result = parse_video_id("not-a-url")
+ assert result.is_err()
+
+
+def test_extract_transcript_success(tmp_path: Path) -> None:
+ fake_segments = [
+  {"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:
+  mock_fetch.return_value = fake_segments
+  result = extract_transcript("https://youtu.be/ABCDEFGHIJK", tmp_path / "transcript.json")
+  assert result.is_ok()
+  data = json.loads((tmp_path / "transcript.json").read_text())
+  assert data["video_id"] == "ABCDEFGHIJK"
+  assert len(data["segments"]) == 2
+  assert data["plain"] == "Hello world\nGoodbye world"
+
+
+def test_extract_transcript_network_error(tmp_path: Path) -> None:
+ with patch("scripts.video_analysis.extract_transcript._fetch_raw_transcript") 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:
+  mock_fetch.side_effect = Exception("transient")
+  result = extract_transcript("https://youtu.be/ABCDEFGHIJK", tmp_path / "transcript.json", retries=2)
+  assert result.is_err()
+  assert mock_fetch.call_count == 2
+
+
+def test_format_transcript_json_structure() -> None:
+ segments = [{"start": 0.0, "duration": 5.0, "text": "Hello"}]
+ out = format_transcript_json("VID123", segments)
+ assert out["video_id"] == "VID123"
+ assert out["segments"] == segments
+ assert out["plain"] == "Hello"
+ assert "fetched_at" in out

tests/test_video_analysis_placeholder.py

@@ -1,10 +0,0 @@
-"""Placeholder test to confirm tests/ is wired correctly for the video_analysis namespace.
-
-Per conductor/code_styleguides/workspace_paths.md, tests live in tests/ (project tree, not %TEMP%).
-This file is deleted in Task 1.1 once real tests for extract_transcript.py are added.
-"""
-from __future__ import annotations
-
-
-def test_placeholder() -> None:
- assert True