diff --git a/.gitignore b/.gitignore index 928417ea..69043bce 100644 --- a/.gitignore +++ b/.gitignore @@ -81,3 +81,6 @@ cli/.venv/ /*.csv /*.ndjson /*.jsonl + +# local-only: override docker-compose ports when the host ports are taken (see docker-compose.override.yml) +docker-compose.override.yml diff --git a/apps/core/Dockerfile b/apps/core/Dockerfile index 8524f28a..1b9c677d 100644 --- a/apps/core/Dockerfile +++ b/apps/core/Dockerfile @@ -25,7 +25,7 @@ ENV PYTHONUNBUFFERED=1 \ # Build deps for psycopg[c] (compiled against libpq). Builder-only. RUN apt-get update \ - && apt-get install -y --no-install-recommends build-essential libpq-dev \ + && apt-get install -y --no-install-recommends build-essential git libpq-dev \ && rm -rf /var/lib/apt/lists/* WORKDIR /app diff --git a/apps/core/feeds/tests_plugin_source_kinds.py b/apps/core/feeds/tests_plugin_source_kinds.py index a0f0b439..95383f2c 100644 --- a/apps/core/feeds/tests_plugin_source_kinds.py +++ b/apps/core/feeds/tests_plugin_source_kinds.py @@ -20,12 +20,14 @@ from feeds.services.sources import _assert_connector_registered from openmagpie_schema.configs import ( _BUILTIN_SOURCE_KINDS, + FacebookGroupSourceSpec, HackerNewsCommentSourceSpec, HackerNewsFeedSourceSpec, PluginSourceSpec, RedditSubredditSourceSpec, RssSourceSpec, SourceSpec, + TwitterSearchSourceSpec, _BuiltinSourceSpec, canonical_spec, ) @@ -295,7 +297,7 @@ class _C2: class BuiltinSourceKindInvariantTests(SimpleTestCase): """The set the plugin fallback rejects is DERIVED from the built-in union, so it - can't drift. Pin that: it equals the SOURCE_KIND of every union member (add a 5th + can't drift. Pin that: it equals the SOURCE_KIND of every union member (add a 6th built-in spec but forget to wire it and this fails loud, mirroring the action side, which derives its set from the WatchActionKind enum).""" @@ -308,7 +310,7 @@ def test_kind_literal_default_matches_source_kind(self) -> None: for m in members: self.assertEqual(m.model_fields["kind"].default, m.SOURCE_KIND, m.__name__) - def test_builtin_source_kinds_are_exactly_the_known_four(self) -> None: + def test_builtin_source_kinds_are_exactly_the_known_builtins(self) -> None: self.assertEqual( _BUILTIN_SOURCE_KINDS, frozenset( @@ -317,6 +319,8 @@ def test_builtin_source_kinds_are_exactly_the_known_four(self) -> None: RssSourceSpec.SOURCE_KIND, HackerNewsFeedSourceSpec.SOURCE_KIND, HackerNewsCommentSourceSpec.SOURCE_KIND, + TwitterSearchSourceSpec.SOURCE_KIND, + FacebookGroupSourceSpec.SOURCE_KIND, } ), ) diff --git a/apps/core/pyproject.toml b/apps/core/pyproject.toml index 39abc14a..739a7c6b 100644 --- a/apps/core/pyproject.toml +++ b/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 + "twikit @ git+https://github.com/unclecode/twikit.git", # X (Twitter) unofficial route (listeningkit-verified 2026 fork of d60/twikit) "ulid>=1.1", ] diff --git a/apps/core/sources/connectors/__init__.py b/apps/core/sources/connectors/__init__.py index 041ba035..5d748e38 100644 --- a/apps/core/sources/connectors/__init__.py +++ b/apps/core/sources/connectors/__init__.py @@ -1,12 +1,16 @@ from .base import Connector +from .facebook import FacebookGroupConnector from .hackernews import HackerNewsCommentConnector, HackerNewsFeedConnector from .reddit import RedditSubRedditConnector from .rss import RssConnector +from .twitter import TwitterSearchConnector __all__ = [ "Connector", + "FacebookGroupConnector", "HackerNewsCommentConnector", "HackerNewsFeedConnector", "RedditSubRedditConnector", "RssConnector", + "TwitterSearchConnector", ] diff --git a/apps/core/sources/connectors/facebook/__init__.py b/apps/core/sources/connectors/facebook/__init__.py new file mode 100644 index 00000000..769341b0 --- /dev/null +++ b/apps/core/sources/connectors/facebook/__init__.py @@ -0,0 +1,16 @@ +"""Facebook connectors, public surface. + +One file per concern: + - `connector.py`, the Connector implementation(s) + polling logic + - `payloads.py`, our internal SourcePayload subclasses + - `client.py`, the subprocess client that calls the facebook-worker.py + - `errors.py`, error taxonomy +""" + +from .connector import FacebookGroupConnector +from .payloads import NewFacebookPostPayload + +__all__ = [ + "FacebookGroupConnector", + "NewFacebookPostPayload", +] diff --git a/apps/core/sources/connectors/facebook/client.py b/apps/core/sources/connectors/facebook/client.py new file mode 100644 index 00000000..d70f44f5 --- /dev/null +++ b/apps/core/sources/connectors/facebook/client.py @@ -0,0 +1,225 @@ +"""Camofox-based Facebook client, unofficial route. + +Spawns the facebook-worker.py shim (sibling checkout +REPOS/facebook-camofox-client/scripts/facebook-worker.py) as a +subprocess with a JSON stdin/stdout contract, mirroring how the +twitter connector wraps twikit. The worker drives one Camofox +(anti-detect Firefox) session per run, injects the account's cookies as +storage state, opens the target Facebook group surface, runs the auth +guard, extracts posts, and returns normalized records + refreshed +cookies. + +Cookies resolution follows the twitter connector's env/file pattern: + +- FACEBOOK_COOKIES_JSON (a full JSON list/dict of facebook.com cookies), +- FACEBOOK_COOKIES_FILE (path to a cookies JSON file), +- FACEBOOK_CREDENTIALS_DIR (dir of *.json cookie exports, each with an + optional *.proxy pin), default credentials/ relative to the core app. + +Proxy: FACEBOOK_PROXY env (or a per-credential .proxy pin in +FACEBOOK_CREDENTIALS_DIR). The worker forwards it to Camofox. + +The connector's poll is a sync iterator, so the subprocess is a +blocking subprocess.run (one bounded run per poll cycle; the worker +launches and tears down its own browser session). +""" + +from __future__ import annotations + +import json +import logging +import os +import subprocess +import sys +from pathlib import Path +from typing import Any + +from .errors import FacebookError, map_worker_error + +log = logging.getLogger("sources.facebook") + + +def _load_cookie_file(path: Path) -> list[dict[str, Any]]: + """Parse a cookies JSON export (array of cookie objects) or a plain dict.""" + data = json.loads(path.read_text(encoding="utf-8")) + if isinstance(data, dict): + return [{"name": str(k), "value": str(v), "domain": ".facebook.com", "path": "/"} for k, v in data.items() if v] + if isinstance(data, list): + return [item for item in data if isinstance(item, dict) and item.get("name") and item.get("value")] + raise ValueError(f"{path}: JSON must be an object or an array of cookie objects") + + +def load_cookies( + *, + cookies_json: str | None = None, + cookies_file: str | None = None, + credentials_dir: str | None = None, +) -> tuple[list[dict[str, Any]], str | None]: + """Resolve the cookie list + proxy for one Facebook session, in priority order. + + Returns ``(cookies, proxy)``. ``cookies`` is empty when nothing is + configured (the worker stays guest-mode; the first open then hits the + login surface and the auth guard returns ``auth_required``, which the + poll loop maps). ``proxy`` comes from `FACEBOOK_PROXY` or a + `.proxy` pin next to the chosen cookie export. + """ + proxy = os.environ.get("FACEBOOK_PROXY", "").strip() or None + + if cookies_json: + try: + data = json.loads(cookies_json) + except json.JSONDecodeError: + log.warning("FACEBOOK_COOKIES_JSON is not valid JSON; ignoring") + else: + if isinstance(data, dict) and data: + out = [{"name": str(k), "value": str(v), "domain": ".facebook.com", "path": "/"} for k, v in data.items() if v] + return out, proxy + if isinstance(data, list) and data: + out = [item for item in data if isinstance(item, dict) and item.get("name") and item.get("value")] + return out, proxy + log.warning("FACEBOOK_COOKIES_JSON is not a non-empty object/array; ignoring") + + if cookies_file and Path(cookies_file).exists(): + try: + return _load_cookie_file(Path(cookies_file)), proxy + except (json.JSONDecodeError, ValueError, OSError) as exc: + log.warning("FACEBOOK_COOKIES_FILE %s unreadable (%s); ignoring", cookies_file, exc) + + directory = Path(credentials_dir or os.environ.get("FACEBOOK_CREDENTIALS_DIR", "credentials")) + if directory.exists(): + for path in sorted(directory.glob("*.json")): + try: + cookies = _load_cookie_file(path) + except (json.JSONDecodeError, ValueError, OSError) as exc: + log.warning("skipping %s: %s", path.name, exc) + continue + if not {"c_user", "xs"}.issubset({c["name"] for c in cookies}): + log.warning("skipping %s: missing c_user/xs (critical pair)", path.name) + continue + proxy_path = path.with_suffix(".proxy") + pin = proxy_path.read_text().strip() if proxy_path.exists() else proxy + return cookies, pin + log.warning("credentials dir %s: no usable cookie set found", directory) + elif credentials_dir: + log.warning("credentials dir %s not found; no sessions loaded", credentials_dir) + + return [], proxy + + +def _resolve_worker_path() -> Path: + """Locate facebook-worker.py: env override, then sibling-checkout, then repo-relative.""" + env_path = os.environ.get("FACEBOOK_WORKER_PATH", "").strip() + if env_path and Path(env_path).exists(): + return Path(env_path) + + # The worker lives in the facebook-camofox-client repo checkout (sibling to + # this repo under REPOS/). Fall back to a few plausible placements. + candidates = [ + Path("facebook-camofox-client/scripts/facebook-worker.py"), + Path(__file__).resolve().parent / ".." / ".." / ".." / ".." / ".." / "facebook-camofox-client" / "scripts" / "facebook-worker.py", + Path(sys.prefix) / "facebook_camofox_client" / "scripts" / "facebook-worker.py", + ] + for candidate in candidates: + resolved = candidate.resolve() + if resolved.exists(): + return resolved + raise FacebookError( + code="worker_not_found", + message="facebook-worker.py not found; set FACEBOOK_WORKER_PATH or clone REPOS/facebook-camofox-client", + retryable=False, + action="install the facebook-camofox-client checkout with its worker script", + ) + + +class FacebookClient: + """Spawns the facebook-worker.py subprocess for one action per call.""" + + def __init__( + self, + *, + cookies: list[dict[str, Any]] | None = None, + cookies_json: str | None = None, + cookies_file: str | None = None, + credentials_dir: str | None = None, + ) -> None: + self._cookies: list[dict[str, Any]] + if cookies is not None: + self._cookies = list(cookies) + self.proxy = os.environ.get("FACEBOOK_PROXY", "").strip() or None + else: + self._cookies, self.proxy = load_cookies( + cookies_json=cookies_json, + cookies_file=cookies_file, + credentials_dir=credentials_dir, + ) + self._worker_path = _resolve_worker_path() + + def search_group( + self, + group_ids: list[str], + terms: list[str] | None = None, + count: int = 20, + ) -> dict[str, Any]: + """Run one Facebook group search via the worker subprocess. + + Returns the worker's parsed output dict (``{ok, result, new_cookies}``). + Raises ``FacebookError`` on a worker-reported failure. + """ + payload = { + "account_id": "openmagpie", + "action": "groups.search", + "cookies": self._cookies, + "params": { + "group_ids": group_ids, + "terms": terms or [], + "limit": count, + }, + "proxy": self.proxy, + } + + try: + proc = subprocess.run( + [sys.executable, str(self._worker_path)], + input=json.dumps(payload), + capture_output=True, + text=True, + timeout=300, # Camofox browser launch + surface open can take minutes + ) + except subprocess.TimeoutExpired as exc: + log.error("facebook worker timed out after 300s (group_ids=%r)", group_ids) + raise FacebookError( + code="worker_timeout", + message="facebook worker timed out after 300s", + retryable=True, + action="retry with backoff; the browser session may hang on login walls", + ) from exc + except OSError as exc: + raise FacebookError( + code="worker_spawn_failed", + message=f"could not spawn facebook worker: {exc}", + retryable=True, + action="check the python runtime and worker path", + ) from exc + + if proc.returncode != 0: + log.error( + "facebook worker exited %d (group_ids=%r): %s", + proc.returncode, + group_ids, + (proc.stderr or proc.stdout)[-2000:], + ) + + try: + data = json.loads(proc.stdout or "{}") + except json.JSONDecodeError as exc: + log.error("facebook worker returned non-JSON stdout: %s", proc.stdout[-2000:]) + raise FacebookError( + code="worker_bad_output", + message="facebook worker returned non-JSON output", + retryable=True, + action="retry; log raw worker output", + ) from exc + + if not data.get("ok"): + raise map_worker_error(data, {"group_ids": group_ids}) + return data diff --git a/apps/core/sources/connectors/facebook/connector.py b/apps/core/sources/connectors/facebook/connector.py new file mode 100644 index 00000000..4574c542 --- /dev/null +++ b/apps/core/sources/connectors/facebook/connector.py @@ -0,0 +1,93 @@ +"""Facebook group search connector, unofficial route (Camofox). + +Polls a `facebook_group` source: one live Facebook group search per cycle +via the facebook-worker.py subprocess (which drives a Camofox anti-detect +browser session), mapping each normalized post record to a +`NewFacebookPostPayload` newer than the source's `since` watermark. + +Error semantics follow the connector contract: any Facebook/Camofox +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 FacebookGroupSourceSpec +from sources.payload_registry import register +from sources.payloads import SourcePayload + +from ..base import BaseConnector, ConnectorParseError +from .client import FacebookClient +from .errors import FacebookError +from .payloads import NewFacebookPostPayload + +log = logging.getLogger("sources.facebook") + + +class FacebookGroupConnector(BaseConnector[FacebookGroupSourceSpec]): + """Polls one Facebook group search stream via the Camofox client. + + Live-mode semantics mirror the other connectors: every cycle yields + posts newer than `since` (the Source row's `last_event_at`). There is + no pagination in phase 1: a search returns up to `spec.count` posts + and the connector filters them by the watermark. The worker session + and cookies are resolved per search via the client's env/file/ + credentials-dir resolution (see client.load_cookies). + """ + + kind = FacebookGroupSourceSpec.SOURCE_KIND + payloads: list[type[SourcePayload]] = [NewFacebookPostPayload] + + # Client created lazily on first poll (avoids worker-path resolution at + # import time, which fails when the sibling checkout isn't present). + _client: FacebookClient | None = None + + @property + def _resolved_client(self) -> FacebookClient: + if self._client is None: + self._client = FacebookClient() + return self._client + + def poll( + self, + spec: FacebookGroupSourceSpec, + since: datetime | None, + field_map: dict[str, str] | None = None, + heartbeat: Callable[[], bool] | None = None, + ) -> Iterator[SourcePayload]: + del field_map + del heartbeat + client = self._resolved_client + try: + data = client.search_group(spec.group_ids, spec.terms, spec.count) + except FacebookError as exc: + log.warning( + "facebook group search failed groups=%r code=%s retryable=%s: %s", + spec.group_ids, + exc.code, + exc.retryable, + exc.message, + ) + raise ConnectorParseError( + f"facebook group search {spec.display()} failed: {exc.code}: {exc.message} ({exc.action})" + ) from exc + + results = data.get("result", {}) + for record in results.get("results", []): + payload = NewFacebookPostPayload.from_record(record, query_terms=spec.terms) + # Watermark filter: only surface posts strictly newer than the + # cursor (the poll op advances the source watermark to the + # newest seen, so a post at the watermark is already recorded). + if since is not None and payload.occurred_at <= since: + continue + yield payload + + +register(FacebookGroupConnector.kind, FacebookGroupConnector.payloads) diff --git a/apps/core/sources/connectors/facebook/errors.py b/apps/core/sources/connectors/facebook/errors.py new file mode 100644 index 00000000..96bed3ad --- /dev/null +++ b/apps/core/sources/connectors/facebook/errors.py @@ -0,0 +1,66 @@ +"""Error taxonomy for the Facebook (camofox) connector. + +Maps subprocess / facebook-camofox-client failures 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 FacebookError(Exception): + """Canonical error shape for one Facebook 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) + + +def map_worker_error(worker_output: dict, context: dict[str, Any] | None = None) -> FacebookError: + """Translate a facebook-worker.py error response into a FacebookError.""" + code = worker_output.get("code", "worker_error") + message = worker_output.get("error", "Unknown worker error") + # Auth failures + if code in ("auth_required",): + return FacebookError( + code="auth_required", + message=message, + retryable=False, + action="refresh Facebook cookies in Twenty _socialAccount record", + context=context or {}, + ) + + # Session expired + if code == "session_expired" or "session expired" in message.lower(): + return FacebookError( + code="session_expired", + message=message, + retryable=False, + action="refresh Facebook cookies; session has expired", + context=context or {}, + ) + + # Browser init failure + if "browser" in message.lower() or "camoufox" in message.lower() or "init" in message.lower(): + return FacebookError( + code="browser_init_failed", + message=message, + retryable=True, + action="retry with backoff; browser binary may be downloading", + context=context or {}, + ) + + # Generic worker error + return FacebookError( + code=code, + message=message, + retryable=True, + action="retry with backoff; log and alert after 5 consecutive", + context=context or {}, + ) diff --git a/apps/core/sources/connectors/facebook/payloads.py b/apps/core/sources/connectors/facebook/payloads.py new file mode 100644 index 00000000..c0bc839e --- /dev/null +++ b/apps/core/sources/connectors/facebook/payloads.py @@ -0,0 +1,121 @@ +"""Facebook payloads: a group post observed via the Camofox client. + +Maps the facebook-camofox-client normalized record shape onto the +openmagpie `SourcePayload` contract: the engine judges `title` + +`content`, so a post's body text goes to `content` and the group ID +becomes the within-kind `source_slug`. Metrics / refs / media stay on +the payload as source-specific fields. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any, ClassVar + +from openmagpie_schema.configs import FacebookGroupSourceSpec +from sources.payloads import SourcePayload + +# Facebook post URL base; permalinks are https://facebook.com/groups//posts/ +FACEBOOK_GROUP_URL = "https://facebook.com/groups" + + +class NewFacebookPostPayload(SourcePayload): + """A single Facebook group post observed by a watched group search. + + `author` is the poster's display name; `group_id` is the Facebook + group ID and the within-kind source slug (grouping items by producing + group). `content` is the post body (the engine's judgeable text). + The rest is source-specific: `metrics` (likes/comments/shares), + `matched_terms`, `raw_extraction`. + """ + + PAYLOAD_KIND: ClassVar[str] = "new_fb_post" + + author: str = "" + group_id: str = "" + metrics: dict[str, int | None] = {} + matched_terms: list[str] = [] + + model_config = {"frozen": True, "extra": "ignore"} + + def source_slug(self) -> str | None: + return self.group_id or None + + @classmethod + def sample(cls, variant: int = 0) -> NewFacebookPostPayload: + n = variant + 1 + post_id = f"fb_post_{n}" + group_id = f"group_{n}" + return cls( + external_id=post_id, + kind=cls.PAYLOAD_KIND, + occurred_at=datetime(2026, 5, 27, 12, 0, tzinfo=UTC), + source=FacebookGroupSourceSpec.SOURCE_KIND, + title="", + content=f"Example Facebook post {n}: the post text that matched this watch.", + url=f"{FACEBOOK_GROUP_URL}/{group_id}/posts/{post_id}", + author=f"Example User {n}", + group_id=group_id, + metrics={"likes": 10 + n, "comments": 2 + n, "shares": n}, + matched_terms=["example"], + ) + + @classmethod + def from_record(cls, record: dict[str, Any], query_terms: list[str] | None = None) -> NewFacebookPostPayload: + """Map a facebook-worker.py normalized record dict to a payload. + + The worker returns records in the NormalizedPostRecord shape + (record_id / external_id / group_id / content / url / author / + occurred_at / metrics). The tests hand in lightweight fakes with + the same key names, so no Camofox import is needed. + """ + external_id = str(record.get("external_id") or record.get("record_id") or "") + group_id = str(record.get("group_id") or "") + author = record.get("author") or {} + author_name = str(author.get("name") or "") if isinstance(author, dict) else str(author or "") + + occurred_at = record.get("occurred_at") + if isinstance(occurred_at, str): + try: + occurred_at = datetime.fromisoformat(occurred_at.replace("Z", "+00:00")) + except ValueError: + occurred_at = datetime.now(UTC) + if not isinstance(occurred_at, datetime): + occurred_at = datetime.now(UTC) + if occurred_at.tzinfo is None: + occurred_at = occurred_at.replace(tzinfo=UTC) + + metrics_raw = record.get("metrics") or {} + metrics = { + "likes": int_or_none(metrics_raw.get("likes")), + "comments": int_or_none(metrics_raw.get("comments")), + "shares": int_or_none(metrics_raw.get("shares")), + } + + url = str(record.get("url") or "") + if not url and group_id and external_id: + url = f"{FACEBOOK_GROUP_URL}/{group_id}/posts/{external_id}" + + return cls( + external_id=external_id, + kind=cls.PAYLOAD_KIND, + occurred_at=occurred_at, + source=FacebookGroupSourceSpec.SOURCE_KIND, + title="", + content=str(record.get("content") or ""), + url=url, + author=author_name, + group_id=group_id, + metrics=metrics, + matched_terms=list(record.get("matched_terms") or query_terms or []), + ) + + +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/connectors/twitter/__init__.py b/apps/core/sources/connectors/twitter/__init__.py new file mode 100644 index 00000000..7643886c --- /dev/null +++ b/apps/core/sources/connectors/twitter/__init__.py @@ -0,0 +1,17 @@ +"""X (Twitter) connector, unofficial route (twikit), ported from listeningkit. + +One file per concern: + - `connector.py` ; the `TwitterSearchConnector` impl (poll loop) + - `client.py` ; `TwikitClient` wrapper (cookies env/file/credentials-dir, + proxy attachment, error translation) + - `payloads.py` ; `NewTweetPayload` (twikit Tweet -> SourcePayload) + - `errors.py` ; twikit error taxonomy -> canonical ListenerError + +Future variants (user timeline, list timeline) reuse `TwikitClient` with +their own spec + payload. +""" + +from .connector import TwitterSearchConnector +from .payloads import NewTweetPayload + +__all__ = ["NewTweetPayload", "TwitterSearchConnector"] diff --git a/apps/core/sources/connectors/twitter/client.py b/apps/core/sources/connectors/twitter/client.py new file mode 100644 index 00000000..48ea7e81 --- /dev/null +++ b/apps/core/sources/connectors/twitter/client.py @@ -0,0 +1,185 @@ +"""Twikit-based X (Twitter) client, unofficial route. + +Ported from REPOS/listeningkit/backend/packages/listeners/twitter/{client, +config,credentials}.py so the openmagpie connector keeps the exact cookie +JSON formats the live listeningkit setup uses (they work today; nothing +about them is omitted or reshaped): + +- `TWITTER_COOKIES_JSON` (a full JSON dict of x.com cookies) — the live + `.env.local` source, +- `TWITTER_COOKIE_AUTH_TOKEN` + `TWITTER_COOKIE_CT0` (the critical pair), +- `TWITTER_COOKIES_FILE` (path to a twikit cookies.json), +- `TWITTER_CREDENTIALS_DIR` (dir of `*.json` cookie exports, each with an + optional `*.proxy` pin), default `credentials/` relative to the core app. + +Proxy: `TWITTER_PROXY` env (or a per-credential `.proxy` pin in +`TWITTER_CREDENTIALS_DIR`). twikit supports `proxy=` natively, so every +request egresses through it (transport-level). + +twikit is async-only; the connector's `poll` is a sync iterator, so +`TwikitClient.search` bridges with a fresh event loop per call +(`asyncio.run`). One search per poll cycle is a bounded, short-lived +loop; no shared loop state to leak across poll cycles. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +from pathlib import Path +from typing import Literal + +from twikit import Client +from twikit.errors import TwitterException + +from .errors import ListenerError, map_bootstrap_failure, map_twikit_error + +# The product string twikit passes to X's search endpoint. The spec's +# `latest` / `top` literals map 1:1 onto these (see connector.py). +TwikitProduct = Literal["Latest", "Top"] + +log = logging.getLogger("sources.twitter") + +# The two cookies that make an authenticated twikit session work. +CRITICAL_COOKIES = ("auth_token", "ct0") + + +def _load_cookie_file(path: Path) -> dict[str, str]: + """Parse a Get-cookies.txt-LOCALLY JSON export (array) or a plain dict.""" + data = json.loads(path.read_text(encoding="utf-8")) + if isinstance(data, dict): + return {str(k): str(v) for k, v in data.items() if v} + if isinstance(data, list): + out: dict[str, str] = {} + for item in data: + name, value = item.get("name"), item.get("value") + if name and value: + out[str(name)] = str(value) + return out + raise ValueError(f"{path}: JSON must be an object or an array of cookie objects") + + +def load_cookies( + *, + cookies_json: str | None = None, + cookies_file: str | None = None, + credentials_dir: str | None = None, +) -> tuple[dict[str, str], str | None]: + """Resolve the cookie dict + proxy for one X session, in priority order. + + Returns ``(cookies, proxy)``. ``cookies`` is empty when nothing is + configured (the caller stays guest-mode; the first search then fails + with a mapped ``unauthorized`` error the poll loop handles). ``proxy`` + comes from `TWITTER_PROXY` or a `.proxy` pin next to the chosen + cookie export. + """ + proxy = os.environ.get("TWITTER_PROXY", "").strip() or None + + if cookies_json: + try: + data = json.loads(cookies_json) + except json.JSONDecodeError: + log.warning("TWITTER_COOKIES_JSON is not valid JSON; ignoring") + else: + if isinstance(data, dict) and data: + return {str(k): str(v) for k, v in data.items()}, proxy + log.warning("TWITTER_COOKIES_JSON is not a non-empty JSON object; ignoring") + + individual = {name: os.environ.get(f"TWITTER_COOKIE_{name.upper()}", "").strip() for name in CRITICAL_COOKIES} + if all(individual.values()): + return individual, proxy + + if cookies_file and Path(cookies_file).exists(): + try: + return _load_cookie_file(Path(cookies_file)), proxy + except (json.JSONDecodeError, ValueError, OSError) as exc: + log.warning("TWITTER_COOKIES_FILE %s unreadable (%s); ignoring", cookies_file, exc) + + directory = Path(credentials_dir or os.environ.get("TWITTER_CREDENTIALS_DIR", "credentials")) + if directory.exists(): + for path in sorted(directory.glob("*.json")): + try: + cookies = _load_cookie_file(path) + except (json.JSONDecodeError, ValueError, OSError) as exc: + log.warning("skipping %s: %s", path.name, exc) + continue + if not {"auth_token", "ct0"}.issubset(cookies): + log.warning("skipping %s: missing auth_token/ct0 (critical pair)", path.name) + continue + proxy_path = path.with_suffix(".proxy") + pin = proxy_path.read_text().strip() if proxy_path.exists() else proxy + return cookies, pin + log.warning("credentials dir %s: no usable cookie set found", directory) + elif credentials_dir: + log.warning("credentials dir %s not found; no sessions loaded", credentials_dir) + + return {}, proxy + + +class ListenerErrorWrapper(Exception): + """Carries a canonical ListenerError through the pipeline.""" + + def __init__(self, err: ListenerError) -> None: + super().__init__(err.message) + self.error = err + + +class TwikitClient: + """Thin, proxy-bound wrapper around the twikit async client. + + One twikit ``Client`` per search call. twikit's ``Client.__init__`` + creates an ``httpx.AsyncClient`` bound to the *currently running* + event loop, and ``search()`` drives twikit with a fresh + ``asyncio.run`` loop per call — so the twikit client must be + constructed inside that loop. A shared instance created at import + time dies with the first loop and later searches fail with + "Event loop is closed" (observed on the second/third source in a + multi-source feed poll). Cookies are resolved once here (cheap env / + file reads); only the twikit client is per-call. + """ + + def __init__( + self, + *, + language: str = "en-US", + proxy: str | None = None, + user_agent: str | None = None, + cookies: dict[str, str] | None = None, + cookies_json: str | None = None, + cookies_file: str | None = None, + credentials_dir: str | None = None, + ) -> None: + self._language = language + self._user_agent = user_agent + self.proxy = proxy + if cookies is not None: + self._cookies = dict(cookies) + else: + self._cookies, self.proxy = load_cookies( + cookies_json=cookies_json, + cookies_file=cookies_file, + credentials_dir=credentials_dir, + ) + + async def _search_async(self, query: str, mode: TwikitProduct, count: int): + # Build the twikit client here, inside the event loop search() + # runs: twikit's Client binds its httpx.AsyncClient to this loop + # at construction (see class docstring). + client = Client(language=self._language, proxy=self.proxy, user_agent=self._user_agent) + if self._cookies: + client.set_cookies(dict(self._cookies), clear_cookies=True) + log.info("session: loaded %d cookie(s)", len(self._cookies)) + else: + log.warning("session: no cookies configured; guest mode only") + try: + return await client.search_tweet(query, mode, count=count) + except TwitterException as exc: + raise ListenerErrorWrapper(map_twikit_error(exc, {"query": query, "mode": mode})) from exc + except Exception as exc: # bootstrap failures (degraded shell) are not TwitterException + raise ListenerErrorWrapper(map_bootstrap_failure(exc, {"query": query})) from exc + + def search(self, query: str, mode: TwikitProduct = "Latest", count: int = 20): + """Run one live search; returns a twikit Result[Tweet] (or a test double).""" + return asyncio.run(self._search_async(query, mode, count)) diff --git a/apps/core/sources/connectors/twitter/connector.py b/apps/core/sources/connectors/twitter/connector.py new file mode 100644 index 00000000..ac4a1e4a --- /dev/null +++ b/apps/core/sources/connectors/twitter/connector.py @@ -0,0 +1,92 @@ +"""X (Twitter) search connector, unofficial route (twikit). + +Polls a `twitter_search` source: one live X search per cycle via the +twikit client (unclecode fork), mapping each result tweet to a +`NewTweetPayload` newer than the source's `since` watermark. + +Error semantics follow the connector contract: any X/twikit 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 TwitterSearchSourceSpec +from sources.payload_registry import register +from sources.payloads import SourcePayload + +from ..base import BaseConnector, ConnectorParseError +from .client import ListenerErrorWrapper, TwikitClient, TwikitProduct +from .payloads import NewTweetPayload + +log = logging.getLogger("sources.twitter") + +# Mode string twikit passes to X's search endpoint. The spec's `latest` / +# `top` literals map 1:1 to twikit's `"Latest"` / `"Top"`. +_TWIKIT_MODES: dict[str, TwikitProduct] = {"latest": "Latest", "top": "Top"} + + +class TwitterSearchConnector(BaseConnector[TwitterSearchSourceSpec]): + """Polls one X (Twitter) search stream via the unofficial twikit route. + + Live-mode semantics mirror the other connectors: every cycle yields + tweets newer than `since` (the Source row's `last_event_at`). There is + no pagination in phase 1: a search returns up to `spec.count` tweets + and the connector filters them by the watermark (X's own recency + ordering makes the first page the newest; a quiet stream needs no + backfill walk). Multi-account session rotation (listeningkit's + session_pool) is a follow-up; phase 1 uses one cookie set via the + client's env/file/credentials-dir resolution. + """ + + kind = TwitterSearchSourceSpec.SOURCE_KIND + payloads: list[type[SourcePayload]] = [NewTweetPayload] + + # One stateless client; cookies + proxy resolved per search from the + # live env / credentials files (see client.load_cookies). + _client = TwikitClient() + + def poll( + self, + spec: TwitterSearchSourceSpec, + 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, _TWIKIT_MODES[spec.mode], spec.count) + except ListenerErrorWrapper as exc: + err = exc.error + log.warning( + "twitter search failed query=%r code=%s retryable=%s: %s", + spec.query, + err.code, + err.retryable, + err.message, + ) + raise ConnectorParseError( + f"twitter search {spec.display()} failed: {err.code}: {err.message} ({err.action})" + ) from exc + + for tweet in results: + payload = NewTweetPayload.from_tweet(tweet, query=spec.query) + # Watermark filter: only surface tweets strictly newer than the + # cursor (the poll op advances the source watermark to the + # newest seen, so a tweet at the watermark is already recorded). + if since is not None and payload.occurred_at <= since: + continue + if spec.lang and payload.lang and payload.lang != spec.lang: + continue + yield payload + + +register(TwitterSearchConnector.kind, TwitterSearchConnector.payloads) diff --git a/apps/core/sources/connectors/twitter/errors.py b/apps/core/sources/connectors/twitter/errors.py new file mode 100644 index 00000000..0fb45b08 --- /dev/null +++ b/apps/core/sources/connectors/twitter/errors.py @@ -0,0 +1,141 @@ +"""Error taxonomy for the X (Twitter) connector, ported from listeningkit. + +Every call into twikit can raise a ``TwitterException`` subclass (or a +bootstrap failure when X serves a degraded shell). This module maps those +to a canonical ``ListenerError``; the connector translates that into +``ConnectorParseError`` at the poll boundary so the feed poll op recovers +per-source (one bad source must not abort the feed cycle). +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from twikit.errors import ( + AccountLocked, + AccountSuspended, + BadRequest, + DuplicateTweet, + Forbidden, + InvalidMedia, + NotFound, + RequestTimeout, + ServerError, + TooManyRequests, + TweetNotAvailable, + TwitterException, + Unauthorized, + UserNotFound, + UserUnavailable, +) + +TWIKIT_ERROR_CODE: dict[type[TwitterException], str] = { + BadRequest: "bad_request", + Unauthorized: "unauthorized", + Forbidden: "forbidden", + NotFound: "not_found", + RequestTimeout: "timeout", + TooManyRequests: "rate_limited", + ServerError: "upstream_error", + AccountSuspended: "account_suspended", + AccountLocked: "account_locked", + DuplicateTweet: "duplicate_tweet", + TweetNotAvailable: "tweet_unavailable", + InvalidMedia: "invalid_media", + UserNotFound: "user_not_found", + UserUnavailable: "user_unavailable", +} + + +@dataclass +class ListenerError: + """Canonical error shape for one X fetch failure.""" + + code: str # stable machine code, see TWIKIT_ERROR_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) + headers: dict[str, str] | None = None + rate_limit_reset: int | None = None # unix ts from x-rate-limit-reset + + +BOOTSTRAP_BLOCKED_MARKERS = ( + "Couldn't get KEY_BYTE indices", + "Couldn't get key from the page source", +) + + +def map_bootstrap_failure(exc: Exception, context: dict[str, Any] | None = None) -> ListenerError: + """X served a degraded shell (bot wall) so twikit could not bootstrap its + ClientTransaction. Cause is almost always egress IP reputation; fix = + residential proxy (see listeningkit docs: proxy.md).""" + msg = str(exc) + code = "bootstrap_blocked" if any(m in msg for m in BOOTSTRAP_BLOCKED_MARKERS) else "internal" + action = ( + "X served a degraded shell to this egress IP (no ondemand.s chunk map). " + "Use a residential proxy + browser fingerprint." + if code == "bootstrap_blocked" + else "unknown failure; log raw and retry with backoff" + ) + return ListenerError( + code=code, + message=msg, + retryable=code == "internal", + action=action, + context=context or {}, + ) + + +def map_twikit_error(exc: TwitterException, context: dict[str, Any] | None = None) -> ListenerError: + """Translate a twikit exception into a canonical ListenerError.""" + code = TWIKIT_ERROR_CODE.get(type(exc), "twitter_error") + reset = getattr(exc, "rate_limit_reset", None) + headers = getattr(exc, "headers", None) + + # X's SearchTimeline intermittently 404s with an EMPTY body (observed + # repeatedly on live polls: same query succeeds on retry seconds later, + # independent of session/cookies/query). That is a transient upstream + # flake on the search endpoint, NOT a deleted tweet/user — a genuine + # not_found carries a message. Retryable so the ops layer backs off + # instead of treating the source as dead. + if isinstance(exc, NotFound) and not str(exc).strip(): + return ListenerError( + code="search_timeline_unavailable", + message="X SearchTimeline returned an empty 404 (transient upstream flake)", + retryable=True, + action="retry with backoff; watermark stays put so the next cycle re-reads", + context=context or {}, + headers=headers, + rate_limit_reset=reset, + ) + + retryable_actions: dict[str, tuple[bool, str]] = { + "bad_request": (False, "fix query / payload; do not retry as-is"), + "unauthorized": (False, "refresh session (guest token / cookies) and retry once"), + "forbidden": (False, "rotate session + proxy pin; alert"), + "not_found": (False, "tweet/user no longer exists; skip"), + "timeout": (True, "retry with backoff"), + "rate_limited": (True, f"backoff until reset ({reset})"), + "upstream_error": (True, "retry with backoff; alert after 5 consecutive"), + "account_suspended": (False, "pause account mode; rotate to a different session; alert"), + "account_locked": (False, "Arkose challenge; pause account mode; alert"), + "duplicate_tweet": (False, "skip (dedupe by design)"), + "tweet_unavailable": (False, "skip"), + "invalid_media": (False, "skip"), + "user_not_found": (False, "skip"), + "user_unavailable": (False, "skip"), + "twitter_error": (True, "unknown upstream error; log raw and retry with backoff"), + } + retryable, action = retryable_actions.get(code, (True, "unknown; log and retry with backoff")) + + return ListenerError( + code=code, + message=str(exc), + retryable=retryable, + action=action, + context=context or {}, + headers=headers, + rate_limit_reset=reset, + ) diff --git a/apps/core/sources/connectors/twitter/payloads.py b/apps/core/sources/connectors/twitter/payloads.py new file mode 100644 index 00000000..57df7602 --- /dev/null +++ b/apps/core/sources/connectors/twitter/payloads.py @@ -0,0 +1,132 @@ +"""X (Twitter) payloads: a tweet observed via the unofficial twikit route. + +Shapes the listeningkit SocialEvent normalization (see +REPOS/listeningkit/docs/okf/backend/domains/twitter/unofficial/parsing.md) +onto the openmagpie `SourcePayload` contract: the engine judges `title` + +`content`, so a tweet's text goes to `content` and the author's handle +becomes the within-kind `source_slug`. Metrics / refs / media stay on the +payload as source-specific fields (available to actions that read them, +omitted from the engine prompt unless included). +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any, ClassVar + +from openmagpie_schema.configs import TwitterSearchSourceSpec +from sources.payloads import SourcePayload + +# Tweet URL base; a tweet's permalink is https://x.com//status/. +X_STATUS_URL = "https://x.com" + + +class NewTweetPayload(SourcePayload): + """A single tweet observed by a watched X search stream. + + `author` is the user's display name; `handle` is the @screen_name and the + within-kind source slug (grouping items by producing account). `content` + is the tweet's full text (the engine's judgeable body). The rest is the + listeningkit event shape carried as payload fields: `metrics`, `refs` + (in_reply_to / quoted / retweet_of), `media`, `lang`. + """ + + PAYLOAD_KIND: ClassVar[str] = "new_tweet" + + author: str = "" + handle: str = "" + lang: str = "" + 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) -> NewTweetPayload: + n = variant + 1 + tweet_id = str(999_000_000_000_000_000 + n) + handle = f"example_user_{n}" + return cls( + external_id=tweet_id, + kind=cls.PAYLOAD_KIND, + occurred_at=datetime(2026, 5, 27, 12, 0, tzinfo=UTC), + source=TwitterSearchSourceSpec.SOURCE_KIND, + title="", + content=f"Example tweet {n}: the post text that matched this watch.", + url=f"{X_STATUS_URL}/{handle}/status/{tweet_id}", + author=f"Example User {n}", + handle=handle, + lang="en", + metrics={"likes": 100 + n, "retweets": 20 + n, "replies": 5 + n, "quotes": 2 + n, "views": 1000 + n}, + refs={"in_reply_to": None, "quoted": None, "retweet_of": None}, + media=[], + ) + + @classmethod + def from_tweet(cls, tweet: Any, query: str | None = None) -> NewTweetPayload: + """Map a twikit `Tweet` (or a duck-typed test double) to a payload. + + Kept attribute-driven (getattr with a default) so the connector's + unit tests can hand in lightweight fakes without importing twikit; + the real twikit Tweet supplies the same attributes. `query` is + recorded nowhere on the payload (the SourceSpec carries it); it is + accepted for symmetry with the listeningkit event's listenId and + future field_map use. + """ + del query + tweet_id = str(getattr(tweet, "id", None) or "") + user = getattr(tweet, "user", None) + handle = "" + author = "" + if user is not None: + handle = str(getattr(user, "screen_name", None) or getattr(user, "username", None) or "") + author = str(getattr(user, "name", None) or "") + text = getattr(tweet, "full_text", None) or getattr(tweet, "text", None) or "" + created = getattr(tweet, "created_at_datetime", None) or getattr(tweet, "created_at", None) + occurred_at = created if isinstance(created, datetime) else datetime.now(UTC) + if occurred_at.tzinfo is None: + occurred_at = occurred_at.replace(tzinfo=UTC) + lang = str(getattr(tweet, "lang", None) or "") + + def _id(obj: Any) -> str | None: + return str(getattr(obj, "id", None)) if obj is not None else None + + media = [] + for m in getattr(tweet, "media", None) or []: + media.append( + { + "type": getattr(m, "type", None), + "url": getattr(m, "media_url_https", None) or getattr(m, "media_url", None), + "thumbnail": getattr(m, "thumbnail_url", None), + } + ) + + return cls( + external_id=tweet_id, + kind=cls.PAYLOAD_KIND, + occurred_at=occurred_at, + source=TwitterSearchSourceSpec.SOURCE_KIND, + title="", + content=text, + url=f"{X_STATUS_URL}/{handle}/status/{tweet_id}" if handle else "", + author=author, + handle=handle, + lang=lang, + metrics={ + "likes": getattr(tweet, "favorite_count", None), + "retweets": getattr(tweet, "retweet_count", None), + "replies": getattr(tweet, "reply_count", None), + "quotes": getattr(tweet, "quote_count", None), + "views": getattr(tweet, "view_count", None), + }, + refs={ + "in_reply_to": _id(getattr(tweet, "in_reply_to", None)), + "quoted": _id(getattr(tweet, "quote", None)), + "retweet_of": _id(getattr(tweet, "retweeted_tweet", None)), + }, + media=media, + ) diff --git a/apps/core/sources/registry.py b/apps/core/sources/registry.py index bfa6713d..2a6c4e99 100644 --- a/apps/core/sources/registry.py +++ b/apps/core/sources/registry.py @@ -11,17 +11,21 @@ from common.models import reject_bad_plugin_kind from sources.connectors import ( Connector, + FacebookGroupConnector, HackerNewsCommentConnector, HackerNewsFeedConnector, RedditSubRedditConnector, RssConnector, + TwitterSearchConnector, ) _REGISTRY: dict[str, Connector[Any]] = { + FacebookGroupConnector.kind: FacebookGroupConnector(), RedditSubRedditConnector.kind: RedditSubRedditConnector(), RssConnector.kind: RssConnector(), HackerNewsFeedConnector.kind: HackerNewsFeedConnector(), HackerNewsCommentConnector.kind: HackerNewsCommentConnector(), + TwitterSearchConnector.kind: TwitterSearchConnector(), } # Core kinds captured before any plugin registers; a plugin can't replace one. diff --git a/apps/core/sources/tests_facebook.py b/apps/core/sources/tests_facebook.py new file mode 100644 index 00000000..d5af6544 --- /dev/null +++ b/apps/core/sources/tests_facebook.py @@ -0,0 +1,278 @@ +"""Facebook group search connector tests (offline, fake worker results). + +The connector's only I/O is the FacebookClient subprocess call; these tests +swap in a fake client result and pin: spec validation (the empty-group-ids +guard), the watermark filter, error translation (FacebookError -> ConnectorParseError), +and payload mapping (a fake normalized record -> NewFacebookPostPayload). +""" + +from datetime import UTC, datetime +from unittest import mock + +from django.test import SimpleTestCase +from pydantic import ValidationError + +from openmagpie_schema.configs import FacebookGroupSourceSpec +from sources.connectors.base import ConnectorParseError +from sources.connectors.facebook.client import FacebookClient +from sources.connectors.facebook.connector import FacebookGroupConnector +from sources.connectors.facebook.errors import FacebookError +from sources.connectors.facebook.payloads import NewFacebookPostPayload + + +class _FakeGroupPost: + """A normalized Facebook post record, matching the facebook-worker.py shape.""" + + def __init__( + self, + post_id: str, + group_id: str = "group_1", + author_name: str = "Alice", + content: str = "Hello from Facebook", + created: datetime | None = None, + likes: int | None = 10, + comments: int | None = 2, + shares: int | None = 1, + ): + self._record = { + "external_id": post_id, + "group_id": group_id, + "author": {"name": author_name}, + "content": content, + "occurred_at": (created or datetime(2026, 6, 1, 12, 0, tzinfo=UTC)).isoformat(), + "url": f"https://facebook.com/groups/{group_id}/posts/{post_id}", + "metrics": { + "likes": likes, + "comments": comments, + "shares": shares, + }, + "matched_terms": ["saas"], + } + + def dict(self) -> dict: + return dict(self._record) + + +class FacebookGroupSourceSpecTests(SimpleTestCase): + def test_empty_group_ids_rejected(self): + with self.assertRaises(ValidationError): + FacebookGroupSourceSpec(kind="facebook_group", group_ids=[]) + + def test_group_ids_whitespace_only_rejected(self): + with self.assertRaises(ValidationError): + FacebookGroupSourceSpec(kind="facebook_group", group_ids=[" "]) + + def test_minimal_spec(self): + spec = FacebookGroupSourceSpec(kind="facebook_group", group_ids=["12345"]) + self.assertEqual(spec.group_ids, ["12345"]) + self.assertEqual(spec.terms, []) + self.assertEqual(spec.count, 20) + + def test_full_spec(self): + spec = FacebookGroupSourceSpec( + kind="facebook_group", + group_ids=["12345", "67890"], + terms=["saas", "fundraising"], + count=50, + ) + self.assertEqual(spec.group_ids, ["12345", "67890"]) + self.assertEqual(spec.terms, ["saas", "fundraising"]) + self.assertEqual(spec.count, 50) + + def test_terms_whitespace_stripped(self): + spec = FacebookGroupSourceSpec( + kind="facebook_group", + group_ids=["12345"], + terms=[" saas ", "fundraising "], + ) + self.assertEqual(spec.terms, ["saas", "fundraising"]) + + def test_count_bounds(self): + with self.assertRaises(ValidationError): + FacebookGroupSourceSpec(kind="facebook_group", group_ids=["x"], count=0) + with self.assertRaises(ValidationError): + FacebookGroupSourceSpec(kind="facebook_group", group_ids=["x"], count=101) + + def test_display(self): + spec = FacebookGroupSourceSpec(kind="facebook_group", group_ids=["12345", "67890"]) + self.assertIn("12345", spec.display()) + + +class FacebookGroupConnectorTests(SimpleTestCase): + def _connector(self, worker_result: dict): + client = mock.Mock(spec=FacebookClient) + client.search_group.return_value = worker_result + conn = FacebookGroupConnector() + conn._client = client # inject the fake before first poll + return conn, client + + def test_yields_payloads_newer_than_since(self): + spec = FacebookGroupSourceSpec(kind="facebook_group", group_ids=["123"]) + newer = _FakeGroupPost("2", created=datetime(2026, 6, 15, 12, 0, tzinfo=UTC)) + older = _FakeGroupPost("1", created=datetime(2026, 5, 1, 12, 0, tzinfo=UTC)) + worker_result = { + "ok": True, + "result": { + "results": [newer.dict(), older.dict()], + "cursor": {}, + "matched_terms": [], + "events": ["groups.search_completed"], + }, + "new_cookies": [], + } + conn, client = self._connector(worker_result) + payloads = list(conn.poll(spec, since=datetime(2026, 5, 15, tzinfo=UTC))) + self.assertEqual(len(payloads), 1) + self.assertEqual(payloads[0].external_id, "2") + client.search_group.assert_called_once_with(["123"], [], 20) + + def test_all_payloads_when_no_since(self): + spec = FacebookGroupSourceSpec(kind="facebook_group", group_ids=["123"]) + p1 = _FakeGroupPost("1") + p2 = _FakeGroupPost("2") + worker_result = { + "ok": True, + "result": { + "results": [p1.dict(), p2.dict()], + "cursor": {}, + "matched_terms": [], + "events": ["groups.search_completed"], + }, + "new_cookies": [], + } + conn, _ = self._connector(worker_result) + payloads = list(conn.poll(spec, since=None)) + self.assertEqual(len(payloads), 2) + + def test_terms_passed_to_client(self): + spec = FacebookGroupSourceSpec( + kind="facebook_group", group_ids=["123"], terms=["saas", "fundraising"] + ) + worker_result = { + "ok": True, + "result": {"results": [], "cursor": {}, "matched_terms": [], "events": ["groups.search_completed"]}, + "new_cookies": [], + } + conn, client = self._connector(worker_result) + list(conn.poll(spec, since=None)) + client.search_group.assert_called_once_with(["123"], ["saas", "fundraising"], 20) + + def test_worker_error_maps_to_connector_parse_error(self): + spec = FacebookGroupSourceSpec(kind="facebook_group", group_ids=["123"]) + err = FacebookError( + code="worker_error", + message="worker failed", + retryable=True, + action="retry", + ) + client = mock.Mock(spec=FacebookClient) + client.search_group.side_effect = err + conn = FacebookGroupConnector() + conn._client = client + with self.assertRaises(ConnectorParseError) as ctx: + list(conn.poll(spec, since=None)) + self.assertIn("worker_error", str(ctx.exception)) + + def test_empty_results_is_not_an_error(self): + spec = FacebookGroupSourceSpec(kind="facebook_group", group_ids=["123"]) + worker_result = { + "ok": True, + "result": {"results": [], "cursor": {}, "matched_terms": [], "events": ["groups.search_completed"]}, + "new_cookies": [], + } + conn, _ = self._connector(worker_result) + payloads = list(conn.poll(spec, since=None)) + self.assertEqual(payloads, []) + + +class NewFacebookPostPayloadTests(SimpleTestCase): + def test_from_record(self): + record = { + "external_id": "123", + "group_id": "group_1", + "author": {"name": "Alice"}, + "content": "Hello from Facebook", + "occurred_at": "2026-06-01T12:00:00+00:00", + "url": "https://facebook.com/groups/group_1/posts/123", + "metrics": {"likes": 10, "comments": 2, "shares": 1}, + "matched_terms": ["saas"], + } + p = NewFacebookPostPayload.from_record(record, query_terms=["saas", "fundraising"]) + self.assertEqual(p.external_id, "123") + self.assertEqual(p.group_id, "group_1") + self.assertEqual(p.author, "Alice") + self.assertEqual(p.content, "Hello from Facebook") + self.assertEqual(p.source, "facebook_group") + self.assertEqual(p.url, "https://facebook.com/groups/group_1/posts/123") + self.assertEqual(p.metrics["likes"], 10) + self.assertEqual(p.metrics["comments"], 2) + self.assertEqual(p.metrics["shares"], 1) + # The record's own matched_terms win; query_terms is only the fallback + # when a record carries none. + self.assertEqual(p.matched_terms, ["saas"]) + + def test_from_record_author_as_string(self): + record = { + "external_id": "456", + "group_id": "group_2", + "author": "Bob", + "content": "Another post", + "occurred_at": "2026-06-02T12:00:00+00:00", + "metrics": {"likes": 5, "comments": 0, "shares": 0}, + } + p = NewFacebookPostPayload.from_record(record) + self.assertEqual(p.author, "Bob") + self.assertEqual(p.matched_terms, []) + + def test_from_record_missing_optional_fields(self): + record = { + "external_id": "789", + "group_id": "group_3", + "content": "Minimal post", + } + p = NewFacebookPostPayload.from_record(record) + self.assertEqual(p.external_id, "789") + self.assertEqual(p.group_id, "group_3") + self.assertEqual(p.author, "") + self.assertEqual(p.metrics, {"likes": None, "comments": None, "shares": None}) + self.assertEqual(p.matched_terms, []) + + def test_from_record_uses_query_terms_as_fallback(self): + record = { + "external_id": "abc", + "group_id": "group_5", + "content": "Fallback terms", + } + p = NewFacebookPostPayload.from_record(record, query_terms=["saas", "fundraising"]) + self.assertEqual(p.matched_terms, ["saas", "fundraising"]) + + def test_from_record_url_built_if_missing(self): + record = { + "external_id": "abc", + "group_id": "group_4", + "content": "Post with generated URL", + } + p = NewFacebookPostPayload.from_record(record) + self.assertEqual(p.url, "https://facebook.com/groups/group_4/posts/abc") + + def test_sample_distinct(self): + a = NewFacebookPostPayload.sample(0) + b = NewFacebookPostPayload.sample(1) + self.assertNotEqual(a.external_id, b.external_id) + self.assertEqual(a.PAYLOAD_KIND, "new_fb_post") + + def test_source_slug_returns_group_id(self): + p = NewFacebookPostPayload.sample(0) + self.assertEqual(p.source_slug(), p.group_id) + + def test_source_slug_none_when_group_id_empty(self): + p = NewFacebookPostPayload( + kind="new_fb_post", + external_id="1", + source="facebook_group", + occurred_at=datetime(2026, 1, 1, tzinfo=UTC), + title="", + content="", + group_id="", + ) + self.assertIsNone(p.source_slug()) diff --git a/apps/core/sources/tests_twitter.py b/apps/core/sources/tests_twitter.py new file mode 100644 index 00000000..aa0bfee0 --- /dev/null +++ b/apps/core/sources/tests_twitter.py @@ -0,0 +1,155 @@ +"""Twitter search connector tests (offline, fake twikit results). + +The connector's only I/O is the twikit client (`TwikitClient.search`); these +tests swap in a fake result iterator and pin: spec validation (the blank-query +firehose guard), the watermark filter, the lang filter, error translation +(ListenerErrorWrapper -> ConnectorParseError), and payload mapping (a duck- +typed fake Tweet -> NewTweetPayload). +""" + +from datetime import UTC, datetime +from unittest import mock + +from django.test import SimpleTestCase +from pydantic import ValidationError + +from openmagpie_schema.configs import TwitterSearchSourceSpec +from sources.connectors.base import ConnectorParseError +from sources.connectors.twitter.client import ListenerErrorWrapper +from sources.connectors.twitter.connector import TwitterSearchConnector +from sources.connectors.twitter.errors import ListenerError +from sources.connectors.twitter.payloads import NewTweetPayload + + +class _FakeUser: + def __init__(self, handle: str, name: str = "Some User"): + self.id = "987654321" + self.screen_name = handle + self.username = handle + self.name = name + + +class _FakeTweet: + def __init__( + self, + tweet_id: str, + handle: str = "alice", + text: str = "hello from x", + created: datetime | None = None, + lang: str = "en", + ): + self.id = tweet_id + self.user = _FakeUser(handle) + self.full_text = text + self.created_at_datetime = created or datetime(2026, 6, 1, 12, 0, tzinfo=UTC) + self.created_at = self.created_at_datetime + self.lang = lang + self.favorite_count = 10 + self.retweet_count = 2 + self.reply_count = 1 + self.quote_count = 0 + self.view_count = 100 + self.media = [] + self.in_reply_to = None + self.quote = None + self.retweeted_tweet = None + + +class TwitterSearchSourceSpecTests(SimpleTestCase): + def test_blank_query_rejected(self): + with self.assertRaises(ValidationError): + TwitterSearchSourceSpec(kind="twitter_search", query=" ") + + def test_count_bounds(self): + with self.assertRaises(ValidationError): + TwitterSearchSourceSpec(kind="twitter_search", query="x", count=0) + with self.assertRaises(ValidationError): + TwitterSearchSourceSpec(kind="twitter_search", query="x", count=101) + + def test_defaults(self): + spec = TwitterSearchSourceSpec(kind="twitter_search", query="social listening") + self.assertEqual(spec.mode, "latest") + self.assertEqual(spec.count, 20) + self.assertEqual(spec.lang, "") + + +class TwitterSearchConnectorTests(SimpleTestCase): + def _connector(self, results): + client = mock.Mock() + client.search.return_value = results + conn = TwitterSearchConnector() + conn._client = client + return conn, client + + def test_yields_payloads_newer_than_since(self): + spec = TwitterSearchSourceSpec(kind="twitter_search", query='"social listening"') + tweets = [ + _FakeTweet("2", text="newer"), + _FakeTweet("1", text="older", created=datetime(2026, 5, 1, 12, 0, tzinfo=UTC)), + ] + conn, client = self._connector(tweets) + payloads = list(conn.poll(spec, since=datetime(2026, 5, 15, tzinfo=UTC))) + self.assertEqual(len(payloads), 1) + self.assertEqual(payloads[0].external_id, "2") + client.search.assert_called_once_with('"social listening"', "Latest", 20) + + def test_lang_filter(self): + spec = TwitterSearchSourceSpec(kind="twitter_search", query="x", lang="es") + tweets = [_FakeTweet("1", lang="en"), _FakeTweet("2", lang="es")] + conn, _ = self._connector(tweets) + payloads = list(conn.poll(spec, since=None)) + self.assertEqual([p.external_id for p in payloads], ["2"]) + + def test_error_maps_to_connector_parse_error(self): + spec = TwitterSearchSourceSpec(kind="twitter_search", query="x") + err = ListenerError(code="rate_limited", message="slow down", retryable=True, action="backoff") + client = mock.Mock() + client.search.side_effect = ListenerErrorWrapper(err) + conn = TwitterSearchConnector() + conn._client = client + with self.assertRaises(ConnectorParseError) as ctx: + list(conn.poll(spec, since=None)) + self.assertIn("rate_limited", str(ctx.exception)) + + def test_empty_404_maps_to_retryable_connector_error(self): + """X SearchTimeline empty-404 is a transient flake, not a dead tweet.""" + from sources.connectors.twitter.errors import ListenerError + + spec = TwitterSearchSourceSpec(kind="twitter_search", query="x") + err = ListenerError( + code="search_timeline_unavailable", + message="X SearchTimeline returned an empty 404 (transient upstream flake)", + retryable=True, + action="retry with backoff", + ) + client = mock.Mock() + client.search.side_effect = ListenerErrorWrapper(err) + conn = TwitterSearchConnector() + conn._client = client + with self.assertRaises(ConnectorParseError) as ctx: + list(conn.poll(spec, since=None)) + self.assertIn("search_timeline_unavailable", str(ctx.exception)) + + def test_mode_top_maps_to_twikit_top(self): + spec = TwitterSearchSourceSpec(kind="twitter_search", query="x", mode="top") + conn, client = self._connector([_FakeTweet("1")]) + list(conn.poll(spec, since=None)) + client.search.assert_called_once_with("x", "Top", 20) + + +class NewTweetPayloadTests(SimpleTestCase): + def test_from_tweet(self): + p = NewTweetPayload.from_tweet(_FakeTweet("123", handle="alice", text="hi")) + self.assertEqual(p.external_id, "123") + self.assertEqual(p.handle, "alice") + self.assertEqual(p.content, "hi") + self.assertEqual(p.source, "twitter_search") + self.assertEqual(p.url, "https://x.com/alice/status/123") + self.assertEqual(p.metrics["likes"], 10) + self.assertEqual(p.refs["in_reply_to"], None) + + def test_sample_distinct(self): + a = NewTweetPayload.sample(0) + b = NewTweetPayload.sample(1) + self.assertNotEqual(a.external_id, b.external_id) + self.assertEqual(a.PAYLOAD_KIND, "new_tweet") diff --git a/packages/openmagpie-schema/schema.json b/packages/openmagpie-schema/schema.json index 86183978..aaa04b25 100644 --- a/packages/openmagpie-schema/schema.json +++ b/packages/openmagpie-schema/schema.json @@ -686,6 +686,44 @@ "title": "ExtractStatus", "type": "string" }, + "FacebookGroupSourceSpec": { + "description": "Identity of one Facebook group search stream. Bound to FacebookGroupConnector.\n\n`group_ids` is the list of Facebook group IDs to search; at least one is\nrequired. `terms` are optional search keywords to filter posts within the\ngroup. `count` caps the per-cycle fetch.", + "properties": { + "count": { + "default": 20, + "maximum": 100, + "minimum": 1, + "title": "Count", + "type": "integer" + }, + "group_ids": { + "items": { + "type": "string" + }, + "minItems": 1, + "title": "Group Ids", + "type": "array" + }, + "kind": { + "const": "facebook_group", + "default": "facebook_group", + "title": "Kind", + "type": "string" + }, + "terms": { + "items": { + "type": "string" + }, + "title": "Terms", + "type": "array" + } + }, + "required": [ + "group_ids" + ], + "title": "FacebookGroupSourceSpec", + "type": "object" + }, "FeedConfigSummary": { "description": "Display-only projection of a feed config for the CLI preview.\n\nBuilt server-side from the typed config (the only place that knows\nthe schema) so the CLI prints it without parsing the `data` blob.\nCurated feeds emit an empty summary because all per-source state\nsurfaces via FeedView.sources / source_count; the class stays as\na hook for future kinds that have non-source-shaped config to\nproject for the CLI.", "properties": {}, @@ -835,6 +873,12 @@ { "$ref": "#/$defs/HackerNewsCommentPayload" }, + { + "$ref": "#/$defs/NewTweetPayload" + }, + { + "$ref": "#/$defs/NewFacebookPostPayload" + }, { "$ref": "#/$defs/FeedItemPayload" } @@ -1654,6 +1698,103 @@ "title": "LogRunWire", "type": "object" }, + "NewFacebookPostPayload": { + "additionalProperties": true, + "description": "`new_fb_post`: one Facebook group post (FacebookGroupConnector, unofficial\nCamofox route). `content` is the post body (the engine's judgeable text);\n`title` is empty (posts have no headline). The facebook-camofox-client\nnormalized record shape is carried as typed fields: `author` (display\nname), `group_id` (the Facebook group ID, also the within-kind source\nslug), `metrics` (likes/comments/shares), `matched_terms` (which search\nterms this post matched).", + "properties": { + "author": { + "default": "", + "title": "Author", + "type": "string" + }, + "content": { + "default": "", + "title": "Content", + "type": "string" + }, + "external_id": { + "default": "", + "title": "External Id", + "type": "string" + }, + "external_url": { + "default": "", + "title": "External Url", + "type": "string" + }, + "group_id": { + "default": "", + "title": "Group Id", + "type": "string" + }, + "kind": { + "const": "new_fb_post", + "title": "Kind", + "type": "string" + }, + "matched_terms": { + "default": [], + "items": { + "type": "string" + }, + "title": "Matched Terms", + "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" + }, + "source": { + "default": "", + "title": "Source", + "type": "string" + }, + "title": { + "default": "", + "title": "Title", + "type": "string" + }, + "url": { + "default": "", + "title": "Url", + "type": "string" + } + }, + "required": [ + "kind" + ], + "title": "NewFacebookPostPayload", + "type": "object" + }, "NewRedditPostPayload": { "additionalProperties": true, "description": "`new_post`: one post off a subreddit's /new (RedditSubredditConnector).", @@ -1733,6 +1874,124 @@ "title": "NewRedditPostPayload", "type": "object" }, + "NewTweetPayload": { + "additionalProperties": true, + "description": "`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}).", + "properties": { + "author": { + "default": "", + "title": "Author", + "type": "string" + }, + "content": { + "default": "", + "title": "Content", + "type": "string" + }, + "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_tweet", + "title": "Kind", + "type": "string" + }, + "lang": { + "default": "", + "title": "Lang", + "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": "NewTweetPayload", + "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": { @@ -2437,10 +2696,12 @@ { "discriminator": { "mapping": { + "facebook_group": "#/$defs/FacebookGroupSourceSpec", "hn_comment": "#/$defs/HackerNewsCommentSourceSpec", "hn_feed": "#/$defs/HackerNewsFeedSourceSpec", "reddit_subreddit": "#/$defs/RedditSubredditSourceSpec", - "rss": "#/$defs/RssSourceSpec" + "rss": "#/$defs/RssSourceSpec", + "twitter_search": "#/$defs/TwitterSearchSourceSpec" }, "propertyName": "kind" }, @@ -2456,6 +2717,12 @@ }, { "$ref": "#/$defs/HackerNewsCommentSourceSpec" + }, + { + "$ref": "#/$defs/TwitterSearchSourceSpec" + }, + { + "$ref": "#/$defs/FacebookGroupSourceSpec" } ] }, @@ -2579,10 +2846,12 @@ { "discriminator": { "mapping": { + "facebook_group": "#/$defs/FacebookGroupSourceSpec", "hn_comment": "#/$defs/HackerNewsCommentSourceSpec", "hn_feed": "#/$defs/HackerNewsFeedSourceSpec", "reddit_subreddit": "#/$defs/RedditSubredditSourceSpec", - "rss": "#/$defs/RssSourceSpec" + "rss": "#/$defs/RssSourceSpec", + "twitter_search": "#/$defs/TwitterSearchSourceSpec" }, "propertyName": "kind" }, @@ -2598,6 +2867,12 @@ }, { "$ref": "#/$defs/HackerNewsCommentSourceSpec" + }, + { + "$ref": "#/$defs/TwitterSearchSourceSpec" + }, + { + "$ref": "#/$defs/FacebookGroupSourceSpec" } ] }, @@ -2647,6 +2922,48 @@ "title": "TelemetryState", "type": "object" }, + "TwitterSearchSourceSpec": { + "description": "Identity of one X (Twitter) search stream. Bound to TwitterSearchConnector.\n\n`query` is the search expression (keywords, quoted phrases, `from:`,\n`lang:`, `filter:` operators, whatever X's search syntax accepts); it is\nREQUIRED and NON-BLANK so a source always carries a server-side pre-filter\nbefore any per-item LLM cost (same discipline as hn_comment: a blank query\nwould be the unfiltered firehose). `mode` picks the result ordering twikit\nasks X for: `latest` (newest first, the listener's default) or `top`\n(ranked). `count` caps the per-cycle fetch. `lang` optionally narrows to\ntweets in one language (ISO 639-1, e.g. \"en\"); empty = no filter.", + "properties": { + "count": { + "default": 20, + "maximum": 100, + "minimum": 1, + "title": "Count", + "type": "integer" + }, + "kind": { + "const": "twitter_search", + "default": "twitter_search", + "title": "Kind", + "type": "string" + }, + "lang": { + "default": "", + "title": "Lang", + "type": "string" + }, + "mode": { + "default": "latest", + "enum": [ + "latest", + "top" + ], + "title": "Mode", + "type": "string" + }, + "query": { + "minLength": 1, + "title": "Query", + "type": "string" + } + }, + "required": [ + "query" + ], + "title": "TwitterSearchSourceSpec", + "type": "object" + }, "WatchActionBackfillState": { "description": "Lifecycle of one WatchActionBackfill job (a queued request to re-run an\naction over the previous step's passes).\n\nThe `process_due_backfills` cron claims a PENDING job (CAS -> RUNNING), does\nthe select/delete/enqueue, then marks it terminal:\n - DONE : the setup finished, and the enqueued runs are now the drain's job.\n - FAILED : same dual meaning as a run's FAILED, transient-until-exhausted, and\n (like a run) readable off `completed_at`: FAILED with `completed_at`\n UNSET is retryable (the reaper cleared it so claim_due re-picks it);\n FAILED with `completed_at` SET is terminal (attempts hit the cap). The\n reaper resets a stale RUNNING to FAILED (retryable, `completed_at`\n cleared), and a permanent setup defect (source action gone) fails with\n attempts bumped to the cap AND `completed_at` stamped, so it isn't\n re-claimed. Terminality is the attempts cap / `completed_at`, not the\n state alone.\nA RUNNING job whose worker died is reaped to FAILED (retryable), so a crash\nmid-setup is retried, safe because the setup is idempotent and guarded by the\njob's `replace_deleted_at` delete-once marker.", "enum": [ @@ -3960,6 +4277,9 @@ { "$ref": "#/$defs/WebhookResult" }, + { + "$ref": "#/$defs/FacebookGroupSourceSpec" + }, { "$ref": "#/$defs/RedditSubredditSourceSpec" }, @@ -3972,6 +4292,9 @@ { "$ref": "#/$defs/HackerNewsCommentSourceSpec" }, + { + "$ref": "#/$defs/TwitterSearchSourceSpec" + }, { "$ref": "#/$defs/EngineStatus" }, diff --git a/packages/openmagpie-schema/src/openmagpie_schema/configs.py b/packages/openmagpie-schema/src/openmagpie_schema/configs.py index 34258ef5..8a7f9259 100644 --- a/packages/openmagpie-schema/src/openmagpie_schema/configs.py +++ b/packages/openmagpie-schema/src/openmagpie_schema/configs.py @@ -149,12 +149,89 @@ def display(self) -> str: return f'HN comments: "{self.query}"' +class TwitterSearchSourceSpec(BaseModel): + """Identity of one X (Twitter) search stream. Bound to TwitterSearchConnector. + + `query` is the search expression (keywords, quoted phrases, `from:`, + `lang:`, `filter:` operators, whatever X's search syntax accepts); it is + REQUIRED and NON-BLANK so a source always carries a server-side pre-filter + before any per-item LLM cost (same discipline as hn_comment: a blank query + would be the unfiltered firehose). `mode` picks the result ordering twikit + asks X for: `latest` (newest first, the listener's default) or `top` + (ranked). `count` caps the per-cycle fetch. `lang` optionally narrows to + tweets in one language (ISO 639-1, e.g. "en"); empty = no filter. + """ + + SOURCE_KIND: ClassVar[str] = "twitter_search" + URL_FIELDS: ClassVar[tuple[str, ...]] = () # no operator-supplied URL to SSRF-check + + kind: Literal["twitter_search"] = "twitter_search" + query: str = Field(min_length=1) + mode: Literal["latest", "top"] = "latest" + count: int = Field(default=20, ge=1, le=100) + lang: str = "" + + @field_validator("query") + @classmethod + def _query_not_blank(cls, v: str) -> str: + v = v.strip() + if not v: + raise ValueError("twitter_search requires a non-blank query (the firehose guard)") + return v + + @field_validator("lang") + @classmethod + def _lang_normalize(cls, v: str) -> str: + return v.strip().lower() + + def display(self) -> str: + return f'X search: "{self.query}"' + + +class FacebookGroupSourceSpec(BaseModel): + """Identity of one Facebook group search stream. Bound to FacebookGroupConnector. + + `group_ids` is the list of Facebook group IDs to search; at least one is + required. `terms` are optional search keywords to filter posts within the + group. `count` caps the per-cycle fetch. + """ + + SOURCE_KIND: ClassVar[str] = "facebook_group" + URL_FIELDS: ClassVar[tuple[str, ...]] = () # no operator-supplied URL to SSRF-check + + kind: Literal["facebook_group"] = "facebook_group" + group_ids: list[str] = Field(min_length=1) + terms: list[str] = Field(default_factory=list) + count: int = Field(default=20, ge=1, le=100) + + @field_validator("group_ids") + @classmethod + def _group_ids_not_empty(cls, v: list[str]) -> list[str]: + stripped = [s.strip() for s in v if s.strip()] + if not stripped: + raise ValueError("facebook_group requires at least one non-empty group_id") + return stripped + + @field_validator("terms") + @classmethod + def _terms_clean(cls, v: list[str]) -> list[str]: + return [s.strip() for s in v if s.strip()] + + def display(self) -> str: + return f'Facebook group search: {", ".join(self.group_ids[:3])}' + + # 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, # so it surfaces as a validation error rather than being absorbed as a raw blob. _BuiltinSourceSpec = Annotated[ - RedditSubredditSourceSpec | RssSourceSpec | HackerNewsFeedSourceSpec | HackerNewsCommentSourceSpec, + RedditSubredditSourceSpec + | RssSourceSpec + | HackerNewsFeedSourceSpec + | HackerNewsCommentSourceSpec + | TwitterSearchSourceSpec + | FacebookGroupSourceSpec, 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 0ee8d27e..7a972bf5 100644 --- a/packages/openmagpie-schema/src/openmagpie_schema/feed_payloads.py +++ b/packages/openmagpie-schema/src/openmagpie_schema/feed_payloads.py @@ -79,6 +79,40 @@ class HackerNewsCommentPayload(FeedItemPayload): story_title: str = "" +class NewTweetPayload(FeedItemPayload): + """`new_tweet`: one X (Twitter) tweet (TwitterSearchConnector, unofficial + twikit route). `content` is the tweet's full text (the engine's judgeable + body); `title` is empty (tweets have no headline). The listeningkit + SocialEvent shape is carried as typed fields: `handle` (the @screen_name, + also the within-kind source slug), `author` (display name), `lang`, + `metrics` (likes/retweets/replies/quotes/views), `refs` + (in_reply_to / quoted / retweet_of), `media` (list of {type,url,thumbnail}).""" + + kind: Literal["new_tweet"] # required, so a non-twitter dump can't match here + author: str = "" + handle: str = "" + lang: str = "" + metrics: dict[str, int | None] = {} + refs: dict[str, str | None] = {} + media: list[dict[str, object]] = [] + + +class NewFacebookPostPayload(FeedItemPayload): + """`new_fb_post`: one Facebook group post (FacebookGroupConnector, unofficial + Camofox route). `content` is the post body (the engine's judgeable text); + `title` is empty (posts have no headline). The facebook-camofox-client + normalized record shape is carried as typed fields: `author` (display + name), `group_id` (the Facebook group ID, also the within-kind source + slug), `metrics` (likes/comments/shares), `matched_terms` (which search + terms this post matched).""" + + kind: Literal["new_fb_post"] # required, so a non-facebook dump can't match here + author: str = "" + group_id: str = "" + metrics: dict[str, int | None] = {} + matched_terms: list[str] = [] + + # 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 @@ -89,6 +123,12 @@ class HackerNewsCommentPayload(FeedItemPayload): # but a consumer keying on `isinstance(data, RssEntryPayload)` won't see the # malformed row (canonical fields like `title` still read off the base). FeedItemData = Annotated[ - RssEntryPayload | NewRedditPostPayload | HackerNewsFeedPayload | HackerNewsCommentPayload | FeedItemPayload, + RssEntryPayload + | NewRedditPostPayload + | HackerNewsFeedPayload + | HackerNewsCommentPayload + | NewTweetPayload + | NewFacebookPostPayload + | FeedItemPayload, Field(union_mode="left_to_right"), ] diff --git a/tools/schema_sync/models.py b/tools/schema_sync/models.py index 5be4e1fc..37837b98 100644 --- a/tools/schema_sync/models.py +++ b/tools/schema_sync/models.py @@ -15,10 +15,12 @@ from openmagpie_schema.auth import AuthUser from openmagpie_schema.backfill import BackfillJob, BackfillListResponse, BackfillPreview from openmagpie_schema.configs import ( + FacebookGroupSourceSpec, HackerNewsCommentSourceSpec, HackerNewsFeedSourceSpec, RedditSubredditSourceSpec, RssSourceSpec, + TwitterSearchSourceSpec, ) from openmagpie_schema.engine import EngineListResponse, EngineStatus from openmagpie_schema.feed import ( @@ -103,10 +105,12 @@ WebhookConfig, WebhookResult, # Source specs (a discriminated union; also reached via SourceFields.spec) + FacebookGroupSourceSpec, RedditSubredditSourceSpec, RssSourceSpec, HackerNewsFeedSourceSpec, HackerNewsCommentSourceSpec, + TwitterSearchSourceSpec, # Engine + telemetry status EngineStatus, EngineListResponse, @@ -175,6 +179,7 @@ WebhookConfig, RedditSubredditSourceSpec, RssSourceSpec, + FacebookGroupSourceSpec, HackerNewsFeedSourceSpec, HackerNewsCommentSourceSpec, ] diff --git a/uv.lock b/uv.lock index 1c5abdf6..4ba751ac 100644 --- a/uv.lock +++ b/uv.lock @@ -67,6 +67,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/df/73/b6e24bd22e6720ca8ee9a85a0c4a2971af8497d8f3193fa05390cbd46e09/backoff-2.2.1-py3-none-any.whl", hash = "sha256:63579f9a0628e06278f7e47b7d7d5b6ce20dc65c5e96a6f3ca99a6adca0396e8", size = 15148, upload-time = "2022-10-05T19:19:30.546Z" }, ] +[[package]] +name = "beautifulsoup4" +version = "4.15.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "soupsieve" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/43/65/318323f98dbee45d42dff61d8f047181bc6f2268a9068cfad035a46be5af/beautifulsoup4-4.15.0.tar.gz", hash = "sha256:288e3ca7d54b06f2ac191970bc275c1939cb46d450b255bf6718b04aa37ab4f7", size = 632571, upload-time = "2026-06-07T16:44:20.453Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/88/c6/92fcd42f1ba33e1184263f25bfabf3d27c383410470f169e4b8163bf9c17/beautifulsoup4-4.15.0-py3-none-any.whl", hash = "sha256:d6f88de62e1d4e38ecb1077eb9724cd0eff29d2a08ca16a401e9b9e93f117cf9", size = 109924, upload-time = "2026-06-07T16:44:21.566Z" }, +] + [[package]] name = "certifi" version = "2026.5.20" @@ -372,6 +385,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/4e/eb/c96d64137e29ae17d83ad2552470bafe3a7a915e85434d9942077d7fd011/feedparser-6.0.12-py3-none-any.whl", hash = "sha256:6bbff10f5a52662c00a2e3f86a38928c37c48f77b3c511aedcd51de933549324", size = 81480, upload-time = "2025-09-10T13:33:58.022Z" }, ] +[[package]] +name = "filetype" +version = "1.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/bb/29/745f7d30d47fe0f251d3ad3dc2978a23141917661998763bebb6da007eb1/filetype-1.2.0.tar.gz", hash = "sha256:66b56cd6474bf41d8c54660347d37afcc3f7d1970648de365c102ef77548aadb", size = 998020, upload-time = "2022-11-02T17:34:04.141Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/18/79/1b8fa1bb3568781e84c9200f951c735f3f157429f44be0495da55894d620/filetype-1.2.0-py2.py3-none-any.whl", hash = "sha256:7ce71b6880181241cf7ac8697a2f1eb6a8bd9b429f7ad6d27b8db9ba5f1c2d25", size = 19970, upload-time = "2022-11-02T17:34:01.425Z" }, +] + [[package]] name = "gunicorn" version = "26.0.0" @@ -437,6 +459,11 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" }, ] +[package.optional-dependencies] +socks = [ + { name = "socksio" }, +] + [[package]] name = "idna" version = "3.18" @@ -500,6 +527,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0e/a4/cf8d779feb133a27a2e3bc833bccb9e13aa332cdf820497ebf72c10ce8c3/jiter-0.15.0-cp314-cp314t-win_arm64.whl", hash = "sha256:66b1880df2d01e206e8339769d1c7c1753bcb653efd6289e203f6f24ebada0c0", size = 191244, upload-time = "2026-05-19T10:09:12.74Z" }, ] +[[package]] +name = "js2py-3-13" +version = "0.74.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pyjsparser" }, + { name = "six" }, + { name = "tzlocal" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e2/9e/17ed2ceebff1539a454b66d3f056001ab37f679e28a336e1ba88407940fe/js2py_3_13-0.74.1.tar.gz", hash = "sha256:91e214f717312f9d510eaf36fcc5325b0b15a22a49831fe2b434bca4a33c1f77", size = 570397, upload-time = "2025-02-07T13:01:33.395Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/47/5f/4bdab35d30055613c58f681f03d3a76e06c874485aced646fac763a1d552/Js2Py_3.13-0.74.1-py3-none-any.whl", hash = "sha256:5c60a80a43197775986c27f33becaf9ebf3731e8e79030c925f44025ed6f0e8b", size = 611795, upload-time = "2025-02-07T13:01:31.087Z" }, +] + [[package]] name = "justext" version = "3.0.2" @@ -604,6 +645,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/6a/bd/6e2b76a6c5dee10397db9c929f0c5066766ec1036046f0335b7ca7ca08b8/lxml_html_clean-0.4.5-py3-none-any.whl", hash = "sha256:c76fcadd1e5bfb9b8bafc2200d51e4e78eb0dad67f56881c21dfb6484c7e7746", size = 14573, upload-time = "2026-05-20T12:17:52.215Z" }, ] +[[package]] +name = "m3u8" +version = "6.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/9b/a5/73697aaa99bb32b610adc1f11d46a0c0c370351292e9b271755084a145e6/m3u8-6.0.0.tar.gz", hash = "sha256:7ade990a1667d7a653bcaf9413b16c3eb5cd618982ff46aaff57fe6d9fa9c0fd", size = 42720, upload-time = "2024-08-07T11:20:06.606Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f8/31/50f3c38b38ff28635ff9c4a4afefddccc5f1b57457b539bdbdf75ce18669/m3u8-6.0.0-py3-none-any.whl", hash = "sha256:566d0748739c552dad10f8c87150078de6a0ec25071fa48e6968e96fc6dcba5d", size = 24133, upload-time = "2024-08-07T11:20:03.96Z" }, +] + [[package]] name = "markdown-it-py" version = "4.2.0" @@ -706,6 +756,7 @@ dependencies = [ { name = "python-dotenv" }, { name = "pyyaml" }, { name = "trafilatura" }, + { name = "twikit" }, { name = "ulid" }, ] @@ -739,6 +790,7 @@ requires-dist = [ { name = "python-dotenv", specifier = ">=1.1" }, { name = "pyyaml", specifier = ">=6.0" }, { name = "trafilatura", specifier = ">=1.7" }, + { name = "twikit", git = "https://github.com/unclecode/twikit.git" }, { name = "ulid", specifier = ">=1.1" }, ] @@ -939,6 +991,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" }, ] +[[package]] +name = "pyjsparser" +version = "2.7.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/48/ef/c72abcfa2c6accd03e7c89c400790fc3d908c5804d50a7c4e9ceabd74d23/pyjsparser-2.7.1.tar.gz", hash = "sha256:be60da6b778cc5a5296a69d8e7d614f1f870faf94e1b1b6ac591f2ad5d729579", size = 24196, upload-time = "2019-04-21T21:56:17.708Z" } + +[[package]] +name = "pyotp" +version = "2.10.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/2d/c6/c5d96a86fd0bf6fa1bbb5c5c341ff3208638b692727a683c8289068d9a11/pyotp-2.10.0.tar.gz", hash = "sha256:d01e9703443616b03c57c700b5cbffd56a1f929c1b0f8f03131bc78c1fca9d3f", size = 18625, upload-time = "2026-06-14T03:48:49.221Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/65/33/7b83bde70eddaaaaef487751a9c3a5cc0c0be54620ded0e120ebdc401ff9/pyotp-2.10.0-py3-none-any.whl", hash = "sha256:1df2f6a1bcc3bb0716172a5215ddc2f8c7c7fd26a13df9927d52e1746934836c", size = 13768, upload-time = "2026-06-14T03:48:47.831Z" }, +] + [[package]] name = "python-dateutil" version = "2.9.0.post0" @@ -1163,6 +1230,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/e9/44/75a9c9421471a6c4805dbf2356f7c181a29c1879239abab1ea2cc8f38b40/sniffio-1.3.1-py3-none-any.whl", hash = "sha256:2f6da418d1f1e0fddd844478f41680e794e6051915791a034ff65e5f100525a2", size = 10235, upload-time = "2024-02-25T23:20:01.196Z" }, ] +[[package]] +name = "socksio" +version = "1.0.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/f8/5c/48a7d9495be3d1c651198fd99dbb6ce190e2274d0f28b9051307bdec6b85/socksio-1.0.0.tar.gz", hash = "sha256:f88beb3da5b5c38b9890469de67d0cb0f9d494b78b106ca1845f96c10b91c4ac", size = 19055, upload-time = "2020-04-17T15:50:34.664Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/37/c3/6eeb6034408dac0fa653d126c9204ade96b819c936e136c5e8a6897eee9c/socksio-1.0.0-py3-none-any.whl", hash = "sha256:95dc1f15f9b34e8d7b16f06d74b8ccf48f609af32ab33c608d08761c5dcbb1f3", size = 12763, upload-time = "2020-04-17T15:50:31.878Z" }, +] + +[[package]] +name = "soupsieve" +version = "2.9.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/69/99/a6ca3beb3ccacb41fb3321d8a60e5566f9e6467601ef8eba6a17e1b89778/soupsieve-2.9.2.tar.gz", hash = "sha256:4a55d8cf158a9c2e587fa4922f1bbb91d68ac829e2d6f25403a85747c71daf74", size = 122445, upload-time = "2026-08-07T00:57:24.801Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/eb/dc/ad025c1ee131eba60c69f4dd5779b18fcf1e6b21a343e2162a84d5d133c7/soupsieve-2.9.2-py3-none-any.whl", hash = "sha256:8089a26fd974ca7a1f30276d3d8492ab266ab15af581642dfe8aa162e0c1c823", size = 37370, upload-time = "2026-08-07T00:57:23.524Z" }, +] + [[package]] name = "sqlparse" version = "0.5.5" @@ -1211,6 +1296,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0e/78/4ad99d79aee2784f49f20fd0a29058ce4c032fe4439047924c43521cd211/trafilatura-2.1.0-py3-none-any.whl", hash = "sha256:0eded5207a806445ddebbe36eae30b9035fe6a2f233c36f6fe82663fca8b9d30", size = 134600, upload-time = "2026-06-07T17:43:28.404Z" }, ] +[[package]] +name = "twikit" +version = "2.3.3" +source = { git = "https://github.com/unclecode/twikit.git#6a73ab97f4de09f79139f6308c9fb80029a9f5f7" } +dependencies = [ + { name = "beautifulsoup4" }, + { name = "filetype" }, + { name = "httpx", extra = ["socks"] }, + { name = "js2py-3-13" }, + { name = "lxml" }, + { name = "m3u8" }, + { name = "pyotp" }, + { name = "webvtt-py" }, +] + [[package]] name = "ty" version = "0.0.43" @@ -1388,3 +1488,12 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/25/91/80908e835e100527a9267147b08c0eee1fa6ab0ffec15edc04d1d44885f7/watchfiles-1.2.0-cp315-cp315-musllinux_1_1_aarch64.whl", hash = "sha256:b718bf356bbc15e559bd8ef41782b573b8ae0e3f177ab244b440568d7ea02cfb", size = 630638, upload-time = "2026-05-18T04:30:49.89Z" }, { url = "https://files.pythonhosted.org/packages/46/4b/95ab2f256bb4af3cb2eb23b9317bda984ee6e0f11733a5c004a6c95b06e3/watchfiles-1.2.0-cp315-cp315-musllinux_1_1_x86_64.whl", hash = "sha256:922c0e019fe68b3ae392965a766b02a71ba1168c932cebc3733cd52c5fe5b377", size = 657684, upload-time = "2026-05-18T04:31:32.027Z" }, ] + +[[package]] +name = "webvtt-py" +version = "0.5.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/5e/f6/7c9c964681fb148e0293e6860108d378e09ccab2218f9063fd3eb87f840a/webvtt-py-0.5.1.tar.gz", hash = "sha256:2040dd325277ddadc1e0c6cc66cbc4a1d9b6b49b24c57a0c3364374c3e8a3dc1", size = 55128, upload-time = "2024-05-30T13:40:17.189Z" } +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" }, +]