| | #!/usr/bin/env python3 | | | """ | | | Build an AI mentor from your favorite creator. | | | | | | Pulls every transcript from a YouTube channel, playlist, or list of videos and | | | writes them as clean, timestamped Markdown files. Drop the output folder into | | | Hippocampus (or any RAG tool) and ask it questions. Every answer can cite the | | | episode and the exact moment it came from. | | | | | | By Brian Jenney (youtube.com/@brianjenney), from the video where he turned | | | every Develop Yourself podcast interview into an AI career mentor. | | | | | | Learn to build RAG systems like this yourself: https://parsity.io | | | Try Hippocampus (20% off for 12 months, code BrianJenney): | | | https://www.thalius.ai/?utm_source=youtube&utm_medium=influeri&utm_campaign=thalius_hippocampus_developer_pilot&utm_content=brianjenney | | | | | | Setup (a virtual environment avoids "externally-managed-environment" errors on Mac): | | | python3 -m venv .venv && source .venv/bin/activate | | | pip install yt-dlp youtube-transcript-api | | | | | | Usage: | | | python build_mentor_corpus.py https://www.youtube.com/@brianjenney | | | python build_mentor_corpus.py https://www.youtube.com/@brianjenney --limit 50 | | | python build_mentor_corpus.py "https://www.youtube.com/playlist?list=PLxxxx" --out my_mentor | | | python build_mentor_corpus.py https://youtu.be/VIDEO_ID https://youtu.be/OTHER_ID | | | | | | Output: | | | mentor_corpus/ | | | 2026-10-28 - Episode title - VIDEO_ID.md (one file per video) | | | ... | | | _manifest.json (what worked, what got skipped) | | | """ | | | | | | import argparse | | | import json | | | import re | | | import sys | | | import time | | | from pathlib import Path | | | | | | try: | | | from yt_dlp import YoutubeDL | | | from youtube_transcript_api import YouTubeTranscriptApi | | | except ImportError: | | | sys.exit( | | | "Missing dependencies. Run:\n" | | | " python3 -m venv .venv && source .venv/bin/activate\n" | | | " pip install yt-dlp youtube-transcript-api" | | | ) | | | | | | | | | # Group caption lines into ~60-second passages. Long enough to carry a full | | | # thought, short enough that a citation points somewhere useful. | | | PASSAGE_SECONDS = 60 | | | | | | | | | def list_videos(urls, limit): | | | """Expand channels/playlists/single videos into a flat list of video metadata.""" |
| | opts = {"quiet": True, "extract_flat": "in_playlist", "skip_download": True} |
| | videos = [] |
| | with YoutubeDL(opts) as ydl: |
| | for url in urls: | | | # A channel URL lists tabs (videos, shorts, live). Ask for long-form videos. | | | if re.search(r"youtube.com/(@|c/|channel/|user/)", url) and not url.rstrip("/").endswith( | | | ("/videos", "/streams", "/shorts") |
| | ): |
| | url = url.rstrip("/") + "/videos" |
| | info = ydl.extract_info(url, download=False) |
| | entries = info.get("entries") or [info] |
| | for e in entries: | | | if not e or not e.get("id"): | | | continue |
| | videos.append({"id": e["id"], "title": e.get("title") or e["id"]}) |
| | if limit and len(videos) >= limit: |
| | return videos | | | return videos | | | | | | | | | def video_details(video_id): | | | """Grab title, channel and upload date for one video (cheap, no download).""" |
| | opts = {"quiet": True, "skip_download": True, "ignore_no_formats_error": True, "no_warnings": True} |
| | with YoutubeDL(opts) as ydl: |
| | info = ydl.extract_info(f"https://www.youtube.com/watch?v={video_id}", download=False) |
| | date = info.get("upload_date") or "" |
| | if len(date) == 8: |
| | date = f"{date[:4]}-{date[4:6]}-{date[6:]}" |
| | return { |
| | "title": info.get("title") or video_id, |
| | "channel": info.get("channel") or info.get("up") or "", |
| | "date": date, | | | "description": (info.get("description") or "").strip(), | | | } | | | | | | | | | def fetch_transcript(video_id, languages): | | | """Return a list of {start, text} caption lines, or None if the video has no captions.""" | | | api = YouTubeTranscriptApi() | | | try: | | | fetched = api.fetch(video_id, languages=languages) | | | except Exception: | | | # Fall back to whatever transcript exists (auto-generated, other language). | | | try: |
| | transcript = next(iter(api.list(video_id))) |
| | fetched = transcript.fetch() |
| | except Exception: | | | return None | | | return [{"start": s.start, "text": s.text} for s in fetched] | | | | | | |
| | def clock(seconds): |
| | seconds = int(seconds) |
| | h, rem = divmod(seconds, 3600) |
| | m, s = divmod(rem, 60) |
| | return f"{h}:{m:02d}:{s:02d}" if h else f"{m}:{s:02d}" |
| | | | | | | | def to_passages(lines): | | | """Merge caption fragments into ~60-second passages with clean text.""" | | | passages, current, start = [], [], None | | | for line in lines: |
| | text = re.sub(r"\[(music\|applause\|laughter)\]", "", line["text"], flags=re.I) |
| | text = re.sub(r"\s+", " ", text.replace("\n", " ")).strip() |
| | if not text: | | | continue | | | if start is None: |
| | start = line["start"] |
| | current.append(text) |
| | if line["start"] - start >= PASSAGE_SECONDS: |
| | passages.append((start, " ".join(current))) |
| | current, start = [], None |
| | if current: | | | passages.append((start, " ".join(current))) | | | return passages | | | | | | |
| | def to_markdown(video_id, meta, passages): |
| | url = f"https://www.youtube.com/watch?v={video_id}" |
| | out = [ |
| | f"# {meta['title']}", |
| | "", |
| | f"- Source: {url}", |
| | f"- Creator: {meta['channel']}" if meta["channel"] else None, |
| | f"- Published: {meta['date']}" if meta["date"] else None, |
| | "", | | | ] | | | if meta["description"]: | | | # The first lines of a description usually name the guest and topic, | | | # which helps retrieval a lot. Skip the link dump below it. |
| | summary = meta["description"].split("\n\n")[0][:600] |
| | out += ["## About this episode", "", summary, ""] |
| | out += ["## Transcript", ""] |
| | for start, text in passages: |
| | stamp = clock(start) |
| | out += [f"### [{stamp}]({url}&t={int(start)}s)", "", text, ""] |
| | return "\n".join(line for line in out if line is not None) | | | | | | |
| | def safe_filename(name): |
| | name = re.sub(r'[<>:"/\\\|?*\x00-\x1f]', "", name) |
| | return re.sub(r"\s+", " ", name).strip()[:90] |
| | | | | | | | def main(): | | | parser = argparse.ArgumentParser( | | | description="Turn a YouTube creator's videos into a timestamped corpus for RAG." | | | ) | | | parser.add_argument("urls", nargs="+", help="Channel, playlist, or video URLs") |
| | parser.add_argument("--out", default="mentor_corpus", help="Output folder (default: mentor_corpus)") |
| | parser.add_argument("--limit", type=int, default=0, help="Max videos to pull (default: all)") |
| | parser.add_argument("--lang", default="en", help="Preferred caption languages, comma separated (default: en)") |
| | parser.add_argument("--delay", type=float, default=1.0, help="Seconds between videos, be polite (default: 1)") |
| | args = parser.parse_args() |
| | |
| | out_dir = Path(args.out) |
| | out_dir.mkdir(parents=True, exist_ok=True) |
| | languages = [l.strip() for l in args.lang.split(",") if l.strip()] |
| | |
| | print("Finding videos...") |
| | videos = list_videos(args.urls, args.limit) |
| | print(f"Found {len(videos)} videos. Pulling transcripts.\n") |
| | |
| | manifest = {"saved": [], "skipped": []} |
| | for i, v in enumerate(videos, 1): |
| | vid = v["id"] |
| | existing = list(out_dir.glob(f"*{vid}.md")) |
| | if existing: |
| | print(f"[{i}/{len(videos)}] already have {v['title']}") |
| | manifest["saved"].append({"id": vid, "file": existing[0].name}) |
| | continue | | | | | | lines = fetch_transcript(vid, languages) | | | if not lines: |
| | print(f"[{i}/{len(videos)}] no captions, skipping: {v['title']}") |
| | manifest["skipped"].append({"id": vid, "title": v["title"], "reason": "no captions"}) |
| | continue | | | | | | try: | | | meta = video_details(vid) | | | except Exception: | | | meta = {"title": v["title"], "channel": "", "date": "", "description": ""} | | | |
| | md = to_markdown(vid, meta, to_passages(lines)) |
| | prefix = f"{meta['date']} - " if meta["date"] else "" |
| | path = out_dir / f"{prefix}{safe_filename(meta['title'])} - {vid}.md" |
| | path.write_text(md, encoding="utf-8") |
| | manifest["saved"].append({"id": vid, "file": path.name}) |
| | print(f"[{i}/{len(videos)}] saved {path.name}") |
| | time.sleep(args.delay) |
| | | | | (out_dir / "_manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8") | | | | | | print( |
| | f"\nDone. {len(manifest['saved'])} transcripts in ./{out_dir}/, " |
| | f"{len(manifest['skipped'])} skipped (see _manifest.json)." |
| | "\n\nNext: upload the folder to Hippocampus (or your own RAG pipeline) and start asking questions." | | | "\nWant to learn how to build the retrieval side yourself? https://parsity.io" | | | ) | | | | | | |
| | if __name__ == "__main__": |
| | main() |