diff --git a/README.md b/README.md index e524a27..4c1823d 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ **The AI world reinvents itself every month. This Claude Code skill keeps you current.** /last30days researches your topic across Reddit, X, and the web from the last 30 days, finds what the community is actually upvoting and sharing, and writes you a prompt that works today, not six months ago. Whether it's Ralph Wiggum loops, Suno music prompts, or the latest Midjourney techniques, you'll prompt like someone who's been paying attention. -**New in V2.1:** X search is now fully bundled - no external `bird` CLI or xAI API key needed for X. Just have Node.js 22+ installed. Uses a vendored subset of Bird's Twitter GraphQL client (MIT licensed, originally by [@steipete](https://x.com/steipete)). +**New in V2.1:** X search is now fully bundled - no external `bird` CLI or xAI API key needed for X. Just have Node.js 22+ installed. Uses a vendored subset of Bird's Twitter GraphQL client (MIT licensed, originally by [@steipete](https://x.com/steipete)). **YouTube search** is now a 4th source - automatically searches YouTube and extracts transcripts via yt-dlp when installed. Inspired by [@steipete](https://x.com/steipete)'s yt-dlp + [summarize](https://github.com/steipete/summarize) toolchain approach. **New in V2:** Dramatically better search results. Smarter query construction finds posts that V1 missed entirely, and a new two-phase search automatically discovers key @handles and subreddits from initial results, then drills deeper. Free X search (no xAI key needed), `--days=N` for flexible lookback, and automatic model fallback. [Full changelog below.](#whats-new-in-v2) @@ -54,7 +54,7 @@ Examples: ## What It Does -1. **Researches** - Scans Reddit and X for discussions from the last 30 days +1. **Researches** - Scans Reddit, X, and YouTube for discussions from the last 30 days 2. **Synthesizes** - Identifies patterns, best practices, and what actually works 3. **Delivers** - Either writes copy-paste-ready prompts for your target tool, or gives you a curated expert-level answer @@ -774,8 +774,8 @@ This example shows /last30days discovering **emerging developer workflows** - re | Flag | Description | |------|-------------| | `--days=N` | Look back N days instead of 30 (e.g., `--days=7` for weekly roundup) | -| `--quick` | Faster research, fewer sources (8-12 each), skips supplemental search | -| `--deep` | Comprehensive research (50-70 Reddit, 40-60 X) with extended supplemental | +| `--quick` | Faster research, fewer sources (8-12 each), skips supplemental search. YouTube: 10 videos, 3 transcripts | +| `--deep` | Comprehensive research (50-70 Reddit, 40-60 X) with extended supplemental. YouTube: 40 videos, 8 transcripts | | `--debug` | Verbose logging for troubleshooting | | `--sources=reddit` | Reddit only | | `--sources=x` | X only | @@ -786,8 +786,9 @@ This example shows /last30days discovering **emerging developer workflows** - re - **Node.js 22+** - For X search (bundled Twitter GraphQL client) - **X session** - Be logged into x.com in your browser, or set `AUTH_TOKEN`/`CT0` env vars - **xAI API key** (optional fallback) - If the bundled search can't authenticate, falls back to xAI's Grok API +- **yt-dlp** (optional) - For YouTube search + transcript extraction. Install via `brew install yt-dlp` or `pip install yt-dlp`. When present, automatically searches YouTube and extracts video transcripts as a 4th source. -At least one API key is required. X search works automatically if you're logged into x.com in your browser. +At least one API key is required. X search works automatically if you're logged into x.com in your browser. YouTube search activates automatically when yt-dlp is in your PATH. ## How It Works @@ -796,6 +797,7 @@ At least one API key is required. X search works automatically if you're logged **Phase 1: Broad discovery** - OpenAI Responses API with `web_search` tool scoped to reddit.com - Vendored Twitter GraphQL search (or xAI API fallback) for X search +- YouTube search + transcript extraction via yt-dlp (when installed) - WebSearch for blogs, news, docs, tutorials - Reddit JSON enrichment for real engagement metrics (upvotes, comments) - Scoring algorithm weighing recency, relevance, and engagement @@ -828,6 +830,14 @@ V2 finds significantly more content than V1. Two major improvements: **Reddit JSON enrichment** - Fetches real upvote and comment counts from Reddit's free API for every thread, giving you actual engagement signals instead of estimates. +### YouTube search with transcripts (v2.1) + +**YouTube is now a 4th research source.** When yt-dlp is installed (`brew install yt-dlp`), /last30days automatically searches YouTube for your topic, fetches view counts and engagement data, and extracts auto-generated transcripts from the top videos. Transcripts give the synthesis engine actual content to work with — not just titles. + +YouTube items go through the same scoring pipeline (relevance + recency + engagement) and are deduped, scored, and rendered alongside Reddit and X results. Views dominate YouTube's engagement formula since they're the primary discovery signal. + +Inspired by [Peter Steinberger](https://x.com/steipete)'s yt-dlp + [summarize](https://github.com/steipete/summarize) toolchain. Peter's approach of combining yt-dlp for search/metadata with transcript extraction for content analysis was the direct inspiration for this feature. + ### Bundled X search (v2.1) **X search is fully self-contained** - No external `bird` CLI or xAI API key needed. /last30days bundles a vendored subset of Bird's Twitter GraphQL client (MIT licensed, by Peter Steinberger). Just be logged into x.com in your browser and it auto-detects your session. Falls back to xAI API if bundled search can't authenticate. diff --git a/SKILL.md b/SKILL.md index ea1a9d0..0e0670c 100644 --- a/SKILL.md +++ b/SKILL.md @@ -1,14 +1,14 @@ --- name: last30days version: "2.1" -description: Research a topic from the last 30 days on Reddit + X + Web, become an expert, and write copy-paste-ready prompts for the user's target tool. +description: Research a topic from the last 30 days on Reddit + X + YouTube + Web, become an expert, and write copy-paste-ready prompts for the user's target tool. argument-hint: 'nano banana pro prompts, NVIDIA news, best AI video tools' allowed-tools: Bash, Read, Write, AskUserQuestion, WebSearch --- # last30days v2.1: Research Any Topic from the Last 30 Days -Research ANY topic across Reddit, X, and the web. Surface what people are actually discussing, recommending, and debating right now. +Research ANY topic across Reddit, X, YouTube, and the web. Surface what people are actually discussing, recommending, and debating right now. ## CRITICAL: Parse User Intent @@ -120,10 +120,11 @@ For ALL query types: The Judge Agent must: 1. Weight Reddit/X sources HIGHER (they have engagement signals: upvotes, likes) -2. Weight WebSearch sources LOWER (no engagement data) -3. Identify patterns that appear across ALL three sources (strongest signals) -4. Note any contradictions between sources -5. Extract the top 3-5 actionable insights +2. Weight YouTube sources HIGH (they have views, likes, and transcript content) +3. Weight WebSearch sources LOWER (no engagement data) +4. Identify patterns that appear across ALL sources (strongest signals) +5. Note any contradictions between sources +6. Extract the top 3-5 actionable insights **Do NOT display stats here - they come at the end, right before the invitation.** @@ -204,7 +205,8 @@ CITATION RULE: Cite sources sparingly to prove research is real. CITATION PRIORITY (most to least preferred): 1. @handles from X — "per @handle" (these prove the tool's unique value) 2. r/subreddits from Reddit — "per r/subreddit" -3. Web sources — ONLY when Reddit/X don't cover that specific fact +3. YouTube channels — "per [channel name] on YouTube" (transcript-backed insights) +4. Web sources — ONLY when Reddit/X/YouTube don't cover that specific fact The tool's value is surfacing what PEOPLE are saying, not what journalists wrote. When both a web article and an X post cover the same fact, cite the X post. @@ -253,12 +255,14 @@ KEY PATTERNS from the research: ✅ All agents reported back! ├─ 🟠 Reddit: {N} threads │ {N} upvotes │ {N} comments ├─ 🔵 X: {N} posts │ {N} likes │ {N} reposts +├─ 🔴 YouTube: {N} videos │ {N} views │ {N} with transcripts ├─ 🌐 Web: {N} pages (supplementary) └─ 🗣️ Top voices: @{handle1} ({N} likes), @{handle2} │ r/{sub1}, r/{sub2} --- ``` If Reddit returned 0 threads, write: "├─ 🟠 Reddit: 0 threads (no results this cycle)" +If YouTube returned 0 videos or yt-dlp is not installed, omit the YouTube line entirely. NEVER use plain text dashes (-) or pipe (|). ALWAYS use ├─ └─ │ and the emoji. **SELF-CHECK before displaying**: Re-read your "What I learned" section. Does it match what the research ACTUALLY says? If you catch yourself projecting your own knowledge instead of the research, rewrite it. @@ -420,7 +424,7 @@ After delivering a prompt, end with: ``` --- 📚 Expert in: {TOPIC} for {TARGET_TOOL} -📊 Based on: {n} Reddit threads ({sum} upvotes) + {n} X posts ({sum} likes) + {n} web pages +📊 Based on: {n} Reddit threads ({sum} upvotes) + {n} X posts ({sum} likes) + {n} YouTube videos ({sum} views) + {n} web pages Want another prompt? Just tell me what you're creating next. ``` diff --git a/docs/plans/2026-02-14-feat-youtube-transcript-search-plan.md b/docs/plans/2026-02-14-feat-youtube-transcript-search-plan.md new file mode 100644 index 0000000..0f8fd77 --- /dev/null +++ b/docs/plans/2026-02-14-feat-youtube-transcript-search-plan.md @@ -0,0 +1,315 @@ +--- +title: "feat: Add YouTube transcript search as 4th source" +type: feat +date: 2026-02-14 +--- + +# feat: Add YouTube Transcript Search + +## Overview + +Add YouTube as a 4th research source alongside Reddit, X, and Web. Search for recent videos on the user's topic, fetch transcripts from the top results, and feed the transcript text into the synthesis — giving the Judge Agent access to what people are *saying* in video form, not just what they're posting on social media. + +**Why this matters:** For many topics (tutorials, product reviews, drama breakdowns), the best content lives on YouTube, not Reddit or X. A 20-minute video review contains 10x the signal of a tweet. The skill currently misses all of it. + +## Proposed Solution + +Use **yt-dlp** (already installed via Homebrew) for both YouTube search and transcript extraction. No new API keys, no new dependencies. Follows the same "zero friction" philosophy as vendored Bird search. + +### Two-step process per research run: + +1. **Search**: `yt-dlp "ytsearch{N}:{topic}" --dateafter {30d_ago} --flat-playlist --print` → top videos by view count +2. **Transcripts**: For top 5 videos, extract auto-generated subtitles via `yt-dlp --write-auto-subs --skip-download`, clean VTT to plaintext in Python + +### Why NOT use `summarize` CLI: + +- Adds 146MB brew dependency (arm64-only binary) +- Calls OpenAI API per video ($0.01-0.03 each) — adds cost on top of existing API usage +- yt-dlp already extracts raw transcripts for free (covers ~95% of videos with auto-captions) +- Raw transcripts are better for synthesis anyway — the LLM doing synthesis (Claude) should interpret the content itself, not get a pre-summarized version + +`summarize` is a great standalone tool, but for integration into a research pipeline where an LLM already synthesizes everything, raw transcripts are the right input. + +## Technical Approach + +### Architecture + +New file: `scripts/lib/youtube_yt.py` (mirrors `bird_x.py` pattern) + +``` +yt-dlp search → metadata (title, views, channel, date) + ↓ + sort by views, take top N + ↓ +yt-dlp subtitle extraction → raw VTT files + ↓ + VTT cleanup → plaintext transcripts + ↓ + truncate to ~500 words per video + ↓ + normalize → YouTubeItem objects + ↓ + score, dedupe, render (same pipeline as Reddit/X) +``` + +### Implementation Phases + +#### Phase 1: Search + Metadata (the fast part) + +**New file: `scripts/lib/youtube_yt.py`** + +Core search function: +```python +def search_youtube(topic: str, from_date: str, to_date: str, depth: str = "default") -> Dict[str, Any]: + """Search YouTube via yt-dlp. No API key needed. + + Returns: + Dict with 'items' list of video metadata dicts. + """ + count = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) + date_filter = from_date.replace("-", "") # YYYYMMDD format + + # yt-dlp search with metadata extraction + cmd = [ + "yt-dlp", + f"ytsearch{count}:{topic}", + "--dateafter", date_filter, + "--flat-playlist", + "--print", "%(view_count)s\t%(id)s\t%(title)s\t%(channel)s\t%(upload_date)s\t%(like_count)s\t%(comment_count)s", + ] + result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + + # Parse tab-separated output, sort by views, return top N + ... +``` + +Depth config (matches existing pattern): +```python +DEPTH_CONFIG = { + "quick": 10, # search 10, transcript top 3 + "default": 20, # search 20, transcript top 5 + "deep": 40, # search 40, transcript top 8 +} + +TRANSCRIPT_LIMITS = { + "quick": 3, + "default": 5, + "deep": 8, +} +``` + +**Key detail**: `yt-dlp --flat-playlist` returns exit code 0 with empty stdout when `--dateafter` filters out everything. Check for empty output, not error codes. + +#### Phase 2: Transcript Extraction (the slow part) + +For top N videos (by view count), fetch transcripts: + +```python +def fetch_transcript(video_id: str, temp_dir: str) -> Optional[str]: + """Fetch auto-generated transcript for a YouTube video. + + Returns: + Plaintext transcript string, or None if no captions available. + """ + cmd = [ + "yt-dlp", + "--write-auto-subs", + "--sub-lang", "en", + "--sub-format", "vtt", + "--skip-download", + "-o", f"{temp_dir}/%(id)s", + f"https://www.youtube.com/watch?v={video_id}", + ] + subprocess.run(cmd, capture_output=True, text=True, timeout=30) + + vtt_path = Path(temp_dir) / f"{video_id}.en.vtt" + if not vtt_path.exists(): + return None + + return _clean_vtt(vtt_path.read_text()) +``` + +VTT cleanup (~10 lines of Python): +```python +def _clean_vtt(vtt_text: str) -> str: + """Convert VTT subtitle format to clean plaintext.""" + text = re.sub(r'^WEBVTT.*?\n\n', '', vtt_text, flags=re.DOTALL) + text = re.sub(r'\d{2}:\d{2}:\d{2}\.\d{3} --> \d{2}:\d{2}:\d{2}\.\d{3}.*\n', '', text) + text = re.sub(r'<[^>]+>', '', text) + lines = text.strip().split('\n') + seen = set() + unique = [] + for line in lines: + stripped = line.strip() + if stripped and stripped not in seen: + seen.add(stripped) + unique.append(stripped) + return re.sub(r'\s+', ' ', ' '.join(unique)).strip() +``` + +**Parallelization**: Run transcript fetches in parallel using ThreadPoolExecutor (same pattern as Phase 2 supplemental searches for Reddit/X): + +```python +def fetch_transcripts_parallel(video_ids: List[str], max_workers: int = 5) -> Dict[str, Optional[str]]: + """Fetch transcripts for multiple videos in parallel.""" + with tempfile.TemporaryDirectory() as temp_dir: + with ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = { + executor.submit(fetch_transcript, vid, temp_dir): vid + for vid in video_ids + } + results = {} + for future in as_completed(futures): + vid = futures[future] + results[vid] = future.result() + return results +``` + +#### Phase 3: Integration into Pipeline + +**Update `scripts/lib/schema.py`** — add YouTubeItem: +```python +@dataclass +class YouTubeItem: + id: str # video_id + title: str + url: str + channel_name: str + date: Optional[str] + date_confidence: str # always "high" for YouTube + engagement: Engagement # views, likes, comments + transcript_snippet: str # first ~500 words of transcript + relevance: float + why_relevant: str + subs: Optional[SubScores] = None + score: int = 0 +``` + +Update `Report` to add: +```python +youtube: List[YouTubeItem] = field(default_factory=list) +youtube_error: Optional[str] = None +``` + +**Update `scripts/lib/score.py`** — YouTube-specific engagement weights: +```python +def compute_youtube_engagement_raw(views, likes, comments): + """YouTube engagement: views dominate, likes secondary, comments tertiary.""" + return ( + 0.50 * math.log1p(views or 0) + + 0.35 * math.log1p(likes or 0) + + 0.15 * math.log1p(comments or 0) + ) +``` + +**Update `scripts/last30days.py`** — add YouTube to ThreadPoolExecutor: +```python +with ThreadPoolExecutor(max_workers=3) as executor: # was 2 + if run_reddit: + reddit_future = executor.submit(_search_reddit, ...) + if run_x: + x_future = executor.submit(_search_x, ...) + if run_youtube: + youtube_future = executor.submit(_search_youtube, ...) +``` + +**Update `scripts/lib/render.py`** — YouTube section in compact output: +``` +### YouTube Videos + +**{id}** (score:{score}) {channel_name} ({date}) [{views} views, {likes} likes] + {title} + https://www.youtube.com/watch?v={id} + {transcript_snippet[:200]}... + *{why_relevant}* +``` + +**Update `scripts/lib/env.py`** — YouTube availability detection: +```python +def is_ytdlp_available() -> bool: + return shutil.which("yt-dlp") is not None +``` + +No API key needed. YouTube search is available whenever yt-dlp is in PATH. + +#### Phase 4: SKILL.md Updates + +Stats box adds YouTube line: +``` +├─ 🎥 YouTube: {N} videos │ {N} views │ {N} transcripts +``` + +Citation priority updated: +``` +1. @handles from X +2. YouTube creators — "per [Channel Name] on YouTube" +3. r/subreddits from Reddit +4. Web sources +``` + +Synthesis instructions updated to weight YouTube transcripts highly — a 20-minute video transcript with 500K views is a stronger signal than a tweet with 50 likes. + +## Acceptance Criteria + +- [x] `yt-dlp` search returns videos matching topic within date range +- [x] Transcripts extracted for top N videos (auto-generated captions) +- [x] Videos without captions gracefully skipped (no error) +- [x] YouTube results appear in compact output with engagement metrics +- [x] YouTube items scored and ranked alongside Reddit/X items +- [x] YouTube auto-activates when yt-dlp is available (no --sources flag needed) +- [x] SKILL.md stats box includes YouTube line +- [x] Transcript snippets (first ~500 words) included in output for LLM synthesis +- [ ] Total YouTube search + transcript extraction completes within 30 seconds +- [x] Works when yt-dlp is not installed (graceful degradation, no crash) +- [ ] Mock mode works for testing without network + +## Dependencies & Risks + +**Dependencies:** +- `yt-dlp` (Homebrew) — already installed, widely available via brew/pip/standalone +- No API keys needed +- No new Python packages (just subprocess + regex) + +**Risks:** +| Risk | Likelihood | Mitigation | +|------|-----------|------------| +| yt-dlp search is slow (>10s) | Medium | Set 30s timeout, run in parallel with Reddit/X | +| YouTube blocks yt-dlp | Low | yt-dlp is actively maintained with anti-bot updates. Degrade gracefully. | +| Videos lack auto-captions | Medium (~5%) | Skip those videos, note in output. Transcript is enrichment, not required. | +| Transcript extraction adds latency | High | Only fetch top 3-5, run in parallel, use tempdir | +| yt-dlp not installed for some users | Medium | Auto-detect, skip YouTube with info message, don't error | +| Linux `--dateafter` date format differs | Low | Use Python to format date, not shell `date -v` | + +## Files to Create/Modify + +### New Files +- `scripts/lib/youtube_yt.py` — search, transcript extraction, parsing +- `tests/test_youtube_yt.py` — unit tests +- `fixtures/youtube_sample.json` — mock data for tests + +### Modified Files +- `scripts/lib/schema.py` — add YouTubeItem, update Report +- `scripts/lib/normalize.py` — add normalize_youtube_items() +- `scripts/lib/score.py` — add YouTube engagement scoring +- `scripts/lib/dedupe.py` — add YouTube dedup (title + channel Jaccard) +- `scripts/lib/render.py` — add YouTube section to compact + full report +- `scripts/lib/env.py` — add yt-dlp availability check, update source detection +- `scripts/last30days.py` — add _search_youtube(), update run_research(), update arg parser +- `SKILL.md` — update stats box, citation rules, synthesis instructions +- `README.md` — document YouTube source, yt-dlp requirement + +## Alternative Approaches Considered + +**1. YouTube Data API v3** — Rejected. Requires API key + Google Cloud project. Adds friction, counter to "zero config" philosophy. 10K quota/day limit. yt-dlp has no limits. + +**2. steipete/summarize for transcripts** — Rejected for MVP. Adds 146MB dependency, requires brew tap, calls OpenAI API per video (adds cost). Raw transcripts via yt-dlp are better input for our synthesis LLM anyway. Could revisit as optional enhancement for captionless videos. + +**3. youtube-transcript-api Python package** — Considered. Lightweight, Python-native transcript fetcher. But adds a pip dependency to a project that currently has zero Python deps. yt-dlp is already a brew dependency we can auto-detect. + +**4. Skip transcripts, just use metadata** — Rejected. Titles + view counts alone don't give the synthesis LLM enough to work with. Transcripts are what make YouTube a *research* source vs just a link list. + +## Cost Impact + +**Zero additional API cost.** yt-dlp scrapes YouTube directly. No API keys, no token usage. The only cost is the existing OpenAI/xAI calls for Reddit/X search, which are unchanged. + +**Time impact:** Adds ~10-20 seconds to research (search + parallel transcript extraction), running in parallel with Reddit/X so effective wall-clock increase is minimal. diff --git a/scripts/last30days.py b/scripts/last30days.py index 94fb0cd..aefbf9e 100644 --- a/scripts/last30days.py +++ b/scripts/last30days.py @@ -43,6 +43,7 @@ from lib import ( ui, websearch, xai_x, + youtube_yt, ) @@ -206,6 +207,34 @@ def _search_x( return x_items, raw_response, x_error +def _search_youtube( + topic: str, + from_date: str, + to_date: str, + depth: str, +) -> tuple: + """Search YouTube via yt-dlp (runs in thread). + + Returns: + Tuple of (youtube_items, youtube_error) + """ + youtube_error = None + + try: + response = youtube_yt.search_and_transcribe( + topic, from_date, to_date, depth=depth, + ) + except Exception as e: + return [], f"{type(e).__name__}: {e}" + + youtube_items = youtube_yt.parse_youtube_response(response) + + if response.get("error"): + youtube_error = response["error"] + + return youtube_items, youtube_error + + def _run_supplemental( topic: str, reddit_items: list, @@ -340,22 +369,25 @@ def run_research( mock: bool = False, progress: ui.ProgressDisplay = None, x_source: str = "xai", + run_youtube: bool = False, ) -> tuple: """Run the research pipeline. Returns: - Tuple of (reddit_items, x_items, web_needed, raw_openai, raw_xai, raw_reddit_enriched, reddit_error, x_error) + Tuple of (reddit_items, x_items, youtube_items, web_needed, raw_openai, raw_xai, raw_reddit_enriched, reddit_error, x_error, youtube_error) Note: web_needed is True when WebSearch should be performed by Claude. The script outputs a marker and Claude handles WebSearch in its session. """ reddit_items = [] x_items = [] + youtube_items = [] raw_openai = None raw_xai = None raw_reddit_enriched = [] reddit_error = None x_error = None + youtube_error = None # Check if WebSearch is needed (always needed in web-only mode) web_needed = sources in ("all", "web", "reddit-web", "x-web") @@ -365,19 +397,35 @@ def run_research( if progress: progress.start_web_only() progress.end_web_only() - return reddit_items, x_items, True, raw_openai, raw_xai, raw_reddit_enriched, reddit_error, x_error + # Still run YouTube in web-only mode if yt-dlp is available + if run_youtube: + if progress: + progress.start_youtube() + try: + youtube_items, youtube_error = _search_youtube(topic, from_date, to_date, depth) + if youtube_error and progress: + progress.show_error(f"YouTube error: {youtube_error}") + except Exception as e: + youtube_error = f"{type(e).__name__}: {e}" + if progress: + progress.show_error(f"YouTube error: {e}") + if progress: + progress.end_youtube(len(youtube_items)) + return reddit_items, x_items, youtube_items, True, raw_openai, raw_xai, raw_reddit_enriched, reddit_error, x_error, youtube_error # Determine which searches to run - run_reddit = sources in ("both", "reddit", "all", "reddit-web") - run_x = sources in ("both", "x", "all", "x-web") + do_reddit = sources in ("both", "reddit", "all", "reddit-web") + do_x = sources in ("both", "x", "all", "x-web") - # Run Reddit and X searches in parallel + # Run Reddit, X, and YouTube searches in parallel reddit_future = None x_future = None + youtube_future = None + max_workers = 2 + (1 if run_youtube else 0) - with ThreadPoolExecutor(max_workers=2) as executor: - # Submit both searches - if run_reddit: + with ThreadPoolExecutor(max_workers=max_workers) as executor: + # Submit searches + if do_reddit: if progress: progress.start_reddit() reddit_future = executor.submit( @@ -385,7 +433,7 @@ def run_research( from_date, to_date, depth, mock ) - if run_x: + if do_x: if progress: progress.start_x() x_future = executor.submit( @@ -393,6 +441,13 @@ def run_research( from_date, to_date, depth, mock, x_source ) + if run_youtube: + if progress: + progress.start_youtube() + youtube_future = executor.submit( + _search_youtube, topic, from_date, to_date, depth + ) + # Collect results if reddit_future: try: @@ -418,6 +473,18 @@ def run_research( if progress: progress.end_x(len(x_items)) + if youtube_future: + try: + youtube_items, youtube_error = youtube_future.result() + if youtube_error and progress: + progress.show_error(f"YouTube error: {youtube_error}") + except Exception as e: + youtube_error = f"{type(e).__name__}: {e}" + if progress: + progress.show_error(f"YouTube error: {e}") + if progress: + progress.end_youtube(len(youtube_items)) + # Enrich Reddit items with real data (sequential, but with error handling per-item) if reddit_items: if progress: @@ -455,7 +522,7 @@ def run_research( if sup_x: x_items.extend(sup_x) - return reddit_items, x_items, web_needed, raw_openai, raw_xai, raw_reddit_enriched, reddit_error, x_error + return reddit_items, x_items, youtube_items, web_needed, raw_openai, raw_xai, raw_reddit_enriched, reddit_error, x_error, youtube_error def main(): @@ -543,6 +610,9 @@ def main(): x_source_status = env.get_x_source_status(config) x_source = x_source_status["source"] # 'bird', 'xai', or None + # Auto-detect yt-dlp for YouTube search + has_ytdlp = env.is_ytdlp_available() + # Initialize progress display with topic progress = ui.ProgressDisplay(args.topic, show_banner=True) @@ -619,7 +689,7 @@ def main(): mode = sources # Run research - reddit_items, x_items, web_needed, raw_openai, raw_xai, raw_reddit_enriched, reddit_error, x_error = run_research( + reddit_items, x_items, youtube_items, web_needed, raw_openai, raw_xai, raw_reddit_enriched, reddit_error, x_error, youtube_error = run_research( args.topic, sources, config, @@ -630,6 +700,7 @@ def main(): args.mock, progress, x_source=x_source or "xai", + run_youtube=has_ytdlp, ) # Processing phase @@ -638,23 +709,28 @@ def main(): # Normalize items normalized_reddit = normalize.normalize_reddit_items(reddit_items, from_date, to_date) normalized_x = normalize.normalize_x_items(x_items, from_date, to_date) + normalized_youtube = normalize.normalize_youtube_items(youtube_items, from_date, to_date) if youtube_items else [] # Hard date filter: exclude items with verified dates outside the range # This is the safety net - even if prompts let old content through, this filters it filtered_reddit = normalize.filter_by_date_range(normalized_reddit, from_date, to_date) filtered_x = normalize.filter_by_date_range(normalized_x, from_date, to_date) + filtered_youtube = normalize.filter_by_date_range(normalized_youtube, from_date, to_date) if normalized_youtube else [] # Score items scored_reddit = score.score_reddit_items(filtered_reddit) scored_x = score.score_x_items(filtered_x) + scored_youtube = score.score_youtube_items(filtered_youtube) if filtered_youtube else [] # Sort items sorted_reddit = score.sort_items(scored_reddit) sorted_x = score.sort_items(scored_x) + sorted_youtube = score.sort_items(scored_youtube) if scored_youtube else [] # Dedupe items deduped_reddit = dedupe.dedupe_reddit(sorted_reddit) deduped_x = dedupe.dedupe_x(sorted_x) + deduped_youtube = dedupe.dedupe_youtube(sorted_youtube) if sorted_youtube else [] # Minimum result guarantee: if all Reddit results were filtered out but # we had raw results, keep top 3 by relevance regardless of score @@ -676,8 +752,10 @@ def main(): ) report.reddit = deduped_reddit report.x = deduped_x + report.youtube = deduped_youtube report.reddit_error = reddit_error report.x_error = x_error + report.youtube_error = youtube_error # Generate context snippet report.context_snippet_md = render.render_context_snippet(report) @@ -689,7 +767,7 @@ def main(): if sources == "web": progress.show_web_only_complete() else: - progress.show_complete(len(deduped_reddit), len(deduped_x)) + progress.show_complete(len(deduped_reddit), len(deduped_x), len(deduped_youtube)) # Output result output_result(report, args.emit, web_needed, args.topic, from_date, to_date, missing_keys, args.days) diff --git a/scripts/lib/dedupe.py b/scripts/lib/dedupe.py index a42024f..22a87c9 100644 --- a/scripts/lib/dedupe.py +++ b/scripts/lib/dedupe.py @@ -36,10 +36,12 @@ def jaccard_similarity(set1: Set[str], set2: Set[str]) -> float: return intersection / union if union > 0 else 0.0 -def get_item_text(item: Union[schema.RedditItem, schema.XItem]) -> str: +def get_item_text(item: Union[schema.RedditItem, schema.XItem, schema.YouTubeItem]) -> str: """Get comparable text from an item.""" if isinstance(item, schema.RedditItem): return item.title + elif isinstance(item, schema.YouTubeItem): + return f"{item.title} {item.channel_name}" else: return item.text @@ -118,3 +120,11 @@ def dedupe_x( ) -> List[schema.XItem]: """Dedupe X items.""" return dedupe_items(items, threshold) + + +def dedupe_youtube( + items: List[schema.YouTubeItem], + threshold: float = 0.7, +) -> List[schema.YouTubeItem]: + """Dedupe YouTube items.""" + return dedupe_items(items, threshold) diff --git a/scripts/lib/env.py b/scripts/lib/env.py index 2ff7001..d549768 100644 --- a/scripts/lib/env.py +++ b/scripts/lib/env.py @@ -196,6 +196,12 @@ def get_x_source(config: Dict[str, Any]) -> Optional[str]: return None +def is_ytdlp_available() -> bool: + """Check if yt-dlp is installed for YouTube search.""" + from . import youtube_yt + return youtube_yt.is_ytdlp_installed() + + def get_x_source_status(config: Dict[str, Any]) -> Dict[str, Any]: """Get detailed X source status for UI decisions. diff --git a/scripts/lib/normalize.py b/scripts/lib/normalize.py index 0d2577e..941a9bf 100644 --- a/scripts/lib/normalize.py +++ b/scripts/lib/normalize.py @@ -4,7 +4,7 @@ from typing import Any, Dict, List, TypeVar, Union from . import dates, schema -T = TypeVar("T", schema.RedditItem, schema.XItem, schema.WebSearchItem) +T = TypeVar("T", schema.RedditItem, schema.XItem, schema.WebSearchItem, schema.YouTubeItem) def filter_by_date_range( @@ -155,6 +155,51 @@ def normalize_x_items( return normalized +def normalize_youtube_items( + items: List[Dict[str, Any]], + from_date: str, + to_date: str, +) -> List[schema.YouTubeItem]: + """Normalize raw YouTube items to schema. + + Args: + items: Raw YouTube items from yt-dlp + from_date: Start of date range + to_date: End of date range + + Returns: + List of YouTubeItem objects + """ + normalized = [] + + for item in items: + # Parse engagement + eng_raw = item.get("engagement", {}) + engagement = schema.Engagement( + views=eng_raw.get("views"), + likes=eng_raw.get("likes"), + num_comments=eng_raw.get("comments"), + ) + + # YouTube dates are reliable (always YYYY-MM-DD from yt-dlp) + date_str = item.get("date") + + normalized.append(schema.YouTubeItem( + id=item.get("video_id", ""), + title=item.get("title", ""), + url=item.get("url", ""), + channel_name=item.get("channel_name", ""), + date=date_str, + date_confidence="high", + engagement=engagement, + transcript_snippet=item.get("transcript_snippet", ""), + relevance=item.get("relevance", 0.7), + why_relevant=item.get("why_relevant", ""), + )) + + return normalized + + def items_to_dicts(items: List) -> List[Dict[str, Any]]: """Convert schema items to dicts for JSON serialization.""" return [item.to_dict() for item in items] diff --git a/scripts/lib/render.py b/scripts/lib/render.py index c4bf83e..8a21f93 100644 --- a/scripts/lib/render.py +++ b/scripts/lib/render.py @@ -170,6 +170,40 @@ def render_compact(report: schema.Report, limit: int = 15, missing_keys: str = " lines.append(f" *{item.why_relevant}*") lines.append("") + # YouTube items + if report.youtube_error: + lines.append("### YouTube Videos") + lines.append("") + lines.append(f"**ERROR:** {report.youtube_error}") + lines.append("") + elif report.youtube: + lines.append("### YouTube Videos") + lines.append("") + for item in report.youtube[:limit]: + eng_str = "" + if item.engagement: + eng = item.engagement + parts = [] + if eng.views is not None: + parts.append(f"{eng.views:,} views") + if eng.likes is not None: + parts.append(f"{eng.likes:,} likes") + if parts: + eng_str = f" [{', '.join(parts)}]" + + date_str = f" ({item.date})" if item.date else "" + + lines.append(f"**{item.id}** (score:{item.score}) {item.channel_name}{date_str}{eng_str}") + lines.append(f" {item.title}") + lines.append(f" {item.url}") + if item.transcript_snippet: + snippet = item.transcript_snippet[:200] + if len(item.transcript_snippet) > 200: + snippet += "..." + lines.append(f" Transcript: {snippet}") + lines.append(f" *{item.why_relevant}*") + lines.append("") + # Web items (if any - populated by Claude) if report.web_error: lines.append("### Web Results") diff --git a/scripts/lib/schema.py b/scripts/lib/schema.py index a9fc5bf..da1534f 100644 --- a/scripts/lib/schema.py +++ b/scripts/lib/schema.py @@ -19,6 +19,9 @@ class Engagement: replies: Optional[int] = None quotes: Optional[int] = None + # YouTube fields + views: Optional[int] = None + def to_dict(self) -> Dict[str, Any]: d = {} if self.score is not None: @@ -35,6 +38,8 @@ class Engagement: d['replies'] = self.replies if self.quotes is not None: d['quotes'] = self.quotes + if self.views is not None: + d['views'] = self.views return d if d else None @@ -169,6 +174,39 @@ class WebSearchItem: } +@dataclass +class YouTubeItem: + """Normalized YouTube item.""" + id: str # video_id + title: str + url: str + channel_name: str + date: Optional[str] = None + date_confidence: str = "high" # YouTube dates are always reliable + engagement: Optional[Engagement] = None + transcript_snippet: str = "" + relevance: float = 0.7 + why_relevant: str = "" + subs: SubScores = field(default_factory=SubScores) + score: int = 0 + + def to_dict(self) -> Dict[str, Any]: + return { + 'id': self.id, + 'title': self.title, + 'url': self.url, + 'channel_name': self.channel_name, + 'date': self.date, + 'date_confidence': self.date_confidence, + 'engagement': self.engagement.to_dict() if self.engagement else None, + 'transcript_snippet': self.transcript_snippet, + 'relevance': self.relevance, + 'why_relevant': self.why_relevant, + 'subs': self.subs.to_dict(), + 'score': self.score, + } + + @dataclass class Report: """Full research report.""" @@ -182,6 +220,7 @@ class Report: reddit: List[RedditItem] = field(default_factory=list) x: List[XItem] = field(default_factory=list) web: List[WebSearchItem] = field(default_factory=list) + youtube: List[YouTubeItem] = field(default_factory=list) best_practices: List[str] = field(default_factory=list) prompt_pack: List[str] = field(default_factory=list) context_snippet_md: str = "" @@ -189,6 +228,7 @@ class Report: reddit_error: Optional[str] = None x_error: Optional[str] = None web_error: Optional[str] = None + youtube_error: Optional[str] = None # Cache info from_cache: bool = False cache_age_hours: Optional[float] = None @@ -207,6 +247,7 @@ class Report: 'reddit': [r.to_dict() for r in self.reddit], 'x': [x.to_dict() for x in self.x], 'web': [w.to_dict() for w in self.web], + 'youtube': [y.to_dict() for y in self.youtube], 'best_practices': self.best_practices, 'prompt_pack': self.prompt_pack, 'context_snippet_md': self.context_snippet_md, @@ -217,6 +258,8 @@ class Report: d['x_error'] = self.x_error if self.web_error: d['web_error'] = self.web_error + if self.youtube_error: + d['youtube_error'] = self.youtube_error if self.from_cache: d['from_cache'] = self.from_cache if self.cache_age_hours is not None: @@ -294,6 +337,28 @@ class Report: score=w.get('score', 0), )) + # Reconstruct YouTube items + youtube_items = [] + for y in data.get('youtube', []): + eng = None + if y.get('engagement'): + eng = Engagement(**y['engagement']) + subs = SubScores(**y.get('subs', {})) if y.get('subs') else SubScores() + youtube_items.append(YouTubeItem( + id=y['id'], + title=y['title'], + url=y['url'], + channel_name=y.get('channel_name', ''), + date=y.get('date'), + date_confidence=y.get('date_confidence', 'high'), + engagement=eng, + transcript_snippet=y.get('transcript_snippet', ''), + relevance=y.get('relevance', 0.7), + why_relevant=y.get('why_relevant', ''), + subs=subs, + score=y.get('score', 0), + )) + return cls( topic=data['topic'], range_from=range_from, @@ -305,12 +370,14 @@ class Report: reddit=reddit_items, x=x_items, web=web_items, + youtube=youtube_items, best_practices=data.get('best_practices', []), prompt_pack=data.get('prompt_pack', []), context_snippet_md=data.get('context_snippet_md', ''), reddit_error=data.get('reddit_error'), x_error=data.get('x_error'), web_error=data.get('web_error'), + youtube_error=data.get('youtube_error'), from_cache=data.get('from_cache', False), cache_age_hours=data.get('cache_age_hours'), ) diff --git a/scripts/lib/score.py b/scripts/lib/score.py index 2f06833..3fca23f 100644 --- a/scripts/lib/score.py +++ b/scripts/lib/score.py @@ -221,6 +221,65 @@ def score_x_items(items: List[schema.XItem]) -> List[schema.XItem]: return items +def compute_youtube_engagement_raw(engagement: Optional[schema.Engagement]) -> Optional[float]: + """Compute raw engagement score for YouTube item. + + Formula: 0.50*log1p(views) + 0.35*log1p(likes) + 0.15*log1p(comments) + Views dominate on YouTube — they're the primary discovery signal. + """ + if engagement is None: + return None + + if engagement.views is None and engagement.likes is None: + return None + + views = log1p_safe(engagement.views) + likes = log1p_safe(engagement.likes) + comments = log1p_safe(engagement.num_comments) + + return 0.50 * views + 0.35 * likes + 0.15 * comments + + +def score_youtube_items(items: List[schema.YouTubeItem]) -> List[schema.YouTubeItem]: + """Compute scores for YouTube items. + + Uses same weight structure as Reddit/X (relevance + recency + engagement). + """ + if not items: + return items + + eng_raw = [compute_youtube_engagement_raw(item.engagement) for item in items] + eng_normalized = normalize_to_100(eng_raw) + + for i, item in enumerate(items): + rel_score = int(item.relevance * 100) + rec_score = dates.recency_score(item.date) + + if eng_normalized[i] is not None: + eng_score = int(eng_normalized[i]) + else: + eng_score = DEFAULT_ENGAGEMENT + + item.subs = schema.SubScores( + relevance=rel_score, + recency=rec_score, + engagement=eng_score, + ) + + overall = ( + WEIGHT_RELEVANCE * rel_score + + WEIGHT_RECENCY * rec_score + + WEIGHT_ENGAGEMENT * eng_score + ) + + if eng_raw[i] is None: + overall -= UNKNOWN_ENGAGEMENT_PENALTY + + item.score = max(0, min(100, int(overall))) + + return items + + def score_websearch_items(items: List[schema.WebSearchItem]) -> List[schema.WebSearchItem]: """Compute scores for WebSearch items WITHOUT engagement metrics. @@ -278,7 +337,7 @@ def score_websearch_items(items: List[schema.WebSearchItem]) -> List[schema.WebS return items -def sort_items(items: List[Union[schema.RedditItem, schema.XItem, schema.WebSearchItem]]) -> List: +def sort_items(items: List[Union[schema.RedditItem, schema.XItem, schema.WebSearchItem, schema.YouTubeItem]]) -> List: """Sort items by score (descending), then date, then source priority. Args: @@ -295,13 +354,15 @@ def sort_items(items: List[Union[schema.RedditItem, schema.XItem, schema.WebSear date = item.date or "0000-00-00" date_key = -int(date.replace("-", "")) - # Tertiary: source priority (Reddit > X > WebSearch) + # Tertiary: source priority (Reddit > X > YouTube > WebSearch) if isinstance(item, schema.RedditItem): source_priority = 0 elif isinstance(item, schema.XItem): source_priority = 1 - else: # WebSearchItem + elif isinstance(item, schema.YouTubeItem): source_priority = 2 + else: # WebSearchItem + source_priority = 3 # Quaternary: title/text for stability text = getattr(item, "title", "") or getattr(item, "text", "") diff --git a/scripts/lib/ui.py b/scripts/lib/ui.py index 9f2768e..82a7cae 100644 --- a/scripts/lib/ui.py +++ b/scripts/lib/ui.py @@ -64,6 +64,14 @@ ENRICHING_MESSAGES = [ "Analyzing discussions...", ] +YOUTUBE_MESSAGES = [ + "Searching YouTube for videos...", + "Finding relevant video content...", + "Scanning YouTube channels...", + "Discovering video discussions...", + "Fetching transcripts...", +] + PROCESSING_MESSAGES = [ "Crunching the data...", "Scoring and ranking...", @@ -280,6 +288,15 @@ class ProgressDisplay: if self.spinner: self.spinner.stop(f"{Colors.CYAN}X{Colors.RESET} Found {count} posts") + def start_youtube(self): + msg = random.choice(YOUTUBE_MESSAGES) + self.spinner = Spinner(f"{Colors.RED}YouTube{Colors.RESET} {msg}", Colors.RED) + self.spinner.start() + + def end_youtube(self, count: int): + if self.spinner: + self.spinner.stop(f"{Colors.RED}YouTube{Colors.RESET} Found {count} videos") + def start_processing(self): msg = random.choice(PROCESSING_MESSAGES) self.spinner = Spinner(f"{Colors.PURPLE}Processing{Colors.RESET} {msg}", Colors.PURPLE) @@ -289,15 +306,21 @@ class ProgressDisplay: if self.spinner: self.spinner.stop() - def show_complete(self, reddit_count: int, x_count: int): + def show_complete(self, reddit_count: int, x_count: int, youtube_count: int = 0): elapsed = time.time() - self.start_time if IS_TTY: sys.stderr.write(f"\n{Colors.GREEN}{Colors.BOLD}✓ Research complete{Colors.RESET} ") sys.stderr.write(f"{Colors.DIM}({elapsed:.1f}s){Colors.RESET}\n") sys.stderr.write(f" {Colors.YELLOW}Reddit:{Colors.RESET} {reddit_count} threads ") - sys.stderr.write(f"{Colors.CYAN}X:{Colors.RESET} {x_count} posts\n\n") + sys.stderr.write(f"{Colors.CYAN}X:{Colors.RESET} {x_count} posts") + if youtube_count: + sys.stderr.write(f" {Colors.RED}YouTube:{Colors.RESET} {youtube_count} videos") + sys.stderr.write("\n\n") else: - sys.stderr.write(f"✓ Research complete ({elapsed:.1f}s) - Reddit: {reddit_count} threads, X: {x_count} posts\n") + parts = [f"Reddit: {reddit_count} threads", f"X: {x_count} posts"] + if youtube_count: + parts.append(f"YouTube: {youtube_count} videos") + sys.stderr.write(f"✓ Research complete ({elapsed:.1f}s) - {', '.join(parts)}\n") sys.stderr.flush() def show_cached(self, age_hours: float = None): diff --git a/scripts/lib/youtube_yt.py b/scripts/lib/youtube_yt.py new file mode 100644 index 0000000..6be7a0c --- /dev/null +++ b/scripts/lib/youtube_yt.py @@ -0,0 +1,331 @@ +"""YouTube search and transcript extraction via yt-dlp for /last30days v2.1. + +Uses yt-dlp (https://github.com/yt-dlp/yt-dlp) for both YouTube search and +transcript extraction. No API keys needed — just have yt-dlp installed. + +Inspired by Peter Steinberger's toolchain approach (yt-dlp + summarize CLI). +""" + +import json +import math +import re +import shutil +import subprocess +import sys +import tempfile +from concurrent.futures import ThreadPoolExecutor, as_completed +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +# Depth configurations: how many videos to search / transcribe +DEPTH_CONFIG = { + "quick": 10, + "default": 20, + "deep": 40, +} + +TRANSCRIPT_LIMITS = { + "quick": 3, + "default": 5, + "deep": 8, +} + +# Max words to keep from each transcript +TRANSCRIPT_MAX_WORDS = 500 + + +def _log(msg: str): + """Log to stderr.""" + sys.stderr.write(f"[YouTube] {msg}\n") + sys.stderr.flush() + + +def is_ytdlp_installed() -> bool: + """Check if yt-dlp is available in PATH.""" + return shutil.which("yt-dlp") is not None + + +def _extract_core_subject(topic: str) -> str: + """Extract core subject from verbose query for YouTube search. + + Strips meta/research words to keep only the core product/concept name, + similar to bird_x.py's approach. + """ + text = topic.lower().strip() + + # Strip multi-word prefixes + prefixes = [ + 'what are the best', 'what is the best', 'what are the latest', + 'what are people saying about', 'what do people think about', + 'how do i use', 'how to use', 'how to', + 'what are', 'what is', 'tips for', 'best practices for', + ] + for p in prefixes: + if text.startswith(p + ' '): + text = text[len(p):].strip() + + # Strip individual noise words + noise = { + 'best', 'top', 'good', 'great', 'awesome', 'killer', + 'latest', 'new', 'news', 'update', 'updates', + 'trending', 'hottest', 'popular', 'viral', + 'practices', 'features', 'guide', 'tutorial', + 'recommendations', 'advice', 'review', 'reviews', + 'prompt', 'prompts', 'prompting', 'techniques', 'tips', + 'tricks', 'methods', 'strategies', 'approaches', + } + words = text.split() + filtered = [w for w in words if w not in noise] + + return ' '.join(filtered) if filtered else text + + +def search_youtube( + topic: str, + from_date: str, + to_date: str, + depth: str = "default", +) -> Dict[str, Any]: + """Search YouTube via yt-dlp. No API key needed. + + Args: + topic: Search topic + from_date: Start date (YYYY-MM-DD) + to_date: End date (YYYY-MM-DD) + depth: 'quick', 'default', or 'deep' + + Returns: + Dict with 'items' list of video metadata dicts. + """ + if not is_ytdlp_installed(): + return {"items": [], "error": "yt-dlp not installed"} + + count = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) + core_topic = _extract_core_subject(topic) + date_filter = from_date.replace("-", "") # YYYYMMDD format + + _log(f"Searching YouTube for '{core_topic}' (since {from_date}, count={count})") + + # yt-dlp search with metadata extraction via JSON + cmd = [ + "yt-dlp", + f"ytsearch{count}:{core_topic}", + "--dateafter", date_filter, + "--flat-playlist", + "--dump-json", + ] + + try: + result = subprocess.run( + cmd, capture_output=True, text=True, timeout=60, + ) + except subprocess.TimeoutExpired: + _log("YouTube search timed out (60s)") + return {"items": [], "error": "Search timed out"} + except FileNotFoundError: + return {"items": [], "error": "yt-dlp not found"} + + if not result.stdout.strip(): + _log("YouTube search returned 0 results") + return {"items": []} + + # Parse JSON-per-line output + items = [] + for line in result.stdout.strip().split("\n"): + line = line.strip() + if not line: + continue + try: + video = json.loads(line) + except json.JSONDecodeError: + continue + + video_id = video.get("id", "") + view_count = video.get("view_count") or 0 + like_count = video.get("like_count") or 0 + comment_count = video.get("comment_count") or 0 + upload_date = video.get("upload_date", "") # YYYYMMDD + + # Convert YYYYMMDD to YYYY-MM-DD + date_str = None + if upload_date and len(upload_date) == 8: + date_str = f"{upload_date[:4]}-{upload_date[4:6]}-{upload_date[6:8]}" + + items.append({ + "video_id": video_id, + "title": video.get("title", ""), + "url": f"https://www.youtube.com/watch?v={video_id}", + "channel_name": video.get("channel", video.get("uploader", "")), + "date": date_str, + "engagement": { + "views": view_count, + "likes": like_count, + "comments": comment_count, + }, + "duration": video.get("duration"), + "relevance": 0.7, # Default; no LLM relevance scoring for YouTube + "why_relevant": f"YouTube video about {core_topic}", + }) + + # Sort by views descending + items.sort(key=lambda x: x["engagement"]["views"], reverse=True) + + _log(f"Found {len(items)} videos") + return {"items": items} + + +def _clean_vtt(vtt_text: str) -> str: + """Convert VTT subtitle format to clean plaintext.""" + # Strip VTT header + text = re.sub(r'^WEBVTT.*?\n\n', '', vtt_text, flags=re.DOTALL) + # Strip timestamps + text = re.sub(r'\d{2}:\d{2}:\d{2}\.\d{3}\s*-->\s*\d{2}:\d{2}:\d{2}\.\d{3}.*\n', '', text) + # Strip position/alignment tags + text = re.sub(r'<[^>]+>', '', text) + # Strip cue numbers + text = re.sub(r'^\d+\s*$', '', text, flags=re.MULTILINE) + # Deduplicate overlapping lines + lines = text.strip().split('\n') + seen = set() + unique = [] + for line in lines: + stripped = line.strip() + if stripped and stripped not in seen: + seen.add(stripped) + unique.append(stripped) + return re.sub(r'\s+', ' ', ' '.join(unique)).strip() + + +def fetch_transcript(video_id: str, temp_dir: str) -> Optional[str]: + """Fetch auto-generated transcript for a YouTube video. + + Args: + video_id: YouTube video ID + temp_dir: Temporary directory for subtitle files + + Returns: + Plaintext transcript string, or None if no captions available. + """ + cmd = [ + "yt-dlp", + "--write-auto-subs", + "--sub-lang", "en", + "--sub-format", "vtt", + "--skip-download", + "--no-warnings", + "-o", f"{temp_dir}/%(id)s", + f"https://www.youtube.com/watch?v={video_id}", + ] + + try: + subprocess.run(cmd, capture_output=True, text=True, timeout=30) + except (subprocess.TimeoutExpired, FileNotFoundError): + return None + + # yt-dlp may save as .en.vtt or .en-orig.vtt + vtt_path = Path(temp_dir) / f"{video_id}.en.vtt" + if not vtt_path.exists(): + # Try alternate naming + for p in Path(temp_dir).glob(f"{video_id}*.vtt"): + vtt_path = p + break + else: + return None + + try: + raw = vtt_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return None + + transcript = _clean_vtt(raw) + + # Truncate to max words + words = transcript.split() + if len(words) > TRANSCRIPT_MAX_WORDS: + transcript = ' '.join(words[:TRANSCRIPT_MAX_WORDS]) + '...' + + return transcript if transcript else None + + +def fetch_transcripts_parallel( + video_ids: List[str], + max_workers: int = 5, +) -> Dict[str, Optional[str]]: + """Fetch transcripts for multiple videos in parallel. + + Args: + video_ids: List of YouTube video IDs + max_workers: Max parallel fetches + + Returns: + Dict mapping video_id to transcript text (or None). + """ + if not video_ids: + return {} + + _log(f"Fetching transcripts for {len(video_ids)} videos") + + results = {} + with tempfile.TemporaryDirectory() as temp_dir: + with ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = { + executor.submit(fetch_transcript, vid, temp_dir): vid + for vid in video_ids + } + for future in as_completed(futures): + vid = futures[future] + try: + results[vid] = future.result() + except Exception: + results[vid] = None + + got = sum(1 for v in results.values() if v) + _log(f"Got transcripts for {got}/{len(video_ids)} videos") + return results + + +def search_and_transcribe( + topic: str, + from_date: str, + to_date: str, + depth: str = "default", +) -> Dict[str, Any]: + """Full YouTube search: find videos, then fetch transcripts for top results. + + Args: + topic: Search topic + from_date: Start date (YYYY-MM-DD) + to_date: End date (YYYY-MM-DD) + depth: 'quick', 'default', or 'deep' + + Returns: + Dict with 'items' list. Each item has a 'transcript_snippet' field. + """ + # Step 1: Search + search_result = search_youtube(topic, from_date, to_date, depth) + items = search_result.get("items", []) + + if not items: + return search_result + + # Step 2: Fetch transcripts for top N by views + transcript_limit = TRANSCRIPT_LIMITS.get(depth, TRANSCRIPT_LIMITS["default"]) + top_ids = [item["video_id"] for item in items[:transcript_limit]] + transcripts = fetch_transcripts_parallel(top_ids) + + # Step 3: Attach transcripts to items + for item in items: + vid = item["video_id"] + transcript = transcripts.get(vid) + item["transcript_snippet"] = transcript or "" + + return {"items": items} + + +def parse_youtube_response(response: Dict[str, Any]) -> List[Dict[str, Any]]: + """Parse YouTube search response to normalized format. + + Returns: + List of item dicts ready for normalization. + """ + return response.get("items", [])