feat(podcasts): make podcasts always available + smarter mention matching

Changes: - Podcasts source is now always available when yt-dlp is installed (same as YouTube). Previously required explicit opt-in via INCLUDE_SOURCES or --search=podcasts. - Smarter mention matching: extract key terms from multi-word topics and use max count across terms. "Kanye West Bully album" now matches episodes mentioning "Kanye" 85 times (previously 0 due to exact phrase). - SKILL.md: add podcast channel resolution to Step 0.55, include --podcast-channels in execution command, update ACTIVE_SOURCES_LIST. Tested: Kanye West query now finds 5 podcast hits including hidden mentions in off-topic episodes (Lost Civilizations, Mike WiLL Made-It). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
feat(podcasts): add YouTube podcast source with transcript-first discovery
2026-04-11 00:29:11 -04:00 · 2026-04-11 00:28:40 -04:00
9 changed files with 798 additions and 1 deletions
@@ -375,7 +375,7 @@ Common patterns:
 - Always active: Reddit, Hacker News, Polymarket
 - If gh CLI is installed (check `which gh`): add GitHub
 - If AUTH_TOKEN/CT0 or XAI_API_KEY or FROM_BROWSER is set: add X
- If yt-dlp is installed (check `which yt-dlp`): add YouTube
+- If yt-dlp is installed (check `which yt-dlp`): add YouTube AND Podcasts
 - If SCRAPECREATORS_API_KEY is set and INCLUDE_SOURCES contains tiktok: add TikTok
 - If SCRAPECREATORS_API_KEY is set and INCLUDE_SOURCES contains instagram: add Instagram
 - If SCRAPECREATORS_API_KEY is set and INCLUDE_SOURCES contains threads: add Threads
@@ -615,6 +615,27 @@ Store as `RESOLVED_IG_CREATORS`.
 Store as `RESOLVED_YT_QUERIES`.
 **6. Podcast channels** — **INFER 6-12 YouTube podcast channel @handles from topic knowledge.** Think in two dimensions:
 1. **Domain podcasts** — What YouTube podcasts focus on this topic's domain?
   - Hip-hop/music → `DrinkChamps,JoeBuddenTV,BreakfastClubPower1051FM,OfficialFlagrant`
   - Tech/AI/startups → `lexfridman,DwarkeshPatel,AllInPod,MyFirstMillionPod,LennysPodcast`
   - Business/finance → `AcquiredFM,InvestLikeTheBest,PatrickBoyleOnFinance,PropGPod`
   - Sports → `PatMcAfeeShowOfficial,ShannonSharpe,ClubShayShay`
   - Culture/celebs → `joerogan,CallHerDaddy,ClubShayShay`
   - Knitting/crafts → `FruityKnitting,VeryPinkKnits,GroceryGirlsKnit`
 2. **Cross-domain podcasts** — What popular interview/deep-dive podcasts might cover this topic even if it's not their main focus?
   - Business-adjacent topics → `AcquiredFM,InvestLikeTheBest` (company deep dives)
   - Tech-adjacent topics → `lexfridman,AllInPod` (broad tech interviews)
   - Culture-adjacent topics → `joerogan,OfficialFlagrant` (celebrity interviews)
 **Rationale:** The engine uses these channels for transcript-first discovery. Even if the topic isn't in an episode title, it may be discussed within the episode. Acquired's "The NFL" episode mentions Taylor Swift 18 times, ESPN 117 times — invisible to YouTube search but found by transcript scanning.
 **Handle accuracy:** Return your best guess at the exact @handle. If wrong, the engine falls back to a search-based lookup. Don't stress the exact spelling — `@AcquiredFM`, `@lexfridman`, `@joerogan` work; `@FLAGRANT` fails but falls back to find `@OfficialFlagrant`.
 Store as `RESOLVED_PODCAST_CHANNELS` (comma-separated, no @ prefix).
 **Concrete examples:**
 | Topic | WebSearches needed | Reddit subs | TikTok hashtags | TikTok creators | IG creators | YT queries |
