Merge pull request #130 from chaosreload/feat/xurl-x-search

feat: add xurl CLI as alternative X search backend (official API v2 via OAuth2)

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"],
     }
 

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

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