diff --git a/README.md b/README.md index 749b3d6b..728cc719 100644 --- a/README.md +++ b/README.md @@ -240,25 +240,7 @@ Social listening is a crowded market (Brand24, Mention, Octolens, Syften, and to | Webhook methods | `POST`, `PUT`, `PATCH` | | Delivery audit | per-attempt `WatchActionDelivery` | -## What we've done - -X/Twitter listening is the first connector added beyond the original Reddit / HN / RSS set. What shipped in this branch: - -- **`twitter_search` source kind** — a twikit-based connector that runs X search queries (mode Top/Latest, count) and maps results to a schema-parity `NewTweetPayload`, registered alongside the existing kinds with the same feed/watch/webhook pipeline. -- **Live-cookie auth** — the connector authenticates with an existing X session cookie JSON (`TWITTER_COOKIES_JSON` → `auth_token`/`ct0`, a cookies file, or a login fallback), so no API key, proxy, or vendor API is required — the same live cookies the listening kit already used keep working. -- **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. +Two connectors ride an unofficial route and need a browser session cookie (X/Twitter always, YouTube only for age-restricted videos); see [apps/core/credentials/README.md](apps/core/credentials/README.md) for setup and the terms-of-service caveat. Per-release history is in the [changelog](CHANGELOG.md). ## Roadmap diff --git a/apps/core/conf/settings/base.py b/apps/core/conf/settings/base.py index 5e833906..477437fc 100644 --- a/apps/core/conf/settings/base.py +++ b/apps/core/conf/settings/base.py @@ -525,6 +525,26 @@ # Google account. Empty (the default) disables it. YOUTUBE_COOKIES_FILE = os.environ.get("YOUTUBE_COOKIES_FILE", "") +# X (Twitter) connector session config. The connector authenticates with an +# existing x.com session, resolved per poll in priority order: +# TWITTER_COOKIES_JSON a full JSON dict of x.com cookies (inline env) +# TWITTER_COOKIE_AUTH_TOKEN + TWITTER_COOKIE_CT0 the critical pair +# TWITTER_COOKIES_FILE path to one cookie export (JSON dict or a +# Get-cookies.txt-LOCALLY array) +# TWITTER_CREDENTIALS_DIR dir of *.json cookie exports, each with an +# optional .proxy pin +# Empty values fall through to the next route; all empty = guest mode (the +# first search fails with a mapped `unauthorized`). See +# apps/core/credentials/README.md for the on-disk convention. +TWITTER_COOKIES_JSON = os.environ.get("TWITTER_COOKIES_JSON", "") +TWITTER_COOKIE_AUTH_TOKEN = os.environ.get("TWITTER_COOKIE_AUTH_TOKEN", "") +TWITTER_COOKIE_CT0 = os.environ.get("TWITTER_COOKIE_CT0", "") +TWITTER_COOKIES_FILE = os.environ.get("TWITTER_COOKIES_FILE", "") +TWITTER_CREDENTIALS_DIR = os.environ.get("TWITTER_CREDENTIALS_DIR", str(BASE_DIR / "credentials" / "twitter")) +# Egress proxy for X requests (twikit passes it to httpx); a per-credential +# .proxy pin in TWITTER_CREDENTIALS_DIR overrides it. +TWITTER_PROXY = os.environ.get("TWITTER_PROXY", "") + # 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 index 847b9bb1..86204eeb 100644 --- a/apps/core/credentials/README.md +++ b/apps/core/credentials/README.md @@ -20,6 +20,58 @@ TWITTER_CREDENTIALS_DIR=/app/apps/core/credentials/twitter YOUTUBE_COOKIES_FILE=/app/apps/core/credentials/youtube/cookies.txt ``` +Settings are read per poll, so a refreshed export applies on the next cycle +without a restart. `.env` changes themselves still need the usual +`docker compose up -d --force-recreate`. + +## X / Twitter: generating a cookie export + +The `twitter_search` connector authenticates with an existing x.com browser +session; there is no API key. + +1. Sign in to x.com in a browser. +2. The minimal route needs just two cookies. In DevTools (Application -> + Cookies -> https://x.com), copy the values of `auth_token` and `ct0` + and set them directly: + + ``` + TWITTER_COOKIE_AUTH_TOKEN= + TWITTER_COOKIE_CT0= + ``` + +3. For the file route instead, export the site's cookies with a browser + extension such as Cookie-Editor (export as JSON) or "Get cookies.txt + LOCALLY" (JSON export). Both shapes are accepted: a plain + `{name: value}` dict, or the extension's array of cookie objects. Save + it as `credentials/twitter/.json`; the connector picks the first + usable export (sorted by filename) that carries the `auth_token`/`ct0` + pair. +4. Optional: pin an egress proxy for one export by writing its URL to + `credentials/twitter/.proxy` (same basename). `TWITTER_PROXY` + sets a global one. + +The full priority order (first configured route wins): +`TWITTER_COOKIES_JSON` (inline JSON dict) -> the `auth_token`/`ct0` pair -> +`TWITTER_COOKIES_FILE` (one export) -> `TWITTER_CREDENTIALS_DIR`. + +Sessions expire when X rotates them (or you log out in that browser); +re-export and the next poll picks it up. + +## YouTube: generating cookies.txt (optional) + +Public YouTube search needs no credentials at all — set this up only if +poll logs show relevant videos skipped as age-gated ("Sign in to confirm +your age"). + +1. Sign in to youtube.com, ideally in a private/incognito window (yt-dlp's + recommendation: export from a private session you then close, so the + browser doesn't rotate the exported cookies out from under you). +2. Export the cookies in **Netscape format** with a "Get cookies.txt + LOCALLY"-style extension while on youtube.com (yt-dlp requires the + cookies.txt format here, not JSON). +3. Save it as `credentials/youtube/cookies.txt` and set + `YOUTUBE_COOKIES_FILE` as above. + 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 diff --git a/apps/core/pyproject.toml b/apps/core/pyproject.toml index 472a8c51..ec37c709 100644 --- a/apps/core/pyproject.toml +++ b/apps/core/pyproject.toml @@ -24,7 +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) + "twikit @ git+https://github.com/unclecode/twikit.git@6a73ab97f4de09f79139f6308c9fb80029a9f5f7", # X (Twitter) unofficial route (listeningkit-verified 2026 fork of d60/twikit), pinned: the fork's default branch floats and this code handles auth "yt-dlp>=2026.07.04", # YouTube search connector (public API only) "ulid>=1.1", ] diff --git a/apps/core/sources/connectors/base.py b/apps/core/sources/connectors/base.py index d48c7c90..cf6d9d46 100644 --- a/apps/core/sources/connectors/base.py +++ b/apps/core/sources/connectors/base.py @@ -103,6 +103,41 @@ def parse_rate_limit_wait(response: httpx.Response) -> float | None: return _as_positive_float(response.headers.get("X-RateLimit-Reset")) +def rate_limit_delay(relative_wait: float | None, attempt: int, *, base: float, cap: float) -> float: + """Seconds to wait before retrying a rate-limited request: the caller's + header-derived relative wait when usable (finite, positive), else + exponential backoff `base * 2**attempt`. Capped at `cap` so a hostile or + far-future value can't stall a poll worker. Each connector converts its + own source shape to a relative wait first (Reddit: `parse_rate_limit_wait`'s + relative seconds; Twitter: an absolute `x-rate-limit-reset` epoch minus + now), keeping `base` / `cap` / retry-count as its own tunables.""" + delay = relative_wait if relative_wait is not None and relative_wait > 0 else base * (2**attempt) + return min(delay, cap) + + +# How often a backoff sleep ticks the caller's poll-lease heartbeat: long +# enough not to thrash, short enough that a minute-scale wait renews the lease +# several times over. +HEARTBEAT_SLEEP_CHUNK_SECONDS = 15.0 + + +def sleep_with_heartbeat(total: float, heartbeat: Callable[[], bool] | None) -> None: + """Sleep `total` seconds, ticking `heartbeat` every chunk so the caller's + poll lease renews through the wait. The return value is deliberately + ignored (see the Connector.poll contract). No heartbeat (direct calls / + tests) = one plain sleep. Shared by every connector that backs off inside + poll() (Reddit's 429 retry, the twikit rate-limit wait).""" + if heartbeat is None: + time.sleep(total) + return + remaining = total + while remaining > 0: + chunk = min(remaining, HEARTBEAT_SLEEP_CHUNK_SECONDS) + time.sleep(chunk) + remaining -= chunk + heartbeat() + + # ── SSRF-safe fetch of the open web ─────────────────────────────────────── # Two callers, one block POLICY (`common.ssrf` / `common.safe_http`): # - RSS feeds (OPERATOR-chosen URLs): `validate_request_url`, an httpx request diff --git a/apps/core/sources/connectors/reddit/connector.py b/apps/core/sources/connectors/reddit/connector.py index 0425a1d6..5bd93601 100644 --- a/apps/core/sources/connectors/reddit/connector.py +++ b/apps/core/sources/connectors/reddit/connector.py @@ -11,7 +11,14 @@ from sources.payload_registry import register from sources.payloads import SourcePayload -from ..base import BaseConnector, ConnectorParseError, parse_rate_limit_wait, read_response_capped +from ..base import ( + BaseConnector, + ConnectorParseError, + parse_rate_limit_wait, + rate_limit_delay, + read_response_capped, + sleep_with_heartbeat, +) from .payloads import NewRedditPostPayload logger = logging.getLogger("sources") @@ -70,40 +77,6 @@ ) RATE_LIMIT_DELAY_CAP_SECONDS = 60.0 -# Backoff sleeps tick the caller's `heartbeat` at this cadence so the poll -# lease renews DURING the wait, not just between sources (the lease detects -# dead holders; a deliberate wait is alive). Far inside the lease window -# (POLL_LOCK_TIMEOUT_SECONDS, 600s), so even a worst-case stack of full -# 60s waits never lets the lease lapse mid-source. -HEARTBEAT_SLEEP_CHUNK_SECONDS = 15.0 - - -def _rate_limit_delay(header_wait: float | None, attempt: int) -> float: - """Seconds to wait before retrying a 429'd page: the wait the response's - rate-limit header asked for (`parse_rate_limit_wait`), else exponential in - the attempt number when no header was usable. Capped so a hostile / buggy - header can't stall a poll worker for minutes. (`parse_rate_limit_wait` - already screens NaN / inf / non-positive, so `header_wait` is a clean - positive float or None.)""" - delay = header_wait if header_wait is not None else RATE_LIMIT_BACKOFF_BASE_SECONDS * (2**attempt) - return min(delay, RATE_LIMIT_DELAY_CAP_SECONDS) - - -def _sleep_with_heartbeat(total: float, heartbeat: Callable[[], bool] | None) -> None: - """Sleep `total` seconds, ticking `heartbeat` every chunk so the - caller's poll lease renews through the wait. The return value is - deliberately ignored (see the Connector.poll contract). No heartbeat - (direct calls / tests) = one plain sleep.""" - if heartbeat is None: - time.sleep(total) - return - remaining = total - while remaining > 0: - chunk = min(remaining, HEARTBEAT_SLEEP_CHUNK_SECONDS) - time.sleep(chunk) - remaining -= chunk - heartbeat() - def _entry_published(entry: Any) -> datetime | None: """feedparser exposes Atom `` as `published_parsed` @@ -188,7 +161,12 @@ def _get_page( if attempt > 0: logger.info("%s succeeded after %d retr%s", url, attempt, "y" if attempt == 1 else "ies") return body - delay = _rate_limit_delay(parse_rate_limit_wait(response), attempt) + delay = rate_limit_delay( + parse_rate_limit_wait(response), + attempt, + base=RATE_LIMIT_BACKOFF_BASE_SECONDS, + cap=RATE_LIMIT_DELAY_CAP_SECONDS, + ) # Sleep AFTER the `with` closes the 429 response, so the wait # never pins the streamed connection open. logger.info( @@ -198,7 +176,7 @@ def _get_page( attempt + 1, MAX_RATE_LIMIT_RETRIES, ) - _sleep_with_heartbeat(delay, heartbeat) + sleep_with_heartbeat(delay, heartbeat) attempt += 1 def poll( diff --git a/apps/core/sources/connectors/twitter/__init__.py b/apps/core/sources/connectors/twitter/__init__.py index 7643886c..c1dcef3e 100644 --- a/apps/core/sources/connectors/twitter/__init__.py +++ b/apps/core/sources/connectors/twitter/__init__.py @@ -5,7 +5,7 @@ - `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 + - `errors.py` ; twikit error taxonomy -> canonical TwitterError Future variants (user timeline, list timeline) reuse `TwikitClient` with their own spec + payload. diff --git a/apps/core/sources/connectors/twitter/client.py b/apps/core/sources/connectors/twitter/client.py index 48ea7e81..eb0fbc2c 100644 --- a/apps/core/sources/connectors/twitter/client.py +++ b/apps/core/sources/connectors/twitter/client.py @@ -1,20 +1,19 @@ """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). +Session material comes from the `TWITTER_*` settings (see +conf/settings/base.py for the priority chain: inline cookie JSON, the +auth_token/ct0 pair, one cookie-export file, or a credentials directory of +exports with optional `.proxy` pins). Cookie formats are the exact ones the +live listeningkit setup used (a plain JSON dict, or a +Get-cookies.txt-LOCALLY array export). + +Resolution happens per search call, not at import: settings are cheap env +reads plus at most a small file, `@override_settings` works in tests, and a +rotated cookie export applies on the next poll without a process restart. + +Proxy: `TWITTER_PROXY` (or a per-credential `.proxy` pin in the 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 @@ -27,14 +26,17 @@ import asyncio import json import logging -import os +import time +from collections.abc import Callable from pathlib import Path from typing import Literal +from django.conf import settings from twikit import Client from twikit.errors import TwitterException -from .errors import ListenerError, map_bootstrap_failure, map_twikit_error +from ..base import rate_limit_delay, sleep_with_heartbeat +from .errors import RATE_LIMITED, TwitterError, 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). @@ -45,6 +47,28 @@ # The two cookies that make an authenticated twikit session work. CRITICAL_COOKIES = ("auth_token", "ct0") +# Rate-limit retry, mirroring the Reddit connector's 429 loop: on a +# `rate_limited` error the client backs off and retries in-cycle rather than +# failing the source immediately. This is brief SMOOTHING, not a full wait to +# reset: X's search window is ~15 min, but the delay is capped well under the +# poll lease (POLL_LOCK_TIMEOUT_SECONDS, 600s), so a genuine rate-limit +# exhausts these retries and defers to the next scheduled poll (whose watermark +# stays put) rather than pinning a worker for a quarter hour. +MAX_RATE_LIMIT_RETRIES = 3 +RATE_LIMIT_BACKOFF_BASE_SECONDS = 5.0 # 5s, 10s, 20s when no usable reset is present +RATE_LIMIT_DELAY_CAP_SECONDS = 60.0 + + +def _reset_delay(reset_epoch: int | None, attempt: int) -> float: + """Delay before retrying a rate-limited search. X sends `x-rate-limit-reset` + as an ABSOLUTE Unix epoch (unlike Reddit's relative seconds), so convert to + a relative wait (`reset - now`) and defer to the shared `rate_limit_delay` + for the exponential fallback + cap. Capped at RATE_LIMIT_DELAY_CAP_SECONDS, + so for X's ~15-min window this returns the cap, not the full reset (see the + retry-loop note above).""" + remaining = reset_epoch - time.time() if reset_epoch is not None else None + return rate_limit_delay(remaining, attempt, base=RATE_LIMIT_BACKOFF_BASE_SECONDS, cap=RATE_LIMIT_DELAY_CAP_SECONDS) + def _load_cookie_file(path: Path) -> dict[str, str]: """Parse a Get-cookies.txt-LOCALLY JSON export (array) or a plain dict.""" @@ -61,13 +85,9 @@ def _load_cookie_file(path: Path) -> dict[str, str]: 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. +def load_cookies() -> tuple[dict[str, str], str | None]: + """Resolve the cookie dict + proxy for one X session from the + `TWITTER_*` settings, in priority order (see conf/settings/base.py). Returns ``(cookies, proxy)``. ``cookies`` is empty when nothing is configured (the caller stays guest-mode; the first search then fails @@ -75,29 +95,31 @@ def load_cookies( comes from `TWITTER_PROXY` or a `.proxy` pin next to the chosen cookie export. """ - proxy = os.environ.get("TWITTER_PROXY", "").strip() or None + proxy = settings.TWITTER_PROXY.strip() or None - if cookies_json: + if settings.TWITTER_COOKIES_JSON: try: - data = json.loads(cookies_json) + data = json.loads(settings.TWITTER_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") + cookies = {str(k): str(v) for k, v in data.items() if v} if isinstance(data, dict) else {} + if set(CRITICAL_COOKIES).issubset(cookies): + return cookies, proxy + log.warning("TWITTER_COOKIES_JSON missing auth_token/ct0 (critical pair); ignoring") - individual = {name: os.environ.get(f"TWITTER_COOKIE_{name.upper()}", "").strip() for name in CRITICAL_COOKIES} + individual = {name: getattr(settings, f"TWITTER_COOKIE_{name.upper()}").strip() for name in CRITICAL_COOKIES} if all(individual.values()): return individual, proxy + cookies_file = settings.TWITTER_COOKIES_FILE 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")) + directory = Path(settings.TWITTER_CREDENTIALS_DIR) if directory.exists(): for path in sorted(directory.glob("*.json")): try: @@ -105,23 +127,29 @@ def load_cookies( except (json.JSONDecodeError, ValueError, OSError) as exc: log.warning("skipping %s: %s", path.name, exc) continue - if not {"auth_token", "ct0"}.issubset(cookies): + if not set(CRITICAL_COOKIES).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 + pin = proxy + if proxy_path.exists(): + try: + # An empty pin file falls back to the global proxy, not "". + pin = proxy_path.read_text(encoding="utf-8").strip() or proxy + except OSError as exc: + log.warning("proxy pin %s unreadable (%s); using default proxy", proxy_path.name, exc) 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) + # A missing directory is a bare install (only the README ships in-repo), + # not misconfiguration; the guest-mode warning fires at search time. return {}, proxy -class ListenerErrorWrapper(Exception): - """Carries a canonical ListenerError through the pipeline.""" +class TwitterErrorWrapper(Exception): + """Carries a canonical TwitterError through the pipeline.""" - def __init__(self, err: ListenerError) -> None: + def __init__(self, err: TwitterError) -> None: super().__init__(err.message) self.error = err @@ -136,8 +164,11 @@ class TwikitClient: 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. + multi-source feed poll). Cookies + proxy are ALSO resolved per call + (from the `TWITTER_*` settings via `load_cookies`), so a rotated + cookie export or env change applies on the next poll without a + restart; an explicit `cookies`/`proxy` constructor arg pins them + instead (tests). """ def __init__( @@ -147,39 +178,65 @@ def __init__( 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, - ) + self._pinned_proxy = proxy + self._pinned_cookies = dict(cookies) if cookies is not None else None async def _search_async(self, query: str, mode: TwikitProduct, count: int): + if self._pinned_cookies is not None: + cookies, proxy = self._pinned_cookies, self._pinned_proxy + else: + cookies, proxy = load_cookies() + proxy = self._pinned_proxy or proxy # 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)) + client = Client(language=self._language, proxy=proxy, user_agent=self._user_agent) + if cookies: + client.set_cookies(dict(cookies), clear_cookies=True) + log.info("session: loaded %d cookie(s)", len(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 + raise TwitterErrorWrapper(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 + raise TwitterErrorWrapper(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).""" + def _run_search(self, query: str, mode: TwikitProduct, count: int): + """One search attempt (a fresh event loop per call, see class + docstring). The rate-limit seam `search` retries around.""" return asyncio.run(self._search_async(query, mode, count)) + + def search( + self, + query: str, + mode: TwikitProduct = "Latest", + count: int = 20, + heartbeat: Callable[[], bool] | None = None, + ): + """Run one live search; returns a twikit Result[Tweet] (or a test double). + + On a `rate_limited` error, back off toward `x-rate-limit-reset` + (capped) and retry, up to MAX_RATE_LIMIT_RETRIES, ticking `heartbeat` + through the wait so the poll lease survives. This is in-cycle + smoothing; a genuine rate-limit exhausts the retries and propagates to + the connector's poll seam, which defers to the next scheduled poll. + Any other error propagates immediately. + """ + attempt = 0 + while True: + try: + return self._run_search(query, mode, count) + except TwitterErrorWrapper as exc: + if exc.error.code != RATE_LIMITED or attempt >= MAX_RATE_LIMIT_RETRIES: + raise + delay = _reset_delay(exc.error.rate_limit_reset, attempt) + log.warning( + "rate limited; retrying in %.0fs (attempt %d/%d)", delay, attempt + 1, MAX_RATE_LIMIT_RETRIES + ) + sleep_with_heartbeat(delay, heartbeat) + attempt += 1 diff --git a/apps/core/sources/connectors/twitter/connector.py b/apps/core/sources/connectors/twitter/connector.py index ac4a1e4a..b20b9e3c 100644 --- a/apps/core/sources/connectors/twitter/connector.py +++ b/apps/core/sources/connectors/twitter/connector.py @@ -23,7 +23,7 @@ from sources.payloads import SourcePayload from ..base import BaseConnector, ConnectorParseError -from .client import ListenerErrorWrapper, TwikitClient, TwikitProduct +from .client import TwikitClient, TwikitProduct, TwitterErrorWrapper from .payloads import NewTweetPayload log = logging.getLogger("sources.twitter") @@ -61,10 +61,9 @@ def poll( 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: + results = self._client.search(spec.query, _TWIKIT_MODES[spec.mode], spec.count, heartbeat=heartbeat) + except TwitterErrorWrapper as exc: err = exc.error log.warning( "twitter search failed query=%r code=%s retryable=%s: %s", @@ -78,11 +77,23 @@ def poll( ) 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: + try: + payload = NewTweetPayload.from_tweet(tweet, query=spec.query) + except Exception: + # One malformed tweet must not fail the source's whole cycle + # (a generator error closes the poll and, since it stays in + # the newest page, re-fails every cycle until it ages out). + log.warning("skipping unmappable tweet id=%s", getattr(tweet, "id", None), exc_info=True) + continue + # Watermark filter: skip only tweets strictly OLDER than the + # cursor. Strict `<`, not `<=`: X `created_at` is second- + # resolution, so two tweets can share a second; the poll op + # advances the watermark to the newest seen, and dropping on tie + # would permanently lose a same-second sibling that arrives in a + # later cycle (its second is already crossed). Re-yielding the + # boundary tweet is idempotent via the external_id dedup (mirrors + # the reddit + youtube connectors). + if since is not None and payload.occurred_at < since: continue if spec.lang and payload.lang and payload.lang != spec.lang: continue diff --git a/apps/core/sources/connectors/twitter/errors.py b/apps/core/sources/connectors/twitter/errors.py index 7f051245..72a7ebf3 100644 --- a/apps/core/sources/connectors/twitter/errors.py +++ b/apps/core/sources/connectors/twitter/errors.py @@ -2,7 +2,7 @@ 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 +to a canonical ``TwitterError``; 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). """ @@ -30,13 +30,18 @@ UserUnavailable, ) +# The one error code the connector branches on (the rate-limit retry loop keys +# off it). Named so the check and the mapping below can't drift to a typo that +# silently disables retries (AGENTS.md: no bare state literals in status checks). +RATE_LIMITED = "rate_limited" + TWIKIT_ERROR_CODE: dict[type[TwitterException], str] = { BadRequest: "bad_request", Unauthorized: "unauthorized", Forbidden: "forbidden", NotFound: "not_found", RequestTimeout: "timeout", - TooManyRequests: "rate_limited", + TooManyRequests: RATE_LIMITED, ServerError: "upstream_error", AccountSuspended: "account_suspended", AccountLocked: "account_locked", @@ -49,7 +54,7 @@ @dataclass -class ListenerError: +class TwitterError: """Canonical error shape for one X fetch failure.""" code: str # stable machine code, see TWIKIT_ERROR_CODE @@ -67,7 +72,7 @@ class ListenerError: ) -def map_bootstrap_failure(exc: Exception, context: dict[str, Any] | None = None) -> ListenerError: +def map_bootstrap_failure(exc: Exception, context: dict[str, Any] | None = None) -> TwitterError: """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).""" @@ -79,7 +84,7 @@ def map_bootstrap_failure(exc: Exception, context: dict[str, Any] | None = None) if code == "bootstrap_blocked" else "unknown failure; log raw and retry with backoff" ) - return ListenerError( + return TwitterError( code=code, message=msg, retryable=code == "internal", @@ -88,8 +93,8 @@ def map_bootstrap_failure(exc: Exception, context: dict[str, Any] | None = None) ) -def map_twikit_error(exc: TwitterException, context: dict[str, Any] | None = None) -> ListenerError: - """Translate a twikit exception into a canonical ListenerError.""" +def map_twikit_error(exc: TwitterException, context: dict[str, Any] | None = None) -> TwitterError: + """Translate a twikit exception into a canonical TwitterError.""" code = TWIKIT_ERROR_CODE.get(type(exc), "twitter_error") reset = getattr(exc, "rate_limit_reset", None) headers = getattr(exc, "headers", None) @@ -108,7 +113,7 @@ def map_twikit_error(exc: TwitterException, context: dict[str, Any] | None = Non # `message: ""`; any other rendering carried response text, so the 404 # is real. if isinstance(exc, NotFound) and str(exc).rstrip().endswith('message: ""'): - return ListenerError( + return TwitterError( code="search_timeline_unavailable", message="X SearchTimeline returned an empty 404 (transient upstream flake)", retryable=True, @@ -124,7 +129,7 @@ def map_twikit_error(exc: TwitterException, context: dict[str, Any] | None = Non "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})"), + 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"), @@ -137,7 +142,7 @@ def map_twikit_error(exc: TwitterException, context: dict[str, Any] | None = Non } retryable, action = retryable_actions.get(code, (True, "unknown; log and retry with backoff")) - return ListenerError( + return TwitterError( code=code, message=str(exc), retryable=retryable, diff --git a/apps/core/sources/connectors/twitter/payloads.py b/apps/core/sources/connectors/twitter/payloads.py index 57df7602..20194afe 100644 --- a/apps/core/sources/connectors/twitter/payloads.py +++ b/apps/core/sources/connectors/twitter/payloads.py @@ -87,7 +87,14 @@ def from_tweet(cls, tweet: Any, query: str | None = None) -> NewTweetPayload: 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 not isinstance(created, datetime): + # A synthetic now() here would advance the source watermark past + # every tweet posted before this instant and strand them (the + # poisoning the YouTube payload floors around). A timestamp-less + # tweet is malformed; refuse it so the connector skips just this + # one (see poll's per-tweet guard). + raise ValueError(f"tweet {tweet_id or ''} carries no created_at datetime") + occurred_at = created if occurred_at.tzinfo is None: occurred_at = occurred_at.replace(tzinfo=UTC) lang = str(getattr(tweet, "lang", None) or "") diff --git a/apps/core/sources/connectors/youtube/errors.py b/apps/core/sources/connectors/youtube/errors.py index 678ae284..67b2d15c 100644 --- a/apps/core/sources/connectors/youtube/errors.py +++ b/apps/core/sources/connectors/youtube/errors.py @@ -1,7 +1,7 @@ """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. +following the same pattern as the Twitter connector's TwitterError. """ from __future__ import annotations diff --git a/apps/core/sources/tests_twitter.py b/apps/core/sources/tests_twitter.py index be6e5b4b..996ff36f 100644 --- a/apps/core/sources/tests_twitter.py +++ b/apps/core/sources/tests_twitter.py @@ -3,10 +3,11 @@ 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- +(TwitterErrorWrapper -> ConnectorParseError), and payload mapping (a duck- typed fake Tweet -> NewTweetPayload). """ +import time from datetime import UTC, datetime from unittest import mock @@ -16,9 +17,9 @@ from openmagpie_schema.configs import TwitterSearchSourceSpec from sources.connectors.base import ConnectorParseError -from sources.connectors.twitter.client import ListenerErrorWrapper +from sources.connectors.twitter.client import TwitterErrorWrapper from sources.connectors.twitter.connector import TwitterSearchConnector -from sources.connectors.twitter.errors import ListenerError, map_twikit_error +from sources.connectors.twitter.errors import TwitterError, map_twikit_error from sources.connectors.twitter.payloads import NewTweetPayload @@ -42,8 +43,8 @@ def __init__( 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.created_at_datetime: datetime | None = created or datetime(2026, 6, 1, 12, 0, tzinfo=UTC) + self.created_at: datetime | None = self.created_at_datetime self.lang = lang self.favorite_count = 10 self.retweet_count = 2 @@ -92,7 +93,17 @@ def test_yields_payloads_newer_than_since(self): 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) + client.search.assert_called_once_with('"social listening"', "Latest", 20, heartbeat=None) + + def test_watermark_boundary_yields(self): + """A tweet AT the watermark re-yields (strict `<`), so a same-second + sibling arriving in a later cycle isn't stranded; external_id dedup + absorbs the re-yield. Dropping on `==` would lose it permanently.""" + at = datetime(2026, 6, 1, 12, 0, tzinfo=UTC) + spec = TwitterSearchSourceSpec(kind="twitter_search", query="x") + conn, _ = self._connector([_FakeTweet("boundary", created=at)]) + payloads = list(conn.poll(spec, since=at)) + self.assertEqual([p.external_id for p in payloads], ["boundary"]) def test_lang_filter(self): spec = TwitterSearchSourceSpec(kind="twitter_search", query="x", lang="es") @@ -103,9 +114,9 @@ def test_lang_filter(self): 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") + err = TwitterError(code="rate_limited", message="slow down", retryable=True, action="backoff") client = mock.Mock() - client.search.side_effect = ListenerErrorWrapper(err) + client.search.side_effect = TwitterErrorWrapper(err) conn = TwitterSearchConnector() conn._client = client with self.assertRaises(ConnectorParseError) as ctx: @@ -119,7 +130,7 @@ def test_empty_404_maps_to_retryable_connector_error(self): # client/client.py: message = f'status: {code}, message: "{text}"'). err = map_twikit_error(NotFound('status: 404, message: ""')) client = mock.Mock() - client.search.side_effect = ListenerErrorWrapper(err) + client.search.side_effect = TwitterErrorWrapper(err) conn = TwitterSearchConnector() conn._client = client with self.assertRaises(ConnectorParseError) as ctx: @@ -130,7 +141,7 @@ 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) + client.search.assert_called_once_with("x", "Top", 20, heartbeat=None) class MapTwikitErrorTests(SimpleTestCase): @@ -149,8 +160,110 @@ def test_message_404_stays_non_retryable_not_found(self): self.assertEqual(err.code, "not_found") self.assertFalse(err.retryable) + def test_rate_limit_reset_extracted_from_header(self): + """A real TooManyRequests carries x-rate-limit-reset; map_twikit_error + must surface it as the int the retry-delay math consumes (not a str).""" + from twikit.errors import TooManyRequests + + from sources.connectors.twitter.client import _reset_delay + from sources.connectors.twitter.errors import RATE_LIMITED + + exc = TooManyRequests("rate limited", headers={"x-rate-limit-reset": "1788888888"}) + err = map_twikit_error(exc) + self.assertEqual(err.code, RATE_LIMITED) + self.assertTrue(err.retryable) + self.assertEqual(err.rate_limit_reset, 1788888888) + # The delay math must accept it without raising (int - float). + self.assertIsInstance(_reset_delay(err.rate_limit_reset, 0), float) + + +class RateLimitRetryTests(SimpleTestCase): + """The client's rate-limit retry loop: wait on `rate_limited` (honoring the + reset epoch) and retry, but pass other errors straight through.""" + + def _rate_limited(self, reset=None): + err = TwitterError(code="rate_limited", message="slow down", retryable=True, action="wait") + err.rate_limit_reset = reset + return TwitterErrorWrapper(err) + + def test_retries_then_succeeds(self): + from sources.connectors.twitter.client import TwikitClient + + client = TwikitClient(cookies={"auth_token": "a", "ct0": "c"}) + calls = {"n": 0} + + def fake_search(q, m, c): + calls["n"] += 1 + if calls["n"] < 3: + raise self._rate_limited() + return ["ok"] + + with ( + mock.patch.object(client, "_run_search", side_effect=fake_search), + mock.patch("sources.connectors.twitter.client.sleep_with_heartbeat") as sleep, + ): + self.assertEqual(client.search("q"), ["ok"]) + self.assertEqual(calls["n"], 3) + self.assertEqual(sleep.call_count, 2) + + def test_gives_up_after_max_retries(self): + from sources.connectors.twitter.client import MAX_RATE_LIMIT_RETRIES, TwikitClient + + client = TwikitClient(cookies={"auth_token": "a", "ct0": "c"}) + with ( + mock.patch.object(client, "_run_search", side_effect=self._rate_limited()), + mock.patch("sources.connectors.twitter.client.sleep_with_heartbeat") as sleep, + self.assertRaises(TwitterErrorWrapper), + ): + client.search("q") + self.assertEqual(sleep.call_count, MAX_RATE_LIMIT_RETRIES) + + def test_non_rate_limit_error_not_retried(self): + from sources.connectors.twitter.client import TwikitClient + + err = TwitterError(code="forbidden", message="no", retryable=False, action="rotate") + client = TwikitClient(cookies={"auth_token": "a", "ct0": "c"}) + with ( + mock.patch.object(client, "_run_search", side_effect=TwitterErrorWrapper(err)), + mock.patch("sources.connectors.twitter.client.sleep_with_heartbeat") as sleep, + self.assertRaises(TwitterErrorWrapper), + ): + client.search("q") + sleep.assert_not_called() + + def test_delay_honors_reset_epoch(self): + from sources.connectors.twitter.client import _reset_delay + + # A reset ~30s in the future is used as the wait; a past reset falls back. + soon = int(time.time()) + 30 + self.assertGreater(_reset_delay(soon, 0), 20) + self.assertLessEqual(_reset_delay(soon, 0), 60) + past = int(time.time()) - 100 + self.assertEqual(_reset_delay(past, 0), 5.0) # base backoff, attempt 0 + class NewTweetPayloadTests(SimpleTestCase): + def test_missing_created_at_raises_instead_of_now(self): + """A timestamp-less tweet must not mint a synthetic now() (it would + advance the watermark past every older-but-new tweet).""" + bad = _FakeTweet("9") + bad.created_at_datetime = None + bad.created_at = None + with self.assertRaises(ValueError): + NewTweetPayload.from_tweet(bad) + + def test_connector_skips_unmappable_tweet_and_keeps_the_rest(self): + bad = _FakeTweet("9") + bad.created_at_datetime = None + bad.created_at = None + spec = TwitterSearchSourceSpec(kind="twitter_search", query="x") + client = mock.Mock() + client.search.return_value = [bad, _FakeTweet("10")] + conn = TwitterSearchConnector() + conn._client = client + payloads = list(conn.poll(spec, since=None)) + self.assertEqual([p.external_id for p in payloads], ["10"]) + def test_from_tweet(self): p = NewTweetPayload.from_tweet(_FakeTweet("123", handle="alice", text="hi")) self.assertEqual(p.external_id, "123") diff --git a/apps/core/sources/tests_twitter_cookies.py b/apps/core/sources/tests_twitter_cookies.py new file mode 100644 index 00000000..978835df --- /dev/null +++ b/apps/core/sources/tests_twitter_cookies.py @@ -0,0 +1,119 @@ +"""X/Twitter cookie/proxy resolution tests (the TWITTER_* settings chain). + +Split from tests_twitter.py (the connector-behavior tests) to keep each file +focused and under the line cap: this module pins load_cookies' priority order +(inline JSON -> auth_token/ct0 pair -> cookie file -> credentials dir), the +per-file .proxy pin precedence, and the critical-pair / empty-pin guards. +""" + +from django.test import SimpleTestCase + + +class LoadCookiesSettingsTests(SimpleTestCase): + """The TWITTER_* settings chain: inline JSON beats the pair, the pair + beats the file, a rotated value applies per call (no import-time cache).""" + + def _settings(self, **overrides): + from django.test import override_settings + + base = { + "TWITTER_COOKIES_JSON": "", + "TWITTER_COOKIE_AUTH_TOKEN": "", + "TWITTER_COOKIE_CT0": "", + "TWITTER_COOKIES_FILE": "", + "TWITTER_CREDENTIALS_DIR": "/nonexistent-for-test", + "TWITTER_PROXY": "", + } + return override_settings(**{**base, **overrides}) + + def test_cookies_json_wins(self): + from sources.connectors.twitter.client import load_cookies + + with self._settings(TWITTER_COOKIES_JSON='{"auth_token": "j", "ct0": "j2"}', TWITTER_COOKIE_AUTH_TOKEN="pair"): + cookies, proxy = load_cookies() + self.assertEqual(cookies, {"auth_token": "j", "ct0": "j2"}) + self.assertIsNone(proxy) + + def test_pair_route_and_proxy(self): + from sources.connectors.twitter.client import load_cookies + + with self._settings(TWITTER_COOKIE_AUTH_TOKEN="a", TWITTER_COOKIE_CT0="c", TWITTER_PROXY="http://p:8080"): + cookies, proxy = load_cookies() + self.assertEqual(cookies, {"auth_token": "a", "ct0": "c"}) + self.assertEqual(proxy, "http://p:8080") + + def test_nothing_configured_is_guest_mode(self): + from sources.connectors.twitter.client import load_cookies + + with self._settings(): + cookies, proxy = load_cookies() + self.assertEqual(cookies, {}) + self.assertIsNone(proxy) + + def test_pair_beats_file(self): + import json + import tempfile + from pathlib import Path + + from sources.connectors.twitter.client import load_cookies + + with tempfile.TemporaryDirectory() as d: + f = Path(d) / "cookies.json" + f.write_text(json.dumps({"auth_token": "file", "ct0": "file2"}), encoding="utf-8") + with self._settings( + TWITTER_COOKIE_AUTH_TOKEN="pair", TWITTER_COOKIE_CT0="pair2", TWITTER_COOKIES_FILE=str(f) + ): + cookies, _ = load_cookies() + self.assertEqual(cookies, {"auth_token": "pair", "ct0": "pair2"}) + + def test_file_route_when_no_pair(self): + import json + import tempfile + from pathlib import Path + + from sources.connectors.twitter.client import load_cookies + + with tempfile.TemporaryDirectory() as d: + f = Path(d) / "cookies.json" + f.write_text(json.dumps({"auth_token": "file", "ct0": "file2"}), encoding="utf-8") + with self._settings(TWITTER_COOKIES_FILE=str(f)): + cookies, _ = load_cookies() + self.assertEqual(cookies, {"auth_token": "file", "ct0": "file2"}) + + def test_credentials_dir_with_proxy_pin(self): + import json + import tempfile + from pathlib import Path + + from sources.connectors.twitter.client import load_cookies + + with tempfile.TemporaryDirectory() as d: + (Path(d) / "acct.json").write_text(json.dumps({"auth_token": "a", "ct0": "c"}), encoding="utf-8") + (Path(d) / "acct.proxy").write_text("http://pin:9090\n", encoding="utf-8") + with self._settings(TWITTER_CREDENTIALS_DIR=d, TWITTER_PROXY="http://global:1"): + cookies, proxy = load_cookies() + self.assertEqual(cookies, {"auth_token": "a", "ct0": "c"}) + self.assertEqual(proxy, "http://pin:9090") # per-file pin overrides the global + + def test_empty_proxy_pin_falls_back_to_global(self): + import json + import tempfile + from pathlib import Path + + from sources.connectors.twitter.client import load_cookies + + with tempfile.TemporaryDirectory() as d: + (Path(d) / "acct.json").write_text(json.dumps({"auth_token": "a", "ct0": "c"}), encoding="utf-8") + (Path(d) / "acct.proxy").write_text(" \n", encoding="utf-8") # empty after strip + with self._settings(TWITTER_CREDENTIALS_DIR=d, TWITTER_PROXY="http://global:1"): + _, proxy = load_cookies() + self.assertEqual(proxy, "http://global:1") # empty pin must not shadow the global with "" + + def test_json_missing_pair_falls_through(self): + from sources.connectors.twitter.client import load_cookies + + # A JSON dict without the critical pair must not return a broken session; + # it falls through to the pair route (here, guest mode). + with self._settings(TWITTER_COOKIES_JSON='{"guest_id": "x"}'): + cookies, _ = load_cookies() + self.assertEqual(cookies, {}) diff --git a/examples/README.md b/examples/README.md index d0faa4bd..a1e5416d 100644 --- a/examples/README.md +++ b/examples/README.md @@ -18,6 +18,12 @@ Available starters: from `hackernews` on purpose: comments are high-volume, so **read the local- processing warning at the top of its `feed.yaml`** and keep the keyword tight before applying: a broad keyword can outrun a local engine. +- `twitter`: an X/Twitter search (`twitter_search`), for catching people talking + about your product or asking for a tool like it. Needs an x.com session cookie + (see [apps/core/credentials/README.md](../apps/core/credentials/README.md)). +- `youtube`: a YouTube search (`youtube_search`) filtered to the last week's + uploads, for videos reviewing or recommending tools in your space. No + credentials needed for public search. Both HN starters take a `query` keyword that runs **server-side as a pre-filter**: it narrows which items the connector pulls into the feed at all, before any watch diff --git a/examples/starters/twitter/feed.yaml b/examples/starters/twitter/feed.yaml new file mode 100644 index 00000000..2a45baec --- /dev/null +++ b/examples/starters/twitter/feed.yaml @@ -0,0 +1,32 @@ +# OpenMagpie starter: X/Twitter brand-mention listening. +# Uses the `twitter_search` source kind (twikit, unofficial route). Needs an +# x.com session cookie; see apps/core/credentials/README.md for generating one +# (and the terms-of-service caveat). Each poll runs the query and surfaces +# tweets 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/twitter/feed.yaml +# magpie watch create -f examples/starters/twitter/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 tweets. +# Full walkthrough: examples/README.md +name: "X/Twitter 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 + # X's search index before any per-item LLM cost, and accepts X's own search + # operators (`from:`, `-word`, `filter:`, quoted phrases). Quote exact + # phrases inside the YAML string, e.g. query: '"social listening"'. + # `mode` picks the result ordering (one of two values): + # latest newest first (the default) - what a listener wants: it pairs + # with the watermark so each poll picks up what's new. + # top X's engagement ranking, not time-ordered - useful for a one-off + # "biggest tweets about X" pull, less so for ongoing listening. + # `count` caps the per-poll fetch (1-100); raise it for a busy query so a + # burst between polls can't overflow past the newest page. + # `lang` (optional): narrow to one language, ISO 639-1, e.g. lang: en. + - spec: {kind: twitter_search, query: '"social listening"', mode: latest, count: 20} + # Watch several phrasings by adding more sources: + # - spec: {kind: twitter_search, query: 'brand monitoring -job', mode: latest, count: 20} diff --git a/examples/starters/twitter/watch.yaml b/examples/starters/twitter/watch.yaml new file mode 100644 index 00000000..fc422d05 --- /dev/null +++ b/examples/starters/twitter/watch.yaml @@ -0,0 +1,22 @@ +# Companion watch for the X/Twitter 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: "X/Twitter product mentions (starter)" +is_active: true +feed_ids: + - REPLACE_WITH_FEED_ID +actions: + - kind: semantic_filter + config: + instructions: "A tweet from someone talking about a social listening or brand monitoring tool: mentioning one by name, asking for a recommendation, or comparing options. The tweet text is the judged body. Not marketing spam or unrelated uses of the words." + threshold: 0.6 + - kind: log + config: + prefix: "[twitter 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/uv.lock b/uv.lock index fb44be6a..a0069e00 100644 --- a/uv.lock +++ b/uv.lock @@ -791,7 +791,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 = "twikit", git = "https://github.com/unclecode/twikit.git?rev=6a73ab97f4de09f79139f6308c9fb80029a9f5f7" }, { name = "ulid", specifier = ">=1.1" }, { name = "yt-dlp", specifier = ">=2026.7.4" }, ] @@ -1301,7 +1301,7 @@ wheels = [ [[package]] name = "twikit" version = "2.3.3" -source = { git = "https://github.com/unclecode/twikit.git#6a73ab97f4de09f79139f6308c9fb80029a9f5f7" } +source = { git = "https://github.com/unclecode/twikit.git?rev=6a73ab97f4de09f79139f6308c9fb80029a9f5f7#6a73ab97f4de09f79139f6308c9fb80029a9f5f7" } dependencies = [ { name = "beautifulsoup4" }, { name = "filetype" },