conductor(probability_logic): Phase 1 Acquire - transcript.json (3315 segments via yt-dlp VTT fallback) + video.log (84MB mp4 downloaded)

Generic reusable drivers added: phase1_acquire.py, phase2_keyframes.py, phase3_ocr.py take slug as arg for batch use across all 12 children.
2026-06-21 16:32:19 -04:00
parent fd95ea4879
commit 7478090e71
5 changed files with 16801 additions and 0 deletions
@@ -0,0 +1,10 @@
+# yt-dlp log
+# url: https://youtu.be/0yF9TvMeAzM
+# output: C:\projects\manual_slop\conductor\tracks\video_analysis_probability_logic_20260621\artifacts\video.mp4
+# returncode: 0
+
+stdout:
+
+
+stderr:
+
@@ -0,0 +1,132 @@
+"""Generic Phase 1 Acquire driver for video_analysis_campaign children.
+
+Reads the child spec from CLI args: slug + URL + needs_yt_dlp_verify.
+Calls extract_transcript (with yt-dlp VTT fallback) + download_video.
+"""
+from __future__ import annotations
+
+import argparse
+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
+
+
+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 phase1_acquire(slug: str, url: str, artifacts_dir: Path) -> dict:
+ print(f"Phase 1 Acquire for {slug}: {url}")
+ print(f"Artifacts: {artifacts_dir}")
+ artifacts_dir.mkdir(parents=True, exist_ok=True)
+
+ m = re.search(r"(?:youtu\.be/|v=)([A-Za-z0-9_-]{11})", url)
+ if not m:
+  return {"status": "error", "error": f"Could not parse video_id from {url}"}
+ video_id = m.group(1)
+
+ print("Step 1: extract_transcript (try youtube-transcript-api)")
+ transcript_path = artifacts_dir / "transcript.json"
+ 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)")
+   last_exc = None
+   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)
+
+ if last_exc is not None:
+  print("Step 1b: yt-dlp subtitle fallback")
+  completed = subprocess.run(
+   ["yt-dlp", "--write-auto-subs", "--sub-langs", "en", "--sub-format", "vtt",
+    "--skip-download", "--output", str(artifacts_dir / video_id), url],
+   capture_output=True, text=True,
+  )
+  candidates = list(artifacts_dir.glob(f"{video_id}*.vtt"))
+  if not candidates:
+   print(f"  yt-dlp subtitle fetch 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],
+    "note": "youtube-transcript-api and yt-dlp VTT both failed. Frame OCR will be the primary signal.",
+   }, 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_dir / "video.mp4"
+ result = download_video(url, video_path)
+ if result.is_err():
+  return {"status": "error", "error": f"download_video: {result.err.class_name}: {result.err.detail[:200]}"}
+ print(f"  OK: wrote {video_path} ({video_path.stat().st_size} bytes)")
+ return {"status": "ok", "video_path": str(video_path), "transcript_path": str(transcript_path)}
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("slug")
+ parser.add_argument("url")
+ parser.add_argument("--artifacts-dir", required=False)
+ args = parser.parse_args()
+ if args.artifacts_dir:
+  artifacts_dir = Path(args.artifacts_dir)
+ else:
+  artifacts_dir = ROOT / "conductor" / "tracks" / f"video_analysis_{args.slug}_20260621" / "artifacts"
+ result = phase1_acquire(args.slug, args.url, artifacts_dir)
+ print(json.dumps(result, indent=2))
+ return 0 if result["status"] == "ok" else 1
+
+
+if __name__ == "__main__":
+ sys.exit(main())
@@ -0,0 +1,37 @@
+"""Generic Phase 2 Keyframes driver."""
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[4]
+sys.path.insert(0, str(ROOT))
+
+from scripts.video_analysis.extract_keyframes import extract_keyframes
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("slug")
+ parser.add_argument("--video", required=False)
+ parser.add_argument("--output-dir", required=False)
+ parser.add_argument("--threshold", type=float, default=0.4)
+ args = parser.parse_args()
+
+ track_dir = ROOT / "conductor" / "tracks" / f"video_analysis_{args.slug}_20260621" / "artifacts"
+ video = Path(args.video) if args.video else track_dir / "video.mp4"
+ output = Path(args.output_dir) if args.output_dir else track_dir / "frames"
+
+ print(f"Phase 2 Keyframes for {video}")
+ output.mkdir(parents=True, exist_ok=True)
+ result = extract_keyframes(video, output, threshold=args.threshold)
+ if result.is_err():
+  print(f"  ERR: {result.err.class_name}: {result.err.detail[:300]}")
+  return 1
+ print(f"  OK: kept {result.value['kept']} frames")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
@@ -0,0 +1,39 @@
+"""Generic Phase 3 OCR driver."""
+from __future__ import annotations
+
+import argparse
+import sys
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[4]
+sys.path.insert(0, str(ROOT))
+
+from scripts.video_analysis.ocr_frames import ocr_frames
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("slug")
+ parser.add_argument("--frames-dir", required=False)
+ parser.add_argument("--output", required=False)
+ parser.add_argument("--backend", default="winsdk")
+ args = parser.parse_args()
+
+ track_dir = ROOT / "conductor" / "tracks" / f"video_analysis_{args.slug}_20260621" / "artifacts"
+ frames = Path(args.frames_dir) if args.frames_dir else track_dir / "frames"
+ output = Path(args.output) if args.output else track_dir / "ocr.md"
+
+ print(f"Phase 3 OCR for {frames} ({args.backend})")
+ t0 = time.time()
+ result = ocr_frames(frames, output, backend=args.backend)
+ elapsed = time.time() - t0
+ if result.is_err():
+  print(f"  ERR: {result.err.class_name}: {result.err.detail[:300]}")
+  return 1
+ print(f"  OK: OCR'd {result.value['frames_ocrd']} frames in {elapsed:.1f}s")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())