diff --git a/.gitignore b/.gitignore index 69043bce..0ddc9afc 100644 --- a/.gitignore +++ b/.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 diff --git a/README.md b/README.md index b1e0de07..749b3d6b 100644 --- a/README.md +++ b/README.md @@ -28,7 +28,7 @@ ## 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. @@ -36,7 +36,7 @@ You curate sources into a feed, write a natural-language description of what's r 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 diff --git a/apps/core/Dockerfile b/apps/core/Dockerfile index 1b9c677d..3d287b2a 100644 --- a/apps/core/Dockerfile +++ b/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 diff --git a/apps/core/conf/settings/base.py b/apps/core/conf/settings/base.py index b37c4007..5e833906 100644 --- a/apps/core/conf/settings/base.py +++ b/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 diff --git a/apps/core/credentials/README.md b/apps/core/credentials/README.md new file mode 100644 index 00000000..847b9bb1 --- /dev/null +++ b/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 .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. diff --git a/apps/core/feeds/tests_plugin_source_kinds.py b/apps/core/feeds/tests_plugin_source_kinds.py index 1252133d..e869b333 100644 --- a/apps/core/feeds/tests_plugin_source_kinds.py +++ b/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, } ), ) diff --git a/apps/core/pyproject.toml b/apps/core/pyproject.toml index 739a7c6b..472a8c51 100644 --- a/apps/core/pyproject.toml +++ b/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", ] diff --git a/apps/core/sources/connectors/__init__.py b/apps/core/sources/connectors/__init__.py index 31939d1b..8b9fba35 100644 --- a/apps/core/sources/connectors/__init__.py +++ b/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", ] diff --git a/apps/core/sources/connectors/youtube/__init__.py b/apps/core/sources/connectors/youtube/__init__.py new file mode 100644 index 00000000..a1dae6f1 --- /dev/null +++ b/apps/core/sources/connectors/youtube/__init__.py @@ -0,0 +1,7 @@ +from .connector import YouTubeSearchConnector +from .payloads import NewVideoPayload + +__all__ = [ + "NewVideoPayload", + "YouTubeSearchConnector", +] diff --git a/apps/core/sources/connectors/youtube/client.py b/apps/core/sources/connectors/youtube/client.py new file mode 100644 index 00000000..8815308d --- /dev/null +++ b/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:` 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 diff --git a/apps/core/sources/connectors/youtube/connector.py b/apps/core/sources/connectors/youtube/connector.py new file mode 100644 index 00000000..ecb6fa80 --- /dev/null +++ b/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) diff --git a/apps/core/sources/connectors/youtube/errors.py b/apps/core/sources/connectors/youtube/errors.py new file mode 100644 index 00000000..678ae284 --- /dev/null +++ b/apps/core/sources/connectors/youtube/errors.py @@ -0,0 +1,83 @@ +"""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 typing import Any + + +class YouTubeError(Exception): + """Canonical error shape for one YouTube fetch failure.""" + + def __init__( + self, + *, + 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] | None = None, + ) -> None: + super().__init__(message) + self.code = code + self.message = message + self.retryable = retryable + self.action = action + self.context = context or {} + + +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 {}, + ) + + # Generic fallback + return YouTubeError( + code="yt_dlp_error", + message=msg, + retryable=True, + action="log and retry with backoff", + context=context or {}, + ) diff --git a/apps/core/sources/connectors/youtube/payloads.py b/apps/core/sources/connectors/youtube/payloads.py new file mode 100644 index 00000000..18730408 --- /dev/null +++ b/apps/core/sources/connectors/youtube/payloads.py @@ -0,0 +1,152 @@ +"""YouTube payloads: a video observed via yt-dlp search. + +Maps YouTube video metadata to the openmagpie SourcePayload contract: +the engine judges title + content, so the video's description goes to +content and the uploader's name becomes the within-kind source_slug. +Metrics / refs / media stay on the payload as source-specific fields. +""" + +from __future__ import annotations + +from contextlib import suppress +from datetime import UTC, datetime +from typing import Any, ClassVar + +from openmagpie_schema.configs import YouTubeSearchSourceSpec +from sources.payloads import SourcePayload + +# YouTube video URL base. +YOUTUBE_VIDEO_URL = "https://www.youtube.com/watch?v=" + + +class NewVideoPayload(SourcePayload): + """A single YouTube video observed by a watched search stream. + + `author` is the channel name; `handle` is the channel ID and the + within-kind source slug (grouping items by producing channel). + `content` is the video description (the engine's judgeable body). + The rest is source-specific: `metrics`, `refs` (related video IDs), + `media` (thumbnails), `duration`. + """ + + PAYLOAD_KIND: ClassVar[str] = "new_video" + + author: str = "" + handle: str = "" + duration: int = 0 # seconds + metrics: dict[str, int | None] = {} + refs: dict[str, str | None] = {} + media: list[dict[str, Any]] = [] + + model_config = {"frozen": True, "extra": "ignore"} + + def source_slug(self) -> str | None: + return self.handle or None + + @classmethod + def sample(cls, variant: int = 0) -> NewVideoPayload: + n = variant + 1 + video_id = str(999_000_000_000_000_000 + n) + handle = f"example_channel_{n}" + return cls( + external_id=video_id, + kind=cls.PAYLOAD_KIND, + occurred_at=datetime(2026, 5, 27, 12, 0, tzinfo=UTC), + source=YouTubeSearchSourceSpec.SOURCE_KIND, + title="", + content=f"Example YouTube video {n}: the description text that matched this watch.", + url=f"{YOUTUBE_VIDEO_URL}{video_id}", + author=f"Example Channel {n}", + handle=handle, + duration=60 * n, + metrics={"views": 1000 + n, "likes": 100 + n, "comments": 10 + n}, + refs={}, + media=[], + ) + + @classmethod + def from_video(cls, video: dict[str, Any]) -> NewVideoPayload: + """Map a yt-dlp video info dict to a payload. + + yt-dlp returns videos as plain dicts when extract_info is called + on a search URI. All attributes are accessed via dict get() with + defaults, so the connector's unit tests can hand in lightweight + fakes without importing yt-dlp. + """ + video_id = str(video.get("id") or "") + uploader = str(video.get("uploader") or video.get("channel") or "") + uploader_id = str(video.get("uploader_id") or video.get("channel_id") or "") + description = str(video.get("description") or "") + upload_date = str(video.get("upload_date") or "") + + # Timestamp resolution, best first: `timestamp`/`release_timestamp` + # (epoch seconds, full resolution) -> `upload_date` (YYYYMMDD, floors + # to midnight UTC) -> today's midnight UTC. The last fallback must + # stay a midnight floor, not now(): a wall-clock value advances the + # source watermark past every same-day midnight-floored video and + # strands them (see the watermark filter in connector.poll). + occurred_at: datetime | None = None + epoch = video.get("timestamp") or video.get("release_timestamp") + if epoch is not None: + with suppress(ValueError, TypeError, OSError, OverflowError): + occurred_at = datetime.fromtimestamp(float(epoch), tz=UTC) + if occurred_at is None and upload_date and len(upload_date) == 8: + with suppress(ValueError): + occurred_at = datetime.strptime(upload_date, "%Y%m%d").replace(tzinfo=UTC) + if occurred_at is None: + occurred_at = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0) + + # Duration in seconds. + duration = int(video.get("duration") or 0) + + # Metrics. + metrics = { + "views": int_or_none(video.get("view_count")), + "likes": int_or_none(video.get("like_count")), + "comments": int_or_none(video.get("comment_count")), + } + + # Media: thumbnails. + media = [] + for thumb in video.get("thumbnails") or []: + url = thumb.get("url") + if url: + media.append( + { + "type": "thumbnail", + "url": url, + "width": int_or_none(thumb.get("width")), + "height": int_or_none(thumb.get("height")), + } + ) + # Fallback to thumbnail field if thumbnails list is empty. + if not media: + thumb_url = video.get("thumbnail") + if thumb_url: + media.append({"type": "thumbnail", "url": thumb_url}) + + return cls( + external_id=video_id, + kind=cls.PAYLOAD_KIND, + occurred_at=occurred_at, + source=YouTubeSearchSourceSpec.SOURCE_KIND, + title=str(video.get("title") or ""), + content=description, + url=str(video.get("webpage_url") or f"{YOUTUBE_VIDEO_URL}{video_id}"), + author=uploader, + handle=uploader_id, + duration=duration, + metrics=metrics, + refs={}, + media=media, + ) + + +def int_or_none(obj: Any) -> int | None: + """Safely convert to int or return None.""" + if obj is None: + return None + try: + return int(obj) + except (ValueError, TypeError): + return None diff --git a/apps/core/sources/registry.py b/apps/core/sources/registry.py index 3ba421b9..89a47fe1 100644 --- a/apps/core/sources/registry.py +++ b/apps/core/sources/registry.py @@ -16,6 +16,7 @@ RedditSubRedditConnector, RssConnector, TwitterSearchConnector, + YouTubeSearchConnector, ) _REGISTRY: dict[str, Connector[Any]] = { @@ -24,6 +25,7 @@ HackerNewsFeedConnector.kind: HackerNewsFeedConnector(), HackerNewsCommentConnector.kind: HackerNewsCommentConnector(), TwitterSearchConnector.kind: TwitterSearchConnector(), + YouTubeSearchConnector.kind: YouTubeSearchConnector(), } # Core kinds captured before any plugin registers; a plugin can't replace one. diff --git a/apps/core/sources/tests_youtube.py b/apps/core/sources/tests_youtube.py new file mode 100644 index 00000000..7c47dc1e --- /dev/null +++ b/apps/core/sources/tests_youtube.py @@ -0,0 +1,184 @@ +"""YouTube search connector tests (offline, fake yt-dlp video dicts). + +The connector's only I/O is the yt-dlp client (`YtDlpClient.search`); these +tests swap in fake video dicts and pin: spec validation (the blank-query +firehose guard), the day-granularity watermark filter (boundary yields, older +skips), error translation (YouTubeError -> ConnectorParseError), the client's +date-ordered search URI, and payload mapping (timestamp fallback chain, +metrics, thumbnails). +""" + +from datetime import UTC, datetime +from typing import Any +from unittest import mock + +from django.test import SimpleTestCase +from pydantic import ValidationError + +from openmagpie_schema.configs import YouTubeSearchSourceSpec +from sources.connectors.base import ConnectorParseError +from sources.connectors.youtube.connector import YouTubeSearchConnector +from sources.connectors.youtube.errors import YouTubeError, map_ytdlp_error +from sources.connectors.youtube.payloads import NewVideoPayload + + +def _fake_video(video_id: str = "abc123", **overrides: Any) -> dict[str, Any]: + video: dict[str, Any] = { + "id": video_id, + "title": "A video title", + "description": "the description text", + "uploader": "Some Channel", + "uploader_id": "UCchannel", + "upload_date": "20260601", + "timestamp": datetime(2026, 6, 1, 15, 30, tzinfo=UTC).timestamp(), + "duration": 120, + "view_count": 1000, + "like_count": 50, + "comment_count": 5, + "webpage_url": f"https://www.youtube.com/watch?v={video_id}", + "thumbnails": [{"url": "https://i.ytimg.com/vi/x/hq.jpg", "width": 480, "height": 360}], + } + video.update(overrides) + return video + + +class YouTubeSearchSourceSpecTests(SimpleTestCase): + def test_blank_query_rejected(self): + with self.assertRaises(ValidationError): + YouTubeSearchSourceSpec(kind="youtube_search", query=" ") + + def test_count_bounds(self): + with self.assertRaises(ValidationError): + YouTubeSearchSourceSpec(kind="youtube_search", query="x", count=0) + with self.assertRaises(ValidationError): + YouTubeSearchSourceSpec(kind="youtube_search", query="x", count=51) + + def test_defaults(self): + spec = YouTubeSearchSourceSpec(kind="youtube_search", query="social listening") + self.assertEqual(spec.count, 20) + + +class YouTubeSearchConnectorTests(SimpleTestCase): + def _connector(self, results): + client = mock.Mock() + client.search.return_value = results + conn = YouTubeSearchConnector() + conn._client = client + return conn, client + + def test_yields_payloads_newer_than_since(self): + spec = YouTubeSearchSourceSpec(kind="youtube_search", query='"social listening"') + videos = [ + _fake_video("new", timestamp=datetime(2026, 6, 1, tzinfo=UTC).timestamp()), + _fake_video("old", timestamp=datetime(2026, 5, 1, tzinfo=UTC).timestamp()), + ] + conn, client = self._connector(videos) + payloads = list(conn.poll(spec, since=datetime(2026, 5, 15, tzinfo=UTC))) + self.assertEqual([p.external_id for p in payloads], ["new"]) + client.search.assert_called_once_with('"social listening"', 20) + + def test_watermark_boundary_yields(self): + """A video AT the watermark re-yields (day-granular timestamps; the + external_id dedup absorbs it). Only strictly-older videos skip.""" + since = datetime(2026, 6, 1, tzinfo=UTC) + spec = YouTubeSearchSourceSpec(kind="youtube_search", query="x") + videos = [_fake_video("same-day", timestamp=None, upload_date="20260601")] + conn, _ = self._connector(videos) + payloads = list(conn.poll(spec, since=since)) + self.assertEqual([p.external_id for p in payloads], ["same-day"]) + + def test_error_maps_to_connector_parse_error(self): + spec = YouTubeSearchSourceSpec(kind="youtube_search", query="x") + client = mock.Mock() + client.search.side_effect = YouTubeError( + code="rate_limited", message="slow down", retryable=True, action="backoff" + ) + conn = YouTubeSearchConnector() + conn._client = client + with self.assertRaises(ConnectorParseError) as ctx: + list(conn.poll(spec, since=None)) + self.assertIn("rate_limited", str(ctx.exception)) + + +class MapYtdlpErrorTests(SimpleTestCase): + def test_rate_limited_retryable(self): + err = map_ytdlp_error(Exception("HTTP Error 429: Too Many Requests")) + self.assertEqual(err.code, "rate_limited") + self.assertTrue(err.retryable) + + def test_unavailable_not_retryable(self): + err = map_ytdlp_error(Exception("Video unavailable")) + self.assertEqual(err.code, "video_unavailable") + self.assertFalse(err.retryable) + + def test_is_raisable_exception(self): + err = map_ytdlp_error(Exception("boom")) + with self.assertRaises(YouTubeError): + raise err + + +class YtDlpClientSearchUriTests(SimpleTestCase): + def test_cookie_file_setting_reaches_ytdlp_opts(self): + from django.test import override_settings + + from sources.connectors.youtube.client import YtDlpClient + + with override_settings(YOUTUBE_COOKIES_FILE="/tmp/yt-cookies.txt"): + self.assertEqual(YtDlpClient()._build_opts(5)["cookies"], "/tmp/yt-cookies.txt") + with override_settings(YOUTUBE_COOKIES_FILE=""): + self.assertNotIn("cookies", YtDlpClient()._build_opts(5)) + # An explicit constructor arg wins over the setting. + with override_settings(YOUTUBE_COOKIES_FILE="/tmp/from-setting.txt"): + self.assertEqual( + YtDlpClient(cookie_file="/tmp/explicit.txt")._build_opts(5)["cookies"], "/tmp/explicit.txt" + ) + + def test_date_ordered_uri_and_count_cap(self): + from sources.connectors.youtube.client import YtDlpClient + + with mock.patch("sources.connectors.youtube.client.yt_dlp.YoutubeDL") as ydl_cls: + ydl = ydl_cls.return_value.__enter__.return_value + ydl.extract_info.return_value = { + "entries": [ + _fake_video(), + None, + {"_type": "playlist", "id": "PLxyz", "title": "a playlist, not a video"}, + ] + } + results = YtDlpClient().search("brand mention", count=99) + ydl.extract_info.assert_called_once_with( + "https://www.youtube.com/results?search_query=brand+mention&sp=EgIIAw%3D%3D", download=False + ) + self.assertEqual(ydl_cls.call_args.args[0]["playlistend"], 50) + self.assertEqual([e["id"] for e in results], ["abc123"]) + + +class NewVideoPayloadTests(SimpleTestCase): + def test_from_video(self): + p = NewVideoPayload.from_video(_fake_video("abc123")) + self.assertEqual(p.external_id, "abc123") + self.assertEqual(p.handle, "UCchannel") + self.assertEqual(p.author, "Some Channel") + self.assertEqual(p.content, "the description text") + self.assertEqual(p.source, "youtube_search") + self.assertEqual(p.url, "https://www.youtube.com/watch?v=abc123") + self.assertEqual(p.metrics["views"], 1000) + self.assertEqual(p.media[0]["type"], "thumbnail") + + def test_timestamp_beats_upload_date(self): + p = NewVideoPayload.from_video(_fake_video()) + self.assertEqual(p.occurred_at, datetime(2026, 6, 1, 15, 30, tzinfo=UTC)) + + def test_upload_date_fallback_floors_to_midnight(self): + p = NewVideoPayload.from_video(_fake_video(timestamp=None, upload_date="20260601")) + self.assertEqual(p.occurred_at, datetime(2026, 6, 1, tzinfo=UTC)) + + def test_missing_dates_floor_to_today_midnight(self): + p = NewVideoPayload.from_video(_fake_video(timestamp=None, upload_date="")) + self.assertEqual(p.occurred_at, p.occurred_at.replace(hour=0, minute=0, second=0, microsecond=0)) + + def test_sample_distinct(self): + a = NewVideoPayload.sample(0) + b = NewVideoPayload.sample(1) + self.assertNotEqual(a.external_id, b.external_id) + self.assertEqual(a.PAYLOAD_KIND, "new_video") diff --git a/examples/starters/youtube/feed.yaml b/examples/starters/youtube/feed.yaml new file mode 100644 index 00000000..33ededf5 --- /dev/null +++ b/examples/starters/youtube/feed.yaml @@ -0,0 +1,29 @@ +# OpenMagpie starter: YouTube brand-mention listening. +# Uses the `youtube_search` source kind (yt-dlp, public search; no API key or +# credentials needed). Each poll runs the query filtered to the last week's +# uploads and surfaces videos newer than the source's watermark. +# Apply by hand (this starter isn't wired into the quickstart seed script): +# magpie feed create -f examples/starters/youtube/feed.yaml +# magpie watch create -f examples/starters/youtube/watch.yaml (after the edits in examples/README.md) +# Then set a past last_event_at on each source, or the first tick only sees +# brand-new videos. NOTE: YouTube timestamps can be day-granular, so a +# same-day item may appear once more across ticks; the item store dedups it. +# Full walkthrough: examples/README.md +name: "YouTube mentions (starter)" +kind: curated +poll_interval_seconds: 900 +data: + retention_days: 30 +sources: + # `query` is REQUIRED (the firehose guard): it pre-filters server-side on + # YouTube's search index before any per-item LLM cost. Quote exact phrases + # inside the YAML string, e.g. query: '"open source social listening"'. + # `count` caps the per-poll fetch (1-50). Each result costs one full + # YouTube page fetch per poll (metadata extraction), so the default 20 at + # this cadence is already ~2k fetches/day for the source. Raise it only + # when a query matches more than `count` videos in a week: results are + # relevance-ranked within the week window, so a busy query can squeeze + # fresh low-view mentions out of the visible slots. + - spec: {kind: youtube_search, query: '"social listening"', count: 20} + # Watch several phrasings by adding more sources: + # - spec: {kind: youtube_search, query: '"brand monitoring" open source', count: 20} diff --git a/examples/starters/youtube/watch.yaml b/examples/starters/youtube/watch.yaml new file mode 100644 index 00000000..42a94272 --- /dev/null +++ b/examples/starters/youtube/watch.yaml @@ -0,0 +1,22 @@ +# Companion watch for the YouTube mentions starter. See feed.yaml and +# examples/README.md. Apply with `magpie watch create -f` after creating the +# feed, and set its real id below (replacing REPLACE_WITH_FEED_ID). +name: "YouTube tool mentions (starter)" +is_active: true +feed_ids: + - REPLACE_WITH_FEED_ID +actions: + - kind: semantic_filter + config: + instructions: "A YouTube video that reviews, demos, or recommends social listening or brand monitoring tools: tutorials, tool roundups, or hands-on walkthroughs. The video description is the judged text. Not general marketing talk that only mentions listening in passing." + threshold: 0.6 + - kind: log + config: + prefix: "[youtube mention]" + # Prefer a push over a log line? Uncomment and point at your notifier (ntfy, or + # a relay like a Slack/Discord webhook or openclaw-style instance). A webhook also + # records a delivery audit you can inspect with `magpie delivery list --action `: + # - kind: webhook + # config: + # url: "https://your-notifier.example/hook" + # method: POST diff --git a/packages/openmagpie-schema/schema.json b/packages/openmagpie-schema/schema.json index 5960ac24..641afde6 100644 --- a/packages/openmagpie-schema/schema.json +++ b/packages/openmagpie-schema/schema.json @@ -838,6 +838,9 @@ { "$ref": "#/$defs/NewTweetPayload" }, + { + "$ref": "#/$defs/NewVideoPayload" + }, { "$ref": "#/$defs/FeedItemPayload" } @@ -1854,6 +1857,124 @@ "title": "NewTweetPayload", "type": "object" }, + "NewVideoPayload": { + "additionalProperties": true, + "description": "`new_video`: one YouTube video (YouTubeSearchConnector, yt-dlp search).\n`content` is the video description (the engine's judgeable body); `title`\nis the video title. `handle` is the channel ID (the within-kind source\nslug), `author` the channel display name, `duration` seconds, `metrics`\n(views/likes/comments), `refs` (related video IDs), `media` (thumbnails).", + "properties": { + "author": { + "default": "", + "title": "Author", + "type": "string" + }, + "content": { + "default": "", + "title": "Content", + "type": "string" + }, + "duration": { + "default": 0, + "title": "Duration", + "type": "integer" + }, + "external_id": { + "default": "", + "title": "External Id", + "type": "string" + }, + "external_url": { + "default": "", + "title": "External Url", + "type": "string" + }, + "handle": { + "default": "", + "title": "Handle", + "type": "string" + }, + "kind": { + "const": "new_video", + "title": "Kind", + "type": "string" + }, + "media": { + "default": [], + "items": { + "additionalProperties": true, + "type": "object" + }, + "title": "Media", + "type": "array" + }, + "metrics": { + "additionalProperties": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ] + }, + "default": {}, + "title": "Metrics", + "type": "object" + }, + "occurred_at": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Occurred At" + }, + "parent_external_id": { + "default": "", + "title": "Parent External Id", + "type": "string" + }, + "refs": { + "additionalProperties": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ] + }, + "default": {}, + "title": "Refs", + "type": "object" + }, + "source": { + "default": "", + "title": "Source", + "type": "string" + }, + "title": { + "default": "", + "title": "Title", + "type": "string" + }, + "url": { + "default": "", + "title": "Url", + "type": "string" + } + }, + "required": [ + "kind" + ], + "title": "NewVideoPayload", + "type": "object" + }, "PluginActionInput": { "description": "Fallback input member for a plugin (non-built-in) action kind. Mirrors\nPluginActionWire on the write path; `config` is required (you can't author an\naction without one) but open (validated server-side by the kind's registry).", "properties": { @@ -2562,7 +2683,8 @@ "hn_feed": "#/$defs/HackerNewsFeedSourceSpec", "reddit_subreddit": "#/$defs/RedditSubredditSourceSpec", "rss": "#/$defs/RssSourceSpec", - "twitter_search": "#/$defs/TwitterSearchSourceSpec" + "twitter_search": "#/$defs/TwitterSearchSourceSpec", + "youtube_search": "#/$defs/YouTubeSearchSourceSpec" }, "propertyName": "kind" }, @@ -2581,6 +2703,9 @@ }, { "$ref": "#/$defs/TwitterSearchSourceSpec" + }, + { + "$ref": "#/$defs/YouTubeSearchSourceSpec" } ] }, @@ -2708,7 +2833,8 @@ "hn_feed": "#/$defs/HackerNewsFeedSourceSpec", "reddit_subreddit": "#/$defs/RedditSubredditSourceSpec", "rss": "#/$defs/RssSourceSpec", - "twitter_search": "#/$defs/TwitterSearchSourceSpec" + "twitter_search": "#/$defs/TwitterSearchSourceSpec", + "youtube_search": "#/$defs/YouTubeSearchSourceSpec" }, "propertyName": "kind" }, @@ -2727,6 +2853,9 @@ }, { "$ref": "#/$defs/TwitterSearchSourceSpec" + }, + { + "$ref": "#/$defs/YouTubeSearchSourceSpec" } ] }, @@ -4033,6 +4162,34 @@ ], "title": "WebhookRunWire", "type": "object" + }, + "YouTubeSearchSourceSpec": { + "description": "Identity of one YouTube search stream. Bound to YouTubeSearchConnector.\n\n`query` is REQUIRED and NON-BLANK so a source always carries a server-side\npre-filter before any per-item LLM cost. `count` caps the per-cycle fetch\n(bounded at 50, the client's search cap).", + "properties": { + "count": { + "default": 20, + "maximum": 50, + "minimum": 1, + "title": "Count", + "type": "integer" + }, + "kind": { + "const": "youtube_search", + "default": "youtube_search", + "title": "Kind", + "type": "string" + }, + "query": { + "minLength": 1, + "title": "Query", + "type": "string" + } + }, + "required": [ + "query" + ], + "title": "YouTubeSearchSourceSpec", + "type": "object" } }, "$schema": "https://json-schema.org/draft/2020-12/schema", diff --git a/packages/openmagpie-schema/src/openmagpie_schema/configs.py b/packages/openmagpie-schema/src/openmagpie_schema/configs.py index 51e46d11..4fcad183 100644 --- a/packages/openmagpie-schema/src/openmagpie_schema/configs.py +++ b/packages/openmagpie-schema/src/openmagpie_schema/configs.py @@ -188,6 +188,32 @@ def display(self) -> str: return f'X search: "{self.query}"' +class YouTubeSearchSourceSpec(BaseModel): + """Identity of one YouTube search stream. Bound to YouTubeSearchConnector. + + `query` is REQUIRED and NON-BLANK so a source always carries a server-side + pre-filter before any per-item LLM cost. `count` caps the per-cycle fetch + (bounded at 50, the client's search cap).""" + + SOURCE_KIND: ClassVar[str] = "youtube_search" + URL_FIELDS: ClassVar[tuple[str, ...]] = () # no operator-supplied URL to SSRF-check + + kind: Literal["youtube_search"] = "youtube_search" + query: str = Field(min_length=1) + count: int = Field(default=20, ge=1, le=50) + + @field_validator("query") + @classmethod + def _query_not_blank(cls, v: str) -> str: + v = v.strip() + if not v: + raise ValueError("youtube_search requires a non-blank query (the firehose guard)") + return v + + def display(self) -> str: + return f'YouTube search: "{self.query}"' + + # The built-ins as a discriminated union over `kind` (defined before the plugin # fallback so the built-in kind set can be derived from it below). A built-in kind # with a malformed spec fails its typed member here and is rejected by the fallback, @@ -197,7 +223,8 @@ def display(self) -> str: | RssSourceSpec | HackerNewsFeedSourceSpec | HackerNewsCommentSourceSpec - | TwitterSearchSourceSpec, + | TwitterSearchSourceSpec + | YouTubeSearchSourceSpec, Field(discriminator="kind"), ] diff --git a/packages/openmagpie-schema/src/openmagpie_schema/feed_payloads.py b/packages/openmagpie-schema/src/openmagpie_schema/feed_payloads.py index e8d7f574..cae15642 100644 --- a/packages/openmagpie-schema/src/openmagpie_schema/feed_payloads.py +++ b/packages/openmagpie-schema/src/openmagpie_schema/feed_payloads.py @@ -97,6 +97,22 @@ class NewTweetPayload(FeedItemPayload): media: list[dict[str, object]] = [] +class NewVideoPayload(FeedItemPayload): + """`new_video`: one YouTube video (YouTubeSearchConnector, yt-dlp search). + `content` is the video description (the engine's judgeable body); `title` + is the video title. `handle` is the channel ID (the within-kind source + slug), `author` the channel display name, `duration` seconds, `metrics` + (views/likes/comments), `refs` (related video IDs), `media` (thumbnails).""" + + kind: Literal["new_video"] # required, so a non-youtube dump can't match here + author: str = "" + handle: str = "" + duration: int = 0 + metrics: dict[str, int | None] = {} + refs: dict[str, str | None] = {} + media: list[dict[str, object]] = [] + + # Tried left-to-right so a dump resolves to its concrete variant (matched on the # required `kind` literal) and only falls to the permissive base when no variant # claims it. Variants REQUIRE their `kind`, so an empty / kind-less dict can't @@ -112,6 +128,7 @@ class NewTweetPayload(FeedItemPayload): | HackerNewsFeedPayload | HackerNewsCommentPayload | NewTweetPayload + | NewVideoPayload | FeedItemPayload, Field(union_mode="left_to_right"), ] diff --git a/uv.lock b/uv.lock index 4ba751ac..fb44be6a 100644 --- a/uv.lock +++ b/uv.lock @@ -758,6 +758,7 @@ dependencies = [ { name = "trafilatura" }, { name = "twikit" }, { name = "ulid" }, + { name = "yt-dlp" }, ] [package.dev-dependencies] @@ -792,6 +793,7 @@ requires-dist = [ { name = "trafilatura", specifier = ">=1.7" }, { name = "twikit", git = "https://github.com/unclecode/twikit.git" }, { name = "ulid", specifier = ">=1.1" }, + { name = "yt-dlp", specifier = ">=2026.7.4" }, ] [package.metadata.requires-dev] @@ -1497,3 +1499,12 @@ sdist = { url = "https://files.pythonhosted.org/packages/5e/f6/7c9c964681fb148e0 wheels = [ { url = "https://files.pythonhosted.org/packages/f3/ed/aad7e0f5a462d679f7b4d2e0d8502c3096740c883b5bbed5103146480937/webvtt_py-0.5.1-py3-none-any.whl", hash = "sha256:9d517d286cfe7fc7825e9d4e2079647ce32f5678eb58e39ef544ffbb932610b7", size = 19802, upload-time = "2024-05-30T13:40:14.661Z" }, ] + +[[package]] +name = "yt-dlp" +version = "2026.8.19" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/1e/e0/832fa4ca334b766a06933a196066edc3dba37cdb6f14cd98d59bcc69a4b4/yt_dlp-2026.8.19.tar.gz", hash = "sha256:9e213e48cea35c66b378e4447903f118f6392a5fa380a2b6d7070ec86f4e0af1", size = 3052025, upload-time = "2026-08-19T23:48:59.291Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/69/b2/8cd1613f56eed7ceb64fbd4df3f1c01246bfb098e6f398228bafda22b80b/yt_dlp-2026.8.19-py3-none-any.whl", hash = "sha256:1d57897e94c6665a0a6f9bc54b34e584284e32c034ffab3a7df25d8f7b24eedf", size = 3185533, upload-time = "2026-08-19T23:48:56.925Z" }, +] diff --git a/web/packages/schema/src/generated.ts b/web/packages/schema/src/generated.ts index 8e07a5f7..2d5a9651 100644 --- a/web/packages/schema/src/generated.ts +++ b/web/packages/schema/src/generated.ts @@ -62,6 +62,9 @@ export type NewRedditPostPayload = z.infer; export const NewTweetPayloadSchema = z.object({ "author": z.string().default(""), "content": z.string().default(""), "external_id": z.string().default(""), "external_url": z.string().default(""), "handle": z.string().default(""), "kind": z.literal("new_tweet"), "lang": z.string().default(""), "media": z.array(z.record(z.string(), z.any())).default([]), "metrics": z.record(z.string(), z.union([z.number().int(), z.null()])).default({}), "occurred_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "parent_external_id": z.string().default(""), "refs": z.record(z.string(), z.union([z.string(), z.null()])).default({}), "source": z.string().default(""), "title": z.string().default(""), "url": z.string().default("") }).catchall(z.any()).describe("`new_tweet`: one X (Twitter) tweet (TwitterSearchConnector, unofficial\ntwikit route). `content` is the tweet's full text (the engine's judgeable\nbody); `title` is empty (tweets have no headline). The listeningkit\nSocialEvent shape is carried as typed fields: `handle` (the @screen_name,\nalso the within-kind source slug), `author` (display name), `lang`,\n`metrics` (likes/retweets/replies/quotes/views), `refs`\n(in_reply_to / quoted / retweet_of), `media` (list of {type,url,thumbnail})."); export type NewTweetPayload = z.infer; +export const NewVideoPayloadSchema = z.object({ "author": z.string().default(""), "content": z.string().default(""), "duration": z.number().int().default(0), "external_id": z.string().default(""), "external_url": z.string().default(""), "handle": z.string().default(""), "kind": z.literal("new_video"), "media": z.array(z.record(z.string(), z.any())).default([]), "metrics": z.record(z.string(), z.union([z.number().int(), z.null()])).default({}), "occurred_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "parent_external_id": z.string().default(""), "refs": z.record(z.string(), z.union([z.string(), z.null()])).default({}), "source": z.string().default(""), "title": z.string().default(""), "url": z.string().default("") }).catchall(z.any()).describe("`new_video`: one YouTube video (YouTubeSearchConnector, yt-dlp search).\n`content` is the video description (the engine's judgeable body); `title`\nis the video title. `handle` is the channel ID (the within-kind source\nslug), `author` the channel display name, `duration` seconds, `metrics`\n(views/likes/comments), `refs` (related video IDs), `media` (thumbnails)."); +export type NewVideoPayload = z.infer; + export const PluginConfigBlobSchema = z.object({}).catchall(z.any()).describe("The write-side config for a plugin action kind: an open blob. The server\nre-validates it against the kind's registered Pydantic config (watches.registry),\nso the wire type stays permissive; `extra=\"allow\"` keeps every submitted key\nthrough `model_dump(mode=\"json\")`."); export type PluginConfigBlob = z.infer; @@ -128,6 +131,9 @@ export type WebhookResult = z.infer; export const WebhookRunWireSchema = z.object({ "action_id": z.string(), "completed_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "created_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "error": z.string().default(""), "feed_item_id": z.string(), "id": z.string(), "kind": z.literal("webhook").default("webhook"), "result": z.union([WebhookResultSchema, z.null()]).default(null), "scheduled_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "started_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "state": WatchActionRunStateSchema, "watch_id": z.string() }); export type WebhookRunWire = z.infer; +export const YouTubeSearchSourceSpecSchema = z.object({ "count": z.number().int().gte(1).lte(50).default(20), "kind": z.literal("youtube_search").default("youtube_search"), "query": z.string().min(1) }).describe("Identity of one YouTube search stream. Bound to YouTubeSearchConnector.\n\n`query` is REQUIRED and NON-BLANK so a source always carries a server-side\npre-filter before any per-item LLM cost. `count` caps the per-cycle fetch\n(bounded at 50, the client's search cap)."); +export type YouTubeSearchSourceSpec = z.infer; + export const BackfillJobSchema = z.object({ "completed_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "completed_since": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "completed_until": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "created_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "deleted": z.number().int().default(0), "enqueued": z.number().int().default(0), "error": z.string().default(""), "id": z.string(), "kind": z.union([WatchActionKindSchema, z.string()]).default(""), "matched": z.number().int().default(0), "occurred_since": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "occurred_until": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "present": z.number().int().default(0), "pruned": z.number().int().default(0), "replace": z.boolean().default(false), "source_action_id": z.string().default(""), "source_is_head": z.boolean().default(false), "started_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "state": WatchActionBackfillStateSchema, "target_action_id": z.string() }).describe("A queued backfill's definition + progress.\n\nThe counts are 0 until `process_due_backfills` runs the job, then reflect what\nit actually did: `matched` source passes in the window, `present`/`pruned`\nsplit, `deleted` terminal runs removed (target + downstream, when `replace`),\n`enqueued` fresh target runs created. `source_action_id` is empty for a\nchain-head target (see `BackfillPreview`).\n\n`kind` is the target action's kind (a `WatchActionKind`; `| str` tolerates a\nsince-removed kind, mirroring the run wire). The four window bounds are the\nresolved absolute datetimes (None = unbounded on that side). `error` carries the\nreason when `state` is FAILED; a FAILED job is retryable while `completed_at` is\nunset and terminal once it's set (the attempts cap was hit), mirroring a run."); export type BackfillJob = z.infer; @@ -146,7 +152,7 @@ export type ExtractResult = z.infer; export const ExtractRunWireSchema = z.object({ "action_id": z.string(), "completed_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "created_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "error": z.string().default(""), "feed_item_id": z.string(), "id": z.string(), "kind": z.literal("extract").default("extract"), "result": z.union([ExtractResultSchema, z.null()]).default(null), "scheduled_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "started_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "state": WatchActionRunStateSchema, "watch_id": z.string() }); export type ExtractRunWire = z.infer; -export const FeedItemWireSchema = z.object({ "data": z.union([RssEntryPayloadSchema, NewRedditPostPayloadSchema, HackerNewsFeedPayloadSchema, HackerNewsCommentPayloadSchema, NewTweetPayloadSchema, FeedItemPayloadSchema]).optional(), "external_id": z.string(), "id": z.string(), "occurred_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "source_kind": z.string(), "source_label": z.string().default("") }).describe("One persisted FeedItem on the wire ; the \"sort by new and go\" unit.\n\n`data` is the connector SourcePayload's dump, parsed into a typed\n`FeedItemData` (canonical fields typed for every kind; connector-specific\nfields typed per known kind; unknown kinds fall to the permissive base).\nDatetimes stay real; the renderer ISO-encodes them.\n\n`source_kind` is the connector kind (e.g. `\"reddit_subreddit\"`),\ndenormalized from the producing Source. `source_label` is the\noperator-visible display string (e.g. `\"r/ClaudeCowork\"`), set on\nthe FeedItem row at record time from the SourceSpec's `.display()`."); +export const FeedItemWireSchema = z.object({ "data": z.union([RssEntryPayloadSchema, NewRedditPostPayloadSchema, HackerNewsFeedPayloadSchema, HackerNewsCommentPayloadSchema, NewTweetPayloadSchema, NewVideoPayloadSchema, FeedItemPayloadSchema]).optional(), "external_id": z.string(), "id": z.string(), "occurred_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "source_kind": z.string(), "source_label": z.string().default("") }).describe("One persisted FeedItem on the wire ; the \"sort by new and go\" unit.\n\n`data` is the connector SourcePayload's dump, parsed into a typed\n`FeedItemData` (canonical fields typed for every kind; connector-specific\nfields typed per known kind; unknown kinds fall to the permissive base).\nDatetimes stay real; the renderer ISO-encodes them.\n\n`source_kind` is the connector kind (e.g. `\"reddit_subreddit\"`),\ndenormalized from the producing Source. `source_label` is the\noperator-visible display string (e.g. `\"r/ClaudeCowork\"`), set on\nthe FeedItem row at record time from the SourceSpec's `.display()`."); export type FeedItemWire = z.infer; export const FeedListResponseSchema = z.object({ "items": z.array(FeedWireSchema).optional(), "next_cursor": z.union([z.string(), z.null()]).default(null) }).describe("`GET /v1/feeds` -> `{\"items\": [...], \"next_cursor\": |None}`.\n\nCursor-paginated by ULID pk, newest-first. Pass `?after=` to fetch\nthe next page; `next_cursor` is the id to send back, or null when the\npage wasn't full (= no more rows)."); @@ -179,7 +185,7 @@ export type SemanticFilterActionWire = z.infer; -export const SourceInputSchema = z.object({ "field_map": z.record(z.string(), z.string()).optional(), "last_event_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "meta": z.record(z.string(), z.string()).optional(), "spec": z.union([z.discriminatedUnion("kind", [RedditSubredditSourceSpecSchema, RssSourceSpecSchema, HackerNewsFeedSourceSpecSchema, HackerNewsCommentSourceSpecSchema, TwitterSearchSourceSpecSchema]), PluginSourceSpecSchema]) }).describe("One source on a feed-create or `feed source set` payload. `spec` / `meta` /\n`field_map` come from `SourceFields`.\n\n`last_event_at` is the optional starting watermark. None means\n\"live mode from now\" (server-policy defaulted at save time); a\npast datetime means \"fetch items newer than this\" ; the operator's\nbackfill knob. Server policy rejects future values."); +export const SourceInputSchema = z.object({ "field_map": z.record(z.string(), z.string()).optional(), "last_event_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "meta": z.record(z.string(), z.string()).optional(), "spec": z.union([z.discriminatedUnion("kind", [RedditSubredditSourceSpecSchema, RssSourceSpecSchema, HackerNewsFeedSourceSpecSchema, HackerNewsCommentSourceSpecSchema, TwitterSearchSourceSpecSchema, YouTubeSearchSourceSpecSchema]), PluginSourceSpecSchema]) }).describe("One source on a feed-create or `feed source set` payload. `spec` / `meta` /\n`field_map` come from `SourceFields`.\n\n`last_event_at` is the optional starting watermark. None means\n\"live mode from now\" (server-policy defaulted at save time); a\npast datetime means \"fetch items newer than this\" ; the operator's\nbackfill knob. Server policy rejects future values."); export type SourceInput = z.infer; export const SourceSetPayloadSchema = z.object({ "sources": z.array(SourceInputSchema).optional(), "version": z.literal("v1").default("v1") }).describe("Round-trip file format for `magpie feed source export` /\n`magpie feed source set`. Operators or their scrape scripts can\nconstruct this directly; the bare `list[SourceInput]` shape is\nalso accepted on input for hand-rolled cases."); @@ -192,7 +198,7 @@ export const SourceWireSchema = z.object({ "created_at": z.union([z.iso.datetime * field instead of re-deriving it per kind from `spec`; it is the same * `SourceSpec.display()` the server records onto FeedItem.source_label. */ -"display": z.string().describe("The source's human label, kind-polymorphic (e.g. `r/foo`, or an RSS\nfeed's name/url). Provided on the wire so a consumer reads one labeled\nfield instead of re-deriving it per kind from `spec`; it is the same\n`SourceSpec.display()` the server records onto FeedItem.source_label.").readonly(), "field_map": z.record(z.string(), z.string()).optional(), "id": z.string(), "last_event_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "meta": z.record(z.string(), z.string()).optional(), "spec": z.union([z.discriminatedUnion("kind", [RedditSubredditSourceSpecSchema, RssSourceSpecSchema, HackerNewsFeedSourceSpecSchema, HackerNewsCommentSourceSpecSchema, TwitterSearchSourceSpecSchema]), PluginSourceSpecSchema]) }).describe("One Source row on the read path. `spec` / `meta` / `field_map` come from\n`SourceFields`.\n\nRelies on pydantic's DEFAULT `extra=\"ignore\"` (no explicit model_config): the\noutput-only `display` computed field is absent from the input schema, so\nre-validating a `model_dump()` (which includes `display`) silently drops it and\nrecomputes, rather than erroring as it would under `extra=\"forbid\"`."); +"display": z.string().describe("The source's human label, kind-polymorphic (e.g. `r/foo`, or an RSS\nfeed's name/url). Provided on the wire so a consumer reads one labeled\nfield instead of re-deriving it per kind from `spec`; it is the same\n`SourceSpec.display()` the server records onto FeedItem.source_label.").readonly(), "field_map": z.record(z.string(), z.string()).optional(), "id": z.string(), "last_event_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "meta": z.record(z.string(), z.string()).optional(), "spec": z.union([z.discriminatedUnion("kind", [RedditSubredditSourceSpecSchema, RssSourceSpecSchema, HackerNewsFeedSourceSpecSchema, HackerNewsCommentSourceSpecSchema, TwitterSearchSourceSpecSchema, YouTubeSearchSourceSpecSchema]), PluginSourceSpecSchema]) }).describe("One Source row on the read path. `spec` / `meta` / `field_map` come from\n`SourceFields`.\n\nRelies on pydantic's DEFAULT `extra=\"ignore\"` (no explicit model_config): the\noutput-only `display` computed field is absent from the input schema, so\nre-validating a `model_dump()` (which includes `display`) silently drops it and\nrecomputes, rather than erroring as it would under `extra=\"forbid\"`."); export type SourceWire = z.infer; export const WatchActionDeliveryViewSchema = z.object({ "action_id": z.string(), "attempt": z.number().int().default(0), "completed_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "created_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "delivery": DeliveryCadenceSchema, "error": z.string().default(""), "http_status": z.union([z.number().int(), z.null()]).default(null), "id": z.string(), "item_count": z.number().int().default(0), "method": WebhookMethodSchema, "request_payload": z.record(z.string(), z.any()).optional(), "started_at": z.union([z.iso.datetime({ offset: true }), z.null()]).default(null), "state": WatchActionDeliveryStateSchema, "target_host": z.string().default(""), "watch_id": z.string() }).describe("`GET /v1/action-deliveries/`, the detail: the list row plus the\nexact `request_payload` we sent (a WebhookPayload dump), stored\npoint-in-time. Opaque here ; headers are NEVER included (auth tokens). Kept\noff the list wire so a list call doesn't ship every batch body.");