feat(sources): add YouTube search connector (yt-dlp)

.gitignore

@@ -34,6 +34,11 @@ venv/
 .env.bak
 *.bak
 
+# Connector credential material (cookie exports, session files); only the
+# README documenting the layout is tracked.
+apps/core/credentials/*
+!apps/core/credentials/README.md
+
 # Django
 *.sqlite3
 db.sqlite3-journal

README.md

@@ -28,15 +28,15 @@
 
 ## What it does
 
-You scan X/Twitter, Reddit, Hacker News, and a few RSS feeds looking for someone hitting a problem your product solves or asking a question you can answer well. Getting there while the conversation is happening is how you build a brand and a community around what you know. OpenMagpie watches the threads for you so you spend your time on engagement instead of searching.
+You scan X/Twitter, YouTube, Reddit, Hacker News, and a few RSS feeds looking for someone talking about your product, hitting a problem it solves, or asking a question you can answer well. Getting there while the conversation is happening is how you build a brand and a community around what you know: a mention answered the day it's posted beats one found in next month's report. OpenMagpie watches the threads for you so you spend your time on engagement instead of searching.
 
 You curate sources into a feed, write a natural-language description of what's relevant (for example, "someone frustrated with manual social monitoring and asking for alternatives"), and a local LLM run via any OpenAI-compatible runner (e.g. Ollama, vLLM, LM Studio) scores each new post against it. Matches go to a webhook or your logs (more integrations coming); everything else is dropped. You read the hits instead of the firehose.
 
 ## Where it listens
 
 OpenMagpie listens wherever communities are having those conversations.
 
-- **Public discussion (today):** X/Twitter, Reddit, Hacker News, and any RSS or Atom feed (news, blogs, Substack publications, and forums that publish feeds).
+- **Public discussion (today):** X/Twitter, YouTube, Reddit, Hacker News, and any RSS or Atom feed (news, blogs, Substack publications, and forums that publish feeds).
 - **Communities you're in (roadmap):** Slack workspaces and LinkedIn you already belong to, so you catch relevant threads in the groups where you participate, no admin or app install required.
 - **Public discussion (soon to be added):** Facebook, TikTok, and Instagram.
 
@@ -144,6 +144,7 @@ A `Feed` is a reusable, curated stream (a set of sources plus an item log). A `W
 graph TD
     subgraph Sources
         TWITTER[X / Twitter]
+        YOUTUBE[YouTube]
         REDDIT[Reddit]
         RSS[RSS / Atom feeds]
         HN[Hacker News]
@@ -170,6 +171,7 @@ graph TD
     end
 
     TWITTER --> FEED
+    YOUTUBE --> FEED
     REDDIT --> FEED
     RSS --> FEED
     HN --> FEED
@@ -231,7 +233,7 @@ Social listening is a crowded market (Brand24, Mention, Octolens, Syften, and to
 
 | Layer | Shipped |
 |---|---|
-| Connectors | X/Twitter (`twitter_search`), Reddit (`reddit_subreddit`), Hacker News (`hn_feed`, `hn_comment`), RSS/Atom (`rss`) |
+| Connectors | X/Twitter (`twitter_search`), YouTube (`youtube_search`), Reddit (`reddit_subreddit`), Hacker News (`hn_feed`, `hn_comment`), RSS/Atom (`rss`) |
 | Engines | Any OpenAI-compatible `/v1` API: Ollama, vLLM, llama.cpp, LM Studio, OpenAI, ... |
 | Action kinds | `semantic_filter` (LLM-judged), `webhook`, `log` |
 | Delivery modes | instant, digest |
@@ -247,6 +249,15 @@ X/Twitter listening is the first connector added beyond the original Reddit / HN
 - **Reliability fixes from live polling** — a per-call twikit client (multi-source polls no longer crash with "Event loop is closed") and X's transient empty-body 404 mapped retryable instead of "tweet deleted", with a regression test. 587 tests green; all CI gates pass.
 - **Verified live end-to-end** — a real X poll through a feed → watch → webhook chain delivered 44/44 items with HTTP 200, payload matched field-for-field against the Twenty `socialEvent` intake contract (`item.handle → actorHandle`, `author → actorName`, `content → eventText`, `occurred_at → occurredAt`, `url → sourceUrl`, `key → dedupeKey`).
 
+YouTube listening followed via yt-dlp:
+
+- **`youtube_search` source kind** — a yt-dlp-based connector that runs YouTube search queries and maps results to a schema-parity `NewVideoPayload`, registered alongside the existing kinds with the same feed/watch/webhook pipeline.
+- **No authentication required** — public YouTube search works without credentials; optional cookie file for age-restricted content.
+- **Error taxonomy** — 5 error codes (`video_unavailable`, `rate_limited`, `js_runtime_missing`, `network_error`, `yt_dlp_error`) with retry semantics.
+- **Watermark-based deduplication** — videos newer than the source's `last_event_at` are surfaced.
+- **Metrics extraction** — views, likes, comments mapped from YouTube metadata.
+- **Thumbnail media** — full thumbnail URLs attached to payloads for rich display.
+
 Next up on the roadmap: **Facebook, TikTok, and Instagram connectors** (soon to be added), then Slack, LinkedIn, GitHub, Bluesky, and Mastodon.
 
 ## Roadmap

apps/core/Dockerfile

@@ -57,9 +57,11 @@ ENV PYTHONUNBUFFERED=1 \
     UV_PYTHON_DOWNLOADS=never \
     UV_PYTHON_PREFERENCE=only-system
 
-# Runtime shared lib for the compiled psycopg[c].
+# Runtime shared lib for the compiled psycopg[c], plus git: the dev compose
+# flow re-syncs the venv in this image (`uv run` without --no-sync over the
+# mounted workspace), which fetches the git-pinned twikit dep.
 RUN apt-get update \
-    && apt-get install -y --no-install-recommends libpq5 \
+    && apt-get install -y --no-install-recommends git libpq5 \
     && rm -rf /var/lib/apt/lists/*
 
 WORKDIR /app

apps/core/conf/settings/base.py

@@ -518,6 +518,13 @@
 # trade-off.
 SOURCE_ALLOW_INSECURE_TLS = env_bool("SOURCE_ALLOW_INSECURE_TLS", "false")
 
+# Path to a Netscape-format cookies.txt for the YouTube connector's yt-dlp
+# extraction. Public search needs no auth; cookies only widen coverage to
+# entries whose extraction requires a signed-in session (age-gated videos,
+# occasional bot challenges), which are otherwise skipped. Use a throwaway
+# Google account. Empty (the default) disables it.
+YOUTUBE_COOKIES_FILE = os.environ.get("YOUTUBE_COOKIES_FILE", "")
+
 # Product telemetry (anonymous, opt-out; see apps/core/telemetry + TELEMETRY.md).
 # POSTHOG_API_KEY defaults to the baked-in PUBLIC, WRITE-ONLY PostHog project key
 # (OpenMagpie's anonymous self-hosted project, PostHog Cloud US) so a self-hoster

apps/core/credentials/README.md

@@ -0,0 +1,26 @@
+# Connector credentials
+
+Session material some connectors can use: cookie exports, per-session proxy
+pins. Everything in this directory except this README is gitignored — nothing
+you put here can be committed.
+
+One subdirectory per connector:
+
+```
+credentials/
+  twitter/    # x.com cookie exports (*.json), optional <name>.proxy pins
+  youtube/    # Netscape-format cookies.txt for age-gated extraction
+```
+
+Point the connector settings here with absolute paths (inside the dev
+containers the repo is mounted at /app):
+
+```
+TWITTER_CREDENTIALS_DIR=/app/apps/core/credentials/twitter
+YOUTUBE_COOKIES_FILE=/app/apps/core/credentials/youtube/cookies.txt
+```
+
+It's recommended to use throwaway accounts for any cookies that land here:
+platforms flag and sometimes lock accounts whose sessions show up in
+automated traffic. These connectors use unofficial routes that may conflict
+with a platform's terms of service — use at your own risk.

apps/core/feeds/tests_plugin_source_kinds.py

@@ -27,6 +27,7 @@
     RssSourceSpec,
     SourceSpec,
     TwitterSearchSourceSpec,
+    YouTubeSearchSourceSpec,
     _BuiltinSourceSpec,
     canonical_spec,
 )
@@ -319,6 +320,7 @@ def test_builtin_source_kinds_are_exactly_the_known_builtins(self) -> None:
                     HackerNewsFeedSourceSpec.SOURCE_KIND,
                     HackerNewsCommentSourceSpec.SOURCE_KIND,
                     TwitterSearchSourceSpec.SOURCE_KIND,
+                    YouTubeSearchSourceSpec.SOURCE_KIND,
                 }
             ),
         )

apps/core/pyproject.toml

@@ -25,6 +25,7 @@ dependencies = [
     "pyyaml>=6.0",  # reads the examples/starters/*.yaml in seed_quickstart
     "trafilatura>=1.7",  # HTML -> readable article text for the engine's lazy external-link fetch
     "twikit @ git+https://github.com/unclecode/twikit.git",  # X (Twitter) unofficial route (listeningkit-verified 2026 fork of d60/twikit)
+    "yt-dlp>=2026.07.04",  # YouTube search connector (public API only)
     "ulid>=1.1",
 ]
 

apps/core/sources/connectors/__init__.py

@@ -3,6 +3,7 @@
 from .reddit import RedditSubRedditConnector
 from .rss import RssConnector
 from .twitter import TwitterSearchConnector
+from .youtube import YouTubeSearchConnector
 
 __all__ = [
     "Connector",
@@ -11,4 +12,5 @@
     "RedditSubRedditConnector",
     "RssConnector",
     "TwitterSearchConnector",
+    "YouTubeSearchConnector",
 ]

apps/core/sources/connectors/youtube/__init__.py

@@ -0,0 +1,7 @@
+from .connector import YouTubeSearchConnector
+from .payloads import NewVideoPayload
+
+__all__ = [
+    "NewVideoPayload",
+    "YouTubeSearchConnector",
+]

apps/core/sources/connectors/youtube/client.py

@@ -0,0 +1,111 @@
+"""yt-dlp-based YouTube client for search extraction.
+
+Wraps yt-dlp's YoutubeDL to perform YouTube searches without downloading
+video content. Uses extract_flat mode for efficiency and handles errors
+via the error taxonomy in errors.py.
+
+Key patterns (ported from listeningkit Twitter client):
+- One YtDlpClient instance per search call; yt-dlp is thread-safe for
+  read-only extraction operations.
+- Search queries use the `ytsearch<N>:<query>` URI scheme.
+- Results are returned as dicts (not downloaded), containing metadata.
+- No authentication required for public search; cookies optional for
+  age-restricted content.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+from urllib.parse import urlencode
+
+import yt_dlp
+from django.conf import settings
+
+from .errors import map_ytdlp_error
+
+log = logging.getLogger("sources.youtube")
+
+# Maximum results per search query. yt-dlp accepts up to 100 but we cap
+# lower to match the Twitter connector's default count.
+MAX_SEARCH_RESULTS = 50
+
+
+class YtDlpClient:
+    """Thin wrapper around yt-dlp for search-only extraction.
+
+    No auth state: YouTube search is public. Optional cookie file can be
+    passed for age-restricted content (not commonly needed for search).
+    """
+
+    def __init__(
+        self,
+        *,
+        quiet: bool = True,
+        no_warnings: bool = True,
+        cookie_file: str | None = None,
+    ) -> None:
+        self._quiet = quiet
+        self._no_warnings = no_warnings
+        self._cookie_file = cookie_file
+
+    def _build_opts(self, count: int) -> dict[str, Any]:
+        opts: dict[str, Any] = {
+            "quiet": self._quiet,
+            "no_warnings": self._no_warnings,
+            "extract_flat": False,  # need full metadata for payloads
+            "skip_download": True,
+            "playlistend": count,  # the count cap for the URL-based search
+            # One unextractable entry (age-gated, region-locked, deleted) must
+            # not abort the whole result page; it comes back as a None entry,
+            # which search() filters out.
+            "ignoreerrors": True,
+        }
+        # Constructor arg wins (tests); else the env-backed setting. Read
+        # per-call, not at import, so @override_settings works and a rotated
+        # cookie file applies without a process restart.
+        cookie_file = self._cookie_file or settings.YOUTUBE_COOKIES_FILE
+        if cookie_file:
+            opts["cookies"] = cookie_file
+        return opts
+
+    def search(
+        self,
+        query: str,
+        count: int = 20,
+    ) -> list[dict[str, Any]]:
+        """Run one YouTube search; returns list of video info dicts.
+
+        Args:
+            query: Search expression (keywords, phrases).
+            count: Max results to fetch (capped at MAX_SEARCH_RESULTS).
+
+        Returns:
+            List of video metadata dicts, newest first.
+
+        Raises:
+            YouTubeError: On extraction failures (mapped from yt-dlp exceptions).
+        """
+        capped_count = min(count, MAX_SEARCH_RESULTS)
+        # sp=EgIIAw= is YouTube's "Upload date: This week" FILTER. YouTube
+        # removed sort-by-upload-date from search entirely (yt-dlp dropped
+        # ytsearchdate for the same reason, yt-dlp/yt-dlp#15898), so recency
+        # comes from restricting the window instead: results are
+        # relevance-ranked but only from the last 7 days, and the watermark +
+        # external_id dedup handle ordering. Old popular videos can't occupy
+        # the N slots; a very busy query should raise `count` since relevance
+        # picks which of the week's matches fill them.
+        search_uri = f"https://www.youtube.com/results?{urlencode({'search_query': query, 'sp': 'EgIIAw=='})}"
+
+        try:
+            with yt_dlp.YoutubeDL(self._build_opts(capped_count)) as ydl:
+                info = ydl.extract_info(search_uri, download=False)
+                entries = (info or {}).get("entries", []) or []
+                # Search results mix in playlists and channels; keep videos
+                # only (full extraction marks them _type "video", or omits
+                # _type on older yt-dlp versions).
+                return [e for e in entries if e is not None and e.get("_type") in (None, "video")]
+        except Exception as exc:
+            err = map_ytdlp_error(exc, {"query": query, "count": capped_count})
+            log.warning("youtube search failed query=%r code=%s: %s", query, err.code, err.message)
+            raise err from exc

apps/core/sources/connectors/youtube/connector.py

@@ -0,0 +1,89 @@
+"""YouTube search connector using yt-dlp.
+
+Polls a `youtube_search` source: one live YouTube search per cycle via
+the yt-dlp client, mapping each result video to a `NewVideoPayload`
+newer than the source's `since` watermark.
+
+Error semantics follow the connector contract: any YouTube/yt-dlp
+failure is raised as `ConnectorParseError` (a `_RECOVERABLE_ERRORS`
+member at the poll seam), so a bad source logs + skips instead of
+aborting the feed cycle. The source's watermark stays put on failure,
+so the next cycle re-reads from the same point and the external_id
+dedup absorbs anything already recorded.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable, Iterator
+from datetime import datetime
+
+from openmagpie_schema.configs import YouTubeSearchSourceSpec
+from sources.payload_registry import register
+from sources.payloads import SourcePayload
+
+from ..base import BaseConnector, ConnectorParseError
+from .client import YtDlpClient
+from .errors import YouTubeError
+from .payloads import NewVideoPayload
+
+log = logging.getLogger("sources.youtube")
+
+
+class YouTubeSearchConnector(BaseConnector[YouTubeSearchSourceSpec]):
+    """Polls one YouTube search stream via yt-dlp.
+
+    Live-mode semantics mirror the other connectors: every cycle yields
+    videos newer than `since` (the Source row's `last_event_at`). There
+    is no pagination in phase 1: a search returns up to `spec.count`
+    videos from the client's last-7-days window (YouTube search has no
+    date SORT any more, only the upload-window filter; see client.search)
+    and the connector filters them by the watermark. The window is the
+    recency guarantee; ordering within it is relevance, which the dedup
+    downstream absorbs.
+    """
+
+    kind = YouTubeSearchSourceSpec.SOURCE_KIND
+    payloads: list[type[SourcePayload]] = [NewVideoPayload]
+
+    # One stateless client; no auth needed for public search.
+    _client = YtDlpClient()
+
+    def poll(
+        self,
+        spec: YouTubeSearchSourceSpec,
+        since: datetime | None,
+        field_map: dict[str, str] | None = None,
+        heartbeat: Callable[[], bool] | None = None,
+    ) -> Iterator[SourcePayload]:
+        del field_map
+        del heartbeat
+        try:
+            results = self._client.search(spec.query, spec.count)
+        except YouTubeError as exc:
+            log.warning(
+                "youtube search failed query=%r code=%s retryable=%s: %s",
+                spec.query,
+                exc.code,
+                exc.retryable,
+                exc.message,
+            )
+            raise ConnectorParseError(
+                f"youtube search {spec.display()} failed: {exc.code}: {exc.message} ({exc.action})"
+            ) from exc
+
+        for video in results:
+            payload = NewVideoPayload.from_video(video)
+            # Watermark filter: skip only videos strictly OLDER than the
+            # cursor. YouTube timestamps can be day-granular (upload_date
+            # floors to midnight UTC), so the `<= since` rule the
+            # full-resolution connectors use would drop every later video
+            # from the same day once the watermark reaches that midnight.
+            # Yielding the boundary (`== since`) instead re-offers already
+            # recorded same-day videos, which the external_id dedup absorbs.
+            if since is not None and payload.occurred_at < since:
+                continue
+            yield payload
+
+
+register(YouTubeSearchConnector.kind, YouTubeSearchConnector.payloads)