@@ -635,6 +656,7 @@ Resolved:
 - Reddit: r/{sub1}, r/{sub2}, r/{sub3}
 - TikTok: #{hashtag1}, #{hashtag2}
 - YouTube: {query1}, {query2}
 - Podcasts: @{channel1}, @{channel2}, @{channel3}
 ```
 Only show lines for platforms where something was resolved. Skip empty lines. This display replaces the old "Parsed intent" block with something more useful.
@@ -760,6 +782,7 @@ fi
 - `--ig-creators={RESOLVED_IG_CREATORS}` (from Step 0.55)
 - `--github-user={RESOLVED_GITHUB_USER}` (from Step 0.5b, person topics only)
 - `--github-repo={RESOLVED_GITHUB_REPOS}` (from Step 0.5c, product/project topics only)
 - `--podcast-channels={RESOLVED_PODCAST_CHANNELS}` (from Step 0.55, 6-12 @handles)
 - Omit any flag where the value was not resolved (empty).
 **If you skipped Steps 0.55 and 0.75 (no WebSearch -- OpenClaw, Codex, etc.), add:**
@@ -0,0 +1,319 @@
 ---
 title: "feat: YouTube podcast source with transcript-first discovery"
 type: feat
 status: active
 date: 2026-04-10
 ---
 # feat: YouTube podcast source with transcript-first discovery
 ## Overview
 Add a "podcasts" source to last30days that discovers podcast content on YouTube by scanning transcripts, not searching titles. The LLM planner resolves topic-relevant podcast channels (e.g., "NVIDIA" -> Acquired, Lex Fridman, Dwarkesh Patel, All-In). The engine fetches recent episodes from those channels, downloads their auto-captions (no video download), and greps for the search topic. Episodes with 5+ topic mentions become podcast results with transcript highlights.
 This finds content invisible to any search engine. Acquired's "The NFL" episode mentions Taylor Swift 18 times, ESPN 117 times, Netflix 102 times - none in the title. A Dwarkesh Patel episode titled "The single biggest bottleneck to scaling AI compute" contains 156 mentions of NVIDIA. No YouTube search finds these. Transcript scanning does.
 Zero new API keys. Zero new dependencies. Reuses existing yt-dlp + transcript pipeline. Podcasts get their own identity in stats and synthesis.
 ## Problem Frame
 YouTube captures a lot of podcast content, but it's mixed with news clips, reaction videos, and shorts. The general YouTube search treats a 2:24:55 Drink Champs interview the same as a 0:30 TMZ clip. Worse, the highest-value podcast content is often invisible to search entirely because the topic is discussed within an episode titled something else.
 Two insights make this solvable:
 1. Podcast episodes are identifiable by duration (>20 minutes) and channel.
 2. YouTube auto-captions are free, downloadable without the video (~7 seconds per episode via yt-dlp), and searchable. Transcript scanning discovers content that title-based search cannot.
 The LLM already resolves subreddits and X handles per topic. Podcast channels are the same pattern.
 ## Requirements Trace
 - R1. LLM resolves topic-relevant podcast YouTube channels dynamically (no hardcoded list)
 - R2. Engine scans recent episode transcripts for the search topic, not just titles
 - R3. Podcast results get their own source identity with own stats line and synthesis treatment
 - R4. Reuses existing yt-dlp transcript pipeline (no new dependencies)
 - R5. Does not duplicate regular YouTube results (dedup by video ID in fusion)
 - R6. Channel resolution works in both the agent layer (SKILL.md) and the Python planner
 ## Scope Boundaries
 - Not building a new API integration (reuses yt-dlp entirely)
 - Not adding PodcastIndex, AssemblyAI, or any podcast-specific API
 - Not changing how the regular YouTube source works
 - Not building a podcast channel database
 - Channels that can't be resolved are skipped silently (graceful degradation)
 ## Context & Research
 ### Relevant Code and Patterns
 - `scripts/lib/youtube_yt.py` - YouTube search + transcript pipeline. Key functions: `search_youtube()`, `fetch_transcripts()`, `extract_transcript_highlights()`
 - `scripts/lib/youtube_yt.py` - `--write-auto-sub --skip-download` fetches captions without downloading video
 - Step 0.55 in `SKILL.md` - subreddit resolution pattern (WebSearch + LLM knowledge -> `--subreddits=`)
 - `scripts/lib/pipeline.py` - source dispatch via if/elif chain in `_retrieve_stream()`, 4-point registration pattern
 - `scripts/lib/normalize.py` - `_normalize_youtube()` handles transcript data, reusable for podcasts
 - `scripts/lib/signals.py` - `SOURCE_QUALITY` dict (YouTube is 0.85)
 - `scripts/lib/planner.py` - `QueryPlan` schema, `SOURCE_CAPABILITIES` dict
 ### Proof of Concept Results (2026-04-10)
 **Transcript-first discovery test:** Fetched auto-captions for 5 recent Acquired episodes (35 seconds total, no video download). Grepped for topics not in any episode title:
 | Topic | Mentions | Episode title | Discoverable by search? |
 |-------|----------|---------------|------------------------|
 | ESPN | 117 | The NFL | No |
 | Super Bowl | 108 | The NFL | No |
 | Netflix | 102 | The NFL | No |
 | Amazon | 87 | The NFL | No |
 | Costco | 63 | The NFL / others | No |
 | Disney | 48 | The NFL | No |
 | LVMH | 27 | Formula 1 / others | No |
 | Taylor Swift | 18 | The NFL | No |
 **Full E2E test (topic: NVIDIA, 4 channels):** LLM resolved Acquired, Lex Fridman, Dwarkesh Patel, All-In. Scanned 14 episodes. Results:
 | Podcast | Episode | NVIDIA mentions | Title mentions NVIDIA? |
 |---------|---------|----------------|----------------------|
 | Lex Fridman | Jensen Huang interview | 159 | Yes |
 | Dwarkesh Patel | Dylan Patel: AI compute bottleneck | 156 | No |
 | Acquired | 10 Years (w/ Michael Lewis) | 24 | No |
 | All-In | SpaceX IPO, Iran, Quantum... | 6 | No |
 3 of 4 hits are invisible to YouTube search. The Dylan Patel episode (156 mentions!) is entirely about NVIDIA's GPU supply chain but the title never says "NVIDIA."
 **Channel handle resolution test:** LLM resolves podcast name + @handle guess. Engine tries @handle first (fast), falls back to `ytsearch1:` if wrong. Tested across 12 channels (tech, hip-hop, knitting): 11/12 resolved on first @handle attempt, 12/12 with fallback. Even niche channels (Fruity Knitting, Grocery Girls Knit, Roxanne Richardson) resolved correctly.
 **Rate limit test:** 4 channels x 3-4 episodes = 14 caption fetches took ~2 minutes sequential. Parallelized with 4 workers: ~30-40 seconds. No YouTube throttling observed. Runs concurrently with Reddit/X/everything else in a 3-minute research run.
 ## Key Technical Decisions
 - **Transcript-first discovery, not title/search-based:** The core innovation. Instead of searching YouTube for `{topic} {podcast_name}` (which only finds episodes titled about the topic), we fetch captions from recent episodes and grep for the topic. This discovers hidden mentions. The approach is validated by POC data showing 3/4 NVIDIA hits were invisible to search.
 - **LLM-resolved channels, not hardcoded:** The LLM planner (agent layer or Python Gemini/OpenAI) resolves 6-12 channels per topic using two-dimensional reasoning: (1) domain podcasts that focus on the topic's area, (2) cross-domain podcasts that might cover it. Tested: the LLM correctly resolved channels for NVIDIA (tech), Kanye (hip-hop), and knitting (craft) - including niche channels like Fruity Knitting and Grocery Girls Knit. Three resolution paths mirror the existing planner architecture:
  - Path 1: Agent layer (SKILL.md with WebSearch) resolves channels in Step 0.55
  - Path 2: Python planner (Gemini/OpenAI) generates channels as a `podcast_channels` field in the QueryPlan
  - Path 3: Fallback (no LLM) uses a small default list of ~5 broad-appeal channels
 - **Handle-first channel resolution with search fallback:** The LLM returns both the podcast name and its best guess at the @handle. The engine tries the @handle first (instant, 92% success rate in testing). If the handle fails, it falls back to `ytsearch1:"{podcast name}" podcast full episode` to find the channel URL. Channels that can't be resolved either way are skipped silently.
 - **New source module wrapping YouTube functions:** `podcast_yt.py` imports `fetch_transcripts()` and `extract_transcript_highlights()` from `youtube_yt.py`. It adds the channel-fetching, caption-scanning, and mention-counting logic. This keeps the regular YouTube source untouched and gives podcasts their own pipeline identity.
 - **Duration filter >= 1200 seconds (20 minutes):** Eliminates clips, shorts, and news segments. Tested empirically - only full podcast episodes survive this filter.
 - **SOURCE_QUALITY: 0.88 (above YouTube's 0.85):** Podcast episodes contain long-form expert discussion with full context. The quality bonus ensures podcast results rank above equivalent YouTube clips when both exist.
 - **Mention count threshold: 5+:** Episodes with fewer than 5 topic mentions are noise (passing references). 5+ indicates substantive discussion. Tested: Taylor Swift at 18 mentions in the NFL episode is substantive discussion of her impact on viewership. "Apple" at 3 mentions in a random episode is just name-dropping.
 ## Open Questions
 ### Resolved During Planning
 - **Can yt-dlp fetch captions without downloading video?** Yes. `yt-dlp --write-auto-sub --sub-lang en --skip-download --sub-format vtt` fetches only the subtitle file. ~7 seconds per episode, ~2MB per 4-hour episode.
 - **Will this double-count YouTube content?** No. Fusion deduplicates by item ID. Both sources use `yt_{video_id}` format.
 - **Can LLMs resolve niche podcast channels?** Yes. Tested with knitting: Fruity Knitting, VeryPink Knits, Grocery Girls Knit, Roxanne Richardson all resolved correctly via @handle.
 - **What about rate limits?** 14 caption fetches across 4 channels showed no throttling. Running in parallel with 4 workers keeps total time under 40 seconds. yt-dlp doesn't use the YouTube Data API (no quota).
 - **How does the LLM know which podcasts to pick?** Two-dimensional prompt: (1) "What YouTube podcasts focus on {topic's domain}?" and (2) "What popular interview/deep-dive podcasts have likely discussed {topic}?" The LLM returns channel names + @handle guesses.
 ### Deferred to Implementation
 - **Exact duration threshold:** Starting with 1200s (20 min). May tune to 900s (15 min) if testing shows missed content.
 - **Mention count threshold tuning:** Starting with 5. May need per-source calibration (a 30-minute podcast with 5 mentions is denser than a 4-hour one with 5 mentions).
 - **Caption language handling:** Starting with English (`--sub-lang en`). Multilingual support deferred.
 - **Parallel worker count:** Starting with 4 workers. May tune based on YouTube throttling behavior at scale.
 ## High-Level Technical Design
 > *This illustrates the intended approach and is directional guidance for review, not implementation specification.*
 ```
 PODCAST DISCOVERY FLOW:
  User query: "NVIDIA"
       |
  LLM planner resolves podcast channels:
  "NVIDIA is a tech/AI company. Domain podcasts: none specific.
   Cross-domain: Acquired (@AcquiredFM), Lex Fridman (@lexfridman),
   Dwarkesh Patel (@DwarkeshPatel), All-In (@AllInPod)"
       |
  Engine receives: --podcast-channels=AcquiredFM,lexfridman,DwarkeshPatel,AllInPod
       |
  For each channel (parallel, 4 workers):
    |
    [1] Resolve @handle -> channel URL
        Try: https://youtube.com/@AcquiredFM/videos
        If fail: ytsearch1:"Acquired podcast full episode" -> extract channel_url
        If fail: skip channel
    |
    [2] Fetch last 3 episode IDs + metadata (duration, date, title)
        yt-dlp --flat-playlist --playlist-end 3
    |
    [3] Filter: duration >= 1200s AND upload_date in date range
    |
    [4] For each surviving episode:
        Fetch auto-captions: yt-dlp --write-auto-sub --skip-download
        Grep captions for "nvidia" (case-insensitive)
        If mentions >= 5: HIT - extract transcript highlights around mentions
    |
  Merge all hits, deduplicate by video_id
  Score: mention_count * log(views)
  Return as source="podcasts" items with transcript_snippet + mention_count
 ```
 ## Implementation Units
 - [ ] **Unit 1: Podcast transcript-scan module**
 **Goal:** Create `scripts/lib/podcast_yt.py` with the channel-fetching, caption-scanning, mention-counting pipeline. Returns podcast episodes discovered via transcript scanning.
 **Requirements:** R2, R3, R4
 **Dependencies:** None (youtube_yt.py already exists)
 **Files:**
 - Create: `scripts/lib/podcast_yt.py`
 - Test: `tests/test_podcast_yt.py`
 **Approach:**
 - `search_podcast_youtube(topic, from_date, to_date, depth, channels)`:
  - For each channel handle (in parallel via ThreadPoolExecutor, max 4 workers):
    1. Resolve handle to channel URL (try @handle first, search fallback)
    2. Fetch last N episode IDs + metadata via `yt-dlp --flat-playlist --playlist-end N`
    3. Filter: `duration >= 1200` and `upload_date` within date range
    4. Fetch auto-captions via `yt-dlp --write-auto-sub --skip-download --sub-lang en`
    5. Grep captions for topic keywords (case-insensitive). Count mentions.
    6. If mentions >= MENTION_THRESHOLD: include as hit. Extract transcript highlights around mentions using `extract_transcript_highlights()` from `youtube_yt`.
  - Merge results, deduplicate by video_id
  - Score: `mention_count * log(views + 1)`
  - Skip channels that can't be resolved or have no recent episodes
 - `resolve_channel(handle)`: Try `@{handle}` URL first. If 404, search `ytsearch1:"{handle}" podcast full episode`, extract channel_url. Return channel_url or None.
 - EPISODES_PER_CHANNEL: quick=2, default=3, deep=4
 - MENTION_THRESHOLD: 5
 - RESULTS_CAP: quick=4, default=8, deep=20
 **Patterns to follow:**
 - `scripts/lib/youtube_yt.py` `search_and_transcribe()` for search-then-enrich flow
 - `scripts/lib/youtube_yt.py` `extract_transcript_highlights()` for highlight extraction
 - `scripts/lib/hackernews.py` for clean module structure with `_log()`, `DEPTH_CONFIG`
 **Test scenarios:**
 - Happy path (hidden mention): topic "Taylor Swift", channels=["AcquiredFM"] -> scans NFL episode, finds 18 mentions, returns episode with highlights about Taylor Swift's NFL viewership impact
 - Happy path (title match): topic "kanye west", channels=["RevoltTV"] -> scans Kanye interview, finds 500+ mentions, returns with highlights
 - Happy path (scoring): episode with 156 mentions and 205K views scores higher than one with 6 mentions and 145K views
 - Happy path (handle resolution): @AcquiredFM resolves directly. @SomeWrongHandle fails, search fallback finds correct channel.
 - Edge case: topic "quantum computing" has <5 mentions in all episodes -> returns empty (threshold not met)
 - Edge case: @handle doesn't exist AND search fallback fails -> channel skipped silently, other channels still scanned
 - Edge case: channel has no episodes in date range -> skipped
 - Edge case: episode has no auto-captions available -> skipped with log warning
 - Error path: yt-dlp not installed -> returns empty items with log warning
 - Error path: caption download times out -> skip that episode, continue
 **Verification:**
 - Discovers episodes where topic is discussed but not in the title (Acquired/NFL/Taylor Swift)
 - Also discovers episodes where topic IS the subject (via same transcript scan)
 - All returned items have duration >= 1200
 - Each item has: video_id, title, channel, url, date, duration, engagement, transcript_snippet, mention_count
 ---
 - [ ] **Unit 2: Pipeline integration**
 **Goal:** Register "podcasts" as a new source in pipeline, normalizer, signals, planner, env, and render.
 **Requirements:** R3, R5, R6
 **Dependencies:** Unit 1
 **Files:**
 - Modify: `scripts/lib/pipeline.py` (import, MOCK_AVAILABLE_SOURCES, available_sources, _retrieve_stream)
 - Modify: `scripts/lib/normalize.py` (add normalizer - reuse `_normalize_youtube` with source override)
 - Modify: `scripts/lib/signals.py` (add SOURCE_QUALITY: 0.88)
 - Modify: `scripts/lib/planner.py` (add SOURCE_CAPABILITIES, extend QueryPlan schema with `podcast_channels` field, add prompt guidance for LLM channel resolution)
 - Modify: `scripts/lib/env.py` (add is_podcast_yt_available - checks yt-dlp installed + "podcasts" in INCLUDE_SOURCES)
 - Modify: `scripts/lib/render.py` (add SOURCE_LABELS: "podcasts" -> "Podcasts")
 - Test: `tests/test_podcast_yt.py` (pipeline dispatch test)
 **Approach:**
 - Availability: yt-dlp installed + "podcasts" in INCLUDE_SOURCES. No API key needed.
 - SOURCE_CAPABILITIES: `{"podcasts": {"discussion", "longform", "expert", "interview"}}`
 - Normalizer: reuse `_normalize_youtube` via lambda wrapper, override source to "podcasts". Add `mention_count` to metadata.
 - CLI flag: `--podcast-channels=handle1,handle2,...` parsed from args
 - Planner: extend QueryPlan with `podcast_channels: list[str]`. Prompt guidance for LLM: "List 6-12 YouTube podcast channel @handles that would discuss this topic. Think in two dimensions: (1) domain podcasts that focus on this area, (2) popular cross-domain interview/deep-dive podcasts that might cover it. Return @handles. If unsure of exact handle, return your best guess."
 - Planner: include "podcasts" source for general/opinion/comparison intents
 - Dedup: podcast items use `yt_{video_id}` ID format (same as YouTube). Fusion dedup handles collisions.
 **Patterns to follow:**
 - 4-point pipeline registration (same as all sources)
 - `_normalize_youtube` reuse via lambda (like tiktok/instagram share `_normalize_shortform_video`)
 - `scripts/lib/env.py` INCLUDE_SOURCES opt-in pattern
 **Test scenarios:**
 - Happy path: "podcasts" in available_sources when yt-dlp installed + INCLUDE_SOURCES contains "podcasts"
 - Happy path: pipeline dispatches to podcast_yt.search_podcast_youtube when source="podcasts"
 - Edge case: yt-dlp not installed -> podcasts not available
 - Edge case: "podcasts" not in INCLUDE_SOURCES -> not available even with yt-dlp
 - Integration: podcast video_id collides with YouTube result -> fusion deduplicates, keeps higher score
 **Verification:**
 - `python3 scripts/last30days.py "NVIDIA" --podcast-channels=AcquiredFM,lexfridman` returns podcast results
 - Stats output shows "Podcasts" line separate from "YouTube"
 ---
 - [ ] **Unit 3: SKILL.md podcast channel resolution + synthesis**
 **Goal:** Add podcast channel resolution to Step 0.55 and podcast-specific synthesis guidance to the Judge Agent section.
 **Requirements:** R1, R3, R6
 **Dependencies:** Unit 2
 **Files:**
 - Modify: `SKILL.md`
 **Approach:**
 - **Step 0.55 addition:** Add "Resolve podcast channels" alongside subreddit, X handle, and TikTok resolution. The agent resolves 6-12 @handles using two-dimensional reasoning (domain + cross-domain). For niche topics, supplement with `WebSearch("{TOPIC} podcast YouTube channel")`. Display resolved channels: "Podcasts: @AcquiredFM, @lexfridman, @DrinkChamps". Pass as `--podcast-channels=AcquiredFM,lexfridman,DrinkChamps`.
 - **Step 0.75 addition:** Add "podcasts" to available sources list. Include in primary subquery sources.
 - **Synthesis guidance addition:** "For podcasts: lead with the guest's name and the podcast name. Quote transcript highlights as direct quotes with speaker attribution. Podcast content represents considered opinion, not hot takes - a 2-hour interview has more nuance than a tweet. When both a podcast and a YouTube clip cover the same topic, prefer the podcast's longer-form analysis."
 - **Stats format:** `├─ 🎙️ Podcasts: {N} episodes │ {N} views │ {N} with transcripts`
 - **INCLUDE_SOURCES:** Add "podcasts" as an option. Note in setup: "Requires yt-dlp (already installed if YouTube works). No API key needed."
 - **Invitation section:** Reference podcast episodes in follow-up suggestions ("Want me to pull more from that Lex Fridman episode?")
 **Patterns to follow:**
 - Step 0.55 subreddit resolution pattern
 - Source-specific synthesis guidance (YouTube highlights, Reddit top comments)
 **Test scenarios:**
 - Test expectation: none - SKILL.md is an instruction document. Verification is manual E2E.
 **Verification:**
 - `/last30days NVIDIA` resolves tech podcast channels and passes them to engine
 - `/last30days Kanye West` resolves hip-hop podcast channels
 - `/last30days knitting` resolves craft podcast channels (Fruity Knitting, etc.)
 - Stats show 🎙️ Podcasts line. Synthesis quotes podcast content with speaker attribution.
 ## Risks & Dependencies
 | Risk | Mitigation |
 |------|------------|
 | LLM guesses wrong @handle | Handle-first resolution with search fallback. 92% first-attempt success in testing, 100% with fallback. Wrong handles fail fast and skip silently. |
 | Transcript scanning adds latency | Runs in parallel with all other sources. 4 channels x 3 episodes = ~30-40s parallelized. Invisible in a 3-minute research run. |
 | Topic mentions below threshold (lots of misses) | LLM picks channels likely to discuss the topic. When it picks well, hit rate is high (4/14 episodes in NVIDIA test). Misses cost ~7s per episode in wasted caption download - acceptable. |
 | YouTube throttles caption downloads | 14 sequential downloads showed no throttling. Capping at 4 parallel workers adds safety margin. If throttled, degrade gracefully (fewer episodes scanned). |
 | Niche topics have no relevant podcast channels | LLM returns fewer channels (3-4 instead of 10-12). If none can be resolved, podcast source returns empty. Other sources (Reddit, X, YouTube) still run. |
 | Same video in both YouTube and podcast results | Fusion deduplicates by `yt_{video_id}`. Podcast version gets 0.88 quality score vs YouTube's 0.85, so podcast version wins dedup. |
 ## Sources & References
 - POC: transcript scan of 5 Acquired episodes found ESPN (117), Netflix (102), Taylor Swift (18), LVMH (27) - all invisible to search
 - POC: E2E NVIDIA test across 4 channels found 5 hits, 3 invisible to search (including 156-mention Dwarkesh Patel episode)
 - POC: handle resolution tested 12 channels (tech, hip-hop, knitting) - 11/12 first-attempt, 12/12 with fallback
 - Related code: `scripts/lib/youtube_yt.py`, `scripts/lib/pipeline.py`, `scripts/lib/hackernews.py`
 - Pattern: SKILL.md Step 0.55 subreddit resolution
 - yt-dlp docs: https://github.com/yt-dlp/yt-dlp
 - Acquired FM: https://www.youtube.com/@AcquiredFM
@@ -170,6 +170,7 @@ def build_parser() -> argparse.ArgumentParser:
                        help="Use web search to discover subreddits/handles before planning (for platforms without WebSearch)")
    parser.add_argument("--github-user", help="GitHub username for person-mode search (e.g., steipete)")
    parser.add_argument("--github-repo", help="Comma-separated owner/repo for project-mode search (e.g., openclaw/openclaw,paperclipai/paperclip)")
    parser.add_argument("--podcast-channels", help="Comma-separated YouTube @handles for podcast transcript scanning (e.g., AcquiredFM,lexfridman,DwarkeshPatel)")
    return parser
@@ -308,6 +309,7 @@ def main() -> int:
        github_user = args.github_user.lstrip("@").lower() if args.github_user else None
        github_repos = [r.strip() for r in args.github_repo.split(",") if r.strip() and "/" in r.strip()] if args.github_repo else None
        podcast_channels = [c.strip().lstrip("@") for c in args.podcast_channels.split(",") if c.strip()] if args.podcast_channels else None
        # --deep-research: auto-enable perplexity source and set deep flag
        if args.deep_research:
@@ -337,6 +339,7 @@ def main() -> int:
            lookback_days=args.lookback_days,
            github_user=github_user,
            github_repos=github_repos,
            podcast_channels=podcast_channels,
        )
    except Exception as exc:
        progress.end_processing()
@@ -53,6 +53,7 @@ def normalize_source_items(
        "xiaohongshu": _normalize_grounding,
        "github": _normalize_github,
        "perplexity": _normalize_grounding,
        "podcasts": lambda s, i, idx, fd, td: _normalize_youtube(s, i, idx, fd, td),
    }
    normalizer = normalizers.get(source)
    if normalizer is None:
@@ -40,6 +40,7 @@ from . import (
    xai_x,
    xiaohongshu_api,
    xquik,
    podcast_yt,
    youtube_yt,
 )
 from .cluster import cluster_candidates
@@ -77,6 +78,7 @@ MOCK_AVAILABLE_SOURCES = [
    "github",
    "perplexity",
    "xquik",
    "podcasts",
 ]
@@ -122,6 +124,11 @@ def available_sources(config: dict[str, Any], requested_sources: list[str] | Non
        available.append("pinterest")
    if env.is_xquik_available(config):
        available.append("xquik")
    # Podcasts: available whenever yt-dlp is installed (same as YouTube).
    # Opt-out only. The source returns empty when no channels are resolved,
    # so there's no cost to having it available.
    if podcast_yt.is_available():
        available.append("podcasts")
    return available
@@ -177,6 +184,7 @@ def run(
    lookback_days: int = 30,
    github_user: str | None = None,
    github_repos: list[str] | None = None,
    podcast_channels: list[str] | None = None,
 ) -> schema.Report:
    settings = DEPTH_SETTINGS[depth]
    requested_sources = normalize_requested_sources(requested_sources)
@@ -318,6 +326,7 @@ def run(
                        tiktok_hashtags=tiktok_hashtags,
                        tiktok_creators=tiktok_creators,
                        ig_creators=ig_creators,
                        podcast_channels=podcast_channels,
                    )
                ] = (subquery, source)
@@ -348,6 +357,7 @@ def run(
                            tiktok_hashtags=tiktok_hashtags,
                            tiktok_creators=tiktok_creators,
                            ig_creators=ig_creators,
                            podcast_channels=podcast_channels,
                        )
                    except Exception as retry_exc:
                        bundle.errors_by_source[source] = f"{exc} (retried once, still failed: {retry_exc})"
@@ -787,6 +797,7 @@ def _retrieve_stream(
    tiktok_hashtags: list[str] | None = None,
    tiktok_creators: list[str] | None = None,
    ig_creators: list[str] | None = None,
    podcast_channels: list[str] | None = None,
 ) -> tuple[list[dict], dict]:
    # Early exit if source was rate-limited by a sibling future
    if rate_limited_sources is not None and source in rate_limited_sources:
@@ -874,6 +885,13 @@ def _retrieve_stream(
            sc_token = config.get("SCRAPECREATORS_API_KEY", "")
            youtube_yt.enrich_with_comments(items, token=sc_token)
        return items, {}
    if source == "podcasts":
        podcast_query = raw_topic or subquery.search_query
        result = podcast_yt.search_podcast_youtube(
            podcast_query, from_date, to_date,
            depth=depth, channels=podcast_channels,
        )
        return result.get("items", []), {}
    if source == "tiktok":
        # Use raw_topic so expand_tiktok_queries() generates diverse variants
        # from the original user topic, not the planner's narrowed search_query.
@@ -71,6 +71,7 @@ SOURCE_CAPABILITIES = {
    "github": {"discussion", "link"},
    "grounding": {"web", "reference", "link"},
    "perplexity": {"web", "reference", "analysis"},
    "podcasts": {"discussion", "video_longform", "expert"},
 }
 DEFAULT_INTENT_CAPABILITIES = {
    "comparison": {"discussion", "video", "web", "reference", "social", "link", "market"},
@@ -0,0 +1,430 @@
 """YouTube podcast discovery via transcript scanning.
 Discovers podcast content by fetching auto-captions from LLM-resolved
 YouTube podcast channels and grepping for the search topic. Finds content
 invisible to title-based search — e.g., Acquired's "The NFL" episode
 mentions Taylor Swift 18 times, ESPN 117 times, Netflix 102 times.
 Uses yt-dlp for channel playlist fetch + caption download. No API keys.
 Reuses transcript highlight extraction from youtube_yt.
 """
 import math
 import os
 import re
 import shutil
 import signal
 import subprocess
 import sys
 import tempfile
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import Any, Dict, List, Optional
 from . import log
 # How many recent episodes to scan per channel, by depth
 EPISODES_PER_CHANNEL = {
    "quick": 2,
    "default": 3,
    "deep": 4,
 }
 # Minimum topic mentions in captions to count as a hit
 MENTION_THRESHOLD = 5
 # Max total results to return
 RESULTS_CAP = {
    "quick": 4,
    "default": 8,
    "deep": 20,
 }
 # Min duration in seconds to qualify as a podcast episode
 MIN_DURATION = 1200  # 20 minutes
 def _log(msg: str):
    log.source_log("Podcasts", msg, tty_only=False)
 def is_available() -> bool:
    """Podcast source is available when yt-dlp is installed."""
    return shutil.which("yt-dlp") is not None
 def resolve_channel(handle: str) -> Optional[str]:
    """Resolve a YouTube @handle to a channel URL.
    Tries the @handle directly first (fast, ~92% success rate).
    Falls back to ytsearch1 if the handle doesn't resolve.
    Returns the channel URL (https://www.youtube.com/channel/...) or None.
    """
    # Try @handle directly - use the channel/videos URL format
    # yt-dlp can fetch from @handle URLs directly for playlist operations
    direct_url = f"https://www.youtube.com/@{handle}/videos"
    try:
        result = subprocess.run(
            ["yt-dlp", "--playlist-end", "1",
             "--print", "%(channel_url)s",
             "--no-download", "--no-warnings", "--ignore-config", "--no-cookies-from-browser",
             direct_url],
            capture_output=True, text=True, timeout=20,
        )
        channel_url = result.stdout.strip().split("\n")[0].strip()
        if channel_url and channel_url.startswith("http"):
            _log(f"Resolved @{handle} -> {channel_url}")
            return channel_url
    except (subprocess.TimeoutExpired, FileNotFoundError):
        pass
    # Fallback: search for the podcast
    _log(f"@{handle} not found, trying search fallback")
    try:
        result = subprocess.run(
            ["yt-dlp", "--flat-playlist", "--playlist-end", "1",
             "--print", "%(channel_url)s",
             f'ytsearch1:"{handle}" podcast full episode'],
            capture_output=True, text=True, timeout=20,
        )
        channel_url = result.stdout.strip()
        if channel_url and channel_url.startswith("http"):
            _log(f"Search fallback resolved {handle} -> {channel_url}")
            return channel_url
    except (subprocess.TimeoutExpired, FileNotFoundError):
        pass
    _log(f"Could not resolve channel: {handle}")
    return None
 def _fetch_recent_episodes(
    channel_url: str,
    limit: int,
    from_date: str,
    to_date: str,
 ) -> List[Dict[str, Any]]:
    """Fetch recent long-form episodes from a channel.
    Returns list of dicts with video_id, title, channel, duration, date, views, likes.
    Filters to episodes with duration >= MIN_DURATION.
    """
    import json as _json
    try:
        result = subprocess.run(
            ["yt-dlp", f"--playlist-end={limit + 2}",
             "--dump-json", "--no-download", "--no-warnings", "--ignore-config", "--no-cookies-from-browser",
             f"{channel_url}/videos"],
            capture_output=True, text=True, timeout=60,
        )
    except (subprocess.TimeoutExpired, FileNotFoundError):
        return []
    episodes = []
    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", "")
        title = video.get("title", "")
        channel = video.get("channel", video.get("uploader", ""))
        duration = video.get("duration") or 0
        upload_date_raw = video.get("upload_date", "")
        views = video.get("view_count") or 0
        likes = video.get("like_count") or 0
        # Convert YYYYMMDD to YYYY-MM-DD
        date_str = None
        if upload_date_raw and len(upload_date_raw) >= 8:
            date_str = f"{upload_date_raw[:4]}-{upload_date_raw[4:6]}-{upload_date_raw[6:8]}"
        # Filter: duration >= MIN_DURATION
        if duration < MIN_DURATION:
            continue
        # Filter: within date range (soft - keep if no date available)
        if date_str and (date_str < from_date or date_str > to_date):
            continue
        episodes.append({
            "video_id": video_id,
            "title": title,
            "channel_name": channel,
            "duration": duration,
            "date": date_str,
            "views": views,
            "likes": likes,
            "url": f"https://www.youtube.com/watch?v={video_id}",
        })
    return episodes[:limit]
 def _fetch_captions(video_id: str, temp_dir: str) -> Optional[str]:
    """Fetch auto-captions for a video. Returns caption text or None."""
    out_template = os.path.join(temp_dir, f"cap_{video_id}")
    try:
        subprocess.run(
            ["yt-dlp", "--write-auto-sub", "--sub-lang", "en",
             "--skip-download", "--sub-format", "vtt",
             "-o", out_template,
             f"https://www.youtube.com/watch?v={video_id}"],
            capture_output=True, text=True, timeout=30,
        )
    except (subprocess.TimeoutExpired, FileNotFoundError):
        return None
    vtt_path = f"{out_template}.en.vtt"
    if not os.path.exists(vtt_path):
        return None
    try:
        with open(vtt_path, "r", encoding="utf-8") as f:
            text = f.read()
        os.remove(vtt_path)
        # Strip VTT formatting: timestamps, alignment, tags, duplicate lines
        # VTT auto-captions repeat lines as they scroll, so deduplicate
        lines = []
        prev_line = ""
        for line in text.split("\n"):
            line = line.strip()
            if not line:
                continue
            if line.startswith("WEBVTT") or line.startswith("Kind:") or line.startswith("Language:"):
                continue
            if re.match(r"^\d{2}:\d{2}:", line):
                continue
            if re.match(r"^NOTE\b", line):
                continue
            if "align:" in line or "position:" in line:
                continue
            # Strip inline VTT tags like <c>, </c>, timestamps
            cleaned = re.sub(r"<[^>]+>", "", line)
            cleaned = cleaned.strip()
            if cleaned and not re.match(r"^\d+$", cleaned) and cleaned != prev_line:
                lines.append(cleaned)
                prev_line = cleaned
        return " ".join(lines)
    except Exception:
        return None
 _NOISE_WORDS = frozenset({
    "the", "a", "an", "of", "and", "or", "for", "to", "in", "on", "at",
    "best", "top", "new", "latest", "review", "news", "vs", "versus",
    "album", "song", "episode", "podcast", "interview", "this", "that",
    "what", "how", "why", "where", "when", "who",
 })
 def _extract_key_terms(topic: str) -> List[str]:
    """Extract meaningful terms from topic for matching.
    For "Kanye West Bully album" -> ["Kanye West", "Bully"] or similar.
    For single words, just returns the word.
    """
    words = [w.strip() for w in topic.split() if w.strip()]
    # Remove noise words
    meaningful = [w for w in words if w.lower() not in _NOISE_WORDS and len(w) > 2]
    if not meaningful:
        return [topic.strip()]
    # If the topic has 2+ meaningful words, also include the full phrase
    # and the first 2 words as a potential entity name
    terms = []
    if len(meaningful) >= 2:
        # Full phrase first (for exact entity matches like "Taylor Swift")
        terms.append(" ".join(meaningful[:2]))
    terms.extend(meaningful)
    return terms
 def _count_mentions(text: str, topic: str) -> int:
    """Count case-insensitive topic mentions in text.
    Uses the maximum mention count across key terms extracted from the topic.
    "Kanye West Bully album" -> max mentions of ["Kanye West", "Kanye", "West", "Bully"].
    This way, an episode mentioning "Kanye" 85 times counts as 85, not 0.
    """
    text_lower = text.lower()
    terms = _extract_key_terms(topic)
    max_count = 0
    for term in terms:
        pattern = re.escape(term.lower())
        count = len(re.findall(pattern, text_lower))
        if count > max_count:
            max_count = count
    return max_count
 def _extract_mention_context(text: str, topic: str, max_excerpts: int = 3) -> List[str]:
    """Extract text snippets around topic mentions for highlights."""
    words = text.split()
    topic_lower = topic.lower()
    excerpts = []
    for i, word in enumerate(words):
        # Check if we're near a mention
        window = " ".join(words[max(0, i - 5):i + 15]).lower()
        if topic_lower in window and len(excerpts) < max_excerpts:
            start = max(0, i - 10)
            end = min(len(words), i + 30)
            excerpt = " ".join(words[start:end])
            # Avoid duplicate excerpts
            if not any(excerpt[:50] in e for e in excerpts):
                excerpts.append(excerpt)
    return excerpts
 def _scan_channel(
    handle: str,
    topic: str,
    from_date: str,
    to_date: str,
    episodes_limit: int,
 ) -> List[Dict[str, Any]]:
    """Scan a single channel's recent episodes for topic mentions.
    Returns list of hit items with mention_count and transcript data.
    """
    # Step 1: Resolve channel handle to URL
    channel_url = resolve_channel(handle)
    if not channel_url:
        return []
    # Step 2: Fetch recent long-form episodes
    episodes = _fetch_recent_episodes(channel_url, episodes_limit, from_date, to_date)
    if not episodes:
        _log(f"No recent long-form episodes from {handle}")
        return []
    _log(f"Scanning {len(episodes)} episodes from {handle}")
    # Step 3: Fetch captions and grep for topic
    hits = []
    with tempfile.TemporaryDirectory() as temp_dir:
        for ep in episodes:
            caption_text = _fetch_captions(ep["video_id"], temp_dir)
            if not caption_text:
                continue
            mention_count = _count_mentions(caption_text, topic)
            if mention_count < MENTION_THRESHOLD:
                continue
            # Extract highlights around the mentions
            from .youtube_yt import extract_transcript_highlights
            highlights = extract_transcript_highlights(caption_text, topic, limit=5)
            mention_excerpts = _extract_mention_context(caption_text, topic)
            # Cap transcript for storage
            words = caption_text.split()
            transcript_snippet = " ".join(words[:5000]) if len(words) > 5000 else caption_text
            hits.append({
                "video_id": ep["video_id"],
                "title": ep["title"],
                "channel_name": ep["channel_name"],
                "url": ep["url"],
                "date": ep["date"],
                "duration": ep["duration"],
                "engagement": {
                    "views": ep["views"],
                    "likes": ep["likes"],
                },
                "mention_count": mention_count,
                "transcript_snippet": transcript_snippet,
                "transcript_highlights": highlights,
                "mention_excerpts": mention_excerpts,
                "relevance": min(1.0, mention_count / 50),
                "why_relevant": f"Podcast: {ep['channel_name']} - {ep['title'][:60]} ({mention_count} mentions)",
            })
            _log(f"  HIT: {ep['title'][:60]} ({mention_count} mentions)")
    return hits
 def search_podcast_youtube(
    topic: str,
    from_date: str,
    to_date: str,
    depth: str = "default",
    channels: Optional[List[str]] = None,
 ) -> Dict[str, Any]:
    """Discover podcast content by scanning transcripts of resolved channels.
    Args:
        topic: Search topic
        from_date: Start date (YYYY-MM-DD)
        to_date: End date (YYYY-MM-DD)
        depth: 'quick', 'default', or 'deep'
        channels: List of YouTube @handles to scan
    Returns:
        Dict with 'items' list. Each item has transcript and mention data.
    """
    if not is_available():
        _log("yt-dlp not installed")
        return {"items": [], "error": "yt-dlp not installed"}
    if not channels:
        _log("No podcast channels provided")
        return {"items": []}
    episodes_limit = EPISODES_PER_CHANNEL.get(depth, EPISODES_PER_CHANNEL["default"])
    results_cap = RESULTS_CAP.get(depth, RESULTS_CAP["default"])
    _log(f"Scanning {len(channels)} podcast channels for '{topic}' (depth={depth}, {episodes_limit} eps/channel)")
    # Scan channels in parallel
    all_hits: List[Dict[str, Any]] = []
    max_workers = min(4, len(channels))
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(
                _scan_channel, handle, topic, from_date, to_date, episodes_limit,
            ): handle
            for handle in channels
        }
        for future in as_completed(futures):
            handle = futures[future]
            try:
                hits = future.result()
                all_hits.extend(hits)
            except Exception as exc:
                _log(f"Error scanning {handle}: {type(exc).__name__}: {exc}")
    # Deduplicate by video_id
    seen = set()
    unique_hits = []
    for hit in all_hits:
        vid = hit["video_id"]
        if vid not in seen:
            seen.add(vid)
            unique_hits.append(hit)
    # Score: mention_count * log(views + 1)
    for hit in unique_hits:
        views = hit["engagement"].get("views", 0)
        hit["_score"] = hit["mention_count"] * math.log(views + 1)
    # Sort by score descending
    unique_hits.sort(key=lambda x: x["_score"], reverse=True)
    # Cap results
    results = unique_hits[:results_cap]
    # Clean up internal scoring field
    for hit in results:
        hit.pop("_score", None)
    _log(f"Found {len(results)} podcast hits across {len(channels)} channels")
    return {"items": results}
@@ -14,6 +14,7 @@ SOURCE_LABELS = {
    "x": "X",
    "github": "GitHub",
    "perplexity": "Perplexity",
    "podcasts": "Podcasts",
 }
@@ -19,6 +19,7 @@ SOURCE_QUALITY = {
    "polymarket": 0.5,
    "instagram": 0.58,
    "tiktok": 0.58,
    "podcasts": 0.88,
 }