#!/usr/bin/env python3
"""yt-learn: YouTube → structured briefing → ready to build.
Usage:
yt-learn <url> Process a video
yt-learn latest Print path to most recent briefing
yt-learn pin [slug] Pin a briefing so auto-cleanup skips it
yt-learn clean Run cleanup now
yt-learn open Open latest briefing in $EDITOR
"""
import argparse
import datetime as dt
import json
import os
import re
import shutil
import subprocess
import sys
from pathlib import Path
--- Config (override via env vars) ---
LLM routing. Default: OpenRouter (good fit for public YouTube content per your tier rules).
DEFAULT_MODELS = {
"openrouter": "anthropic/claude-3.5-haiku", # cheap fallback; set YT_LEARN_MODEL to your Owl Alpha slug
"anthropic": "claude-sonnet-4-6",
}
---------------- Helpers ----------------
def slugify(text: str, max_len: int = 50) -> str:
text = re.sub(r"[^\w\s-]", "", text or "").strip().lower()
text = re.sub(r"[\s_-]+", "-", text)
return text[:max_len].strip("-") or "untitled"
def fmt_ts(seconds: float) -> str:
s = int(seconds or 0)
if s < 3600:
return f"{s // 60:02d}:{s % 60:02d}"
return f"{s // 3600}:{(s % 3600) // 60:02d}:{s % 60:02d}"
def run(cmd, **kw):
return subprocess.run(cmd, check=True, capture_output=True, text=True, **kw).stdout
def check_deps():
missing = [t for t in ("yt-dlp", "ffmpeg") if not shutil.which(t)]
if missing:
sys.exit(f"Missing dependencies: {', '.join(missing)}. Run install.sh.")
---------------- Fetch + transcribe ----------------
def get_video_meta(url: str) -> dict:
return json.loads(run(["yt-dlp", "--dump-single-json", "--no-warnings", url]))
def fetch_subs(url: str, folder: Path):
"""Try official + auto English subs. Returns VTT path or None."""
try:
run([
"yt-dlp",
"--write-subs", "--write-auto-subs",
"--sub-langs", "en.*,en",
"--sub-format", "vtt",
"--skip-download",
"--no-warnings",
"-o", str(folder / "subs.%(ext)s"),
url,
])
except subprocess.CalledProcessError:
return None
vtts = list(folder.glob("subs*.vtt"))
return vtts[0] if vtts else None
def vtt_to_segments(vtt_path: Path):
"""Parse VTT into [(seconds, text), ...] with simple dedup."""
segments = []
lines = vtt_path.read_text(encoding="utf-8", errors="ignore").splitlines()
i = 0
ts_re = re.compile(r"(\d{1,2}):(\d{2}):(\d{2})(?:\.\d{1,3})?\s*-->")
while i < len(lines):
m = ts_re.match(lines[i].strip())
if m:
h, mn, s = map(int, m.groups())
seconds = h * 3600 + mn * 60 + s
text_parts = []
i += 1
while i < len(lines) and lines[i].strip():
clean = re.sub(r"<[^>]+>", "", lines[i]).strip()
if clean:
text_parts.append(clean)
i += 1
if text_parts:
segments.append((seconds, " ".join(text_parts)))
i += 1
Dedup: drop segments whose text was already emitted by the previous one
def fetch_audio_and_whisper(url: str, folder: Path):
"""Fallback when no subs: download audio, transcribe with faster-whisper."""
try:
from faster_whisper import WhisperModel
except ImportError:
sys.exit("No subs found and faster-whisper not installed.\n"
"Install: pip install --user faster-whisper")
run([
"yt-dlp", "-f", "bestaudio",
"-x", "--audio-format", "mp3",
"--no-warnings",
"-o", str(folder / "audio.%(ext)s"),
url,
])
audio = next(folder.glob("audio.*"), None)
if not audio:
sys.exit("Audio extraction failed.")
print(f" Loading whisper ({WHISPER_SIZE})...")
model = WhisperModel(WHISPER_SIZE, device="cpu", compute_type="int8")
segs, _ = model.transcribe(str(audio), language="en")
out = [(s.start, s.text.strip()) for s in segs if s.text.strip()]
audio.unlink(missing_ok=True)
return out
def write_transcript(segments, folder: Path):
path = folder / "transcript.txt"
path.write_text("\n".join(f"[{fmt_ts(ts)}] {text}" for ts, text in segments),
encoding="utf-8")
return path
---------------- Frames ----------------
def fetch_frames(url: str, folder: Path, max_frames: int = 6):
"""Download lowest-res video, extract scene-change keyframes, delete video."""
frames_dir = folder / "frames"
frames_dir.mkdir(exist_ok=True)
video_path = folder / "video.mp4"
try:
run([
"yt-dlp",
"-f", "worstvideo[ext=mp4]/worst[ext=mp4]/worst",
"--no-warnings",
"-o", str(video_path),
url,
])
except subprocess.CalledProcessError:
return []
if not video_path.exists():
return []
try:
subprocess.run([
"ffmpeg", "-y", "-i", str(video_path),
"-vf", "select='gt(scene,0.4)',scale=640:-1",
"-vsync", "vfr",
"-frames:v", str(max_frames),
str(frames_dir / "frame_%03d.jpg"),
], check=True, capture_output=True)
except subprocess.CalledProcessError:
pass
# If scene-detect found nothing, fall back to N evenly-spaced frames
if not list(frames_dir.glob("frame_*.jpg")):
try:
duration = float(run([
"ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "default=nw=1:nk=1", str(video_path),
]).strip())
step = max(1, int(duration // max_frames))
subprocess.run([
"ffmpeg", "-y", "-i", str(video_path),
"-vf", f"fps=1/{step},scale=640:-1",
"-frames:v", str(max_frames),
str(frames_dir / "frame_%03d.jpg"),
], check=True, capture_output=True)
except Exception:
pass
video_path.unlink(missing_ok=True)
return sorted(frames_dir.glob("frame_*.jpg"))
---------------- LLM brief ----------------
BRIEF_PROMPT = """You are summarizing a YouTube video for a Dubai-based AI builder who watches multiple videos a day to extract buildable ideas. He ships working prototypes, not pitch decks. He values terse bullets, concrete specifics, and timestamps so he can verify or jump back in.
Output strict markdown with ONLY these sections (omit any that are genuinely empty):
TL;DR
Core claims
Mental models / frameworks
Build hooks
Open questions
Rules:
- No "this video discusses..." preambles. Start at TL;DR.
- No padding. No restating obvious things.
- Bullets over prose.
- Quote sparingly (under 15 words) when the exact wording matters.
VIDEO METADATA:
Title: {title}
Channel: {uploader}
Duration: {duration}
URL: {url}
TIMESTAMPED TRANSCRIPT:
{transcript}
"""
def call_llm(transcript: str, meta: dict) -> str:
prompt = BRIEF_PROMPT.format(
title=meta.get("title", "Unknown"),
uploader=meta.get("uploader", "Unknown"),
duration=fmt_ts(meta.get("duration", 0)),
url=meta.get("webpage_url", ""),
transcript=transcript[:400_000], # ~100k token cap
)
if LLM_PROVIDER == "anthropic":
return _call_anthropic(prompt)
return _call_openrouter(prompt)
def _call_anthropic(prompt: str) -> str:
import urllib.request
api_key = os.environ.get("ANTHROPIC_API_KEY")
if not api_key:
sys.exit("ANTHROPIC_API_KEY not set.")
model = LLM_MODEL or DEFAULT_MODELS["anthropic"]
req = urllib.request.Request(
"https://api.anthropic.com/v1/messages",
method="POST",
headers={
"x-api-key": api_key,
"anthropic-version": "2023-06-01",
"content-type": "application/json",
},
data=json.dumps({
"model": model,
"max_tokens": 4000,
"messages": [{"role": "user", "content": prompt}],
}).encode(),
)
with urllib.request.urlopen(req, timeout=120) as r:
data = json.loads(r.read())
return data["content"][0]["text"]
def _call_openrouter(prompt: str) -> str:
import urllib.request
api_key = os.environ.get("OPENROUTER_API_KEY")
if not api_key:
sys.exit("OPENROUTER_API_KEY not set. Set it, or YT_LEARN_LLM=anthropic + ANTHROPIC_API_KEY.")
model = LLM_MODEL or DEFAULT_MODELS["openrouter"]
req = urllib.request.Request(
"https://openrouter.ai/api/v1/chat/completions",
method="POST",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
data=json.dumps({
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 4000,
}).encode(),
)
with urllib.request.urlopen(req, timeout=120) as r:
data = json.loads(r.read())
return data["choices"][0]["message"]["content"]
---------------- Cleanup ----------------
def cleanup():
if not ROOT.exists():
return []
cutoff = dt.datetime.now() - dt.timedelta(days=CLEANUP_DAYS)
removed = []
for folder in ROOT.iterdir():
if not folder.is_dir() or folder.name.startswith("."):
continue
if (folder / ".pinned").exists():
continue
if dt.datetime.fromtimestamp(folder.stat().st_mtime) < cutoff:
shutil.rmtree(folder, ignore_errors=True)
removed.append(folder.name)
return removed
---------------- Commands ----------------
def cmd_process(url: str):
check_deps()
ROOT.mkdir(parents=True, exist_ok=True)
print("→ Fetching metadata...")
meta = get_video_meta(url)
title = meta.get("title") or "untitled"
vid = meta.get("id", "")
date = dt.date.today().isoformat()
slug = f"{date}-{slugify(title)}-{vid[:6]}"
folder = ROOT / slug
folder.mkdir(parents=True, exist_ok=True)
print(f" Folder: {folder}")
print("→ Pulling subs...")
vtt = fetch_subs(url, folder)
if vtt:
segments = vtt_to_segments(vtt)
for f in folder.glob("subs*.vtt"):
f.unlink()
print(f" ✓ {len(segments)} subtitle segments")
else:
print(" ✗ No subs. Falling back to whisper...")
segments = fetch_audio_and_whisper(url, folder)
print(f" ✓ {len(segments)} transcribed segments")
if not segments:
sys.exit("Transcription empty. Bailing.")
transcript_path = write_transcript(segments, folder)
timestamped = transcript_path.read_text()
print("→ Extracting keyframes...")
frames = fetch_frames(url, folder)
print(f" ✓ {len(frames)} frames")
print(f"→ Generating brief via {LLM_PROVIDER}...")
body = call_llm(timestamped, meta)
brief_md = f"""# {title}
**Source:** {meta.get("webpage_url", "")}
**Channel:** {meta.get("uploader", "?")}
**Duration:** {fmt_ts(meta.get("duration", 0))}
**Watched:** {date}
**Frames:** {len(frames)} in `frames/`
**Transcript:** `transcript.txt`
---
{body}
"""
(folder / "brief.md").write_text(brief_md, encoding="utf-8")
(folder / ".meta.json").write_text(json.dumps({
"url": url,
"id": vid,
"title": title,
"uploader": meta.get("uploader"),
"duration": meta.get("duration"),
"watched": date,
}, indent=2))
Update .latest symlink
removed = cleanup()
print()
print(f"✓ Brief written: {folder / 'brief.md'}")
print(f"✓ Latest symlink: {LATEST}")
if removed:
print(f" Cleaned {len(removed)} old briefings")
print()
print("In Claude Code, load with:")
print(f" @~/yt-learn/.latest/brief.md")
print("Then: \"now build me X based on this\"")
def cmd_latest():
if LATEST.exists():
print(LATEST.resolve())
else:
sys.exit("No briefings yet.")
def cmd_pin(slug):
target = ROOT / slug if slug else (LATEST.resolve() if LATEST.exists() else None)
if not target or not target.exists():
sys.exit("Nothing to pin.")
(target / ".pinned").touch()
print(f"✓ Pinned: {target}")
def cmd_clean():
removed = cleanup()
print(f"Removed {len(removed)} folder(s).")
for r in removed:
print(f" - {r}")
def cmd_open():
if not LATEST.exists():
sys.exit("No briefings yet.")
editor = os.environ.get("EDITOR", "less")
subprocess.run([editor, str(LATEST / "brief.md")])
---------------- Main ----------------
def main():
# Shorthand: yt-learn <url>
if len(sys.argv) >= 2 and sys.argv[1].startswith(("http://", "https://")):
return cmd_process(sys.argv[1])
p = argparse.ArgumentParser(prog="yt-learn", description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
sub = p.add_subparsers(dest="cmd")
pp = sub.add_parser("process")
pp.add_argument("url")
pin = sub.add_parser("pin")
pin.add_argument("slug", nargs="?")
sub.add_parser("latest")
sub.add_parser("clean")
sub.add_parser("open")
ns = p.parse_args()
if ns.cmd == "process":
cmd_process(ns.url)
elif ns.cmd == "pin":
cmd_pin(ns.slug)
elif ns.cmd == "latest":
cmd_latest()
elif ns.cmd == "clean":
cmd_clean()
elif ns.cmd == "open":
cmd_open()
else:
p.print_help()
if __name__ == "__main__":
main()