feat: Add YouTube as 4th research source via yt-dlp
YouTube search and transcript extraction runs automatically when yt-dlp is installed. Searches for topic videos from the last N days, fetches auto-generated transcripts for top results, and feeds them through the same scoring pipeline (relevance + recency + engagement) as Reddit/X. New files: - youtube_yt.py: search, transcript extraction, VTT cleanup Modified files: - schema.py: YouTubeItem dataclass, updated Report - normalize.py: normalize_youtube_items() - score.py: YouTube engagement scoring (views-dominated) - dedupe.py: YouTube deduplication - render.py: YouTube section in compact output - env.py: is_ytdlp_available() check - ui.py: YouTube progress messages - last30days.py: _search_youtube(), parallel execution with Reddit/X - SKILL.md: YouTube in stats box, citation priority - README.md: YouTube docs, yt-dlp requirement, Peter shoutout Inspired by Peter Steinberger's yt-dlp + summarize toolchain approach. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Matt Van Horn
@@ -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", [])
|
||||
Reference in New Issue
Block a user