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

apps/core/pyproject.toml

@@ -24,6 +24,7 @@ dependencies = [
     "python-dotenv>=1.1",
     "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
+    "yt-dlp>=2026.07.04",  # YouTube search connector (public API only)
     "ulid>=1.1",
 ]
 

apps/core/sources/connectors/__init__.py

@@ -2,11 +2,13 @@
 from .hackernews import HackerNewsCommentConnector, HackerNewsFeedConnector
 from .reddit import RedditSubRedditConnector
 from .rss import RssConnector
+from .youtube import YouTubeSearchConnector
 
 __all__ = [
     "Connector",
     "HackerNewsCommentConnector",
     "HackerNewsFeedConnector",
     "RedditSubRedditConnector",
     "RssConnector",
+    "YouTubeSearchConnector",
 ]

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

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

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

@@ -0,0 +1,95 @@
+"""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
+
+import yt_dlp
+
+from .errors import YouTubeError, 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) -> 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,
+        }
+        if self._cookie_file:
+            opts["cookies"] = self._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)
+        search_uri = f"ytsearch{capped_count}:{query}"
+
+        try:
+            with yt_dlp.YoutubeDL(self._build_opts()) as ydl:
+                info = ydl.extract_info(search_uri, download=False)
+                entries = info.get("entries", []) or []
+                return [e for e in entries if e is not None]
+        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 YouTubeError(
+                code=err.code,
+                message=err.message,
+                retryable=err.retryable,
+                action=err.action,
+                context=err.context,
+            ) from exc

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

@@ -0,0 +1,83 @@
+"""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 and the connector filters them by the watermark (YouTube's
+    search ordering is newest-first; a quiet stream needs no backfill
+    walk).
+    """
+
+    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: only surface videos strictly newer than the
+            # cursor (the poll op advances the source watermark to the
+            # newest seen, so a video at the watermark is already recorded).
+            if since is not None and payload.occurred_at <= since:
+                continue
+            yield payload
+
+
+register(YouTubeSearchConnector.kind, YouTubeSearchConnector.payloads)

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

@@ -0,0 +1,92 @@
+"""Error taxonomy for the YouTube (yt-dlp) connector.
+
+Maps yt-dlp exceptions to canonical error shapes with retry semantics,
+following the same pattern as the Twitter connector's ListenerError.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class YouTubeError:
+    """Canonical error shape for one YouTube fetch failure."""
+
+    code: str  # stable machine code
+    message: str  # human-readable
+    retryable: bool  # safe to retry with backoff?
+    action: str  # what the ops layer should do
+    context: dict[str, Any] = field(default_factory=dict)
+
+
+# Error codes for YouTube-specific failures.
+YT_DLP_ERROR_CODES: dict[type[Exception], str] = {
+    # yt-dlp DownloadError subclasses
+    Exception: "yt_dlp_error",  # catch-all
+}
+
+
+def map_ytdlp_error(exc: Exception, context: dict[str, Any] | None = None) -> YouTubeError:
+    """Translate an yt-dlp exception into a canonical YouTubeError."""
+    msg = str(exc)
+
+    # Video not available (region-restricted, deleted, private)
+    if "This video is not available" in msg or "Video unavailable" in msg:
+        return YouTubeError(
+            code="video_unavailable",
+            message=msg,
+            retryable=False,
+            action="skip (video no longer available)",
+            context=context or {},
+        )
+
+    # Rate limiting / throttling
+    if "rate limited" in msg.lower() or "too many requests" in msg.lower():
+        return YouTubeError(
+            code="rate_limited",
+            message=msg,
+            retryable=True,
+            action="retry with exponential backoff",
+            context=context or {},
+        )
+
+    # Missing JavaScript runtime (warning only, still works in degraded mode)
+    if "No supported JavaScript runtime" in msg:
+        return YouTubeError(
+            code="js_runtime_missing",
+            message=msg,
+            retryable=False,
+            action="install deno or node; proceeding in degraded mode",
+            context=context or {},
+        )
+
+    # Network/connection errors
+    if any(marker in msg.lower() for marker in ["connection", "timeout", "network", "urlopen"]):
+        return YouTubeError(
+            code="network_error",
+            message=msg,
+            retryable=True,
+            action="retry with backoff",
+            context=context or {},
+        )
+
+    # Upload date parsing failures
+    if "upload_date" in msg.lower() or "date" in msg.lower():
+        return YouTubeError(
+            code="date_parse_error",
+            message=msg,
+            retryable=False,
+            action="use current timestamp as fallback",
+            context=context or {},
+        )
+
+    # Generic fallback
+    return YouTubeError(
+        code="yt_dlp_error",
+        message=msg,
+        retryable=True,
+        action="log and retry with backoff",
+        context=context or {},
+    )