conductor(cs229): Phase 1 Acquire - transcript.json (5397 segments via yt-dlp VTT fallback) + video.log (yt-dlp success for 336MB mp4, R5 verified)
Fix extract_transcript.py: YouTubeTranscriptApi.get_transcript() (not .fetch()). youtube-transcript-api v1.2.4 uses class method get_transcript(video_id), not instance .fetch(). R5 mitigation: yt-dlp's VTT auto-sub extraction works where youtube-transcript-api fails (XML parse error on empty response). 5397 segments recovered. Add gitignore patterns for video_analysis artifacts: *.mp4, *.vtt (regenerable). video.log intentionally tracked.
This commit is contained in:
+26993
File diff suppressed because one or more lines are too long
@@ -0,0 +1,10 @@
|
||||
# yt-dlp log
|
||||
# url: https://youtu.be/9vM4p9NN0Ts
|
||||
# output: C:\projects\manual_slop\conductor\tracks\video_analysis_cs229_building_llms_20260621\artifacts\video.mp4
|
||||
# returncode: 0
|
||||
|
||||
stdout:
|
||||
|
||||
|
||||
stderr:
|
||||
|
||||
@@ -0,0 +1,117 @@
|
||||
"""Phase 1 Acquire driver for video_analysis_cs229_building_llms_20260621.
|
||||
|
||||
Strategy: youtube-transcript-api fails for this video (R5: XML parse error on empty response,
|
||||
likely a YouTube API restriction). Fall back to yt-dlp's own subtitle extraction.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[4]
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from scripts.video_analysis.download_video import download_video
|
||||
from scripts.video_analysis.extract_transcript import _fetch_raw_transcript
|
||||
|
||||
URL = "https://youtu.be/9vM4p9NN0Ts"
|
||||
ARTIFACTS = ROOT / "conductor" / "tracks" / "video_analysis_cs229_building_llms_20260621" / "artifacts"
|
||||
ARTIFACTS.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
||||
def _parse_vtt_segments(vtt_path: Path) -> list[dict]:
|
||||
text = vtt_path.read_text(encoding="utf-8")
|
||||
segments: list[dict] = []
|
||||
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 main() -> int:
|
||||
print(f"Phase 1 Acquire for {URL}")
|
||||
print(f"Artifacts: {ARTIFACTS}")
|
||||
|
||||
print("Step 1: extract_transcript (try youtube-transcript-api)")
|
||||
transcript_path = ARTIFACTS / "transcript.json"
|
||||
video_id = "9vM4p9NN0Ts"
|
||||
last_exc = None
|
||||
for attempt in range(3):
|
||||
try:
|
||||
segments = _fetch_raw_transcript(video_id)
|
||||
data = {
|
||||
"video_id": video_id,
|
||||
"segments": segments,
|
||||
"plain": "\n".join(s["text"] for s in segments),
|
||||
"fetched_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
"source": "youtube-transcript-api",
|
||||
}
|
||||
transcript_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
|
||||
print(f" OK: wrote {transcript_path} ({len(segments)} segments)")
|
||||
break
|
||||
except Exception as e:
|
||||
last_exc = e
|
||||
print(f" attempt {attempt+1} failed: {type(e).__name__}: {str(e)[:200]}")
|
||||
if attempt < 2:
|
||||
time.sleep(2 ** attempt)
|
||||
else:
|
||||
print(f" youtube-transcript-api failed after 3 attempts. Falling back to yt-dlp subtitles.")
|
||||
print("Step 1b: yt-dlp subtitle fallback")
|
||||
vtt_path = ARTIFACTS / f"{video_id}.en.vtt"
|
||||
completed = subprocess.run(
|
||||
["yt-dlp", "--write-auto-subs", "--sub-langs", "en", "--sub-format", "vtt",
|
||||
"--skip-download", "--output", str(ARTIFACTS / video_id), URL],
|
||||
capture_output=True, text=True,
|
||||
)
|
||||
candidates = list(ARTIFACTS.glob(f"{video_id}*.vtt"))
|
||||
if not candidates:
|
||||
print(f" yt-dlp subtitle fetch also failed: {completed.stderr[:300]}")
|
||||
print(f" No transcript available. Continuing with download only.")
|
||||
transcript_path.write_text(json.dumps({
|
||||
"video_id": video_id,
|
||||
"segments": [],
|
||||
"plain": "",
|
||||
"fetched_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
"source": "unavailable",
|
||||
"error": str(last_exc)[:500] if last_exc else None,
|
||||
"note": "youtube-transcript-api failed with XML parse error (R5). yt-dlp subtitles also unavailable. Frame OCR will be the primary signal for this video.",
|
||||
}, indent=2, ensure_ascii=False), encoding="utf-8")
|
||||
else:
|
||||
vtt_path = candidates[0]
|
||||
segments = _parse_vtt_segments(vtt_path)
|
||||
data = {
|
||||
"video_id": video_id,
|
||||
"segments": segments,
|
||||
"plain": "\n".join(s["text"] for s in segments),
|
||||
"fetched_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
"source": "yt-dlp-vtt",
|
||||
}
|
||||
transcript_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
|
||||
print(f" OK: parsed {len(segments)} segments from {vtt_path.name}")
|
||||
|
||||
print("Step 2: download_video")
|
||||
video_path = ARTIFACTS / "video.mp4"
|
||||
result = download_video(URL, video_path)
|
||||
if result.is_err():
|
||||
print(f" ERR: {result.err.class_name}: {result.err.detail[:200]}")
|
||||
return 1
|
||||
print(f" OK: wrote {video_path} ({video_path.stat().st_size} bytes)")
|
||||
print(f" log: {result.value['log']}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -70,10 +70,9 @@ def format_transcript_json(video_id: str, segments: list[dict[str, Any]]) -> dic
|
||||
|
||||
|
||||
def _fetch_raw_transcript(video_id: str) -> list[dict[str, Any]]:
|
||||
api = YouTubeTranscriptApi()
|
||||
fetched = api.fetch(video_id)
|
||||
fetched = YouTubeTranscriptApi.get_transcript(video_id)
|
||||
return [
|
||||
{"start": float(s.start), "duration": float(s.duration), "text": str(s.text)}
|
||||
{"start": float(s["start"]), "duration": float(s["duration"]), "text": str(s["text"])}
|
||||
for s in fetched
|
||||
]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user