| 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 stat |
| 18 | import subprocess |
| 19 | from pathlib import Path |
| 20 | from typing import Any, Dict, List, Optional, Tuple |
| 21 | |
| 22 | from . import health, http, log |
| 23 | from . import x_api |
| 24 | # One X API v2 depth table and one parser: xurl_x re-imports both. |
| 25 | from .x_api import DEPTH_CONFIG |
| 26 | |
| 27 | # xurl auth status marks a configured app-only bearer as "bearer: ✓". |
| 28 | # Search uses --auth app, so availability must require this — oauth1 alone |
| 29 | # is not enough. |
| 30 | _BEARER_CONFIGURED_RE = re.compile(r"bearer:\s*✓") |
| 31 | |
| 32 | |
| 33 | def _log(msg: str) -> None: |
| 34 | log.source_log("xurl", msg, tty_only=False) |
| 35 | |
| 36 | |
| 37 | |
| 38 | # Memoized availability, mirroring health.py's per-process dependency-probe |
| 39 | # cache: each uncached is_available() check spawns an `xurl auth status` |
| 40 | # subprocess (local credential status; no network). The doctor/safe-diagnose |
| 41 | # path never uses it — see stored_auth_status()/has_stored_auth() below — |
| 42 | # but research-time callers may consult it more than once per process. |
| 43 | # None means "not yet probed". |
| 44 | _availability_cache: Optional[bool] = None |
| 45 | |
| 46 | |
| 47 | def clear_availability_cache() -> None: |
| 48 | """Reset the memoized is_available() result (tests, or a re-check after auth).""" |
| 49 | global _availability_cache |
| 50 | _availability_cache = None |
| 51 | |
| 52 | |
| 53 | def is_available() -> bool: |
| 54 | """Check if xurl is installed and has app-only bearer auth. |
| 55 | |
| 56 | Returns True only if xurl binary is found AND ``xurl auth status`` |
| 57 | exits 0 with a configured app-only bearer (``bearer: ✓``). OAuth1 |
| 58 | alone is insufficient — ``search_x`` pins ``--auth app``. |
| 59 | Memoized per process; ``clear_availability_cache()`` resets. |
| 60 | """ |
| 61 | global _availability_cache |
| 62 | if _availability_cache is None: |
| 63 | _availability_cache = _is_available_uncached() |
| 64 | return _availability_cache |
| 65 | |
| 66 | |
| 67 | def _is_available_uncached() -> bool: |
| 68 | try: |
| 69 | result = subprocess.run( |
| 70 | ["xurl", "auth", "status"], |
| 71 | capture_output=True, |
| 72 | text=True, |
| 73 | timeout=10, |
| 74 | ) |
| 75 | return ( |
| 76 | result.returncode == 0 |
| 77 | and _BEARER_CONFIGURED_RE.search(result.stdout) is not None |
| 78 | ) |
| 79 | except (OSError, subprocess.TimeoutExpired): |
| 80 | # OSError covers FileNotFoundError (no xurl on PATH) and |
| 81 | # PermissionError (a non-executable match on PATH, e.g. WSL's |
| 82 | # /mnt/c/.../WindowsApps shim returning EACCES on exec). |
| 83 | return False |
| 84 | |
| 85 | |
| 86 | # --------------------------------------------------------------------------- |
| 87 | # Local auth evidence (doctor / safe-diagnose path — no subprocess, no |
| 88 | # network). |
| 89 | # |
| 90 | # xurl persists OAuth credentials to an on-disk token store at ~/.xurl |
| 91 | # (YAML in current releases; legacy versions wrote JSON — see the upstream |
| 92 | # store package at github.com/xdevplatform/xurl). A populated store is the |
| 93 | # strongest LOCAL evidence of authentication obtainable without spending a |
| 94 | # network call, so doctor keys on it and reports "auth not live-verified" |
| 95 | # instead of running `xurl whoami` (a real, authenticated X API request |
| 96 | # that would violate doctor's no-network guarantee). |
| 97 | # --------------------------------------------------------------------------- |
| 98 | |
| 99 | AUTH_OK = "ok" # token store present with stored credentials |
| 100 | AUTH_MISSING = "missing" # no token store, or no credentials stored in it |
| 101 | AUTH_ERROR = "error" # token store exists but could not be read |
| 102 | |
| 103 | # Substrings a populated store carries in both the YAML and legacy JSON |
| 104 | # formats (per-user oauth2 token blocks, or an app-only bearer token). |
| 105 | _TOKEN_STORE_MARKERS = ( |
| 106 | "access_token", |
| 107 | "bearer_token", |
| 108 | "oauth2_tokens", |
| 109 | "oauth1_tokens", |
| 110 | ) |
| 111 | |
| 112 | |
| 113 | def _is_file(path: Path) -> bool: |
| 114 | """True when *path* is a regular file; raise on an unreadable stat. |
| 115 | |
| 116 | ``pathlib.Path.is_file()`` swallows ``OSError`` into ``False``, which |
| 117 | would silently misreport a permission-denied (EACCES) store as absent. |
| 118 | Probing via ``stat`` keeps the typed ``AUTH_ERROR`` path reachable for |
| 119 | real stat failures. A missing path (``FileNotFoundError``) returns False — |
| 120 | absence is a normal state (this also covers a dangling symlink whose |
| 121 | target is gone). Path stubs without ``stat``/``is_file`` support (test |
| 122 | doubles) fall back to their own behavior. |
| 123 | """ |
| 124 | try: |
| 125 | return stat.S_ISREG(path.stat().st_mode) |
| 126 | except FileNotFoundError: |
| 127 | return False |
| 128 | except (AttributeError, TypeError): |
| 129 | try: |
| 130 | return bool(path.is_file()) |
| 131 | except AttributeError: |
| 132 | return False |
| 133 | |
| 134 | |
| 135 | def _is_dir(path: Path) -> bool: |
| 136 | """True when *path* is a directory; raise on an unreadable stat. |
| 137 | |
| 138 | Same rationale as :func:`_is_file` — pathlib's ``is_dir()`` hides |
| 139 | ``OSError``, so a permission-denied parent would read as a missing store. |
| 140 | """ |
| 141 | try: |
| 142 | return stat.S_ISDIR(path.stat().st_mode) |
| 143 | except FileNotFoundError: |
| 144 | return False |
| 145 | except (AttributeError, TypeError): |
| 146 | try: |
| 147 | return bool(path.is_dir()) |
| 148 | except AttributeError: |
| 149 | return False |
| 150 | |
| 151 | |
| 152 | def token_store_path() -> Path: |
| 153 | """xurl's on-disk OAuth token store (~/.xurl/auth.yml). |
| 154 | |
| 155 | Current xurl (>=1.1) stores credentials in a YAML file at |
| 156 | ``~/.xurl/auth.yml``; ``~/.xurl`` itself is a directory. Legacy |
| 157 | releases wrote a flat file at ``~/.xurl``. ``stored_auth_status()`` |
| 158 | resolves both layouts; this returns the canonical current path. |
| 159 | """ |
| 160 | return Path.home() / ".xurl" / "auth.yml" |
| 161 | |
| 162 | |
| 163 | def stored_auth_status() -> Tuple[str, str]: |
| 164 | """Local-only evidence of xurl authentication: ``(status, detail)``. |
| 165 | |
| 166 | Reads only the on-disk token store — never spawns xurl, never touches |
| 167 | the network. ``status`` is AUTH_OK (store holds credentials), |
| 168 | AUTH_MISSING (no store / empty store / no credential markers), or |
| 169 | AUTH_ERROR (store exists but cannot be read — surfaced as a typed |
| 170 | error, not as "unconfigured"). |
| 171 | """ |
| 172 | # Resolve the actual credential file across layouts. Current xurl keeps |
| 173 | # ~/.xurl/auth.yml inside the ~/.xurl directory; legacy installs wrote a |
| 174 | # flat ~/.xurl file. Honor whatever token_store_path() resolves to and |
| 175 | # check both shapes so a valid directory layout is never misread as a |
| 176 | # missing token store. Path stubs that cannot combine (e.g. an unreadable |
| 177 | # store stub without / support) still reach the typed error path below. |
| 178 | path = token_store_path() |
| 179 | try: |
| 180 | base = path.parent if path.name == "auth.yml" else path |
| 181 | except (AttributeError, TypeError): |
| 182 | # A path stub with no parent/name support: treat it as the file itself. |
| 183 | base = path |
| 184 | try: |
| 185 | # Use stat rather than is_file()/is_dir(): pathlib swallows OSError |
| 186 | # into False, which would silently turn a permission-denied store into |
| 187 | # "no token store" instead of the typed AUTH_ERROR. |
| 188 | candidates = [base / "auth.yml", base] if _is_dir(base) else [base, base / "auth.yml"] |
| 189 | except (TypeError, AttributeError): |
| 190 | # A path stub that cannot combine (no __truediv__): the path itself is |
| 191 | # the only candidate. |
| 192 | candidates = [path] |
| 193 | except OSError as exc: |
| 194 | return ( |
| 195 | AUTH_ERROR, |
| 196 | f"token store {base} unreadable: {type(exc).__name__}: {exc}", |
| 197 | ) |
| 198 | try: |
| 199 | path = next(c for c in candidates if _is_file(c)) |
| 200 | except StopIteration: |
| 201 | return AUTH_MISSING, f"no token store at {token_store_path()}" |
| 202 | except OSError as exc: |
| 203 | return ( |
| 204 | AUTH_ERROR, |
| 205 | f"token store {base} unreadable: {type(exc).__name__}: {exc}", |
| 206 | ) |
| 207 | try: |
| 208 | content = path.read_text(encoding="utf-8", errors="replace") |
| 209 | except OSError as exc: |
| 210 | return ( |
| 211 | AUTH_ERROR, |
| 212 | f"token store {path} unreadable: {type(exc).__name__}: {exc}", |
| 213 | ) |
| 214 | if any(marker in content for marker in _TOKEN_STORE_MARKERS): |
| 215 | return AUTH_OK, f"stored OAuth credentials found in {path}" |
| 216 | return AUTH_MISSING, f"token store {path} has no stored credentials" |
| 217 | |
| 218 | |
| 219 | def has_stored_auth() -> bool: |
| 220 | """Local-only availability: xurl on PATH with stored credentials. |
| 221 | |
| 222 | The doctor/safe-diagnose counterpart of ``is_available()`` — the same |
| 223 | "installed and authenticated" question answered from local evidence |
| 224 | only (PATH lookup + token store), never a live ``xurl whoami``. A |
| 225 | broken token store reads as unavailable here; the doctor probe layer |
| 226 | (``backends._probe_xurl``) reports that case as a typed error. |
| 227 | """ |
| 228 | return shutil.which("xurl") is not None and stored_auth_status()[0] == AUTH_OK |
| 229 | |
| 230 | |
| 231 | # Engine-authored fixed error strings (the same rule as x_api): xurl's |
| 232 | # stderr can echo the request, the bearer, or an account id, so it never |
| 233 | # reaches the outcome detail. Each string carries a marker that |
| 234 | # http.classify_failure recognizes. |
| 235 | ERR_PAYMENT_REQUIRED = "xurl: payment required (X API credits exhausted)" |
| 236 | ERR_UNAUTHORIZED = "xurl: unauthorized (bearer token rejected)" |
| 237 | ERR_FORBIDDEN = "xurl: forbidden (bearer token lacks access)" |
| 238 | ERR_RATE_LIMITED = "xurl: rate limit exceeded (X API)" |
| 239 | ERR_FAILED = "xurl: search failed" |
| 240 | ERR_INVALID_JSON = "xurl: invalid JSON from xurl" |
| 241 | ERR_NOT_FOUND = "xurl not found in PATH" |
| 242 | ERR_TIMED_OUT = "xurl search timed out (30s)" |
| 243 | |
| 244 | |
| 245 | def _classify_cli_failure(output: str) -> str: |
| 246 | """Map xurl's stderr/stdout to a fixed string via the shared classifier.""" |
| 247 | text = output or "" |
| 248 | state = http.classify_failure(message=text) |
| 249 | if state == health.PAYMENT_REQUIRED: |
| 250 | return ERR_PAYMENT_REQUIRED |
| 251 | if state == health.AUTH_FAILED: |
| 252 | # The shared vocabulary has one auth state; split the wording only. |
| 253 | lowered = text.lower() |
| 254 | if "401" in lowered or "unauthorized" in lowered: |
| 255 | return ERR_UNAUTHORIZED |
| 256 | return ERR_FORBIDDEN |
| 257 | if state == health.RATE_LIMITED: |
| 258 | return ERR_RATE_LIMITED |
| 259 | return ERR_FAILED |
| 260 | |
| 261 | |
| 262 | def search_x( |
| 263 | query: str, |
| 264 | depth: str = "default", |
| 265 | ) -> Dict[str, Any]: |
| 266 | """Search X via xurl CLI using X API v2 search/recent. |
| 267 | |
| 268 | Args: |
| 269 | query: Search query string |
| 270 | depth: "quick", "default", or "deep" |
| 271 | |
| 272 | Returns: |
| 273 | Raw JSON response from X API v2 tweets/search/recent, or a dict |
| 274 | with an "error" key holding a fixed string on failure. The CLI's |
| 275 | own output reaches only the debug log, never the error. |
| 276 | """ |
| 277 | max_results = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 278 | # X API v2 search/recent requires max_results in 10–100 range |
| 279 | max_results = max(10, min(100, max_results)) |
| 280 | |
| 281 | try: |
| 282 | # --auth app (app-only bearer): xurl >=1.1 mis-signs OAuth1 requests |
| 283 | # whose query needs percent-encoding (spaces, parens, ...) -> 401. |
| 284 | # Bearer auth sends no signature, so multi-word queries work. |
| 285 | result = subprocess.run( |
| 286 | ["xurl", "search", query, "-n", str(max_results), "--auth", "app"], |
| 287 | capture_output=True, |
| 288 | text=True, |
| 289 | timeout=30, |
| 290 | ) |
| 291 | |
| 292 | if result.returncode != 0: |
| 293 | error_text = result.stderr.strip() or result.stdout.strip() |
| 294 | classified = _classify_cli_failure(error_text) |
| 295 | # The raw CLI output can echo the bearer; only the fixed string |
| 296 | # and its size reach the debug log. |
| 297 | log.debug(f"xurl exit {result.returncode}: {classified} ({len(error_text)} chars)") |
| 298 | return {"error": classified} |
| 299 | |
| 300 | return json.loads(result.stdout) |
| 301 | |
| 302 | except FileNotFoundError: |
| 303 | return {"error": ERR_NOT_FOUND} |
| 304 | except subprocess.TimeoutExpired: |
| 305 | return {"error": ERR_TIMED_OUT} |
| 306 | except json.JSONDecodeError: |
| 307 | return {"error": ERR_INVALID_JSON} |
| 308 | except Exception as exc: |
| 309 | return {"error": f"xurl: {type(exc).__name__}"} |
| 310 | |
| 311 | |
| 312 | def parse_x_response( |
| 313 | response: Dict[str, Any], |
| 314 | topic: str = "", |
| 315 | ) -> List[Dict[str, Any]]: |
| 316 | """Parse an xurl search response into normalized item dicts. |
| 317 | |
| 318 | Delegates to the shared X API v2 parser (``x_api.parse_v2_response``) |
| 319 | so xurl and xapi never drift; only the id prefix differs. |
| 320 | |
| 321 | Args: |
| 322 | response: Raw X API v2 response dict from search_x() |
| 323 | topic: Original search topic (used for relevance scoring) |
| 324 | |
| 325 | Returns: |
| 326 | List of item dicts. Empty list on error or no results. |
| 327 | """ |
| 328 | if isinstance(response, dict) and "error" in response: |
| 329 | _log(f"Error in response: {response['error']}") |
| 330 | return [] |
| 331 | return x_api.parse_v2_response(response, topic, None, id_prefix="XURL") |
| 332 |