| 1 | """X (Twitter) search via xurl CLI — official X API v2. |
| 2 | |
| 3 | xurl is X's official CLI for the X API |
| 4 | (https://github.com/xdevplatform/xurl). It requires only a free |
| 5 | X Developer App. No xAI subscription or browser cookies needed. |
| 6 | |
| 7 | Install: npm install -g @xdevplatform/xurl |
| 8 | Auth: xurl auth app-only <bearer-token> (search / availability) |
| 9 | xurl auth oauth1 ... (optional; not used for search) |
| 10 | |
| 11 | Priority: xAI API > Bird/GraphQL > xurl > web-only fallback |
| 12 | """ |
| 13 | |
| 14 | import json |
| 15 | import re |
| 16 | import shutil |
| 17 | import subprocess |
| 18 | from pathlib import Path |
| 19 | from typing import Any, Dict, List, Optional, Tuple |
| 20 | |
| 21 | from . import log |
| 22 | from .relevance import token_overlap_relevance as _compute_relevance |
| 23 | |
| 24 | # xurl auth status marks a configured app-only bearer as "bearer: ✓". |
| 25 | # Search uses --auth app, so availability must require this — oauth1 alone |
| 26 | # is not enough. |
| 27 | _BEARER_CONFIGURED_RE = re.compile(r"bearer:\s*✓") |
| 28 | |
| 29 | |
| 30 | def _log(msg: str) -> None: |
| 31 | log.source_log("xurl", msg, tty_only=False) |
| 32 | |
| 33 | |
| 34 | # Depth configurations: number of results to request |
| 35 | DEPTH_CONFIG = { |
| 36 | "quick": 10, |
| 37 | "default": 30, |
| 38 | "deep": 60, |
| 39 | } |
| 40 | |
| 41 | |
| 42 | # Memoized availability, mirroring health.py's per-process dependency-probe |
| 43 | # cache: each uncached is_available() check spawns an `xurl auth status` |
| 44 | # subprocess (local credential status; no network). The doctor/safe-diagnose |
| 45 | # path never uses it — see stored_auth_status()/has_stored_auth() below — |
| 46 | # but research-time callers may consult it more than once per process. |
| 47 | # None means "not yet probed". |
| 48 | _availability_cache: Optional[bool] = None |
| 49 | |
| 50 | |
| 51 | def clear_availability_cache() -> None: |
| 52 | """Reset the memoized is_available() result (tests, or a re-check after auth).""" |
| 53 | global _availability_cache |
| 54 | _availability_cache = None |
| 55 | |
| 56 | |
| 57 | def is_available() -> bool: |
| 58 | """Check if xurl is installed and has app-only bearer auth. |
| 59 | |
| 60 | Returns True only if xurl binary is found AND ``xurl auth status`` |
| 61 | exits 0 with a configured app-only bearer (``bearer: ✓``). OAuth1 |
| 62 | alone is insufficient — ``search_x`` pins ``--auth app``. |
| 63 | Memoized per process; ``clear_availability_cache()`` resets. |
| 64 | """ |
| 65 | global _availability_cache |
| 66 | if _availability_cache is None: |
| 67 | _availability_cache = _is_available_uncached() |
| 68 | return _availability_cache |
| 69 | |
| 70 | |
| 71 | def _is_available_uncached() -> bool: |
| 72 | try: |
| 73 | result = subprocess.run( |
| 74 | ["xurl", "auth", "status"], |
| 75 | capture_output=True, |
| 76 | text=True, |
| 77 | timeout=10, |
| 78 | ) |
| 79 | return ( |
| 80 | result.returncode == 0 |
| 81 | and _BEARER_CONFIGURED_RE.search(result.stdout) is not None |
| 82 | ) |
| 83 | except (OSError, subprocess.TimeoutExpired): |
| 84 | # OSError covers FileNotFoundError (no xurl on PATH) and |
| 85 | # PermissionError (a non-executable match on PATH, e.g. WSL's |
| 86 | # /mnt/c/.../WindowsApps shim returning EACCES on exec). |
| 87 | return False |
| 88 | |
| 89 | |
| 90 | # --------------------------------------------------------------------------- |
| 91 | # Local auth evidence (doctor / safe-diagnose path — no subprocess, no |
| 92 | # network). |
| 93 | # |
| 94 | # xurl persists OAuth credentials to an on-disk token store at ~/.xurl |
| 95 | # (YAML in current releases; legacy versions wrote JSON — see the upstream |
| 96 | # store package at github.com/xdevplatform/xurl). A populated store is the |
| 97 | # strongest LOCAL evidence of authentication obtainable without spending a |
| 98 | # network call, so doctor keys on it and reports "auth not live-verified" |
| 99 | # instead of running `xurl whoami` (a real, authenticated X API request |
| 100 | # that would violate doctor's no-network guarantee). |
| 101 | # --------------------------------------------------------------------------- |
| 102 | |
| 103 | AUTH_OK = "ok" # token store present with stored credentials |
| 104 | AUTH_MISSING = "missing" # no token store, or no credentials stored in it |
| 105 | AUTH_ERROR = "error" # token store exists but could not be read |
| 106 | |
| 107 | # Substrings a populated store carries in both the YAML and legacy JSON |
| 108 | # formats (per-user oauth2 token blocks, or an app-only bearer token). |
| 109 | _TOKEN_STORE_MARKERS = ( |
| 110 | "access_token", |
| 111 | "bearer_token", |
| 112 | "oauth2_tokens", |
| 113 | "oauth1_tokens", |
| 114 | ) |
| 115 | |
| 116 | |
| 117 | def token_store_path() -> Path: |
| 118 | """xurl's on-disk OAuth token store (~/.xurl).""" |
| 119 | return Path.home() / ".xurl" |
| 120 | |
| 121 | |
| 122 | def stored_auth_status() -> Tuple[str, str]: |
| 123 | """Local-only evidence of xurl authentication: ``(status, detail)``. |
| 124 | |
| 125 | Reads only the on-disk token store — never spawns xurl, never touches |
| 126 | the network. ``status`` is AUTH_OK (store holds credentials), |
| 127 | AUTH_MISSING (no store / empty store / no credential markers), or |
| 128 | AUTH_ERROR (store exists but cannot be read — surfaced as a typed |
| 129 | error, not as "unconfigured"). |
| 130 | """ |
| 131 | path = token_store_path() |
| 132 | try: |
| 133 | if not path.is_file(): |
| 134 | return AUTH_MISSING, f"no token store at {path}" |
| 135 | content = path.read_text(encoding="utf-8", errors="replace") |
| 136 | except OSError as exc: |
| 137 | return ( |
| 138 | AUTH_ERROR, |
| 139 | f"token store {path} unreadable: {type(exc).__name__}: {exc}", |
| 140 | ) |
| 141 | if any(marker in content for marker in _TOKEN_STORE_MARKERS): |
| 142 | return AUTH_OK, f"stored OAuth credentials found in {path}" |
| 143 | return AUTH_MISSING, f"token store {path} has no stored credentials" |
| 144 | |
| 145 | |
| 146 | def has_stored_auth() -> bool: |
| 147 | """Local-only availability: xurl on PATH with stored credentials. |
| 148 | |
| 149 | The doctor/safe-diagnose counterpart of ``is_available()`` — the same |
| 150 | "installed and authenticated" question answered from local evidence |
| 151 | only (PATH lookup + token store), never a live ``xurl whoami``. A |
| 152 | broken token store reads as unavailable here; the doctor probe layer |
| 153 | (``backends._probe_xurl``) reports that case as a typed error. |
| 154 | """ |
| 155 | return shutil.which("xurl") is not None and stored_auth_status()[0] == AUTH_OK |
| 156 | |
| 157 | |
| 158 | def search_x( |
| 159 | query: str, |
| 160 | depth: str = "default", |
| 161 | ) -> Dict[str, Any]: |
| 162 | """Search X via xurl CLI using X API v2 search/recent. |
| 163 | |
| 164 | Args: |
| 165 | query: Search query string |
| 166 | depth: "quick", "default", or "deep" |
| 167 | |
| 168 | Returns: |
| 169 | Raw JSON response from X API v2 tweets/search/recent, or a dict |
| 170 | with an "error" key on failure. |
| 171 | """ |
| 172 | max_results = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 173 | # X API v2 search/recent requires max_results in 10–100 range |
| 174 | max_results = max(10, min(100, max_results)) |
| 175 | |
| 176 | try: |
| 177 | # --auth app (app-only bearer): xurl >=1.1 mis-signs OAuth1 requests |
| 178 | # whose query needs percent-encoding (spaces, parens, ...) -> 401. |
| 179 | # Bearer auth sends no signature, so multi-word queries work. |
| 180 | result = subprocess.run( |
| 181 | ["xurl", "search", query, "-n", str(max_results), "--auth", "app"], |
| 182 | capture_output=True, |
| 183 | text=True, |
| 184 | timeout=30, |
| 185 | ) |
| 186 | |
| 187 | if result.returncode != 0: |
| 188 | error_text = result.stderr.strip() or result.stdout.strip() |
| 189 | return {"error": f"xurl search failed: {error_text}"} |
| 190 | |
| 191 | return json.loads(result.stdout) |
| 192 | |
| 193 | except FileNotFoundError: |
| 194 | return {"error": "xurl not found in PATH"} |
| 195 | except subprocess.TimeoutExpired: |
| 196 | return {"error": "xurl search timed out (30s)"} |
| 197 | except json.JSONDecodeError as exc: |
| 198 | return {"error": f"Invalid JSON from xurl: {exc}"} |
| 199 | except Exception as exc: |
| 200 | return {"error": f"{type(exc).__name__}: {exc}"} |
| 201 | |
| 202 | |
| 203 | def parse_x_response( |
| 204 | response: Dict[str, Any], |
| 205 | topic: str = "", |
| 206 | ) -> List[Dict[str, Any]]: |
| 207 | """Parse xurl search response into normalized item dicts. |
| 208 | |
| 209 | Output format matches the existing XItem schema used by xai_x and bird_x: |
| 210 | id, text, url, author_handle, date, engagement, why_relevant, relevance. |
| 211 | |
| 212 | Args: |
| 213 | response: Raw X API v2 response dict from search_x() |
| 214 | topic: Original search topic (used for relevance scoring) |
| 215 | |
| 216 | Returns: |
| 217 | List of item dicts. Empty list on error or no results. |
| 218 | """ |
| 219 | items: List[Dict[str, Any]] = [] |
| 220 | |
| 221 | if "error" in response: |
| 222 | _log(f"Error in response: {response['error']}") |
| 223 | return items |
| 224 | |
| 225 | data = response.get("data") or [] |
| 226 | if not data: |
| 227 | return items |
| 228 | |
| 229 | # Build author lookup from includes.users |
| 230 | authors: Dict[str, Dict[str, Any]] = {} |
| 231 | for user in (response.get("includes") or {}).get("users") or []: |
| 232 | authors[user["id"]] = user |
| 233 | |
| 234 | for i, tweet in enumerate(data): |
| 235 | author_id = tweet.get("author_id", "") |
| 236 | author = authors.get(author_id, {}) |
| 237 | username = author.get("username", "") |
| 238 | |
| 239 | tweet_id = tweet.get("id", "") |
| 240 | url = f"https://x.com/{username}/status/{tweet_id}" if username else "" |
| 241 | |
| 242 | # Parse public_metrics |
| 243 | engagement: Optional[Dict[str, Any]] = None |
| 244 | metrics = tweet.get("public_metrics") or {} |
| 245 | if metrics: |
| 246 | engagement = { |
| 247 | "likes": metrics.get("like_count", 0), |
| 248 | "reposts": metrics.get("retweet_count", 0), |
| 249 | "replies": metrics.get("reply_count", 0), |
| 250 | "quotes": metrics.get("quote_count", 0), |
| 251 | } |
| 252 | |
| 253 | # Parse ISO 8601 date → YYYY-MM-DD |
| 254 | date: Optional[str] = None |
| 255 | created = tweet.get("created_at", "") |
| 256 | if created: |
| 257 | m = re.match(r"(\d{4}-\d{2}-\d{2})", created) |
| 258 | if m: |
| 259 | date = m.group(1) |
| 260 | |
| 261 | text = tweet.get("text", "").strip() |
| 262 | |
| 263 | # Relevance score via shared token-overlap function |
| 264 | relevance = _compute_relevance(topic, text) if topic else 0.5 |
| 265 | |
| 266 | items.append({ |
| 267 | "id": f"XURL{i + 1}", |
| 268 | "text": text[:500], |
| 269 | "url": url, |
| 270 | "author_handle": username, |
| 271 | "date": date, |
| 272 | "engagement": engagement, |
| 273 | "why_relevant": "", |
| 274 | "relevance": relevance, |
| 275 | }) |
| 276 | |
| 277 | return items |
| 278 |