| 1 | """Host-fetched X envelope: the ``--x-posts`` lane. |
| 2 | |
| 3 | The hosting model fetches posts through its own X connector and writes them |
| 4 | to a ``.json`` file of flat rows. The engine ingests that file as its X |
| 5 | source for one run: strict at the envelope level (a malformed, stale, or |
| 6 | off-topic file fails closed with an exit-2 contract error), lenient per row |
| 7 | (a bad row is dropped and counted), and trusting nothing a row asserts about |
| 8 | itself beyond a numeric id and a grammar-valid handle. |
| 9 | |
| 10 | Shape (``schema: "last30days-x-posts/1"``):: |
| 11 | |
| 12 | { |
| 13 | "schema": "last30days-x-posts/1", |
| 14 | "generated_at": "<ISO-8601>", |
| 15 | "topic": "<the run topic>", |
| 16 | "window": {"from": "YYYY-MM-DD", "to": "YYYY-MM-DD"}, |
| 17 | "provider": "<free text>", |
| 18 | "status": "ok" | "partial" | "error", |
| 19 | "error": "<short category: credits | not-connected | unavailable | window-unsupported>", |
| 20 | "calls": [{"lane": "topic|from|mention|related", "handles": [...], "posts": [ROW, ...]}] |
| 21 | } |
| 22 | |
| 23 | A ROW is a flat object with exactly ``id``, ``author_handle``, ``created_at``, |
| 24 | ``text``, ``likes``, ``reposts``, ``replies``, ``quotes``. Any other key is |
| 25 | ignored (and counted once per row as ``extra-fields``); in particular a |
| 26 | row-supplied ``url`` is never used for the citation, only cross-checked. |
| 27 | |
| 28 | The envelope is single-serve: the first X subquery takes the topic-lane rows |
| 29 | and the handle-lane section takes the lane calls; later X subqueries, |
| 30 | judge-retry, and thin-retry get nothing (``take_topic`` / ``take_lanes`` |
| 31 | return ``None`` once consumed). |
| 32 | |
| 33 | Security contract: contract errors name the path and the field, never a |
| 34 | field value; the envelope's raw ``error`` text reaches stderr only under |
| 35 | ``LAST30DAYS_DEBUG``; every outcome detail is one of the fixed strings below. |
| 36 | """ |
| 37 | |
| 38 | from __future__ import annotations |
| 39 | |
| 40 | import hashlib |
| 41 | import json |
| 42 | import os |
| 43 | import re |
| 44 | import stat |
| 45 | import threading |
| 46 | from dataclasses import dataclass, field |
| 47 | from datetime import date, datetime, timedelta, timezone |
| 48 | from pathlib import Path |
| 49 | from typing import Any |
| 50 | |
| 51 | from . import env, health, log, schema |
| 52 | from .relevance import token_overlap_relevance as _compute_relevance |
| 53 | from .query import leading_mentions |
| 54 | from .x_api import _clean_handle, _decode_snowflake, _looks_generated, is_own_post |
| 55 | |
| 56 | SCHEMA = "last30days-x-posts/1" |
| 57 | ID_PREFIX = "XHOST" |
| 58 | |
| 59 | # Input bounds. Named so the recipe and the receipts can quote them. |
| 60 | MAX_BYTES = 8 * 1024 * 1024 |
| 61 | MAX_CALLS = 20 |
| 62 | MAX_ROWS_PER_CALL = 500 |
| 63 | MAX_ROWS_TOTAL = 1000 |
| 64 | MAX_TEXT_CHARS = 10_000 |
| 65 | MAX_AGE_SECONDS = 6 * 60 * 60 |
| 66 | MAX_ID_DIGITS = 20 |
| 67 | |
| 68 | LANES = ("topic", "from", "mention", "related") |
| 69 | STATUSES = ("ok", "partial", "error") |
| 70 | ROW_FIELDS = frozenset( |
| 71 | {"id", "author_handle", "created_at", "text", "likes", "reposts", "replies", "quotes"} |
| 72 | ) |
| 73 | COUNTERS = ( |
| 74 | "out-of-window", |
| 75 | "missing-id-text", |
| 76 | "uncitable", |
| 77 | "date-mismatch", |
| 78 | "handle-mismatch", |
| 79 | "duplicate", |
| 80 | "lane-mismatch", |
| 81 | "truncated", |
| 82 | "extra-fields", |
| 83 | ) |
| 84 | |
| 85 | # Engine-authored fixed outcome details. The envelope's own error text |
| 86 | # never becomes a detail string. |
| 87 | DETAIL_CREDITS = "X connector reported no credits" |
| 88 | DETAIL_NOT_CONNECTED = "X connector not connected" |
| 89 | DETAIL_UNAVAILABLE = "X connector unavailable" |
| 90 | DETAIL_ERROR = "X connector error" |
| 91 | DETAIL_GENERATED = "X connector rows rejected (id sequence looks generated)" |
| 92 | DETAIL_PARTIAL = "X connector returned partial results" |
| 93 | DETAIL_NOT_PASSED = "connector result not passed" |
| 94 | |
| 95 | CATEGORY_CREDITS = "credits" |
| 96 | CATEGORY_NOT_CONNECTED = "not-connected" |
| 97 | CATEGORY_UNAVAILABLE = "unavailable" |
| 98 | CATEGORY_WINDOW_UNSUPPORTED = "window-unsupported" |
| 99 | _CATEGORY_GENERATED = "generated" |
| 100 | |
| 101 | _REMEDY = ( |
| 102 | "Rewrite the file to the last30days-x-posts/1 shape, or re-run without --x-posts." |
| 103 | ) |
| 104 | |
| 105 | # Directories the envelope must never be read from: the engine's own config |
| 106 | # dir (which holds .env) and the credential stores of the X tooling and the |
| 107 | # usual shell neighbours. Checked after realpath so a symlink cannot reach in. |
| 108 | _CREDENTIAL_STORES = ( |
| 109 | "~/.xurl", |
| 110 | "~/.grok", |
| 111 | "~/.config/last30days", |
| 112 | "~/.config/gh", |
| 113 | "~/.aws", |
| 114 | "~/.ssh", |
| 115 | "~/.netrc", |
| 116 | ) |
| 117 | |
| 118 | # Snowflake ids predate nothing before X's epoch; a decode outside |
| 119 | # [2010-11-04, now + 1 day] is not a real post id. |
| 120 | _SNOWFLAKE_FLOOR = datetime(2010, 11, 4, tzinfo=timezone.utc) |
| 121 | |
| 122 | # C0 and C1 control characters. Text keeps newlines and tabs; handles and the |
| 123 | # error field keep nothing. |
| 124 | _CONTROL_ANY_RE = re.compile(r"[\x00-\x1f\x7f-\x9f]") |
| 125 | _CONTROL_TEXT_RE = re.compile(r"[\x00-\x08\x0b-\x1f\x7f-\x9f]") |
| 126 | _PLACEHOLDER_HANDLES = frozenset( |
| 127 | {"unknown", "n/a", "na", "none", "null", "user", "anonymous", "placeholder", "handle"} |
| 128 | ) |
| 129 | _STATUS_URL_RE = re.compile( |
| 130 | r"^https?://(?:www\.|mobile\.)?(?:x\.com|twitter\.com)/(?P<handle>[A-Za-z0-9_]{1,15})/status(?:es)?/(?P<id>\d+)", |
| 131 | re.IGNORECASE, |
| 132 | ) |
| 133 | _TOPIC_STRIP = "\"'“”‘’ \t" |
| 134 | |
| 135 | |
| 136 | class EnvelopeContractError(Exception): |
| 137 | """The envelope failed its contract: unreadable, out of bounds, malformed, |
| 138 | stale, off-topic, or outside the research window. The CLI maps this to |
| 139 | exit code 2. The message names the path and the offending field only.""" |
| 140 | |
| 141 | def __init__(self, message: str) -> None: |
| 142 | super().__init__(message) |
| 143 | self.message = message |
| 144 | |
| 145 | |
| 146 | @dataclass |
| 147 | class EnvelopeCall: |
| 148 | """One validated handle-lane call (``from``, ``mention``, or ``related``).""" |
| 149 | |
| 150 | index: int |
| 151 | lane: str |
| 152 | handles: list[str] |
| 153 | posts: list[dict[str, Any]] |
| 154 | |
| 155 | |
| 156 | @dataclass |
| 157 | class Envelope: |
| 158 | """A validated ``--x-posts`` file, ready to be served once.""" |
| 159 | |
| 160 | path: str |
| 161 | sha256: str |
| 162 | status: str |
| 163 | error_category: str |
| 164 | provider: str |
| 165 | window: tuple[str, str] |
| 166 | topic_items: list[dict[str, Any]] |
| 167 | lane_calls: list[EnvelopeCall] |
| 168 | counters: dict[str, int] |
| 169 | accepted: int |
| 170 | total: int |
| 171 | lane_counts: dict[str, int] |
| 172 | call_lanes: list[str] |
| 173 | warnings: list[str] = field(default_factory=list) |
| 174 | notes: list[str] = field(default_factory=list) |
| 175 | _lock: threading.Lock = field(default_factory=threading.Lock, repr=False, compare=False) |
| 176 | _topic_served: bool = field(default=False, repr=False, compare=False) |
| 177 | _lanes_served: bool = field(default=False, repr=False, compare=False) |
| 178 | |
| 179 | def take_topic(self) -> list[dict[str, Any]] | None: |
| 180 | """Topic-lane rows on the first call; ``None`` once consumed.""" |
| 181 | with self._lock: |
| 182 | if self._topic_served: |
| 183 | return None |
| 184 | self._topic_served = True |
| 185 | return list(self.topic_items) |
| 186 | |
| 187 | def take_lanes(self) -> list[EnvelopeCall] | None: |
| 188 | """Handle-lane calls on the first call; ``None`` once consumed.""" |
| 189 | with self._lock: |
| 190 | if self._lanes_served: |
| 191 | return None |
| 192 | self._lanes_served = True |
| 193 | return list(self.lane_calls) |
| 194 | |
| 195 | def outcome(self) -> tuple[schema.RunOutcomeState, str] | None: |
| 196 | """The fixed source outcome for a non-ok envelope, else ``None``.""" |
| 197 | if self.status == "error": |
| 198 | return _error_outcome(self.error_category) |
| 199 | if self.status == "partial": |
| 200 | detail = DETAIL_PARTIAL |
| 201 | if self.error_category: |
| 202 | detail += f": {self.error_category}" |
| 203 | if self.call_lanes: |
| 204 | detail += f" (calls: {', '.join(self.call_lanes)})" |
| 205 | return schema.PARTIAL, detail |
| 206 | return None |
| 207 | |
| 208 | def receipt(self) -> str: |
| 209 | dropped = ", ".join( |
| 210 | f"{name} {count}" for name, count in self.counters.items() if count |
| 211 | ) or "none" |
| 212 | lanes = ", ".join(f"{lane} {self.lane_counts.get(lane, 0)}" for lane in LANES) |
| 213 | line = ( |
| 214 | f"host-fetched X: accepted {self.accepted} of {self.total} " |
| 215 | f"(dropped: {dropped}); lanes: {lanes}" |
| 216 | ) |
| 217 | if self.status != "ok": |
| 218 | line += f"; status {self.status}" |
| 219 | if self.error_category: |
| 220 | line += f" ({self.error_category})" |
| 221 | if self.notes: |
| 222 | line += "; notes: " + "; ".join(self.notes) |
| 223 | return line |
| 224 | |
| 225 | |
| 226 | def _error_outcome(category: str) -> tuple[schema.RunOutcomeState, str]: |
| 227 | if category == CATEGORY_CREDITS: |
| 228 | return schema.PAYMENT_REQUIRED, DETAIL_CREDITS |
| 229 | if category == CATEGORY_NOT_CONNECTED: |
| 230 | return health.ERROR, DETAIL_NOT_CONNECTED |
| 231 | if category == CATEGORY_UNAVAILABLE: |
| 232 | return health.ERROR, DETAIL_UNAVAILABLE |
| 233 | if category == _CATEGORY_GENERATED: |
| 234 | return health.ERROR, DETAIL_GENERATED |
| 235 | return health.ERROR, DETAIL_ERROR |
| 236 | |
| 237 | |
| 238 | # --------------------------------------------------------------------------- |
| 239 | # Helpers |
| 240 | # --------------------------------------------------------------------------- |
| 241 | |
| 242 | |
| 243 | def _stat(path: Path) -> os.stat_result: |
| 244 | """Seam for tests: the size and type check runs before any read.""" |
| 245 | return os.stat(path) |
| 246 | |
| 247 | |
| 248 | def _log(msg: str) -> None: |
| 249 | log.source_log("x", msg, tty_only=False) |
| 250 | |
| 251 | |
| 252 | def _protected_roots() -> list[Path]: |
| 253 | roots: list[Path] = [] |
| 254 | for candidate in (env.CONFIG_DIR, getattr(env, "CONFIG_FILE", None)): |
| 255 | if candidate: |
| 256 | roots.append(Path(candidate)) |
| 257 | for store in _CREDENTIAL_STORES: |
| 258 | roots.append(Path(store).expanduser()) |
| 259 | return [Path(os.path.realpath(root)) for root in roots] |
| 260 | |
| 261 | |
| 262 | def _inside_protected(real: Path) -> bool: |
| 263 | for root in _protected_roots(): |
| 264 | if real == root or root in real.parents: |
| 265 | return True |
| 266 | return False |
| 267 | |
| 268 | |
| 269 | def _normalize_topic(value: str) -> str: |
| 270 | cleaned = _CONTROL_ANY_RE.sub("", str(value or "")) |
| 271 | cleaned = cleaned.strip(_TOPIC_STRIP).casefold() |
| 272 | return " ".join(cleaned.split()) |
| 273 | |
| 274 | |
| 275 | def _parse_day(value: Any) -> date | None: |
| 276 | if not isinstance(value, str): |
| 277 | return None |
| 278 | try: |
| 279 | return date.fromisoformat(value.strip()[:10]) |
| 280 | except ValueError: |
| 281 | return None |
| 282 | |
| 283 | |
| 284 | def _parse_iso(value: str) -> datetime | None: |
| 285 | text = value.strip() |
| 286 | if text.endswith(("Z", "z")): |
| 287 | text = text[:-1] + "+00:00" |
| 288 | try: |
| 289 | parsed = datetime.fromisoformat(text) |
| 290 | except ValueError: |
| 291 | return None |
| 292 | if parsed.tzinfo is None: |
| 293 | parsed = parsed.replace(tzinfo=timezone.utc) |
| 294 | return parsed |
| 295 | |
| 296 | |
| 297 | def _category(raw: str) -> str: |
| 298 | """The short error class: first token, lowercased, [a-z0-9-] only.""" |
| 299 | cleaned = _CONTROL_ANY_RE.sub("", raw or "").strip().lower() |
| 300 | if not cleaned: |
| 301 | return "" |
| 302 | token = cleaned.split()[0] |
| 303 | token = re.sub(r"[^a-z0-9-]", "", token) |
| 304 | return token[:40] |
| 305 | |
| 306 | |
| 307 | def _clean_text(value: str) -> str: |
| 308 | """Strip C0/C1 controls (newlines and tabs survive) and defang markup. |
| 309 | |
| 310 | Envelope text is model-written and post-authored, so the two Markdown |
| 311 | constructs that stay active in the saved report are neutralised without |
| 312 | hiding the characters: a link tail ``](`` becomes ``]\\(`` (renders as |
| 313 | the same text, never as a link) and an HTML comment opener ``<!--`` |
| 314 | becomes ``<! --`` (never a comment, so a pasted META marker cannot be |
| 315 | promoted). Raw tags are left to the HTML renderer's escaping. |
| 316 | """ |
| 317 | text = value.replace("\r\n", "\n").replace("\r", "\n") |
| 318 | text = _CONTROL_TEXT_RE.sub("", text) |
| 319 | return text.replace("](", "]\\(").replace("<!--", "<! --") |
| 320 | |
| 321 | |
| 322 | def _int(value: Any) -> int: |
| 323 | if isinstance(value, bool): |
| 324 | return 0 |
| 325 | try: |
| 326 | return max(0, int(value)) |
| 327 | except (TypeError, ValueError): |
| 328 | return 0 |
| 329 | |
| 330 | |
| 331 | def _post_id(value: Any) -> str: |
| 332 | if isinstance(value, bool): |
| 333 | return "" |
| 334 | if isinstance(value, int): |
| 335 | candidate = str(value) |
| 336 | elif isinstance(value, str): |
| 337 | candidate = value.strip() |
| 338 | else: |
| 339 | return "" |
| 340 | if not candidate.isdigit() or len(candidate) > MAX_ID_DIGITS: |
| 341 | return "" |
| 342 | return candidate |
| 343 | |
| 344 | |
| 345 | def _handle(value: Any) -> str: |
| 346 | if not isinstance(value, str) or _CONTROL_ANY_RE.search(value): |
| 347 | return "" |
| 348 | handle = _clean_handle(value) |
| 349 | if handle.lower() in _PLACEHOLDER_HANDLES: |
| 350 | return "" |
| 351 | return handle |
| 352 | |
| 353 | |
| 354 | def _clean_handles(value: Any) -> list[str] | None: |
| 355 | """Grammar-valid, lowercased handles; ``None`` when any entry fails.""" |
| 356 | if not isinstance(value, list): |
| 357 | return None |
| 358 | cleaned: list[str] = [] |
| 359 | for raw in value: |
| 360 | handle = _handle(raw) |
| 361 | if not handle: |
| 362 | return None |
| 363 | if handle.lower() not in cleaned: |
| 364 | cleaned.append(handle.lower()) |
| 365 | return cleaned |
| 366 | |
| 367 | |
| 368 | # --------------------------------------------------------------------------- |
| 369 | # Reader |
| 370 | # --------------------------------------------------------------------------- |
| 371 | |
| 372 | |
| 373 | # Clock skew a host may legitimately show; anything further ahead is a |
| 374 | # stamp chosen to outlive the six-hour freshness gate. |
| 375 | FUTURE_SKEW_SECONDS = 5 * 60 |
| 376 | |
| 377 | |
| 378 | def _generated_in_future(value: Any) -> bool: |
| 379 | """True when ``generated_at`` is more than ``FUTURE_SKEW_SECONDS`` ahead of now.""" |
| 380 | try: |
| 381 | stamp = datetime.fromisoformat(str(value)) |
| 382 | except (TypeError, ValueError): |
| 383 | return False |
| 384 | if stamp.tzinfo is None: |
| 385 | stamp = stamp.replace(tzinfo=timezone.utc) |
| 386 | ahead = stamp.astimezone(timezone.utc) - datetime.now(timezone.utc) |
| 387 | return ahead.total_seconds() > FUTURE_SKEW_SECONDS |
| 388 | |
| 389 | |
| 390 | def read( |
| 391 | path: str | os.PathLike[str], |
| 392 | window: tuple[str, str], |
| 393 | topic: str, |
| 394 | handles: list[str] | None = None, |
| 395 | related: list[str] | None = None, |
| 396 | ) -> Envelope: |
| 397 | """Validate and ingest a ``--x-posts`` file for this run. |
| 398 | |
| 399 | ``window`` is the engine's ``(from_date, to_date)``; ``topic`` is the run |
| 400 | topic (or comparison entity); ``handles`` are the run's ``--x-handle`` |
| 401 | handles and ``related`` its ``--x-related`` handles, which together bound |
| 402 | what a call's ``handles`` may claim. |
| 403 | |
| 404 | Raises ``EnvelopeContractError`` (exit 2) for every fail-closed case. |
| 405 | """ |
| 406 | raw_path = os.fspath(path) |
| 407 | |
| 408 | def fail(what: str) -> EnvelopeContractError: |
| 409 | return EnvelopeContractError(f"--x-posts file {raw_path} {what}. {_REMEDY}") |
| 410 | |
| 411 | real = Path(os.path.realpath(raw_path)) |
| 412 | try: |
| 413 | info = _stat(real) |
| 414 | except OSError as exc: |
| 415 | raise fail(f"cannot be read ({type(exc).__name__})") from None |
| 416 | if not stat.S_ISREG(info.st_mode): |
| 417 | raise fail("is not a regular file") |
| 418 | if real.suffix.lower() != ".json": |
| 419 | raise fail("must be a .json file") |
| 420 | if _inside_protected(real): |
| 421 | raise fail("is inside a configuration or credential directory") |
| 422 | if info.st_size > MAX_BYTES: |
| 423 | raise fail(f"exceeds the {MAX_BYTES // (1024 * 1024)} MiB limit") |
| 424 | |
| 425 | try: |
| 426 | data = real.read_bytes() |
| 427 | except OSError as exc: |
| 428 | raise fail(f"cannot be read ({type(exc).__name__})") from None |
| 429 | digest = hashlib.sha256(data).hexdigest() |
| 430 | try: |
| 431 | text = data.decode("utf-8") |
| 432 | except UnicodeDecodeError: |
| 433 | raise fail("is not valid UTF-8") from None |
| 434 | try: |
| 435 | payload = json.loads(text) |
| 436 | except (RecursionError, MemoryError): |
| 437 | raise fail("is too deeply nested or too large to parse") from None |
| 438 | except ValueError: |
| 439 | raise fail("is not valid JSON") from None |
| 440 | |
| 441 | if not isinstance(payload, dict): |
| 442 | raise fail("must be a JSON object") |
| 443 | if payload.get("schema") != SCHEMA: |
| 444 | raise fail('has a missing or unsupported "schema" field') |
| 445 | status = payload.get("status") |
| 446 | if status not in STATUSES: |
| 447 | raise fail('has a "status" field that is not ok, partial, or error') |
| 448 | if not env.is_timestamp_fresh(payload.get("generated_at"), MAX_AGE_SECONDS): |
| 449 | raise fail( |
| 450 | 'has a "generated_at" field that is missing, malformed, or older than ' |
| 451 | f"{MAX_AGE_SECONDS // 3600} hours" |
| 452 | ) |
| 453 | if _generated_in_future(payload.get("generated_at")): |
| 454 | raise fail('has a "generated_at" field in the future') |
| 455 | if _normalize_topic(payload.get("topic")) != _normalize_topic(topic) or not _normalize_topic(topic): |
| 456 | raise fail('has a "topic" field that does not match this run\'s topic') |
| 457 | |
| 458 | from_date, to_date = window |
| 459 | window_raw = payload.get("window") |
| 460 | if not isinstance(window_raw, dict): |
| 461 | raise fail('has a missing or malformed "window" field') |
| 462 | window_from = _parse_day(window_raw.get("from")) |
| 463 | window_to = _parse_day(window_raw.get("to")) |
| 464 | if window_from is None or window_to is None: |
| 465 | raise fail('has a "window" field without valid from/to dates') |
| 466 | engine_from = _parse_day(from_date) or date.min |
| 467 | engine_to = _parse_day(to_date) or date.max |
| 468 | if window_to < engine_from: |
| 469 | raise fail('has a "window" that ends before the research window starts') |
| 470 | warnings: list[str] = [] |
| 471 | if window_from > engine_from or window_to < engine_to: |
| 472 | warnings.append( |
| 473 | f"host-fetched X window {window_from.isoformat()}..{window_to.isoformat()} " |
| 474 | f"is narrower than the research window {from_date}..{to_date}; " |
| 475 | "the research window is authoritative" |
| 476 | ) |
| 477 | |
| 478 | provider = _CONTROL_ANY_RE.sub("", str(payload.get("provider") or ""))[:64].strip() |
| 479 | |
| 480 | error_raw = payload.get("error") |
| 481 | if error_raw is not None and not isinstance(error_raw, str): |
| 482 | raise fail('has an "error" field that is not a string') |
| 483 | if error_raw and ("\n" in error_raw or "\r" in error_raw): |
| 484 | raise fail('has an "error" field that is not a single line') |
| 485 | error_category = _category(error_raw or "") |
| 486 | if error_raw and log.is_debug(): |
| 487 | _log(f"envelope error text (debug only): {_CONTROL_ANY_RE.sub('', error_raw)[:500]}") |
| 488 | |
| 489 | calls_raw = payload.get("calls") |
| 490 | if not isinstance(calls_raw, list): |
| 491 | raise fail('has a "calls" field that is not a list') |
| 492 | if len(calls_raw) > MAX_CALLS: |
| 493 | raise fail(f'has more than {MAX_CALLS} entries in "calls"') |
| 494 | if not calls_raw and status == "ok": |
| 495 | raise fail('has an empty "calls" list with status ok') |
| 496 | |
| 497 | allowed = {h.lower() for h in (_clean_handles(list(handles or [])) or [])} |
| 498 | related_set = {h.lower() for h in (_clean_handles(list(related or [])) or [])} |
| 499 | allowed |= related_set |
| 500 | |
| 501 | counters: dict[str, int] = {name: 0 for name in COUNTERS} |
| 502 | lane_counts: dict[str, int] = {lane: 0 for lane in LANES} |
| 503 | notes: list[str] = [] |
| 504 | call_lanes: list[str] = [] |
| 505 | topic_items: list[dict[str, Any]] = [] |
| 506 | lane_calls: list[EnvelopeCall] = [] |
| 507 | seen_ids: set[str] = set() |
| 508 | kept_ids: list[str] = [] |
| 509 | total = 0 |
| 510 | now_plus = datetime.now(timezone.utc) + timedelta(days=1) |
| 511 | |
| 512 | for index, call in enumerate(calls_raw, start=1): |
| 513 | if not isinstance(call, dict): |
| 514 | raise fail(f"has a call {index} that is not an object") |
| 515 | posts = call.get("posts") |
| 516 | if not isinstance(posts, list) or any(not isinstance(row, dict) for row in posts): |
| 517 | raise fail(f'has a call {index} whose "posts" is not a list of objects') |
| 518 | if len(posts) > MAX_ROWS_PER_CALL: |
| 519 | raise fail(f"has a call {index} with more than {MAX_ROWS_PER_CALL} rows") |
| 520 | total += len(posts) |
| 521 | if total > MAX_ROWS_TOTAL: |
| 522 | raise fail(f"has more than {MAX_ROWS_TOTAL} rows in total") |
| 523 | |
| 524 | lane_raw = call.get("lane") |
| 525 | lane = lane_raw.strip().lower() if isinstance(lane_raw, str) else "" |
| 526 | call_handles = _clean_handles(call.get("handles")) |
| 527 | if lane not in LANES: |
| 528 | if lane_raw is not None: |
| 529 | counters["lane-mismatch"] += 1 |
| 530 | notes.append(f"call {index}: unknown lane served as topic") |
| 531 | lane = "topic" |
| 532 | call_handles = [] |
| 533 | elif lane != "topic": |
| 534 | valid = bool(call_handles) and set(call_handles) <= allowed |
| 535 | if valid and lane == "related": |
| 536 | valid = set(call_handles) <= related_set |
| 537 | if not valid: |
| 538 | counters["lane-mismatch"] += 1 |
| 539 | notes.append( |
| 540 | f"call {index}: {lane} lane served as topic " |
| 541 | "(handles are not in --x-handle/--x-related)" |
| 542 | ) |
| 543 | lane = "topic" |
| 544 | call_handles = [] |
| 545 | else: |
| 546 | call_handles = [] |
| 547 | call_lanes.append(lane) |
| 548 | |
| 549 | kept: list[dict[str, Any]] = [] |
| 550 | for row in posts: |
| 551 | item = _ingest_row( |
| 552 | row, topic=topic, from_date=from_date, to_date=to_date, |
| 553 | now_plus=now_plus, seen_ids=seen_ids, counters=counters, |
| 554 | ) |
| 555 | if item is None: |
| 556 | continue |
| 557 | author = item["author_handle"].lower() |
| 558 | if lane == "from" or lane == "related": |
| 559 | if author not in call_handles: |
| 560 | counters["lane-mismatch"] += 1 |
| 561 | seen_ids.discard(item["post_id"]) |
| 562 | continue |
| 563 | elif lane == "mention": |
| 564 | if author in call_handles or any( |
| 565 | is_own_post(item["url"], handle) for handle in call_handles |
| 566 | ): |
| 567 | counters["lane-mismatch"] += 1 |
| 568 | seen_ids.discard(item["post_id"]) |
| 569 | continue |
| 570 | kept.append(item) |
| 571 | kept_ids.append(item["post_id"]) |
| 572 | |
| 573 | lane_counts[lane] += len(kept) |
| 574 | if lane == "topic": |
| 575 | topic_items.extend(kept) |
| 576 | else: |
| 577 | lane_calls.append(EnvelopeCall(index=index, lane=lane, handles=call_handles, posts=kept)) |
| 578 | |
| 579 | accepted = len(kept_ids) |
| 580 | if status == "error": |
| 581 | # An error envelope carries no evidence: the outcome is the story. |
| 582 | topic_items, lane_calls = [], [] |
| 583 | lane_counts = {lane: 0 for lane in LANES} |
| 584 | accepted = 0 |
| 585 | elif kept_ids and _looks_generated(kept_ids): |
| 586 | status = "error" |
| 587 | error_category = _CATEGORY_GENERATED |
| 588 | topic_items, lane_calls = [], [] |
| 589 | lane_counts = {lane: 0 for lane in LANES} |
| 590 | accepted = 0 |
| 591 | notes.append("id sequence looks generated; envelope rejected") |
| 592 | |
| 593 | for position, item in enumerate( |
| 594 | [*topic_items, *(post for call in lane_calls for post in call.posts)], start=1 |
| 595 | ): |
| 596 | item["id"] = f"{ID_PREFIX}{position}" |
| 597 | |
| 598 | envelope = Envelope( |
| 599 | path=raw_path, |
| 600 | sha256=digest, |
| 601 | status=status, |
| 602 | error_category=error_category, |
| 603 | provider=provider, |
| 604 | window=(window_from.isoformat(), window_to.isoformat()), |
| 605 | topic_items=topic_items, |
| 606 | lane_calls=lane_calls, |
| 607 | counters=counters, |
| 608 | accepted=accepted, |
| 609 | total=total, |
| 610 | lane_counts=lane_counts, |
| 611 | call_lanes=call_lanes, |
| 612 | warnings=warnings, |
| 613 | notes=notes, |
| 614 | ) |
| 615 | _log(envelope.receipt()) |
| 616 | for warning in warnings: |
| 617 | _log(warning) |
| 618 | return envelope |
| 619 | |
| 620 | |
| 621 | def _ingest_row( |
| 622 | row: dict[str, Any], |
| 623 | *, |
| 624 | topic: str, |
| 625 | from_date: str, |
| 626 | to_date: str, |
| 627 | now_plus: datetime, |
| 628 | seen_ids: set[str], |
| 629 | counters: dict[str, int], |
| 630 | ) -> dict[str, Any] | None: |
| 631 | """Rebuild one flat row into the engine's X item shape, or drop it. |
| 632 | |
| 633 | Returns ``None`` after incrementing the matching counter. The citation |
| 634 | URL is always engine-built from the id and the validated handle. |
| 635 | """ |
| 636 | if set(row) - ROW_FIELDS: |
| 637 | counters["extra-fields"] += 1 |
| 638 | |
| 639 | post_id = _post_id(row.get("id")) |
| 640 | text_raw = row.get("text") |
| 641 | text = _clean_text(text_raw) if isinstance(text_raw, str) else "" |
| 642 | if not post_id or not text.strip(): |
| 643 | counters["missing-id-text"] += 1 |
| 644 | return None |
| 645 | text = text.strip() |
| 646 | if len(text) > MAX_TEXT_CHARS: |
| 647 | text = text[:MAX_TEXT_CHARS] |
| 648 | counters["truncated"] += 1 |
| 649 | |
| 650 | decoded = _decode_snowflake(post_id) |
| 651 | if decoded is None or decoded < _SNOWFLAKE_FLOOR or decoded > now_plus: |
| 652 | counters["out-of-window"] += 1 |
| 653 | return None |
| 654 | snowflake_day = decoded.date() |
| 655 | |
| 656 | created_raw = row.get("created_at") |
| 657 | if isinstance(created_raw, str) and created_raw.strip(): |
| 658 | created = _parse_iso(created_raw) |
| 659 | if created is not None and abs((created.date() - snowflake_day).days) > 1: |
| 660 | counters["date-mismatch"] += 1 |
| 661 | return None |
| 662 | |
| 663 | day = snowflake_day.isoformat() |
| 664 | if (from_date and day < from_date) or (to_date and day > to_date): |
| 665 | counters["out-of-window"] += 1 |
| 666 | return None |
| 667 | |
| 668 | if post_id in seen_ids: |
| 669 | counters["duplicate"] += 1 |
| 670 | return None |
| 671 | seen_ids.add(post_id) |
| 672 | |
| 673 | handle = _handle(row.get("author_handle")) |
| 674 | url_raw = row.get("url") |
| 675 | if isinstance(url_raw, str) and url_raw.strip(): |
| 676 | match = _STATUS_URL_RE.match(url_raw.strip()) |
| 677 | url_handle = match.group("handle").lower() if match else "" |
| 678 | consistent = bool(match) and match.group("id") == post_id and ( |
| 679 | url_handle == "i" or not handle or url_handle == handle.lower() |
| 680 | ) |
| 681 | if not consistent: |
| 682 | counters["handle-mismatch"] += 1 |
| 683 | handle = "" |
| 684 | |
| 685 | if handle: |
| 686 | url = f"https://x.com/{handle}/status/{post_id}" |
| 687 | else: |
| 688 | counters["uncitable"] += 1 |
| 689 | url = f"https://x.com/i/status/{post_id}" |
| 690 | |
| 691 | return { |
| 692 | "id": "", |
| 693 | "text": text, |
| 694 | "url": url, |
| 695 | "author_handle": handle, |
| 696 | "date": day, |
| 697 | "engagement": { |
| 698 | "likes": _int(row.get("likes")), |
| 699 | "reposts": _int(row.get("reposts")), |
| 700 | "replies": _int(row.get("replies")), |
| 701 | "quotes": _int(row.get("quotes")), |
| 702 | }, |
| 703 | "mentioned_handles": leading_mentions(text), |
| 704 | "why_relevant": "", |
| 705 | "relevance": _compute_relevance(topic, text) if topic else 0.5, |
| 706 | "post_id": post_id, |
| 707 | } |
| 708 |