| 1 | """Direct X API v2 backend (``xapi``) and the shared v2 response parser. |
| 2 | |
| 3 | ``xapi`` searches X through ``api.x.com/2`` with an app-only bearer token |
| 4 | (``X_BEARER_TOKEN``). Full-archive search is tried first; when the developer |
| 5 | project is not enrolled for it (HTTP 403 with an enrollment marker), the |
| 6 | search retries once against recent search with the window clamped to the |
| 7 | last seven days and the truncation named in the result. |
| 8 | |
| 9 | This module also owns the one X API v2 parser (``parse_v2_response``) that |
| 10 | ``xurl_x`` delegates to, plus the snowflake, handle-grammar, and |
| 11 | generated-sequence helpers that ``grok_x`` imports back. |
| 12 | |
| 13 | Security contract: the bearer travels only in the Authorization |
| 14 | header; every failure becomes an engine-authored fixed string chosen by |
| 15 | status code plus a marker match on the body, and the response body, reason |
| 16 | phrase, and headers never enter an error string, a log line, or an |
| 17 | exception. Topic and handle text is sanitized before it enters a query, and |
| 18 | the research window is always carried by request parameters. |
| 19 | """ |
| 20 | |
| 21 | from __future__ import annotations |
| 22 | |
| 23 | import re |
| 24 | import time |
| 25 | from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait |
| 26 | from datetime import datetime, timedelta, timezone |
| 27 | from typing import Any, Callable, Dict, List, Optional |
| 28 | |
| 29 | from . import health, http, log |
| 30 | from .relevance import token_overlap_relevance as _compute_relevance |
| 31 | |
| 32 | _API_BASE = "https://api.x.com/2" |
| 33 | _SEARCH_ALL_URL = f"{_API_BASE}/tweets/search/all" |
| 34 | _SEARCH_RECENT_URL = f"{_API_BASE}/tweets/search/recent" |
| 35 | |
| 36 | # Depth configurations: number of posts to collect per query (shared with |
| 37 | # xurl_x, which re-imports this table). The API minimum per page is 10. |
| 38 | DEPTH_CONFIG = { |
| 39 | "quick": 10, |
| 40 | "default": 30, |
| 41 | "deep": 60, |
| 42 | } |
| 43 | |
| 44 | # X API v2 caps a search query at 512 characters (Basic) and pages at 10..100. |
| 45 | MAX_QUERY_CHARS = 512 |
| 46 | MIN_PAGE_RESULTS = 10 |
| 47 | MAX_PAGE_RESULTS = 100 |
| 48 | # Recent search reaches back seven days; end_time must be at least ten |
| 49 | # seconds before now, so a "today" window keeps a small safety margin. |
| 50 | RECENT_WINDOW_DAYS = 7 |
| 51 | END_TIME_SAFETY_SECONDS = 30 |
| 52 | TIMEOUT_SECONDS = 30 |
| 53 | RETRIES = 2 |
| 54 | # Wall-clock budget for one search (all pages, including the recent-search |
| 55 | # fallback). A throttled or slow walk returns what it has instead of holding |
| 56 | # the whole run; the source's per-request timeout still bounds each page. |
| 57 | DEADLINE_SECONDS = 90 |
| 58 | # One budget shared by every supplemental handle lane of a run (explicit |
| 59 | # from, extracted from, about, related), started once by the caller: the |
| 60 | # point is to bound the total, not each part. |
| 61 | LANE_BUDGET_SECONDS = 150.0 |
| 62 | |
| 63 | TRUNCATION_DETAIL = "window truncated to 7 days" |
| 64 | # Receipt when the lane budget stopped a search before ``count`` posts were |
| 65 | # collected: the posts kept are real, the coverage is not complete. |
| 66 | DEADLINE_DETAIL = "search stopped at the lane deadline; results may be incomplete" |
| 67 | |
| 68 | # The one description of what an app-only bearer can reach. Doctor and the |
| 69 | # prescriptions quote it; the bearer path is never described as parity with |
| 70 | # the connector lane. |
| 71 | BEARER_COVERAGE_NOTE = ( |
| 72 | "recent posts, about the last week, unless your X developer project has " |
| 73 | "full-archive access" |
| 74 | ) |
| 75 | |
| 76 | # Engine-authored fixed error strings. Each carries a marker that |
| 77 | # http.classify_failure recognizes, because the X retrieval branch classifies |
| 78 | # by message text only. |
| 79 | ERR_PAYMENT_REQUIRED = "xapi: payment required (X API credits exhausted)" |
| 80 | ERR_UNAUTHORIZED = "xapi: unauthorized (bearer token rejected)" |
| 81 | ERR_FORBIDDEN = "xapi: forbidden (bearer token lacks access)" |
| 82 | ERR_RATE_LIMITED = "xapi: rate limit exceeded (X API)" |
| 83 | ERR_TIMED_OUT = "xapi: timed out" |
| 84 | ERR_UNREACHABLE = "xapi: connection error (X API unreachable)" |
| 85 | ERR_NO_TOKEN = "No X_BEARER_TOKEN configured" |
| 86 | ERR_EMPTY_QUERY = "xapi: empty query after sanitizing" |
| 87 | |
| 88 | # Failures that end every remaining lane for the run (the key itself is the |
| 89 | # problem, not this request). |
| 90 | _FATAL_ERRORS = frozenset({ERR_PAYMENT_REQUIRED, ERR_UNAUTHORIZED, ERR_FORBIDDEN}) |
| 91 | |
| 92 | # A 403 whose body carries one of these is "this project is not enrolled for |
| 93 | # full-archive search", which recent search can still serve. |
| 94 | _ENROLLMENT_MARKERS = ( |
| 95 | "client-not-enrolled", |
| 96 | "not enrolled", |
| 97 | "enrolled", |
| 98 | "access level", |
| 99 | "subset of", |
| 100 | ) |
| 101 | # X-specific body markers for credit exhaustion, matched on any status on |
| 102 | # top of http.classify_failure's shared payment-required markers: X reports |
| 103 | # a depleted pay-per-use balance under more than one code and title. |
| 104 | _X_CREDIT_MARKERS = ("creditsdepleted", "credits depleted") |
| 105 | |
| 106 | # Twitter/X snowflake epoch (2010-11-04T01:42:54.657Z) in milliseconds. |
| 107 | _SNOWFLAKE_EPOCH_MS = 1288834974657 |
| 108 | |
| 109 | # X's real handle grammar. Handles are interpolated into post URLs and into |
| 110 | # search operators (from:, @), so anything outside this charset is rejected |
| 111 | # rather than passed through. entity_extract applies the same rule to |
| 112 | # @mentions. |
| 113 | _HANDLE_RE = re.compile(r"[A-Za-z0-9_]{1,15}") |
| 114 | |
| 115 | # Grouping syntax that carries no lexical meaning in a topic, mirroring |
| 116 | # bird_x. Double quotes are handled separately: xapi strips them all and |
| 117 | # wraps the core in exactly one phrase pair, so a planner quote can never |
| 118 | # close the phrase early. |
| 119 | _GROUPING_CHARS = "“”()[]{}" |
| 120 | _QUOTE_CHARS = "\"„‟" |
| 121 | _APOSTROPHES = "'‘’" |
| 122 | |
| 123 | |
| 124 | def _log(msg: str) -> None: |
| 125 | log.source_log("xapi", msg, tty_only=False) |
| 126 | |
| 127 | |
| 128 | def _utcnow() -> datetime: |
| 129 | """Current UTC time; a seam tests patch to pin the window arithmetic.""" |
| 130 | return datetime.now(timezone.utc) |
| 131 | |
| 132 | |
| 133 | # --------------------------------------------------------------------------- |
| 134 | # Shared helpers (moved here from grok_x; grok_x imports them back) |
| 135 | # --------------------------------------------------------------------------- |
| 136 | |
| 137 | |
| 138 | def _decode_snowflake(post_id: str) -> Optional[datetime]: |
| 139 | """Recover a post's creation time from its id, with no network call.""" |
| 140 | try: |
| 141 | value = int(str(post_id).strip()) |
| 142 | except (TypeError, ValueError): |
| 143 | return None |
| 144 | if value <= 0: |
| 145 | return None |
| 146 | try: |
| 147 | return datetime.fromtimestamp( |
| 148 | ((value >> 22) + _SNOWFLAKE_EPOCH_MS) / 1000, tz=timezone.utc |
| 149 | ) |
| 150 | except (OverflowError, OSError, ValueError): |
| 151 | return None |
| 152 | |
| 153 | |
| 154 | def _looks_generated(ids: List[str]) -> bool: |
| 155 | """True when ids form a near-uniform arithmetic run. |
| 156 | |
| 157 | Real ranked results are not evenly spaced in time. A fabricated set often |
| 158 | is, because a model interpolates a plausible-looking id sequence. Four |
| 159 | ids is the minimum used here: three gaps are needed before a near-uniform |
| 160 | step reads as generated rather than coincidental. |
| 161 | """ |
| 162 | numeric = [] |
| 163 | for pid in ids: |
| 164 | try: |
| 165 | numeric.append(int(pid)) |
| 166 | except (TypeError, ValueError): |
| 167 | return False |
| 168 | if len(numeric) < 4: |
| 169 | return False |
| 170 | numeric.sort() |
| 171 | gaps = [b - a for a, b in zip(numeric, numeric[1:])] |
| 172 | if any(g <= 0 for g in gaps): |
| 173 | return False |
| 174 | mean = sum(gaps) / len(gaps) |
| 175 | if mean <= 0: |
| 176 | return False |
| 177 | # Every gap within 5% of the mean is not something real timelines do. |
| 178 | return all(abs(g - mean) / mean < 0.05 for g in gaps) |
| 179 | |
| 180 | |
| 181 | def _clean_handle(value: str) -> str: |
| 182 | """Return a grammar-valid handle, or '' when the value is not one.""" |
| 183 | candidate = str(value or "").strip().lstrip("@") |
| 184 | return candidate if _HANDLE_RE.fullmatch(candidate) else "" |
| 185 | |
| 186 | |
| 187 | # --------------------------------------------------------------------------- |
| 188 | # Parser |
| 189 | # --------------------------------------------------------------------------- |
| 190 | |
| 191 | |
| 192 | def _safe_int(value: Any) -> Optional[int]: |
| 193 | if value is None: |
| 194 | return None |
| 195 | try: |
| 196 | return int(value) |
| 197 | except (ValueError, TypeError): |
| 198 | return None |
| 199 | |
| 200 | |
| 201 | def parse_v2_response( |
| 202 | response: Dict[str, Any], |
| 203 | topic: str = "", |
| 204 | window: Optional[tuple[str, str]] = None, |
| 205 | *, |
| 206 | id_prefix: str = "XAPI", |
| 207 | index_offset: int = 0, |
| 208 | seen_ids: Optional[set[str]] = None, |
| 209 | ) -> List[Dict[str, Any]]: |
| 210 | """Parse an X API v2 search response into the engine's X item shape. |
| 211 | |
| 212 | Output matches the XItem schema used by xai_x and bird_x (id, text, url, |
| 213 | author_handle, date, engagement, mentioned_handles, why_relevant, |
| 214 | relevance) plus ``post_id`` (the numeric snowflake as a string). The |
| 215 | citation URL is rebuilt from the id and username; URL fields in the |
| 216 | response are never used. A post whose author is not in |
| 217 | ``includes.users`` (or whose username fails the handle grammar) keeps |
| 218 | the ``https://x.com/i/status/<id>`` form with an empty handle. |
| 219 | |
| 220 | ``window`` is an optional ``(from_date, to_date)`` pair of YYYY-MM-DD |
| 221 | strings; dated posts outside it are dropped. ``seen_ids`` and |
| 222 | ``index_offset`` let several calls share one accumulator with unique ids. |
| 223 | """ |
| 224 | items: List[Dict[str, Any]] = [] |
| 225 | if not isinstance(response, dict): |
| 226 | return items |
| 227 | if "error" in response: |
| 228 | _log("error in response; no items parsed") |
| 229 | return items |
| 230 | |
| 231 | data = response.get("data") or [] |
| 232 | if not isinstance(data, list) or not data: |
| 233 | return items |
| 234 | |
| 235 | authors: Dict[str, str] = {} |
| 236 | for user in (response.get("includes") or {}).get("users") or []: |
| 237 | if isinstance(user, dict) and user.get("id") is not None: |
| 238 | authors[str(user["id"])] = _clean_handle(user.get("username", "")) |
| 239 | |
| 240 | from .query import leading_mentions |
| 241 | |
| 242 | window_from, window_to = (window or (None, None)) |
| 243 | seen = seen_ids if seen_ids is not None else set() |
| 244 | |
| 245 | for tweet in data: |
| 246 | if not isinstance(tweet, dict): |
| 247 | continue |
| 248 | tweet_id = str(tweet.get("id") or "").strip() |
| 249 | if not tweet_id.isdigit(): |
| 250 | continue |
| 251 | if tweet_id in seen: |
| 252 | continue |
| 253 | |
| 254 | username = authors.get(str(tweet.get("author_id") or ""), "") |
| 255 | if username: |
| 256 | url = f"https://x.com/{username}/status/{tweet_id}" |
| 257 | else: |
| 258 | url = f"https://x.com/i/status/{tweet_id}" |
| 259 | |
| 260 | note = tweet.get("note_tweet") |
| 261 | text = "" |
| 262 | if isinstance(note, dict) and note.get("text"): |
| 263 | text = str(note["text"]) |
| 264 | else: |
| 265 | text = str(tweet.get("text") or "") |
| 266 | text = text.strip() |
| 267 | |
| 268 | engagement: Optional[Dict[str, Any]] = None |
| 269 | metrics = tweet.get("public_metrics") or {} |
| 270 | if isinstance(metrics, dict) and metrics: |
| 271 | engagement = { |
| 272 | "likes": metrics.get("like_count", 0), |
| 273 | "reposts": metrics.get("retweet_count", 0), |
| 274 | "replies": metrics.get("reply_count", 0), |
| 275 | "quotes": metrics.get("quote_count", 0), |
| 276 | } |
| 277 | bookmarks = _safe_int(metrics.get("bookmark_count")) |
| 278 | if bookmarks is not None: |
| 279 | engagement["bookmarks"] = bookmarks |
| 280 | views = _safe_int(metrics.get("impression_count")) |
| 281 | if views is not None: |
| 282 | engagement["views"] = views |
| 283 | |
| 284 | date: Optional[str] = None |
| 285 | created = str(tweet.get("created_at") or "") |
| 286 | if created: |
| 287 | m = re.match(r"(\d{4}-\d{2}-\d{2})", created) |
| 288 | if m: |
| 289 | date = m.group(1) |
| 290 | if date and window_from and date < window_from: |
| 291 | continue |
| 292 | if date and window_to and date > window_to: |
| 293 | continue |
| 294 | |
| 295 | seen.add(tweet_id) |
| 296 | items.append({ |
| 297 | "id": f"{id_prefix}{index_offset + len(items) + 1}", |
| 298 | "text": text[:500], |
| 299 | "url": url, |
| 300 | "author_handle": username, |
| 301 | "date": date, |
| 302 | "engagement": engagement, |
| 303 | "mentioned_handles": leading_mentions(text), |
| 304 | "why_relevant": "", |
| 305 | "relevance": _compute_relevance(topic, text) if topic else 0.5, |
| 306 | "post_id": tweet_id, |
| 307 | }) |
| 308 | |
| 309 | return items |
| 310 | |
| 311 | |
| 312 | # --------------------------------------------------------------------------- |
| 313 | # Query compilation |
| 314 | # --------------------------------------------------------------------------- |
| 315 | |
| 316 | |
| 317 | def _topic_tokens(topic: str) -> List[str]: |
| 318 | """Sanitized topic tokens: no quotes, grouping, operators, or negation.""" |
| 319 | separators = str.maketrans({char: " " for char in _GROUPING_CHARS + _QUOTE_CHARS}) |
| 320 | cleaned = str(topic or "").translate(separators) |
| 321 | tokens: List[str] = [] |
| 322 | for token in cleaned.split(): |
| 323 | clean = token.strip(_APOSTROPHES) |
| 324 | if not clean: |
| 325 | continue |
| 326 | if ":" in clean or clean.startswith("-"): |
| 327 | continue |
| 328 | if clean.upper() in ("OR", "AND"): |
| 329 | continue |
| 330 | tokens.append(clean) |
| 331 | return tokens |
| 332 | |
| 333 | |
| 334 | def _compile(tokens: List[str]) -> str: |
| 335 | return f'"{" ".join(tokens)}" -is:retweet' |
| 336 | |
| 337 | |
| 338 | def build_query(topic: str) -> str: |
| 339 | """Compile a topic into one quoted phrase plus ``-is:retweet``. |
| 340 | |
| 341 | Mirrors ``bird_x.build_topic_query`` (grouping characters become spaces, |
| 342 | quotes are balanced) and additionally drops every operator-shaped token |
| 343 | (anything with a colon, anything starting with ``-``, bare OR/AND) so a |
| 344 | topic can never smuggle ``from:``, ``since:``, or a phrase break into the |
| 345 | request. The window is never part of the query (see ``_window``). The |
| 346 | result stays under ``MAX_QUERY_CHARS``, cut at a token boundary. Returns |
| 347 | "" when nothing lexical survives. |
| 348 | """ |
| 349 | tokens = _topic_tokens(topic) |
| 350 | if not tokens: |
| 351 | return "" |
| 352 | compiled = _compile(tokens) |
| 353 | while len(compiled) > MAX_QUERY_CHARS and len(tokens) > 1: |
| 354 | tokens.pop() |
| 355 | compiled = _compile(tokens) |
| 356 | if len(compiled) > MAX_QUERY_CHARS: |
| 357 | # One token longer than the whole budget: keep as much of it as fits. |
| 358 | overhead = len(_compile([""])) |
| 359 | tokens = [tokens[0][: MAX_QUERY_CHARS - overhead]] |
| 360 | compiled = _compile(tokens) |
| 361 | return compiled |
| 362 | |
| 363 | |
| 364 | # --------------------------------------------------------------------------- |
| 365 | # Window |
| 366 | # --------------------------------------------------------------------------- |
| 367 | |
| 368 | |
| 369 | def _parse_day(value: str) -> Optional[datetime]: |
| 370 | try: |
| 371 | return datetime.strptime(str(value)[:10], "%Y-%m-%d").replace(tzinfo=timezone.utc) |
| 372 | except (TypeError, ValueError): |
| 373 | return None |
| 374 | |
| 375 | |
| 376 | def _iso(when: datetime) -> str: |
| 377 | return when.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") |
| 378 | |
| 379 | |
| 380 | def _window(from_date: str, to_date: str) -> tuple[str, str]: |
| 381 | """``(start_time, end_time)`` for the request, ISO 8601 with Z. |
| 382 | |
| 383 | ``start_time`` is midnight UTC of ``from_date``; ``end_time`` is the end |
| 384 | of ``to_date`` or, when that would be now or later ("today"), now minus a |
| 385 | safety margin. |
| 386 | """ |
| 387 | now = _utcnow() |
| 388 | start = _parse_day(from_date) or (now - timedelta(days=30)) |
| 389 | end_day = _parse_day(to_date) |
| 390 | latest = now - timedelta(seconds=END_TIME_SAFETY_SECONDS) |
| 391 | if end_day is None: |
| 392 | end = latest |
| 393 | else: |
| 394 | end = min(end_day + timedelta(hours=23, minutes=59, seconds=59), latest) |
| 395 | if start >= end: |
| 396 | start = end - timedelta(days=1) |
| 397 | return _iso(start), _iso(end) |
| 398 | |
| 399 | |
| 400 | def _recent_floor() -> str: |
| 401 | """The earliest ``start_time`` recent search accepts, with a margin.""" |
| 402 | return _iso( |
| 403 | _utcnow() - timedelta(days=RECENT_WINDOW_DAYS) + timedelta(seconds=END_TIME_SAFETY_SECONDS) |
| 404 | ) |
| 405 | |
| 406 | |
| 407 | # --------------------------------------------------------------------------- |
| 408 | # Transport |
| 409 | # --------------------------------------------------------------------------- |
| 410 | |
| 411 | |
| 412 | class _XApiFailure(Exception): |
| 413 | """A request failure already reduced to its fixed string. |
| 414 | |
| 415 | ``enrollment`` marks the one 403 that recent search can still serve. |
| 416 | The message is always one of the ERR_* literals (or ``xapi: http <n>``), |
| 417 | never response text. |
| 418 | """ |
| 419 | |
| 420 | def __init__(self, message: str, *, enrollment: bool = False): |
| 421 | super().__init__(message) |
| 422 | self.enrollment = enrollment |
| 423 | |
| 424 | |
| 425 | def _failure_for(exc: http.HTTPError) -> _XApiFailure: |
| 426 | status = getattr(exc, "status_code", None) |
| 427 | body = str(getattr(exc, "body", "") or "").lower() |
| 428 | if ( |
| 429 | status == 402 |
| 430 | or http.classify_failure(message=body) == health.PAYMENT_REQUIRED |
| 431 | or any(marker in body for marker in _X_CREDIT_MARKERS) |
| 432 | ): |
| 433 | return _XApiFailure(ERR_PAYMENT_REQUIRED) |
| 434 | if status == 401: |
| 435 | return _XApiFailure(ERR_UNAUTHORIZED) |
| 436 | if status == 403: |
| 437 | enrolled = any(marker in body for marker in _ENROLLMENT_MARKERS) |
| 438 | return _XApiFailure(ERR_FORBIDDEN, enrollment=enrolled) |
| 439 | if status == 429: |
| 440 | return _XApiFailure(ERR_RATE_LIMITED) |
| 441 | state = getattr(exc, "outcome_state", None) |
| 442 | if status == 408 or state == health.TIMEOUT: |
| 443 | return _XApiFailure(ERR_TIMED_OUT) |
| 444 | if status: |
| 445 | return _XApiFailure(f"xapi: http {status}") |
| 446 | if state == health.UNREACHABLE: |
| 447 | return _XApiFailure(ERR_UNREACHABLE) |
| 448 | return _XApiFailure("xapi: request failed (HTTPError)") |
| 449 | |
| 450 | |
| 451 | def _get( |
| 452 | token: str, url: str, params: Dict[str, Any], deadline: Optional[float] = None, |
| 453 | ) -> Dict[str, Any]: |
| 454 | """One authenticated GET; every failure surfaces as ``_XApiFailure``. |
| 455 | |
| 456 | ``deadline`` is handed to the transport as its wall deadline, so a |
| 457 | request started just before it cannot keep the full per-request timeout |
| 458 | and retry cycle: the socket wait and every backoff sleep stop at the |
| 459 | deadline (``http.DeadlineExceeded`` -> ``ERR_TIMED_OUT``). |
| 460 | """ |
| 461 | try: |
| 462 | response = http.get( |
| 463 | url, |
| 464 | headers={"Authorization": f"Bearer {token}"}, |
| 465 | params=params, |
| 466 | timeout=TIMEOUT_SECONDS, |
| 467 | retries=RETRIES, |
| 468 | deadline_monotonic=deadline, |
| 469 | ) |
| 470 | except http.HTTPError as exc: |
| 471 | raise _failure_for(exc) from None |
| 472 | except TimeoutError: |
| 473 | raise _XApiFailure(ERR_TIMED_OUT) from None |
| 474 | except Exception as exc: # noqa: BLE001 - the message must never leak |
| 475 | raise _XApiFailure(f"xapi: request failed ({type(exc).__name__})") from None |
| 476 | return response if isinstance(response, dict) else {} |
| 477 | |
| 478 | |
| 479 | def _page_size(count: int) -> int: |
| 480 | return max(MIN_PAGE_RESULTS, min(MAX_PAGE_RESULTS, int(count))) |
| 481 | |
| 482 | |
| 483 | def _search_pages( |
| 484 | token: str, |
| 485 | url: str, |
| 486 | params: Dict[str, Any], |
| 487 | count: int, |
| 488 | deadline: Optional[float] = None, |
| 489 | ) -> Dict[str, Any]: |
| 490 | """Follow ``next_token`` until ``count`` posts are collected. |
| 491 | |
| 492 | ``deadline`` is a ``time.monotonic()`` instant: the first page always |
| 493 | runs; a later page is skipped once the deadline has passed and the |
| 494 | posts collected so far are returned with ``"truncated": True`` so the |
| 495 | caller can report the incomplete coverage. |
| 496 | """ |
| 497 | data: List[Dict[str, Any]] = [] |
| 498 | users: Dict[str, Dict[str, Any]] = {} |
| 499 | truncated = False |
| 500 | page_params = dict(params) |
| 501 | page_params["max_results"] = _page_size(count) |
| 502 | # Bound the walk even when every page carries a next_token. |
| 503 | max_pages = max(1, -(-count // MIN_PAGE_RESULTS)) |
| 504 | for page_index in range(max_pages): |
| 505 | if page_index and deadline is not None and time.monotonic() >= deadline: |
| 506 | _log(f"search deadline ({DEADLINE_SECONDS}s) reached; keeping {len(data)} posts") |
| 507 | truncated = True |
| 508 | break |
| 509 | # A fresh dict per page: the transport must never see a later |
| 510 | # page's next_token on an earlier request. |
| 511 | try: |
| 512 | response = _get(token, url, dict(page_params), deadline) |
| 513 | except _XApiFailure as exc: |
| 514 | if data and str(exc) == ERR_TIMED_OUT: |
| 515 | # The budget ran out mid-walk: the pages already collected |
| 516 | # are the result, reported as truncated, not a failure. |
| 517 | _log(f"search deadline reached mid-walk; keeping {len(data)} posts") |
| 518 | truncated = True |
| 519 | break |
| 520 | raise |
| 521 | page = response.get("data") or [] |
| 522 | if isinstance(page, list): |
| 523 | data.extend(t for t in page if isinstance(t, dict)) |
| 524 | for user in (response.get("includes") or {}).get("users") or []: |
| 525 | if isinstance(user, dict) and user.get("id") is not None: |
| 526 | users[str(user["id"])] = user |
| 527 | next_token = (response.get("meta") or {}).get("next_token") |
| 528 | if not next_token or len(data) >= count: |
| 529 | break |
| 530 | page_params["next_token"] = next_token |
| 531 | return { |
| 532 | "data": data[:count], |
| 533 | "includes": {"users": list(users.values())}, |
| 534 | "truncated": truncated, |
| 535 | } |
| 536 | |
| 537 | |
| 538 | def _run_search( |
| 539 | token: str, |
| 540 | query: str, |
| 541 | from_date: str, |
| 542 | to_date: str, |
| 543 | count: int, |
| 544 | *, |
| 545 | topic: str, |
| 546 | id_prefix: str, |
| 547 | label: str, |
| 548 | deadline: Optional[float] = None, |
| 549 | ) -> Dict[str, Any]: |
| 550 | """Full-archive search with the recent-search fallback. |
| 551 | |
| 552 | ``deadline`` (a ``time.monotonic()`` instant) is shared with the caller's |
| 553 | other lanes; without one the search gets its own ``DEADLINE_SECONDS``. |
| 554 | A shared deadline that has already passed sends no request at all. |
| 555 | |
| 556 | Returns ``{"items": [...]}`` (plus ``"warning"`` when the window was |
| 557 | truncated or the deadline stopped the walk early) or ``{"items": [], |
| 558 | "error": <fixed string>}``. |
| 559 | """ |
| 560 | start, end = _window(from_date, to_date) |
| 561 | params = { |
| 562 | "query": query, |
| 563 | "start_time": start, |
| 564 | "end_time": end, |
| 565 | "sort_order": "recency", |
| 566 | "expansions": "author_id", |
| 567 | "tweet.fields": "created_at,public_metrics,note_tweet,entities", |
| 568 | "user.fields": "username", |
| 569 | } |
| 570 | if deadline is not None and time.monotonic() >= deadline: |
| 571 | _log(f"{label}: lane budget ({LANE_BUDGET_SECONDS:.0f}s) exhausted before the search started") |
| 572 | return {"items": [], "warning": DEADLINE_DETAIL} |
| 573 | _log(f"Searching: {label}") |
| 574 | warnings: List[str] = [] |
| 575 | if deadline is None: |
| 576 | deadline = time.monotonic() + DEADLINE_SECONDS |
| 577 | try: |
| 578 | response = _search_pages(token, _SEARCH_ALL_URL, params, count, deadline) |
| 579 | except _XApiFailure as exc: |
| 580 | if not exc.enrollment: |
| 581 | _log(f"{label}: {exc}") |
| 582 | return {"items": [], "error": str(exc)} |
| 583 | _log(f"{label}: full-archive search not enrolled; retrying recent search ({TRUNCATION_DETAIL})") |
| 584 | floor = _recent_floor() |
| 585 | if floor >= end: |
| 586 | # The whole window predates what recent search can reach: an |
| 587 | # empty, truncated result, never a start_time after end_time. |
| 588 | _log(f"{label}: window ends before the recent-search floor; nothing to fetch") |
| 589 | return {"items": [], "warning": TRUNCATION_DETAIL} |
| 590 | params["start_time"] = max(start, floor) |
| 591 | try: |
| 592 | response = _search_pages(token, _SEARCH_RECENT_URL, params, count, deadline) |
| 593 | except _XApiFailure as exc2: |
| 594 | _log(f"{label}: {exc2}") |
| 595 | return {"items": [], "error": str(exc2)} |
| 596 | warnings.append(TRUNCATION_DETAIL) |
| 597 | if response.get("truncated"): |
| 598 | warnings.append(DEADLINE_DETAIL) |
| 599 | items = parse_v2_response(response, topic, (from_date, to_date), id_prefix=id_prefix) |
| 600 | result: Dict[str, Any] = {"items": items} |
| 601 | if warnings: |
| 602 | result["warning"] = "; ".join(warnings) |
| 603 | return result |
| 604 | |
| 605 | |
| 606 | # --------------------------------------------------------------------------- |
| 607 | # Public search entry points |
| 608 | # --------------------------------------------------------------------------- |
| 609 | |
| 610 | |
| 611 | def search_x( |
| 612 | token: str, |
| 613 | query: str, |
| 614 | from_date: str, |
| 615 | to_date: str, |
| 616 | depth: str = "default", |
| 617 | ) -> Dict[str, Any]: |
| 618 | """Topic search via X API v2. |
| 619 | |
| 620 | Returns ``{"items": [...]}`` or ``{"items": [], "error": "..."}`` (the |
| 621 | xquik shape); ``"warning"`` carries the truncation detail after the |
| 622 | recent-search fallback. |
| 623 | """ |
| 624 | if not token: |
| 625 | return {"items": [], "error": ERR_NO_TOKEN} |
| 626 | count = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 627 | compiled = build_query(query) |
| 628 | if not compiled: |
| 629 | _log("topic compiled to an empty query after sanitizing") |
| 630 | return {"items": [], "error": ERR_EMPTY_QUERY} |
| 631 | return _run_search( |
| 632 | token, compiled, from_date, to_date, count, |
| 633 | topic=query, id_prefix="XAPI", label="topic", |
| 634 | ) |
| 635 | |
| 636 | |
| 637 | def is_own_post(url: str, handle: str) -> bool: |
| 638 | """True when a post URL is authored by ``handle`` (their own post). |
| 639 | |
| 640 | The ABOUT lanes (here, xquik, the host envelope) drop the subject's own |
| 641 | posts so only mentions *by others* remain. Handles both x.com and |
| 642 | twitter.com permalinks. |
| 643 | """ |
| 644 | u = (url or "").lower() |
| 645 | h = handle.lower().lstrip("@").strip() |
| 646 | return bool(h) and (f"x.com/{h}/status" in u or f"twitter.com/{h}/status" in u) |
| 647 | |
| 648 | |
| 649 | |
| 650 | def _lane_handle(raw: str, lane: str) -> str: |
| 651 | handle = _clean_handle(raw) |
| 652 | if not handle: |
| 653 | _log(f"skipping {lane} lane for a handle outside the X handle grammar") |
| 654 | return handle |
| 655 | |
| 656 | |
| 657 | _MAX_LANE_WORKERS = 5 |
| 658 | |
| 659 | |
| 660 | def _run_handle_lanes( |
| 661 | handles: List[str], |
| 662 | search_one: Callable[[str], Dict[str, Any]], |
| 663 | *, |
| 664 | id_prefix: str, |
| 665 | keep: Optional[Callable[[Dict[str, Any], str], bool]] = None, |
| 666 | warnings: Optional[List[str]] = None, |
| 667 | ) -> List[Dict[str, Any]]: |
| 668 | """Run one search per handle on a bounded pool and merge in handle order. |
| 669 | |
| 670 | Per-handle receipts (``"warning"`` on a result, e.g. a deadline stop) |
| 671 | are appended to ``warnings`` once each, so the caller can report |
| 672 | incomplete lane coverage instead of presenting it as complete. |
| 673 | |
| 674 | Mirrors ``bird_x.search_handles``: at most five handles in flight. A |
| 675 | fatal auth/payment result stops further handles from being scheduled; |
| 676 | whatever was already fetched is kept. Results are merged on the calling |
| 677 | thread in ``handles`` order, deduped by post id across handles, and |
| 678 | renumbered ``<id_prefix><n>`` so the output is deterministic regardless |
| 679 | of completion order. ``keep(item, handle)`` filters a handle's items |
| 680 | after the post id is recorded, so a dropped post still dedupes later. |
| 681 | """ |
| 682 | results: List[Optional[Dict[str, Any]]] = [None] * len(handles) |
| 683 | if handles: |
| 684 | max_workers = min(_MAX_LANE_WORKERS, len(handles)) |
| 685 | fatal = False |
| 686 | with ThreadPoolExecutor(max_workers=max_workers) as pool: |
| 687 | pending: Dict[Any, int] = {} |
| 688 | next_index = 0 |
| 689 | while pending or (next_index < len(handles) and not fatal): |
| 690 | while next_index < len(handles) and not fatal and len(pending) < max_workers: |
| 691 | pending[pool.submit(search_one, handles[next_index])] = next_index |
| 692 | next_index += 1 |
| 693 | done, _ = wait(list(pending), return_when=FIRST_COMPLETED) |
| 694 | for future in done: |
| 695 | index = pending.pop(future) |
| 696 | result = future.result() |
| 697 | results[index] = result |
| 698 | if result.get("error") in _FATAL_ERRORS: |
| 699 | fatal = True # the key itself failed; keep what we have |
| 700 | |
| 701 | items: List[Dict[str, Any]] = [] |
| 702 | seen_ids: set[str] = set() |
| 703 | for handle, result in zip(handles, results): |
| 704 | note = (result or {}).get("warning") |
| 705 | if note and warnings is not None and note not in warnings: |
| 706 | warnings.append(note) |
| 707 | for item in (result or {}).get("items", []): |
| 708 | post_id = item.get("post_id", "") |
| 709 | if post_id in seen_ids: |
| 710 | continue |
| 711 | seen_ids.add(post_id) |
| 712 | if keep is not None and not keep(item, handle): |
| 713 | continue |
| 714 | item["id"] = f"{id_prefix}{len(items) + 1}" |
| 715 | items.append(item) |
| 716 | return items |
| 717 | |
| 718 | |
| 719 | def search_handles( |
| 720 | handles: List[str], |
| 721 | topic: str, |
| 722 | from_date: str, |
| 723 | to_date: str, |
| 724 | *, |
| 725 | count_per: int = 8, |
| 726 | token: str = "", |
| 727 | deadline: Optional[float] = None, |
| 728 | warnings: Optional[List[str]] = None, |
| 729 | ) -> List[Dict[str, Any]]: |
| 730 | """FROM lane: posts authored BY each handle (their own timeline). |
| 731 | |
| 732 | The topic is NOT AND'd into the query; it ranks relevance only (mirrors |
| 733 | ``xquik.search_handles``). A handle outside the X grammar skips its lane |
| 734 | with a receipt line. |
| 735 | """ |
| 736 | if not token or not handles: |
| 737 | return [] |
| 738 | clean = [h for h in (_lane_handle(raw, "from:") for raw in handles) if h] |
| 739 | |
| 740 | def _search_one(handle: str) -> Dict[str, Any]: |
| 741 | return _run_search( |
| 742 | token, f"from:{handle} -is:retweet", from_date, to_date, count_per, |
| 743 | topic=topic, id_prefix="XF", label=f"from:{handle}", deadline=deadline, |
| 744 | ) |
| 745 | |
| 746 | return _run_handle_lanes(clean, _search_one, id_prefix="XF", warnings=warnings) |
| 747 | |
| 748 | |
| 749 | def search_mentions( |
| 750 | handles: List[str], |
| 751 | from_date: str, |
| 752 | to_date: str, |
| 753 | *, |
| 754 | topic: str = "", |
| 755 | count_per: int = 5, |
| 756 | token: str = "", |
| 757 | deadline: Optional[float] = None, |
| 758 | warnings: Optional[List[str]] = None, |
| 759 | ) -> List[Dict[str, Any]]: |
| 760 | """ABOUT lane: posts mentioning each handle, authored by OTHERS. |
| 761 | |
| 762 | The query is ``@handle -from:handle`` and the handle's own posts are |
| 763 | dropped again client-side (``is_own_post``) so only third-party mentions |
| 764 | remain. |
| 765 | """ |
| 766 | if not token or not handles: |
| 767 | return [] |
| 768 | clean = [h for h in (_lane_handle(raw, "mention") for raw in handles) if h] |
| 769 | |
| 770 | def _search_one(handle: str) -> Dict[str, Any]: |
| 771 | return _run_search( |
| 772 | token, f"@{handle} -from:{handle} -is:retweet", from_date, to_date, count_per, |
| 773 | topic=topic, id_prefix="XA", label=f"@{handle}", deadline=deadline, |
| 774 | ) |
| 775 | |
| 776 | return _run_handle_lanes( |
| 777 | clean, _search_one, id_prefix="XA", |
| 778 | keep=lambda item, handle: not is_own_post(item.get("url", ""), handle), |
| 779 | warnings=warnings, |
| 780 | ) |
| 781 |