| 1 | """Bird X search client for the v3.0.0 last30days pipeline. |
| 2 | |
| 3 | Uses a vendored subset of @steipete/bird v0.8.0 (MIT License) to search X |
| 4 | via Twitter's GraphQL API. No external `bird` CLI binary needed - just Node.js. |
| 5 | See scripts/lib/vendor/bird-search/package.json for authoritative version. |
| 6 | """ |
| 7 | |
| 8 | import json |
| 9 | import os |
| 10 | import shutil |
| 11 | import sys |
| 12 | import time |
| 13 | from pathlib import Path |
| 14 | |
| 15 | from . import env, health, http, log, subproc |
| 16 | from datetime import datetime |
| 17 | from typing import Any, Dict, List, Optional, Tuple |
| 18 | |
| 19 | from .relevance import token_overlap_relevance as _compute_relevance |
| 20 | |
| 21 | # How many times to retry the bird-search subprocess when stdout is non-JSON |
| 22 | # (typically an HTML anti-bot interstitial from Twitter's edge). |
| 23 | MAX_JSON_DECODE_RETRIES = 2 |
| 24 | JSON_DECODE_RETRY_DELAY = 5.0 # seconds between retry attempts |
| 25 | |
| 26 | |
| 27 | def _leading_mentions(text: str) -> list: |
| 28 | """Leading-run @mention parse, shared with other X-shaped sources (xquik). |
| 29 | |
| 30 | Thin wrapper over ``query.leading_mentions`` so bird and xquik share one |
| 31 | implementation; kept here for existing call sites and tests. |
| 32 | """ |
| 33 | from .query import leading_mentions |
| 34 | return leading_mentions(text) |
| 35 | |
| 36 | |
| 37 | def _first_of(*values): |
| 38 | """Return first value that is not None.""" |
| 39 | for v in values: |
| 40 | if v is not None: |
| 41 | return v |
| 42 | return None |
| 43 | |
| 44 | # Path to the vendored bird-search wrapper |
| 45 | _BIRD_SEARCH_MJS = Path(__file__).parent / "vendor" / "bird-search" / "bird-search.mjs" |
| 46 | |
| 47 | # Depth configurations: number of results to request |
| 48 | DEPTH_CONFIG = { |
| 49 | "quick": 12, |
| 50 | "default": 30, |
| 51 | "deep": 60, |
| 52 | } |
| 53 | |
| 54 | # Module-level credentials injected from .env config |
| 55 | _credentials: Dict[str, str] = {} |
| 56 | |
| 57 | |
| 58 | def set_credentials(auth_token: Optional[str], ct0: Optional[str]): |
| 59 | """Inject AUTH_TOKEN/CT0 from .env config so Node subprocesses can use them.""" |
| 60 | if auth_token: |
| 61 | _credentials['AUTH_TOKEN'] = auth_token |
| 62 | if ct0: |
| 63 | _credentials['CT0'] = ct0 |
| 64 | |
| 65 | |
| 66 | def _has_injected_credentials() -> bool: |
| 67 | """Return True when both X session cookies were injected from config.""" |
| 68 | return bool(_credentials.get('AUTH_TOKEN') and _credentials.get('CT0')) |
| 69 | |
| 70 | |
| 71 | def _has_process_credentials() -> bool: |
| 72 | """Return True when AUTH_TOKEN/CT0 are present in process env.""" |
| 73 | return bool(env.read_secret_env("AUTH_TOKEN") and env.read_secret_env("CT0")) |
| 74 | |
| 75 | |
| 76 | def _subprocess_env() -> Dict[str, str]: |
| 77 | """Build env dict for Node subprocesses, merging injected credentials.""" |
| 78 | env = os.environ.copy() |
| 79 | env.update(_credentials) |
| 80 | # Hard-disable browser-cookie fallback so normal pipeline runs never hit |
| 81 | # Safari/Chrome Keychain prompts during source detection or search. |
| 82 | env["BIRD_DISABLE_BROWSER_COOKIES"] = "1" |
| 83 | return env |
| 84 | |
| 85 | |
| 86 | def _log(msg: str): |
| 87 | log.source_log("Bird", msg, tty_only=False) |
| 88 | |
| 89 | |
| 90 | def classify_run_failure(detail: str) -> str: |
| 91 | """Map Bird's subprocess-only failure shapes to run outcome states.""" |
| 92 | text = detail.lower() |
| 93 | if any(marker in text for marker in ("interstitial", "non-json", "invalid json")): |
| 94 | return health.SCHEMA_DRIFT |
| 95 | if any( |
| 96 | marker in text |
| 97 | for marker in ("cookie expired", "expired cookie", "unauthorized", "forbidden", "login required") |
| 98 | ): |
| 99 | return health.AUTH_FAILED |
| 100 | return http.classify_failure(message=detail) |
| 101 | |
| 102 | |
| 103 | def _extract_core_subject(topic: str) -> str: |
| 104 | """Extract core subject from verbose query for X search. |
| 105 | |
| 106 | X search is literal keyword AND matching — all words must appear. |
| 107 | Aggressively strip question/meta/research words to keep only the |
| 108 | core product/concept name (max 5 words). |
| 109 | """ |
| 110 | from .query import extract_core_subject |
| 111 | return extract_core_subject(topic, max_words=5, strip_suffixes=True) |
| 112 | |
| 113 | |
| 114 | def _plain_query_tokens(text: str) -> list[str]: |
| 115 | """Return lexical tokens without Bird query grouping syntax.""" |
| 116 | separators = str.maketrans({char: " " for char in '\"“”()[]{}'}) |
| 117 | return [ |
| 118 | clean |
| 119 | for token in text.translate(separators).split() |
| 120 | if (clean := token.strip("'‘’")) |
| 121 | ] |
| 122 | |
| 123 | |
| 124 | def is_bird_installed() -> bool: |
| 125 | """Check if vendored Bird search module is available. |
| 126 | |
| 127 | Returns: |
| 128 | True if bird-search.mjs exists and Node.js is in PATH. |
| 129 | """ |
| 130 | if not _BIRD_SEARCH_MJS.exists(): |
| 131 | return False |
| 132 | return shutil.which("node") is not None |
| 133 | |
| 134 | |
| 135 | def is_bird_authenticated() -> Optional[str]: |
| 136 | """Check if explicit X credentials are available. |
| 137 | |
| 138 | Returns: |
| 139 | Auth source string if authenticated, None otherwise. |
| 140 | """ |
| 141 | if not is_bird_installed(): |
| 142 | return None |
| 143 | |
| 144 | if _has_injected_credentials(): |
| 145 | return "env AUTH_TOKEN" |
| 146 | if _has_process_credentials(): |
| 147 | return "env AUTH_TOKEN" |
| 148 | return None |
| 149 | |
| 150 | |
| 151 | _probe_cache: Optional[Optional[bool]] = "unset" # "unset" | True | False | None |
| 152 | |
| 153 | |
| 154 | def probe_works(timeout: int = 8) -> Optional[bool]: |
| 155 | """Cheap runtime check that X auth actually returns data. |
| 156 | |
| 157 | Returns True when a 1-result probe comes back without an error, False on a |
| 158 | clear failure (auth error / generic search failure), and None when the |
| 159 | result is inconclusive (network timeout) so callers can fail open and keep |
| 160 | the static credential-presence status rather than reporting a false-down. |
| 161 | Cached per process so repeated diagnose calls don't re-probe. |
| 162 | """ |
| 163 | global _probe_cache |
| 164 | if _probe_cache != "unset": |
| 165 | return _probe_cache # type: ignore[return-value] |
| 166 | if not (_has_injected_credentials() or _has_process_credentials()): |
| 167 | _probe_cache = False |
| 168 | return False |
| 169 | from datetime import datetime, timedelta, timezone |
| 170 | since = (datetime.now(timezone.utc) - timedelta(days=30)).strftime("%Y-%m-%d") |
| 171 | # @x (the platform's own account) posts frequently, so a no-error response |
| 172 | # means auth works even if this particular window is quiet. |
| 173 | resp = _run_bird_search(f"from:x since:{since}", count=1, timeout=timeout) |
| 174 | if isinstance(resp, dict) and resp.get("error"): |
| 175 | err = str(resp.get("error")).lower() |
| 176 | if "timed out" in err or "timeout" in err: |
| 177 | _probe_cache = None # inconclusive — don't downgrade on a transient timeout |
| 178 | return None |
| 179 | _probe_cache = False |
| 180 | return False |
| 181 | _probe_cache = True |
| 182 | return True |
| 183 | |
| 184 | |
| 185 | def check_npm_available() -> bool: |
| 186 | """Check if npm is available (kept for API compatibility). |
| 187 | |
| 188 | Returns: |
| 189 | True if 'npm' command is available in PATH, False otherwise. |
| 190 | """ |
| 191 | return shutil.which("npm") is not None |
| 192 | |
| 193 | |
| 194 | def install_bird() -> Tuple[bool, str]: |
| 195 | """No-op. Bird search is vendored in v3.0.0, no installation needed. |
| 196 | |
| 197 | Returns: |
| 198 | Tuple of (success, message). |
| 199 | """ |
| 200 | if is_bird_installed(): |
| 201 | return True, "Bird search is bundled with /last30days v3.0.0 - no installation needed." |
| 202 | if not shutil.which("node"): |
| 203 | return False, "Node.js 22+ is required for X search. Install Node.js first." |
| 204 | return False, f"Vendored bird-search.mjs not found at {_BIRD_SEARCH_MJS}" |
| 205 | |
| 206 | |
| 207 | def get_bird_status() -> Dict[str, Any]: |
| 208 | """Get comprehensive Bird search status. |
| 209 | |
| 210 | Returns: |
| 211 | Dict with keys: installed, authenticated, username, can_install |
| 212 | """ |
| 213 | installed = is_bird_installed() |
| 214 | auth_source = is_bird_authenticated() if installed else None |
| 215 | |
| 216 | return { |
| 217 | "installed": installed, |
| 218 | "authenticated": auth_source is not None, |
| 219 | "username": auth_source, # Now returns auth source (e.g., "Safari", "env AUTH_TOKEN") |
| 220 | "can_install": True, # Always vendored in v3.0.0 |
| 221 | } |
| 222 | |
| 223 | |
| 224 | def _invoke_bird_subprocess(query: str, count: int, timeout: int): |
| 225 | """Invoke the vendored bird-search.mjs subprocess once. |
| 226 | |
| 227 | Returns (result, error_dict). If error_dict is non-None, treat it as the |
| 228 | final result and do not retry — those errors are terminal (timeout, |
| 229 | spawn failure). If error_dict is None, the subprocess ran to completion |
| 230 | and `result` is the SubprocResult; the caller decides whether to retry |
| 231 | based on the result.stdout content. |
| 232 | """ |
| 233 | cmd = [ |
| 234 | "node", str(_BIRD_SEARCH_MJS), |
| 235 | query, |
| 236 | "--count", str(count), |
| 237 | "--json", |
| 238 | ] |
| 239 | |
| 240 | pid_holder: list[int] = [] |
| 241 | |
| 242 | def _register(pid: int) -> None: |
| 243 | pid_holder.append(pid) |
| 244 | try: |
| 245 | from last30days import register_child_pid |
| 246 | register_child_pid(pid) |
| 247 | except ImportError: |
| 248 | pass |
| 249 | |
| 250 | try: |
| 251 | result = subproc.run_with_timeout( |
| 252 | cmd, |
| 253 | timeout=timeout, |
| 254 | env=_subprocess_env(), |
| 255 | on_pid=_register, |
| 256 | ) |
| 257 | except subproc.SubprocTimeout: |
| 258 | return None, {"error": f"Search timed out after {timeout}s", "items": []} |
| 259 | except Exception as e: |
| 260 | return None, {"error": str(e), "items": []} |
| 261 | finally: |
| 262 | if pid_holder: |
| 263 | try: |
| 264 | from last30days import unregister_child_pid |
| 265 | unregister_child_pid(pid_holder[0]) |
| 266 | except Exception: |
| 267 | pass |
| 268 | |
| 269 | return result, None |
| 270 | |
| 271 | |
| 272 | def _run_bird_search(query: str, count: int, timeout: int) -> Dict[str, Any]: |
| 273 | """Run a search using the vendored bird-search.mjs module. |
| 274 | |
| 275 | Retries the subprocess on JSON-decode failure (typically a Twitter |
| 276 | anti-bot HTML interstitial in stdout) up to MAX_JSON_DECODE_RETRIES |
| 277 | times with JSON_DECODE_RETRY_DELAY seconds between attempts. Terminal |
| 278 | errors (subprocess timeout, non-zero return code) are returned |
| 279 | immediately without retry. |
| 280 | |
| 281 | Args: |
| 282 | query: Full search query string (including since: filter) |
| 283 | count: Number of results to request |
| 284 | timeout: Timeout in seconds (per attempt) |
| 285 | |
| 286 | Returns: |
| 287 | Raw Bird JSON response or error dict. |
| 288 | """ |
| 289 | last_decode_error: Optional[str] = None |
| 290 | |
| 291 | for attempt in range(MAX_JSON_DECODE_RETRIES): |
| 292 | result, terminal_error = _invoke_bird_subprocess(query, count, timeout) |
| 293 | if terminal_error is not None: |
| 294 | return terminal_error |
| 295 | |
| 296 | output = result.stdout.strip() |
| 297 | if result.returncode != 0: |
| 298 | if not output: |
| 299 | error = result.stderr.strip() or "Bird search failed" |
| 300 | return {"error": error, "items": []} |
| 301 | # Windows/Node 24: the vendored Bird CLI uses native fetch (undici), |
| 302 | # and calling process.exit() while keep-alive sockets are still |
| 303 | # closing trips a libuv assertion -> non-zero exit code AFTER it has |
| 304 | # already written a complete, valid JSON result to stdout. Trust |
| 305 | # stdout when it has content; only treat a non-zero exit as a real |
| 306 | # failure when stdout is empty. |
| 307 | |
| 308 | if not output: |
| 309 | return {"items": []} |
| 310 | |
| 311 | try: |
| 312 | parsed = json.loads(output) |
| 313 | except json.JSONDecodeError as e: |
| 314 | # Twitter's edge sometimes serves an HTML anti-bot interstitial |
| 315 | # in place of JSON. Tag the failure shape so it's distinguishable |
| 316 | # from "no results" in logs, then retry the subprocess. |
| 317 | looks_html = output.lstrip().lower().startswith(("<!doctype", "<html", "<")) |
| 318 | attempt_num = attempt + 1 |
| 319 | log_msg = ( |
| 320 | f"Bird search returned non-JSON stdout " |
| 321 | f"(looks_html={looks_html}, attempt {attempt_num}/{MAX_JSON_DECODE_RETRIES}, " |
| 322 | f"first 80 chars: {output[:80]!r})" |
| 323 | ) |
| 324 | last_decode_error = str(e) |
| 325 | if attempt_num < MAX_JSON_DECODE_RETRIES: |
| 326 | log.source_log( |
| 327 | "X/bird", |
| 328 | f"{log_msg}; retrying in {JSON_DECODE_RETRY_DELAY:.0f}s", |
| 329 | tty_only=False, |
| 330 | ) |
| 331 | time.sleep(JSON_DECODE_RETRY_DELAY) |
| 332 | continue |
| 333 | log.source_log("X/bird", log_msg, tty_only=False) |
| 334 | return { |
| 335 | "error": ( |
| 336 | f"Invalid JSON response after {MAX_JSON_DECODE_RETRIES} attempts " |
| 337 | f"(likely Twitter anti-bot interstitial): {e}" |
| 338 | ), |
| 339 | "items": [], |
| 340 | } |
| 341 | |
| 342 | if isinstance(parsed, list): |
| 343 | return {"items": parsed} |
| 344 | return parsed |
| 345 | |
| 346 | # Defensive fallthrough — loop should always return above. |
| 347 | return { |
| 348 | "error": f"Bird search exhausted retries: {last_decode_error}", |
| 349 | "items": [], |
| 350 | } |
| 351 | |
| 352 | |
| 353 | def search_x( |
| 354 | topic: str, |
| 355 | from_date: str, |
| 356 | to_date: str, |
| 357 | depth: str = "default", |
| 358 | ) -> Dict[str, Any]: |
| 359 | """Search X using Bird CLI with automatic retry on 0 results. |
| 360 | |
| 361 | Args: |
| 362 | topic: Search topic |
| 363 | from_date: Start date (YYYY-MM-DD) |
| 364 | to_date: End date (YYYY-MM-DD) - unused but kept for API compatibility |
| 365 | depth: Research depth - "quick", "default", or "deep" |
| 366 | |
| 367 | Returns: |
| 368 | Raw Bird JSON response or error dict. |
| 369 | """ |
| 370 | count = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 371 | timeout = 30 if depth == "quick" else 45 if depth == "default" else 60 |
| 372 | |
| 373 | # Extract core subject - X search is literal, not semantic |
| 374 | core_words = _plain_query_tokens(_extract_core_subject(topic)) |
| 375 | core_topic = " ".join(core_words) |
| 376 | query = f"{core_topic} since:{from_date}" |
| 377 | |
| 378 | _log(f"Searching: {query}") |
| 379 | response = _run_bird_search(query, count, timeout) |
| 380 | last_clean_response = response if not response.get("error") else None |
| 381 | |
| 382 | # Check if we got results |
| 383 | items = parse_bird_response(response, query=core_topic) |
| 384 | |
| 385 | # Retry with OR groups for multi-word queries (X supports OR operator) |
| 386 | if not items and len(core_words) >= 2: |
| 387 | from .query import extract_compound_terms |
| 388 | compounds = extract_compound_terms(topic) |
| 389 | if compounds: |
| 390 | # Build OR-group query: ("multi-agent" OR "agent simulation") since:DATE |
| 391 | or_parts = ' OR '.join(f'"{t}"' for t in compounds[:3]) |
| 392 | _log(f"0 results for '{core_topic}', retrying with OR groups: {or_parts}") |
| 393 | query = f"({or_parts}) since:{from_date}" |
| 394 | response = _run_bird_search(query, count, timeout) |
| 395 | if not response.get("error"): |
| 396 | last_clean_response = response |
| 397 | items = parse_bird_response(response, query=core_topic) |
| 398 | |
| 399 | # Retry with fewer keywords if still 0 results and query has 3+ words |
| 400 | if not items and len(core_words) > 2: |
| 401 | shorter = ' '.join(core_words[:2]) |
| 402 | _log(f"0 results for '{core_topic}', retrying with '{shorter}'") |
| 403 | query = f"{shorter} since:{from_date}" |
| 404 | response = _run_bird_search(query, count, timeout) |
| 405 | if not response.get("error"): |
| 406 | last_clean_response = response |
| 407 | items = parse_bird_response(response, query=core_topic) |
| 408 | |
| 409 | # Last-chance retry: use strongest remaining token (often the product name) |
| 410 | if not items and core_words: |
| 411 | low_signal = { |
| 412 | 'trendiest', 'trending', 'hottest', 'hot', 'popular', 'viral', |
| 413 | 'best', 'top', 'latest', 'new', 'plugin', 'plugins', |
| 414 | 'skill', 'skills', 'tool', 'tools', |
| 415 | } |
| 416 | candidates = [w for w in core_words if w not in low_signal] |
| 417 | if candidates: |
| 418 | # Keep an entity anchor (the first distinctive topic token) in the |
| 419 | # retry so it can't collapse to a bare generic token like "compound" |
| 420 | # and flood the X pool with off-topic noise. Add the strongest |
| 421 | # (longest) distinctive token when it differs from the anchor; |
| 422 | # otherwise query the anchor alone. Better to return 0 than to |
| 423 | # over-broaden to an unanchored generic term. |
| 424 | anchor = candidates[0] |
| 425 | strongest = max(candidates, key=len) |
| 426 | retry_terms = anchor if strongest == anchor else f"{anchor} {strongest}" |
| 427 | _log(f"0 results for '{core_topic}', retrying anchored on '{retry_terms}'") |
| 428 | query = f"{retry_terms} since:{from_date}" |
| 429 | response = _run_bird_search(query, count, timeout) |
| 430 | if not response.get("error"): |
| 431 | last_clean_response = response |
| 432 | |
| 433 | if response.get("error") and last_clean_response is not None: |
| 434 | _log("Optional retry failed after a clean empty response; preserving no-results outcome") |
| 435 | return last_clean_response |
| 436 | return response |
| 437 | |
| 438 | |
| 439 | def search_handles( |
| 440 | handles: List[str], |
| 441 | topic: Optional[str], |
| 442 | from_date: str, |
| 443 | count_per: int = 5, |
| 444 | ) -> List[Dict[str, Any]]: |
| 445 | """Search specific X handles for topic-related content. |
| 446 | |
| 447 | Pulls each handle's actual timeline via `from:handle since:` — the FROM |
| 448 | lane (tweets BY the person), engagement-weighted downstream. The topic is |
| 449 | used for relevance RANKING, never AND'd into the query: X search is literal, |
| 450 | so `from:handle <their name>` only matched tweets where they wrote their own |
| 451 | name and returned ~0. Used in Phase 2 after entity extraction. |
| 452 | |
| 453 | Args: |
| 454 | handles: List of X handles to search (without @) |
| 455 | topic: Search topic — used for relevance ranking only, not the query |
| 456 | from_date: Start date (YYYY-MM-DD) |
| 457 | count_per: Results to request per handle |
| 458 | |
| 459 | Returns: |
| 460 | List of raw item dicts (same format as parse_bird_response output). |
| 461 | """ |
| 462 | core_topic = _extract_core_subject(topic) if topic else None |
| 463 | |
| 464 | def _search_one_handle(handle: str) -> List[Dict[str, Any]]: |
| 465 | handle = handle.lstrip("@") |
| 466 | # Always unfiltered: pull the timeline, rank by topic relevance below. |
| 467 | query = f"from:{handle} since:{from_date}" |
| 468 | |
| 469 | cmd = [ |
| 470 | "node", str(_BIRD_SEARCH_MJS), |
| 471 | query, |
| 472 | "--count", str(count_per), |
| 473 | "--json", |
| 474 | ] |
| 475 | |
| 476 | try: |
| 477 | result = subproc.run_with_timeout(cmd, timeout=15, env=_subprocess_env()) |
| 478 | except subproc.SubprocTimeout: |
| 479 | _log(f"Handle search timed out for @{handle}") |
| 480 | return [] |
| 481 | except OSError as e: |
| 482 | _log(f"Handle search error for @{handle}: {e}") |
| 483 | return [] |
| 484 | |
| 485 | output = result.stdout.strip() |
| 486 | if result.returncode != 0: |
| 487 | if not output: |
| 488 | _log(f"Handle search failed for @{handle}: {result.stderr.strip()}") |
| 489 | return [] |
| 490 | # Windows/Node 24: benign libuv assertion can cause non-zero exit |
| 491 | # AFTER valid JSON is written to stdout. Trust stdout content. |
| 492 | |
| 493 | if not output: |
| 494 | return [] |
| 495 | |
| 496 | try: |
| 497 | response = json.loads(output) |
| 498 | except json.JSONDecodeError: |
| 499 | _log(f"Invalid JSON from handle search for @{handle}") |
| 500 | return [] |
| 501 | items = parse_bird_response(response, query=core_topic) |
| 502 | # Log on success/empty too (not only on failure): a silent handle search |
| 503 | # made the from: query look like it never ran and caused wrong diagnoses. |
| 504 | _log(f"Searching: {query} -> {len(items)} results") |
| 505 | return items |
| 506 | |
| 507 | from concurrent.futures import ThreadPoolExecutor, as_completed |
| 508 | |
| 509 | all_items: List[Dict[str, Any]] = [] |
| 510 | with ThreadPoolExecutor(max_workers=min(5, len(handles))) as executor: |
| 511 | futures = {executor.submit(_search_one_handle, h): h for h in handles} |
| 512 | for future in as_completed(futures): |
| 513 | all_items.extend(future.result()) |
| 514 | |
| 515 | return all_items |
| 516 | |
| 517 | |
| 518 | def search_mentions( |
| 519 | handles: List[str], |
| 520 | from_date: str, |
| 521 | count_per: int = 5, |
| 522 | ) -> List[Dict[str, Any]]: |
| 523 | """Search for tweets ABOUT/TO each handle — the mention lane. |
| 524 | |
| 525 | Queries `@handle since:` (tweets that mention the account) and excludes the |
| 526 | handle's OWN tweets (those belong to the FROM lane via search_handles), so |
| 527 | this surfaces what OTHERS are saying about the person. Engagement-weighted |
| 528 | downstream; deduped against the FROM lane by URL at normalize time. |
| 529 | |
| 530 | Args: |
| 531 | handles: List of X handles (without @) |
| 532 | from_date: Start date (YYYY-MM-DD) |
| 533 | count_per: Results to request per handle |
| 534 | |
| 535 | Returns: |
| 536 | List of raw item dicts (same format as parse_bird_response output). |
| 537 | """ |
| 538 | def _search_one(handle: str) -> List[Dict[str, Any]]: |
| 539 | handle = handle.lstrip("@") |
| 540 | query = f"@{handle} since:{from_date}" |
| 541 | cmd = [ |
| 542 | "node", str(_BIRD_SEARCH_MJS), |
| 543 | query, |
| 544 | "--count", str(count_per), |
| 545 | "--json", |
| 546 | ] |
| 547 | try: |
| 548 | result = subproc.run_with_timeout(cmd, timeout=15, env=_subprocess_env()) |
| 549 | except subproc.SubprocTimeout: |
| 550 | _log(f"Mention search timed out for @{handle}") |
| 551 | return [] |
| 552 | except OSError as e: |
| 553 | _log(f"Mention search error for @{handle}: {e}") |
| 554 | return [] |
| 555 | if result.returncode != 0: |
| 556 | _log(f"Mention search failed for @{handle}: {result.stderr.strip()}") |
| 557 | return [] |
| 558 | output = result.stdout.strip() |
| 559 | if not output: |
| 560 | return [] |
| 561 | try: |
| 562 | response = json.loads(output) |
| 563 | except json.JSONDecodeError: |
| 564 | _log(f"Invalid JSON from mention search for @{handle}") |
| 565 | return [] |
| 566 | items = parse_bird_response(response, query=None) |
| 567 | # ABOUT lane = OTHERS mentioning the handle. Drop the handle's own tweets |
| 568 | # (the FROM lane already covers those); identify by the status URL author. |
| 569 | hl = handle.lower() |
| 570 | # The Bird API may return either x.com or twitter.com permalinks, so |
| 571 | # match both when excluding the handle's own tweets. |
| 572 | def _is_own(url): |
| 573 | u = (url or "").lower() |
| 574 | return f"x.com/{hl}/status" in u or f"twitter.com/{hl}/status" in u |
| 575 | about = [it for it in items if not _is_own(it.get("url"))] |
| 576 | _log(f"Searching: {query} -> {len(about)} mentions") |
| 577 | return about |
| 578 | |
| 579 | from concurrent.futures import ThreadPoolExecutor, as_completed |
| 580 | |
| 581 | all_items: List[Dict[str, Any]] = [] |
| 582 | with ThreadPoolExecutor(max_workers=min(5, len(handles))) as executor: |
| 583 | futures = {executor.submit(_search_one, h): h for h in handles} |
| 584 | for future in as_completed(futures): |
| 585 | all_items.extend(future.result()) |
| 586 | return all_items |
| 587 | |
| 588 | |
| 589 | def parse_bird_response(response: Dict[str, Any], query: str = "") -> List[Dict[str, Any]]: |
| 590 | """Parse Bird response to match xai_x output format. |
| 591 | |
| 592 | Args: |
| 593 | response: Raw Bird JSON response |
| 594 | query: Original search query for relevance scoring |
| 595 | |
| 596 | Returns: |
| 597 | List of normalized item dicts matching xai_x.parse_x_response() format. |
| 598 | """ |
| 599 | items = [] |
| 600 | |
| 601 | # Check for errors |
| 602 | if "error" in response and response["error"]: |
| 603 | _log(f"Bird error: {response['error']}") |
| 604 | return items |
| 605 | |
| 606 | # Bird returns a list of tweets directly or under a key |
| 607 | raw_items = response if isinstance(response, list) else response.get("items", response.get("tweets", [])) |
| 608 | |
| 609 | if not isinstance(raw_items, list): |
| 610 | return items |
| 611 | |
| 612 | for i, tweet in enumerate(raw_items): |
| 613 | if not isinstance(tweet, dict): |
| 614 | continue |
| 615 | |
| 616 | # Extract URL - Bird uses permanent_url or we construct from id |
| 617 | url = tweet.get("permanent_url") or tweet.get("url", "") |
| 618 | if not url and tweet.get("id"): |
| 619 | # Try different field structures Bird might use |
| 620 | author = tweet.get("author", {}) or tweet.get("user", {}) |
| 621 | screen_name = author.get("username") or author.get("screen_name", "") |
| 622 | if screen_name: |
| 623 | url = f"https://x.com/{screen_name}/status/{tweet['id']}" |
| 624 | |
| 625 | if not url: |
| 626 | continue |
| 627 | |
| 628 | # Parse date from created_at/createdAt (e.g., "Wed Jan 15 14:30:00 +0000 2026") |
| 629 | date = None |
| 630 | created_at = tweet.get("createdAt") or tweet.get("created_at", "") |
| 631 | if created_at: |
| 632 | try: |
| 633 | # Try ISO format first (e.g., "2026-02-03T22:33:32Z") |
| 634 | # Check for ISO date separator, not just "T" (which appears in "Tue") |
| 635 | if len(created_at) > 10 and created_at[10] == "T": |
| 636 | dt = datetime.fromisoformat(created_at.replace("Z", "+00:00")) |
| 637 | else: |
| 638 | # Twitter format: "Wed Jan 15 14:30:00 +0000 2026" |
| 639 | dt = datetime.strptime(created_at, "%a %b %d %H:%M:%S %z %Y") |
| 640 | date = dt.strftime("%Y-%m-%d") |
| 641 | except (ValueError, TypeError): |
| 642 | pass |
| 643 | |
| 644 | # Extract user info (Bird uses author.username, older format uses user.screen_name) |
| 645 | author = tweet.get("author", {}) or tweet.get("user", {}) |
| 646 | author_handle = author.get("username") or author.get("screen_name", "") or tweet.get("author_handle", "") |
| 647 | |
| 648 | # Build engagement dict (Bird uses camelCase: likeCount, retweetCount, etc.) |
| 649 | engagement = { |
| 650 | "likes": _first_of(tweet.get("likeCount"), tweet.get("like_count"), tweet.get("favorite_count")), |
| 651 | "reposts": _first_of(tweet.get("retweetCount"), tweet.get("retweet_count")), |
| 652 | "replies": _first_of(tweet.get("replyCount"), tweet.get("reply_count")), |
| 653 | "quotes": _first_of(tweet.get("quoteCount"), tweet.get("quote_count")), |
| 654 | } |
| 655 | # Convert to int where possible |
| 656 | for key in engagement: |
| 657 | if engagement[key] is not None: |
| 658 | try: |
| 659 | engagement[key] = int(engagement[key]) |
| 660 | except (ValueError, TypeError): |
| 661 | engagement[key] = None |
| 662 | |
| 663 | # Build normalized item |
| 664 | text = str(tweet.get("text", tweet.get("full_text", ""))).strip()[:500] |
| 665 | item = { |
| 666 | "id": f"X{i+1}", |
| 667 | "text": text, |
| 668 | "url": url, |
| 669 | "author_handle": author_handle.lstrip("@"), |
| 670 | # Leading @mentions parsed from the post text identify who a reply is |
| 671 | # directed at (X replies open with the target handle(s)). Used by the |
| 672 | # interaction-signal classifier in rerank. |
| 673 | "mentioned_handles": _leading_mentions(text), |
| 674 | "date": date, |
| 675 | "engagement": engagement if any(v is not None for v in engagement.values()) else None, |
| 676 | "why_relevant": "", # Bird doesn't provide relevance explanations |
| 677 | "relevance": _compute_relevance(query, str(tweet.get("text", ""))) if query else 0.7, |
| 678 | } |
| 679 | |
| 680 | items.append(item) |
| 681 | |
| 682 | return items |
| 683 |