From adac4c377a34437f696383a889dcd52c199a9f83 Mon Sep 17 00:00:00 2001 From: weichao Date: Mon, 30 Mar 2026 03:54:29 +0000 Subject: [PATCH] feat: add xurl CLI as alternative X search backend Adds xurl (https://github.com/openclaw/xurl) as a third X search backend, sitting after xAI API and Bird/GraphQL in the priority chain. xurl uses the official X API v2 with OAuth2+PKCE authentication, requiring only a free X Developer App. It auto-refreshes tokens and works reliably as a stable fallback when xAI API key or browser cookies are not available. Limitations: - X API search/recent returns last 7 days only (vs Bird's full archive) - No AI-powered relevance scoring (uses token_overlap_relevance instead) - Free tier: 180 requests per 15-minute window New files: - scripts/lib/xurl_x.py: xurl CLI wrapper with search + parse - tests/test_xurl_x.py: 30 unit tests (all passing) Modified files: - scripts/lib/env.py: detect xurl in get_x_source_with_method(), get_missing_keys(), and get_x_source_status() - scripts/last30days.py: add xurl_x import and xurl branch in _search_x() priority chain - SKILL.md: document xurl setup option --- SKILL.md | 4 +- scripts/lib/env.py | 16 ++- scripts/lib/pipeline.py | 4 + scripts/lib/xurl_x.py | 171 +++++++++++++++++++++++++++ tests/test_xurl_x.py | 254 ++++++++++++++++++++++++++++++++++++++++ 5 files changed, 446 insertions(+), 3 deletions(-) create mode 100644 scripts/lib/xurl_x.py create mode 100644 tests/test_xurl_x.py diff --git a/SKILL.md b/SKILL.md index 9f986eb..982c0f9 100644 --- a/SKILL.md +++ b/SKILL.md @@ -291,7 +291,7 @@ Common patterns: - Always active: Reddit, Hacker News, Polymarket - If gh CLI is installed (check `which gh`): add GitHub -- If AUTH_TOKEN/CT0 or XAI_API_KEY or FROM_BROWSER is set: add X +- If AUTH_TOKEN/CT0 or XAI_API_KEY or FROM_BROWSER is set, or xurl CLI is installed and authenticated: add X - If yt-dlp is installed (check `which yt-dlp`): add YouTube - If SCRAPECREATORS_API_KEY is set and INCLUDE_SOURCES contains tiktok: add TikTok - If SCRAPECREATORS_API_KEY is set and INCLUDE_SOURCES contains instagram: add Instagram @@ -1506,7 +1506,7 @@ Want another prompt? Just tell me what you're creating next. **What this skill does:** - Sends search queries to ScrapeCreators API (`api.scrapecreators.com`) for TikTok and Instagram search, and as a Reddit backup when public Reddit is unavailable (requires SCRAPECREATORS_API_KEY) - Legacy: Sends search queries to OpenAI's Responses API (`api.openai.com`) for Reddit discovery (fallback if no SCRAPECREATORS_API_KEY) -- Sends search queries to Twitter's GraphQL API (via optional user-provided AUTH_TOKEN/CT0 env vars - no browser session access) or xAI's API (`api.x.ai`) for X search +- Sends search queries to Twitter's GraphQL API (via optional user-provided AUTH_TOKEN/CT0 env vars - no browser session access), xAI's API (`api.x.ai`), or the official X API v2 via xurl CLI (OAuth2, auto-detected when installed and authenticated) for X search - Sends search queries to Algolia HN Search API (`hn.algolia.com`) for Hacker News story and comment discovery (free, no auth) - Sends search queries to Polymarket Gamma API (`gamma-api.polymarket.com`) for prediction market discovery (free, no auth) - Runs `yt-dlp` locally for YouTube search and transcript extraction (no API key, public data) diff --git a/scripts/lib/env.py b/scripts/lib/env.py index 601c163..f864d7e 100644 --- a/scripts/lib/env.py +++ b/scripts/lib/env.py @@ -356,6 +356,10 @@ def get_x_source_with_method(config: dict[str, Any]) -> tuple[str | None, str]: if config.get("AUTH_TOKEN") and config.get("CT0"): method = config.get("_AUTH_TOKEN_SOURCE", "env") return "bird", method + # Fall back to xurl CLI (official X API v2, OAuth2, free developer app) + from . import xurl_x + if xurl_x.is_available(): + return "xurl", "oauth2" return None, "none" @@ -401,6 +405,7 @@ def get_x_source(config: dict[str, Any]) -> str | None: Returns: 'bird' if Bird is installed and explicit cookies are configured, 'xai' if XAI_API_KEY is configured, + 'xurl' if xurl CLI is installed and authenticated, None if no X source available. """ # Import here to avoid circular dependency @@ -421,6 +426,11 @@ def get_x_source(config: dict[str, Any]) -> str | None: if has_bird_creds and bird_x.is_bird_installed(): return 'bird' + # Fall back to xurl CLI (official X API v2, OAuth2, free developer app) + from . import xurl_x + if xurl_x.is_available(): + return 'xurl' + return None @@ -602,14 +612,18 @@ def get_x_source_status(config: dict[str, Any]) -> dict[str, Any]: elif xai_available: source = 'xai' else: - source = None + # Fall back to xurl CLI + from . import xurl_x as _xurl_check + source = 'xurl' if _xurl_check.is_available() else None + from . import xurl_x as _xurl_x return { "source": source, "bird_installed": bird_status["installed"], "bird_authenticated": bird_status["authenticated"], "bird_username": bird_status["username"], "xai_available": xai_available, + "xurl_available": _xurl_x.is_available(), "can_install_bird": bird_status["can_install"], } diff --git a/scripts/lib/pipeline.py b/scripts/lib/pipeline.py index d759475..b50b94a 100644 --- a/scripts/lib/pipeline.py +++ b/scripts/lib/pipeline.py @@ -40,6 +40,7 @@ from . import ( xai_x, xiaohongshu_api, xquik, + xurl_x, youtube_yt, ) from .cluster import cluster_candidates @@ -895,6 +896,9 @@ def _retrieve_stream( depth=depth, ) return xai_x.parse_x_response(result), {} + if backend == "xurl": + result = xurl_x.search_x(subquery.search_query, depth=depth) + return xurl_x.parse_x_response(result, topic=subquery.search_query), {} raise RuntimeError("No X backend is available.") if source == "youtube": # Use raw_topic so expand_youtube_queries() generates diverse variants diff --git a/scripts/lib/xurl_x.py b/scripts/lib/xurl_x.py new file mode 100644 index 0000000..994b58c --- /dev/null +++ b/scripts/lib/xurl_x.py @@ -0,0 +1,171 @@ +"""X (Twitter) search via xurl CLI — official X API v2 with OAuth2. + +xurl is an open-source CLI for the X API (https://github.com/openclaw/xurl). +It uses OAuth2 with PKCE and automatic token refresh, requiring only a free +X Developer App. No xAI subscription or browser cookies needed. + +Install: npm install -g xurl +Auth: xurl auth oauth2 login + +Priority: xAI API > Bird/GraphQL > xurl > web-only fallback +""" + +import json +import re +import subprocess +import sys +from typing import Any, Dict, List, Optional + +from .relevance import token_overlap_relevance as _compute_relevance + + +def _log(msg: str) -> None: + sys.stderr.write(f"[xurl] {msg}\n") + sys.stderr.flush() + + +# Depth configurations: number of results to request +DEPTH_CONFIG = { + "quick": 10, + "default": 30, + "deep": 60, +} + + +def is_available() -> bool: + """Check if xurl is installed and has valid authentication. + + Returns True only if xurl binary is found AND the user is authenticated + (i.e. ``xurl whoami`` exits 0 and returns a username field). + """ + try: + result = subprocess.run( + ["xurl", "whoami"], + capture_output=True, + text=True, + timeout=10, + ) + return result.returncode == 0 and '"username"' in result.stdout + except FileNotFoundError: + return False + except subprocess.TimeoutExpired: + return False + + +def search_x( + query: str, + depth: str = "default", +) -> Dict[str, Any]: + """Search X via xurl CLI using X API v2 search/recent. + + Args: + query: Search query string + depth: "quick", "default", or "deep" + + Returns: + Raw JSON response from X API v2 tweets/search/recent, or a dict + with an "error" key on failure. + """ + max_results = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) + # X API v2 search/recent requires max_results in 10–100 range + max_results = max(10, min(100, max_results)) + + try: + result = subprocess.run( + ["xurl", "search", query, "-n", str(max_results)], + capture_output=True, + text=True, + timeout=30, + ) + + if result.returncode != 0: + error_text = result.stderr.strip() or result.stdout.strip() + return {"error": f"xurl search failed: {error_text}"} + + return json.loads(result.stdout) + + except FileNotFoundError: + return {"error": "xurl not found in PATH"} + except subprocess.TimeoutExpired: + return {"error": "xurl search timed out (30s)"} + except json.JSONDecodeError as exc: + return {"error": f"Invalid JSON from xurl: {exc}"} + except Exception as exc: + return {"error": f"{type(exc).__name__}: {exc}"} + + +def parse_x_response( + response: Dict[str, Any], + topic: str = "", +) -> List[Dict[str, Any]]: + """Parse xurl search response into normalized item dicts. + + Output format matches the existing XItem schema used by xai_x and bird_x: + id, text, url, author_handle, date, engagement, why_relevant, relevance. + + Args: + response: Raw X API v2 response dict from search_x() + topic: Original search topic (used for relevance scoring) + + Returns: + List of item dicts. Empty list on error or no results. + """ + items: List[Dict[str, Any]] = [] + + if "error" in response: + _log(f"Error in response: {response['error']}") + return items + + data = response.get("data") or [] + if not data: + return items + + # Build author lookup from includes.users + authors: Dict[str, Dict[str, Any]] = {} + for user in (response.get("includes") or {}).get("users") or []: + authors[user["id"]] = user + + for i, tweet in enumerate(data): + author_id = tweet.get("author_id", "") + author = authors.get(author_id, {}) + username = author.get("username", "") + + tweet_id = tweet.get("id", "") + url = f"https://x.com/{username}/status/{tweet_id}" if username else "" + + # Parse public_metrics + engagement: Optional[Dict[str, Any]] = None + metrics = tweet.get("public_metrics") or {} + if metrics: + engagement = { + "likes": metrics.get("like_count", 0), + "reposts": metrics.get("retweet_count", 0), + "replies": metrics.get("reply_count", 0), + "quotes": metrics.get("quote_count", 0), + } + + # Parse ISO 8601 date → YYYY-MM-DD + date: Optional[str] = None + created = tweet.get("created_at", "") + if created: + m = re.match(r"(\d{4}-\d{2}-\d{2})", created) + if m: + date = m.group(1) + + text = tweet.get("text", "").strip() + + # Relevance score via shared token-overlap function + relevance = _compute_relevance(topic, text) if topic else 0.5 + + items.append({ + "id": f"XURL{i + 1}", + "text": text[:500], + "url": url, + "author_handle": username, + "date": date, + "engagement": engagement, + "why_relevant": "", + "relevance": relevance, + }) + + return items diff --git a/tests/test_xurl_x.py b/tests/test_xurl_x.py new file mode 100644 index 0000000..84fcf59 --- /dev/null +++ b/tests/test_xurl_x.py @@ -0,0 +1,254 @@ +"""Tests for xurl_x module.""" + +import json +import sys +import unittest +from pathlib import Path +from unittest import mock + +sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) + +from lib import xurl_x + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_api_response(tweets=None, users=None): + """Build a minimal X API v2 search/recent response.""" + tweets = tweets or [] + users = users or [] + resp = {"data": tweets} + if users: + resp["includes"] = {"users": users} + return resp + + +# --------------------------------------------------------------------------- +# is_available +# --------------------------------------------------------------------------- + +class TestIsAvailable(unittest.TestCase): + def test_returns_true_when_xurl_authenticated(self): + completed = mock.Mock(returncode=0, stdout='{"username": "testuser"}') + with mock.patch("subprocess.run", return_value=completed): + self.assertTrue(xurl_x.is_available()) + + def test_returns_false_when_not_authenticated(self): + completed = mock.Mock(returncode=1, stdout="") + with mock.patch("subprocess.run", return_value=completed): + self.assertFalse(xurl_x.is_available()) + + def test_returns_false_when_not_installed(self): + with mock.patch("subprocess.run", side_effect=FileNotFoundError): + self.assertFalse(xurl_x.is_available()) + + def test_returns_false_on_timeout(self): + import subprocess + with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired("xurl", 10)): + self.assertFalse(xurl_x.is_available()) + + def test_returns_false_when_no_username_in_output(self): + # returncode=0 but output does not contain '"username"' + completed = mock.Mock(returncode=0, stdout='{"id": "123"}') + with mock.patch("subprocess.run", return_value=completed): + self.assertFalse(xurl_x.is_available()) + + +# --------------------------------------------------------------------------- +# search_x +# --------------------------------------------------------------------------- + +class TestSearchX(unittest.TestCase): + def test_returns_parsed_json_on_success(self): + payload = {"data": [{"id": "1", "text": "hello world", "author_id": "u1"}]} + completed = mock.Mock(returncode=0, stdout=json.dumps(payload)) + with mock.patch("subprocess.run", return_value=completed): + result = xurl_x.search_x("hello world") + self.assertEqual(result["data"][0]["id"], "1") + + def test_returns_error_on_non_zero_exit(self): + completed = mock.Mock(returncode=1, stdout="", stderr="rate limit exceeded") + with mock.patch("subprocess.run", return_value=completed): + result = xurl_x.search_x("test") + self.assertIn("error", result) + self.assertIn("rate limit exceeded", result["error"]) + + def test_returns_error_on_invalid_json(self): + completed = mock.Mock(returncode=0, stdout="NOT JSON") + with mock.patch("subprocess.run", return_value=completed): + result = xurl_x.search_x("test") + self.assertIn("error", result) + self.assertIn("Invalid JSON", result["error"]) + + def test_returns_error_when_not_installed(self): + with mock.patch("subprocess.run", side_effect=FileNotFoundError): + result = xurl_x.search_x("test") + self.assertIn("error", result) + self.assertIn("not found", result["error"]) + + def test_returns_error_on_timeout(self): + import subprocess + with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired("xurl", 30)): + result = xurl_x.search_x("test") + self.assertIn("error", result) + self.assertIn("timed out", result["error"]) + + def test_max_results_clamped_to_100(self): + # DEPTH_CONFIG["deep"] = 60, should stay at 60 (within 10-100 range) + completed = mock.Mock(returncode=0, stdout=json.dumps({})) + with mock.patch("subprocess.run", return_value=completed) as run_mock: + xurl_x.search_x("test", depth="deep") + call_args = run_mock.call_args[0][0] + n_idx = call_args.index("-n") + self.assertLessEqual(int(call_args[n_idx + 1]), 100) + + def test_max_results_at_least_10(self): + completed = mock.Mock(returncode=0, stdout=json.dumps({})) + with mock.patch("subprocess.run", return_value=completed) as run_mock: + xurl_x.search_x("test", depth="quick") + call_args = run_mock.call_args[0][0] + n_idx = call_args.index("-n") + self.assertGreaterEqual(int(call_args[n_idx + 1]), 10) + + def test_unknown_depth_falls_back_to_default(self): + completed = mock.Mock(returncode=0, stdout=json.dumps({})) + with mock.patch("subprocess.run", return_value=completed) as run_mock: + xurl_x.search_x("test", depth="nonexistent") + call_args = run_mock.call_args[0][0] + n_idx = call_args.index("-n") + self.assertEqual(int(call_args[n_idx + 1]), xurl_x.DEPTH_CONFIG["default"]) + + +# --------------------------------------------------------------------------- +# parse_x_response +# --------------------------------------------------------------------------- + +class TestParseXResponse(unittest.TestCase): + def _tweet(self, id_, text, author_id, created_at=None, metrics=None): + t = {"id": id_, "text": text, "author_id": author_id} + if created_at: + t["created_at"] = created_at + if metrics: + t["public_metrics"] = metrics + return t + + def _user(self, id_, username): + return {"id": id_, "username": username} + + def test_empty_response_returns_empty_list(self): + self.assertEqual(xurl_x.parse_x_response({}), []) + + def test_error_response_returns_empty_list(self): + self.assertEqual(xurl_x.parse_x_response({"error": "oops"}), []) + + def test_parses_basic_tweet(self): + resp = _make_api_response( + tweets=[self._tweet("111", "Hello AI", "u1")], + users=[self._user("u1", "alice")], + ) + items = xurl_x.parse_x_response(resp) + self.assertEqual(len(items), 1) + self.assertEqual(items[0]["text"], "Hello AI") + self.assertEqual(items[0]["author_handle"], "alice") + self.assertIn("alice", items[0]["url"]) + self.assertIn("111", items[0]["url"]) + + def test_parses_date_from_iso(self): + resp = _make_api_response( + tweets=[self._tweet("1", "text", "u1", created_at="2024-06-15T12:00:00Z")], + ) + items = xurl_x.parse_x_response(resp) + self.assertEqual(items[0]["date"], "2024-06-15") + + def test_date_none_when_missing(self): + resp = _make_api_response(tweets=[self._tweet("1", "text", "u1")]) + items = xurl_x.parse_x_response(resp) + self.assertIsNone(items[0]["date"]) + + def test_parses_engagement_metrics(self): + metrics = { + "like_count": 42, + "retweet_count": 10, + "reply_count": 5, + "quote_count": 2, + } + resp = _make_api_response( + tweets=[self._tweet("1", "text", "u1", metrics=metrics)], + ) + items = xurl_x.parse_x_response(resp) + self.assertEqual(items[0]["engagement"]["likes"], 42) + self.assertEqual(items[0]["engagement"]["reposts"], 10) + self.assertEqual(items[0]["engagement"]["replies"], 5) + self.assertEqual(items[0]["engagement"]["quotes"], 2) + + def test_engagement_none_when_no_metrics(self): + resp = _make_api_response(tweets=[self._tweet("1", "text", "u1")]) + items = xurl_x.parse_x_response(resp) + self.assertIsNone(items[0]["engagement"]) + + def test_text_truncated_to_500_chars(self): + long_text = "x" * 600 + resp = _make_api_response(tweets=[self._tweet("1", long_text, "u1")]) + items = xurl_x.parse_x_response(resp) + self.assertLessEqual(len(items[0]["text"]), 500) + + def test_id_prefixed_with_xurl(self): + resp = _make_api_response(tweets=[self._tweet("1", "text", "u1")]) + items = xurl_x.parse_x_response(resp) + self.assertTrue(items[0]["id"].startswith("XURL")) + + def test_relevance_computed_when_topic_given(self): + resp = _make_api_response( + tweets=[self._tweet("1", "Claude Code is great for AI coding", "u1")], + ) + items = xurl_x.parse_x_response(resp, topic="Claude Code") + self.assertGreater(items[0]["relevance"], 0.5) + + def test_relevance_neutral_when_no_topic(self): + resp = _make_api_response(tweets=[self._tweet("1", "some text", "u1")]) + items = xurl_x.parse_x_response(resp) + self.assertEqual(items[0]["relevance"], 0.5) + + def test_url_empty_when_no_username(self): + # author_id not in includes.users → username="" + resp = _make_api_response(tweets=[self._tweet("999", "text", "unknown_uid")]) + items = xurl_x.parse_x_response(resp) + self.assertEqual(items[0]["url"], "") + + def test_multiple_tweets_parsed(self): + tweets = [self._tweet(str(i), f"tweet {i}", "u1") for i in range(5)] + resp = _make_api_response(tweets=tweets, users=[self._user("u1", "bob")]) + items = xurl_x.parse_x_response(resp) + self.assertEqual(len(items), 5) + + def test_empty_data_list(self): + resp = _make_api_response(tweets=[]) + self.assertEqual(xurl_x.parse_x_response(resp), []) + + def test_why_relevant_is_empty_string(self): + # xurl doesn't provide LLM-generated why_relevant (unlike xai_x) + resp = _make_api_response(tweets=[self._tweet("1", "text", "u1")]) + items = xurl_x.parse_x_response(resp) + self.assertEqual(items[0]["why_relevant"], "") + + +# --------------------------------------------------------------------------- +# DEPTH_CONFIG +# --------------------------------------------------------------------------- + +class TestDepthConfig(unittest.TestCase): + def test_all_standard_depths_present(self): + for depth in ("quick", "default", "deep"): + self.assertIn(depth, xurl_x.DEPTH_CONFIG) + + def test_deep_greater_than_quick(self): + self.assertGreater( + xurl_x.DEPTH_CONFIG["deep"], + xurl_x.DEPTH_CONFIG["quick"], + ) + + +if __name__ == "__main__": + unittest.main()