972 lines
35 KiB
Python
972 lines
35 KiB
Python
"""Offline TweetDetail parser, pure cursor walker, and global
|
|
human-paced :class:`RequestGovernor` for ``twitter.fetch_thread``.
|
|
|
|
This module owns three typed components:
|
|
|
|
1. The typed boundary dataclasses (:class:`Cursor`,
|
|
:class:`TweetDetailPage`) and the pure JSON-to-typed-data
|
|
parser :func:`parse_tweet_detail_page`. The parser inspects
|
|
the TweetDetail wire path
|
|
``data.threaded_conversation_with_injections_v2.instructions``
|
|
and converts typed timeline entries into existing
|
|
:class:`twitter.models.PostData` instances. Cursors with
|
|
eligible kinds (``Bottom``, ``ShowMore``, ``ShowMoreThreads``)
|
|
are kept; Top and unknown cursors are dropped without
|
|
raising. Tweets wrapped in
|
|
:class:`TweetWithVisibilityResults` are unwrapped via their
|
|
inner ``tweet`` field.
|
|
|
|
2. The pure FIFO continuation walker
|
|
:func:`walk_continuation_queue` and the pure post-group
|
|
merger :func:`merge_post_groups`. The walker has no
|
|
transport, no sleeper, no request cap, and no roles of its
|
|
own; the future orchestration wraps every provider call in
|
|
:meth:`RequestGovernor.issue`.
|
|
|
|
3. The global human-paced :class:`RequestGovernor` (and its
|
|
:class:`RequestBudgetExceeded` budget exception). The
|
|
governor owns the ONLY global request count, the ONLY
|
|
pacing state, and the ONLY ordered ``roles`` log. One
|
|
governor per run wraps every provider call later
|
|
(bootstrap homepage, bootstrap JS, tip TweetDetail, root
|
|
TweetDetail, every continuation).
|
|
|
|
This module deliberately does NOT contain HTTP code, cookie
|
|
loading, transaction helpers, the production transport, or
|
|
the CLI. Those live in later tasks of the same track and
|
|
reuse the parser, the walker, and the governor declared here.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from collections import deque
|
|
from collections.abc import Callable, Mapping, Sequence
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Protocol
|
|
|
|
# --- canonical package-or-script import shim ---
|
|
# Both `python -m twitter.fetch_thread` (package-mode) and
|
|
# `python scripts/twitter/fetch_thread.py` (script-mode) must work.
|
|
try:
|
|
from .models import PostData, PostMetrics
|
|
except ImportError: # pragma: no cover - script-mode import
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from models import PostData, PostMetrics
|
|
|
|
|
|
# --- fetcher-local types ---
|
|
|
|
|
|
# --- provider-error taxonomy (Task 9) ---
|
|
|
|
|
|
class FetchError(RuntimeError):
|
|
"""Base class for fetcher errors that fail closed.
|
|
|
|
The fetcher never retries, never logs, and never echoes
|
|
response bodies, URLs, headers, cursor values, or secret
|
|
values in exception messages. Status code and exception
|
|
type are the only diagnostic surfaces.
|
|
"""
|
|
|
|
|
|
class RequestBudgetExceeded(FetchError):
|
|
"""Raised when the global request budget is exhausted.
|
|
|
|
Inherits from :class:`FetchError` so callers can catch the
|
|
full fetcher-error family with one ``except`` clause. The
|
|
class remains a :class:`RuntimeError` transitively because
|
|
:class:`FetchError` is a :class:`RuntimeError`. Existing
|
|
governor tests that assert the ``RuntimeError`` lineage
|
|
continue to pass.
|
|
"""
|
|
|
|
|
|
class AuthError(FetchError):
|
|
"""HTTP 401 or 403. The authenticated session is no longer valid."""
|
|
|
|
|
|
class RateLimitError(FetchError):
|
|
"""HTTP 429. Twitter rate limit hit; no automatic retry."""
|
|
|
|
|
|
class ProviderError(FetchError):
|
|
"""Any non-2xx that is not 401/403/429, or a transport-layer
|
|
failure (``TimeoutError``, ``ConnectionError``, ``OSError``)
|
|
raised by the wrapped operation."""
|
|
|
|
|
|
ELIGIBLE_CURSOR_TYPES: frozenset[str] = frozenset(
|
|
{"Bottom", "ShowMore", "ShowMoreThreads"},
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Cursor:
|
|
"""A typed cursor entry from a TweetDetail response."""
|
|
value: str
|
|
cursor_type: str
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.value:
|
|
raise ValueError("Cursor.value must be non-empty")
|
|
if self.cursor_type not in ELIGIBLE_CURSOR_TYPES:
|
|
raise ValueError(
|
|
f"Cursor.cursor_type {self.cursor_type!r} is not in eligible set"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class TweetDetailPage:
|
|
"""Parsed TweetDetail page: typed posts, eligible cursors, and
|
|
the focal post's ``legacy.conversation_id_str``."""
|
|
posts: tuple[PostData, ...]
|
|
cursors: tuple[Cursor, ...]
|
|
focal_conversation_id: str | None = None
|
|
|
|
|
|
# --- typed raw-boundary helpers ---
|
|
|
|
|
|
def _from_wire_text(node: Mapping[str, object], key: str) -> str:
|
|
raw = node.get(key)
|
|
if not isinstance(raw, str):
|
|
raise KeyError(f"required string field {key!r}")
|
|
return raw
|
|
|
|
|
|
def _from_wire_int(node: Mapping[str, object], key: str) -> int:
|
|
raw = node.get(key)
|
|
if isinstance(raw, bool) or not isinstance(raw, int):
|
|
raise KeyError(f"required int field {key!r}")
|
|
return raw
|
|
|
|
|
|
def _from_wire_optional_int(
|
|
node: Mapping[str, object], key: str,
|
|
) -> int | None:
|
|
"""Accept ``int`` or numeric-string. Anything else is rejected
|
|
with ``KeyError``. ``None`` propagates as ``None``."""
|
|
raw = node.get(key)
|
|
if raw is None:
|
|
return None
|
|
if isinstance(raw, bool) or not isinstance(raw, int):
|
|
if isinstance(raw, str) and raw.lstrip("-").isdigit():
|
|
return int(raw)
|
|
raise KeyError(f"optional int field {key!r} has wrong type")
|
|
return raw
|
|
|
|
|
|
# --- per-tweet normalization ---
|
|
|
|
|
|
def _find_item_content(
|
|
entry: Mapping[str, object],
|
|
) -> Mapping[str, object] | None:
|
|
"""Return the entry's ``itemContent``-shaped payload, accepting
|
|
all three layouts seen in real TweetDetail wire responses:
|
|
``entry.content.itemContent``, ``entry.item.itemContent``,
|
|
and ``entry.itemContent`` directly."""
|
|
content = entry.get("content")
|
|
if isinstance(content, Mapping):
|
|
ic = content.get("itemContent")
|
|
if isinstance(ic, Mapping):
|
|
return ic
|
|
item = entry.get("item")
|
|
if isinstance(item, Mapping):
|
|
ic = item.get("itemContent")
|
|
if isinstance(ic, Mapping):
|
|
return ic
|
|
ic = entry.get("itemContent")
|
|
if isinstance(ic, Mapping):
|
|
return ic
|
|
return None
|
|
|
|
|
|
def _find_cursor_source(entry: Mapping[str, object]) -> Mapping[str, object] | None:
|
|
"""Return the entry's cursor source mapping, accepting all
|
|
three layouts (top-level content, module item.itemContent,
|
|
direct itemContent)."""
|
|
content = entry.get("content")
|
|
if isinstance(content, Mapping) and "cursorType" in content:
|
|
return content
|
|
item = entry.get("item")
|
|
if isinstance(item, Mapping):
|
|
item_content = item.get("itemContent")
|
|
if (
|
|
isinstance(item_content, Mapping)
|
|
and "cursorType" in item_content
|
|
):
|
|
return item_content
|
|
ic = entry.get("itemContent")
|
|
if isinstance(ic, Mapping) and "cursorType" in ic:
|
|
return ic
|
|
return None
|
|
|
|
|
|
def _entry_type(entry: Mapping[str, object]) -> object:
|
|
content = entry.get("content")
|
|
if isinstance(content, Mapping):
|
|
return content.get("entryType")
|
|
return None
|
|
|
|
|
|
def _resolve_post_id(tweet: Mapping[str, object]) -> str:
|
|
rest_id = tweet.get("rest_id")
|
|
if isinstance(rest_id, str) and rest_id:
|
|
return rest_id
|
|
legacy = tweet.get("legacy")
|
|
if isinstance(legacy, Mapping):
|
|
legacy_id = legacy.get("id_str")
|
|
if isinstance(legacy_id, str) and legacy_id:
|
|
return legacy_id
|
|
raise KeyError("tweet: missing rest_id and legacy.id_str")
|
|
|
|
|
|
def _resolve_user(tweet: Mapping[str, object]) -> tuple[str, str]:
|
|
"""Return ``(author, handle)`` for the tweet.
|
|
|
|
Current shape first: ``user_results.result.core.name`` and
|
|
``user_results.result.core.screen_name``. BOTH fields must be
|
|
non-empty strings for the current shape to win; otherwise
|
|
fall back to the legacy pair
|
|
``user_results.result.legacy.name`` and
|
|
``user_results.result.legacy.screen_name``. The legacy pair
|
|
is mandatory once the current shape is rejected.
|
|
"""
|
|
core = tweet.get("core")
|
|
user_results_node: object = None
|
|
if isinstance(core, Mapping):
|
|
user_results_node = core.get("user_results")
|
|
user: object = None
|
|
if isinstance(user_results_node, Mapping):
|
|
user = user_results_node.get("result")
|
|
if isinstance(user, Mapping):
|
|
# Current shape: user.core.name / user.core.screen_name.
|
|
# Wins only when BOTH fields are non-empty strings;
|
|
# otherwise fall through to the legacy pair.
|
|
user_core = user.get("core")
|
|
if isinstance(user_core, Mapping):
|
|
name_node = user_core.get("name")
|
|
screen_node = user_core.get("screen_name")
|
|
if (
|
|
isinstance(name_node, str)
|
|
and name_node
|
|
and isinstance(screen_node, str)
|
|
and screen_node
|
|
):
|
|
return str(name_node), str(screen_node)
|
|
# Fallback: user.legacy.name / user.legacy.screen_name
|
|
user_legacy = user.get("legacy")
|
|
if isinstance(user_legacy, Mapping):
|
|
name = _from_wire_text(user_legacy, "name")
|
|
handle = _from_wire_text(user_legacy, "screen_name")
|
|
return name, handle
|
|
raise KeyError("tweet: missing user_results.result core or legacy")
|
|
|
|
|
|
def _resolve_text(tweet: Mapping[str, object]) -> str:
|
|
note_tweet = tweet.get("note_tweet")
|
|
if isinstance(note_tweet, Mapping):
|
|
results = note_tweet.get("note_tweet_results")
|
|
if isinstance(results, Mapping):
|
|
inner = results.get("result")
|
|
if isinstance(inner, Mapping):
|
|
txt = inner.get("text")
|
|
if isinstance(txt, str) and txt:
|
|
return txt
|
|
legacy = tweet.get("legacy")
|
|
if isinstance(legacy, Mapping):
|
|
full_text = legacy.get("full_text")
|
|
if isinstance(full_text, str) and full_text:
|
|
return full_text
|
|
raise KeyError("tweet: missing note_tweet text and legacy.full_text")
|
|
|
|
|
|
def _normalize_created_at(created_at: str) -> str:
|
|
"""Twitter ``created_at`` → ``YYYY-MM-DD HH:MM:SS``."""
|
|
dt = datetime.strptime(created_at, "%a %b %d %H:%M:%S %z %Y")
|
|
return dt.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
def _resolve_media_urls(legacy: Mapping[str, object]) -> tuple[str, ...]:
|
|
"""Collect media URLs from ``legacy.extended_entities.media[]``
|
|
(falling back to ``entities.media[]``). Each photo URL is added;
|
|
each video contributes its highest-bitrate ``video/mp4``
|
|
variant URL. Duplicates are dropped, preserving order."""
|
|
ee_node = legacy.get("extended_entities")
|
|
ents: object = ee_node if isinstance(ee_node, Mapping) else None
|
|
if ents is None:
|
|
ents = legacy.get("entities")
|
|
if not isinstance(ents, Mapping):
|
|
return ()
|
|
media_node = ents.get("media")
|
|
if not isinstance(media_node, list):
|
|
return ()
|
|
urls: list[str] = []
|
|
for m in media_node:
|
|
if not isinstance(m, Mapping):
|
|
continue
|
|
photo_url = m.get("media_url_https")
|
|
if not isinstance(photo_url, str):
|
|
photo_url = m.get("media_url")
|
|
if isinstance(photo_url, str) and photo_url:
|
|
urls.append(photo_url)
|
|
video_info = m.get("video_info")
|
|
if isinstance(video_info, Mapping):
|
|
variants_node = video_info.get("variants")
|
|
if isinstance(variants_node, list):
|
|
best: tuple[int, str] | None = None
|
|
for v in variants_node:
|
|
if not isinstance(v, Mapping):
|
|
continue
|
|
if v.get("content_type") != "video/mp4":
|
|
continue
|
|
vurl = v.get("url")
|
|
if not isinstance(vurl, str):
|
|
continue
|
|
br_node = v.get("bitrate")
|
|
if isinstance(br_node, bool) or not isinstance(br_node, int):
|
|
br = 0
|
|
else:
|
|
br = br_node
|
|
if best is None or br > best[0]:
|
|
best = (br, vurl)
|
|
if best is not None:
|
|
urls.append(best[1])
|
|
seen: set[str] = set()
|
|
deduped: list[str] = []
|
|
for u in urls:
|
|
if u not in seen:
|
|
seen.add(u)
|
|
deduped.append(u)
|
|
return tuple(deduped)
|
|
|
|
|
|
def _build_post_data(tweet: Mapping[str, object]) -> PostData:
|
|
"""Convert an unwrapped tweet mapping into a typed
|
|
:class:`PostData`. Raises ``KeyError`` or ``TypeError`` on
|
|
missing or wrong-typed required fields."""
|
|
post_id = _resolve_post_id(tweet)
|
|
author, handle = _resolve_user(tweet)
|
|
text = _resolve_text(tweet)
|
|
legacy_node = tweet.get("legacy")
|
|
if not isinstance(legacy_node, Mapping):
|
|
raise KeyError(f"tweet {post_id}: missing legacy")
|
|
created_at = _from_wire_text(legacy_node, "created_at")
|
|
timestamp = _normalize_created_at(created_at)
|
|
reply_to_node = legacy_node.get("in_reply_to_status_id_str")
|
|
reply_to_id: str | None = (
|
|
str(reply_to_node)
|
|
if isinstance(reply_to_node, str) and reply_to_node
|
|
else None
|
|
)
|
|
quote_node = legacy_node.get("quoted_status_id_str")
|
|
quote_of_id: str | None = (
|
|
str(quote_node)
|
|
if isinstance(quote_node, str) and quote_node
|
|
else None
|
|
)
|
|
media_urls = _resolve_media_urls(legacy_node)
|
|
views_node = tweet.get("views")
|
|
views_count: int | None
|
|
if isinstance(views_node, Mapping):
|
|
views_count = _from_wire_optional_int(views_node, "count")
|
|
else:
|
|
views_count = None
|
|
metrics = PostMetrics(
|
|
reply_count=_from_wire_int(legacy_node, "reply_count"),
|
|
repost_count=_from_wire_int(legacy_node, "retweet_count"),
|
|
like_count=_from_wire_int(legacy_node, "favorite_count"),
|
|
view_count=views_count,
|
|
)
|
|
return PostData(
|
|
post_id=post_id,
|
|
author=author,
|
|
handle=handle,
|
|
text=text,
|
|
timestamp=timestamp,
|
|
media_urls=media_urls,
|
|
reply_to_id=reply_to_id,
|
|
quote_of_id=quote_of_id,
|
|
metrics=metrics,
|
|
)
|
|
|
|
|
|
def _normalize_tweet_entry(entry: Mapping[str, object]) -> PostData | None:
|
|
"""Return a :class:`PostData` if ``entry`` carries a Tweet.
|
|
Returns ``None`` for cursors, modules, tombstone / unavailable
|
|
results, or entries whose inner ``__typename`` is not ``Tweet``."""
|
|
entry_type = _entry_type(entry)
|
|
item_type = entry.get("itemType")
|
|
if (
|
|
entry_type not in ("TimelineTimelineItem", "TimelineTweet")
|
|
and item_type not in ("TimelineTimelineItem", "TimelineTweet")
|
|
):
|
|
return None
|
|
ic = _find_item_content(entry)
|
|
if ic is None:
|
|
return None
|
|
tweet_results = ic.get("tweet_results")
|
|
if not isinstance(tweet_results, Mapping):
|
|
return None
|
|
inner = tweet_results.get("result")
|
|
if not isinstance(inner, Mapping):
|
|
return None
|
|
if inner.get("__typename") == "TweetWithVisibilityResults":
|
|
inner = inner.get("tweet")
|
|
if not isinstance(inner, Mapping):
|
|
return None
|
|
if inner.get("__typename") != "Tweet":
|
|
return None
|
|
return _build_post_data(inner)
|
|
|
|
|
|
def _normalize_cursor_entry(entry: Mapping[str, object]) -> Cursor | None:
|
|
"""Return a :class:`Cursor` for ``entry`` if it is a
|
|
TimelineTimelineCursor with an eligible ``cursorType``.
|
|
Top and unknown kinds are silently dropped."""
|
|
cursor_source = _find_cursor_source(entry)
|
|
if cursor_source is None:
|
|
return None
|
|
entry_type = _entry_type(entry)
|
|
item_type = entry.get("itemType")
|
|
if (
|
|
entry_type != "TimelineTimelineCursor"
|
|
and item_type != "TimelineTimelineCursor"
|
|
):
|
|
return None
|
|
cursor_type_node = cursor_source.get("cursorType")
|
|
if not isinstance(cursor_type_node, str):
|
|
return None
|
|
if cursor_type_node not in ELIGIBLE_CURSOR_TYPES:
|
|
return None
|
|
value_node = cursor_source.get("value")
|
|
if isinstance(value_node, str):
|
|
value = value_node
|
|
elif isinstance(value_node, int) and not isinstance(value_node, bool):
|
|
value = str(value_node)
|
|
else:
|
|
return None
|
|
try:
|
|
return Cursor(value=value, cursor_type=cursor_type_node)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _extract_focal_conversation_id(
|
|
entry: Mapping[str, object], focal_id: str,
|
|
) -> str | None:
|
|
"""Return ``legacy.conversation_id_str`` from the wrapped tweet
|
|
whose id equals ``focal_id``. Inspects only the current entry;
|
|
iterators like :func:`_walk_entries` are responsible for
|
|
visiting every nested timeline node and calling this function
|
|
on each."""
|
|
ic = _find_item_content(entry)
|
|
if ic is None:
|
|
return None
|
|
tweet_results = ic.get("tweet_results")
|
|
if not isinstance(tweet_results, Mapping):
|
|
return None
|
|
inner = tweet_results.get("result")
|
|
if not isinstance(inner, Mapping):
|
|
return None
|
|
if inner.get("__typename") == "TweetWithVisibilityResults":
|
|
inner = inner.get("tweet")
|
|
if not isinstance(inner, Mapping):
|
|
return None
|
|
cand_id = inner.get("rest_id")
|
|
if not isinstance(cand_id, str) or not cand_id:
|
|
legacy = inner.get("legacy")
|
|
if isinstance(legacy, Mapping):
|
|
cand_id = legacy.get("id_str")
|
|
if not isinstance(cand_id, str) or cand_id != focal_id:
|
|
return None
|
|
legacy = inner.get("legacy")
|
|
if not isinstance(legacy, Mapping):
|
|
return None
|
|
cid = legacy.get("conversation_id_str")
|
|
if isinstance(cid, str) and cid:
|
|
return cid
|
|
return None
|
|
|
|
|
|
def _walk_entries(
|
|
entries: Sequence[object], focal_id: str | None,
|
|
) -> tuple[list[PostData], list[Cursor], str | None]:
|
|
"""Walk entries / modules iteratively via an explicit stack.
|
|
|
|
The walk is bounded to the typed entries / modules tree. It
|
|
does NOT recursively scan tweet / quote / user payloads for
|
|
cursor-looking strings; cursors are only read from the
|
|
typed cursor source locations at the entry / module level.
|
|
|
|
Source order is preserved: the stack is initialised with
|
|
the entries in reverse so the first ``pop`` yields the
|
|
first source entry, and module items are pushed in reverse
|
|
so the same invariant applies recursively.
|
|
|
|
Returns ``(posts, cursors, focal_conversation_id_or_none)``.
|
|
"""
|
|
posts: list[PostData] = []
|
|
cursors: list[Cursor] = []
|
|
focal_conversation_id: str | None = None
|
|
stack: list[object] = list(reversed(list(entries)))
|
|
while stack:
|
|
node = stack.pop()
|
|
if not isinstance(node, Mapping):
|
|
continue
|
|
post = _normalize_tweet_entry(node)
|
|
if post is not None:
|
|
posts.append(post)
|
|
cursor = _normalize_cursor_entry(node)
|
|
if cursor is not None:
|
|
cursors.append(cursor)
|
|
if focal_id is not None and focal_conversation_id is None:
|
|
cid = _extract_focal_conversation_id(node, focal_id)
|
|
if cid is not None:
|
|
focal_conversation_id = cid
|
|
content = node.get("content")
|
|
if isinstance(content, Mapping) and content.get(
|
|
"entryType"
|
|
) == "TimelineTimelineModule":
|
|
items_node = content.get("items")
|
|
if isinstance(items_node, list):
|
|
for item in reversed(items_node):
|
|
stack.append(item)
|
|
return posts, cursors, focal_conversation_id
|
|
|
|
|
|
def parse_tweet_detail_page(
|
|
raw: object, *, focal_id: str | None = None,
|
|
) -> TweetDetailPage:
|
|
"""Parse a TweetDetail response into typed posts, eligible
|
|
cursors, and the focal post's ``legacy.conversation_id_str``.
|
|
|
|
The wire path is
|
|
``data.threaded_conversation_with_injections_v2.instructions``;
|
|
this function walks only those instructions and the module
|
|
contents they reference. It does not recursively scan
|
|
tweet / quote / user payloads for cursor-looking strings.
|
|
|
|
Args:
|
|
raw: The deserialised wire JSON. Accepts ``object`` so
|
|
the caller can hand anything to the parser; a
|
|
non-mapping ``raw`` produces an empty page rather
|
|
than raising.
|
|
focal_id: When set, the parser walks all entries
|
|
(including those nested in TimelineTimelineModule
|
|
content.items[]) to find the entry whose id equals
|
|
``focal_id`` and exposes its
|
|
``legacy.conversation_id_str`` as
|
|
``page.focal_conversation_id``.
|
|
|
|
Returns:
|
|
A :class:`TweetDetailPage`. ``posts`` is deduped by id
|
|
with last-seen data preserved (first-seen order intact);
|
|
``cursors`` is deduped by value with first-seen wins.
|
|
"""
|
|
if not isinstance(raw, Mapping):
|
|
return TweetDetailPage(
|
|
posts=(), cursors=(), focal_conversation_id=None,
|
|
)
|
|
data_node = raw.get("data")
|
|
if not isinstance(data_node, Mapping):
|
|
return TweetDetailPage(
|
|
posts=(), cursors=(), focal_conversation_id=None,
|
|
)
|
|
threaded = data_node.get("threaded_conversation_with_injections_v2")
|
|
if not isinstance(threaded, Mapping):
|
|
return TweetDetailPage(
|
|
posts=(), cursors=(), focal_conversation_id=None,
|
|
)
|
|
instructions = threaded.get("instructions")
|
|
if not isinstance(instructions, list):
|
|
return TweetDetailPage(
|
|
posts=(), cursors=(), focal_conversation_id=None,
|
|
)
|
|
all_posts: list[PostData] = []
|
|
all_cursors: list[Cursor] = []
|
|
final_focal: str | None = None
|
|
for instruction in instructions:
|
|
if not isinstance(instruction, Mapping):
|
|
continue
|
|
itype = instruction.get("type")
|
|
if itype == "TimelineAddEntries":
|
|
entries = instruction.get("entries")
|
|
if isinstance(entries, list):
|
|
posts, cursors, focal = _walk_entries(entries, focal_id)
|
|
all_posts.extend(posts)
|
|
all_cursors.extend(cursors)
|
|
if final_focal is None and focal is not None:
|
|
final_focal = focal
|
|
elif itype == "TimelineAddToModule":
|
|
module_items = instruction.get("moduleItems")
|
|
if isinstance(module_items, list):
|
|
posts, cursors, focal = _walk_entries(module_items, focal_id)
|
|
all_posts.extend(posts)
|
|
all_cursors.extend(cursors)
|
|
if final_focal is None and focal is not None:
|
|
final_focal = focal
|
|
elif itype == "TimelineReplaceEntry":
|
|
entry = instruction.get("entry")
|
|
if isinstance(entry, Mapping):
|
|
posts, cursors, focal = _walk_entries([entry], focal_id)
|
|
all_posts.extend(posts)
|
|
all_cursors.extend(cursors)
|
|
if final_focal is None and focal is not None:
|
|
final_focal = focal
|
|
by_id: dict[str, PostData] = {}
|
|
order: list[str] = []
|
|
for post in all_posts:
|
|
if post.post_id not in by_id:
|
|
order.append(post.post_id)
|
|
by_id[post.post_id] = post
|
|
deduped_posts: list[PostData] = [by_id[oid] for oid in order]
|
|
seen_cursor_values: set[str] = set()
|
|
deduped_cursors: list[Cursor] = []
|
|
for c in all_cursors:
|
|
if c.value not in seen_cursor_values:
|
|
seen_cursor_values.add(c.value)
|
|
deduped_cursors.append(c)
|
|
return TweetDetailPage(
|
|
posts=tuple(deduped_posts),
|
|
cursors=tuple(deduped_cursors),
|
|
focal_conversation_id=final_focal,
|
|
)
|
|
|
|
|
|
# --- root resolution (Task 3) ---
|
|
|
|
|
|
def resolve_root_from_tip(tip: str, page: TweetDetailPage) -> str:
|
|
"""Resolve the conversation root from a parsed tip page.
|
|
|
|
Returns ``page.focal_conversation_id``, which
|
|
:func:`parse_tweet_detail_page` extracted from the focal tip
|
|
post's ``legacy.conversation_id_str``.
|
|
|
|
Verifies that ``tip`` is present in ``page.posts`` and that
|
|
``page.focal_conversation_id`` is a non-empty string. Neither
|
|
a parent-chain walk nor a ``reply_to_id is None`` heuristic is
|
|
used; the resolved root must come from the focal post's
|
|
conversation id. Parent-chain closure is validated later after
|
|
the tip+root page merge.
|
|
|
|
Args:
|
|
tip: The post id of the input tip (the user's ``--tip``
|
|
value).
|
|
page: A parsed :class:`TweetDetailPage` from
|
|
:func:`parse_tweet_detail_page`.
|
|
|
|
Returns:
|
|
The resolved conversation root post id.
|
|
|
|
Raises:
|
|
ValueError: ``tip`` is not present in ``page.posts``.
|
|
ValueError: ``page.focal_conversation_id`` is ``None`` or
|
|
an empty string.
|
|
"""
|
|
tip_ids = {post.post_id for post in page.posts}
|
|
if tip not in tip_ids:
|
|
raise ValueError(f"tip {tip!r} not present in page.posts")
|
|
focal = page.focal_conversation_id
|
|
if not focal:
|
|
raise ValueError(
|
|
"page.focal_conversation_id is missing or empty; "
|
|
"cannot resolve conversation root"
|
|
)
|
|
return focal
|
|
|
|
|
|
# --- tip + root page merge (Task 4) ---
|
|
|
|
|
|
def merge_post_groups(
|
|
*groups: Sequence[PostData],
|
|
) -> tuple[PostData, ...]:
|
|
"""Merge multiple post sequences into a deduped tuple.
|
|
|
|
Last-seen data wins: when the same ``post_id`` appears in more
|
|
than one group, the later group's :class:`PostData` replaces the
|
|
earlier one. First-insertion order is preserved: the position of
|
|
each id in the output is the position of its first appearance in
|
|
the iteration order. Pure function; no I/O, network, or sample
|
|
constants.
|
|
"""
|
|
seen: dict[str, PostData] = {}
|
|
order: list[str] = []
|
|
for group in groups:
|
|
for post in group:
|
|
if post.post_id not in seen:
|
|
order.append(post.post_id)
|
|
seen[post.post_id] = post
|
|
return tuple(seen[pid] for pid in order)
|
|
|
|
|
|
# --- continuation queue walker (Task 5) ---
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ContinuationWalkResult:
|
|
"""Outcome of :func:`walk_continuation_queue`.
|
|
|
|
``posts`` is the deduped union of the initial page's posts and
|
|
every continuation page's posts, with last-seen data wins and
|
|
first-insertion order preserved. ``fetched_cursor_count`` is
|
|
the number of times the walker invoked ``fetch_page`` (i.e.,
|
|
the number of unseen eligible cursors it followed).
|
|
"""
|
|
posts: tuple[PostData, ...]
|
|
fetched_cursor_count: int
|
|
|
|
|
|
def walk_continuation_queue(
|
|
initial_page: TweetDetailPage,
|
|
*,
|
|
fetch_page: Callable[[Cursor], TweetDetailPage],
|
|
) -> ContinuationWalkResult:
|
|
"""Walk the eligible explicit reply cursor queue, FIFO.
|
|
|
|
Seeds the queue with ``initial_page.cursors``. For each unseen
|
|
cursor value (deduped by ``Cursor.value``), invokes
|
|
``fetch_page`` exactly once, merges the returned page's posts
|
|
into the accumulator via :func:`merge_post_groups`, and
|
|
enqueues the returned page's own cursors. Iteration terminates
|
|
when the queue empties.
|
|
|
|
The walker has no HTTP, sleep, retry, request cap, root/tip
|
|
special case, or logging. The future global ``RequestGovernor``
|
|
wraps ``fetch_page`` and is the only request cap; depth /
|
|
loop behaviour is unbounded here.
|
|
|
|
Exceptions raised by ``fetch_page`` propagate immediately on
|
|
the call that raised them. The walker does not catch, swallow,
|
|
or re-invoke the callback after a failure.
|
|
|
|
Args:
|
|
initial_page: The starting :class:`TweetDetailPage`. Its
|
|
``posts`` seed the accumulator; its ``cursors`` seed
|
|
the FIFO queue.
|
|
fetch_page: Callable invoked with one :class:`Cursor` and
|
|
returning the parsed page for that continuation.
|
|
|
|
Returns:
|
|
A :class:`ContinuationWalkResult` with the merged posts
|
|
and the count of callback invocations.
|
|
"""
|
|
queue: deque[Cursor] = deque(initial_page.cursors)
|
|
seen_cursor_values: set[str] = set()
|
|
accumulator: tuple[PostData, ...] = initial_page.posts
|
|
fetched_cursor_count: int = 0
|
|
while queue:
|
|
cursor = queue.popleft()
|
|
if cursor.value in seen_cursor_values:
|
|
continue
|
|
seen_cursor_values.add(cursor.value)
|
|
next_page = fetch_page(cursor)
|
|
fetched_cursor_count += 1
|
|
accumulator = merge_post_groups(accumulator, next_page.posts)
|
|
for next_cursor in next_page.cursors:
|
|
if next_cursor.value not in seen_cursor_values:
|
|
queue.append(next_cursor)
|
|
return ContinuationWalkResult(
|
|
posts=accumulator,
|
|
fetched_cursor_count=fetched_cursor_count,
|
|
)
|
|
|
|
|
|
# --- global human-paced request governor (Task 8) ---
|
|
|
|
|
|
class Sleeper(Protocol):
|
|
"""The duck-typed surface the governor depends on.
|
|
|
|
Production passes a real sleeper that blocks for
|
|
``seconds``; tests pass ``RecordingSleeper`` from
|
|
:mod:`twitter.tests.fetch_thread_helpers`, which records
|
|
durations without blocking.
|
|
"""
|
|
|
|
def sleep(self, seconds: float) -> None:
|
|
"""Block (or record) for ``seconds`` seconds."""
|
|
|
|
|
|
class RequestGovernor:
|
|
"""Global, human-paced request envelope.
|
|
|
|
Exactly one :class:`RequestGovernor` wraps every provider
|
|
call later in the orchestration (bootstrap homepage,
|
|
bootstrap JS, tip TweetDetail, root TweetDetail, every
|
|
continuation). The governor owns the **only** global request
|
|
count and the **only** pacing state. The pure cursor walker
|
|
:func:`walk_continuation_queue` carries no
|
|
``max_requests``, no request index, no sleeper, and no role
|
|
log of its own.
|
|
|
|
Behaviour:
|
|
|
|
- ``max_requests >= 1`` and ``min_delay >= 0`` are enforced
|
|
at construction; non-empty ``role`` is enforced at every
|
|
``issue``.
|
|
- Before every ``issue``: if ``count >= max_requests`` the
|
|
governor raises :class:`RequestBudgetExceeded` **before**
|
|
sleeping and **before** invoking the operation. No sleep
|
|
is appended, no role is appended.
|
|
- The very first attempted call does not sleep. Every later
|
|
attempted call sleeps exactly once with the configured
|
|
``min_delay`` before invoking the operation.
|
|
- The role is appended and counted exactly once,
|
|
immediately before invoking the operation. If the
|
|
operation raises, the attempted call remains counted; the
|
|
governor does not catch, retry, or strip the role.
|
|
- No logging, no HTTP, no URL inspection, no cursor
|
|
inspection, no retry, no concurrency, no time polling,
|
|
no automatic backoff.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
max_requests: int = 8,
|
|
min_delay: float = 5.0,
|
|
sleeper: Sleeper,
|
|
) -> None:
|
|
if not isinstance(max_requests, int) or isinstance(max_requests, bool):
|
|
raise ValueError(
|
|
f"max_requests must be int, got {type(max_requests).__name__}",
|
|
)
|
|
if max_requests < 1:
|
|
raise ValueError(
|
|
f"max_requests must be >= 1, got {max_requests}",
|
|
)
|
|
if not isinstance(min_delay, (int, float)) or isinstance(min_delay, bool):
|
|
raise ValueError(
|
|
f"min_delay must be numeric, got {type(min_delay).__name__}",
|
|
)
|
|
if float(min_delay) < 0.0:
|
|
raise ValueError(
|
|
f"min_delay must be >= 0, got {min_delay}",
|
|
)
|
|
if sleeper is None:
|
|
raise ValueError("sleeper must not be None")
|
|
self._max_requests: int = max_requests
|
|
self._min_delay: float = float(min_delay)
|
|
self._sleeper: Sleeper = sleeper
|
|
self._roles: list[str] = []
|
|
|
|
@property
|
|
def count(self) -> int:
|
|
"""The number of attempted calls counted so far."""
|
|
return len(self._roles)
|
|
|
|
@property
|
|
def roles(self) -> tuple[str, ...]:
|
|
"""The ordered tuple of roles appended for attempted calls."""
|
|
return tuple(self._roles)
|
|
|
|
def issue(
|
|
self,
|
|
role: str,
|
|
operation: Callable[[], tuple[int, str]],
|
|
) -> tuple[int, str]:
|
|
"""Wrap ``operation`` in the pacing + budget envelope.
|
|
|
|
Args:
|
|
role: A non-empty label for the attempted call
|
|
(e.g., ``"bootstrap_homepage"``,
|
|
``"tip_tweet_detail"``,
|
|
``"continuation_tweet_detail"``). Appended to
|
|
:attr:`roles` exactly once on attempted calls.
|
|
operation: A zero-argument callable returning
|
|
``(status_code, body_text)``. The governor
|
|
invokes it at most once.
|
|
|
|
Returns:
|
|
The ``(status_code, body_text)`` tuple returned by
|
|
``operation``.
|
|
|
|
Raises:
|
|
ValueError: ``role`` is empty or not a string.
|
|
RequestBudgetExceeded: ``count >= max_requests``
|
|
before this call. Raised **before** sleeping and
|
|
**before** invoking ``operation``; the operation
|
|
is not called, no role is appended, no sleep is
|
|
recorded.
|
|
Exception: any exception raised by ``operation``
|
|
propagates immediately. The attempted call is
|
|
counted (the role was already appended), and the
|
|
governor does not retry.
|
|
"""
|
|
if not isinstance(role, str):
|
|
raise ValueError(
|
|
f"role must be str, got {type(role).__name__}",
|
|
)
|
|
if not role:
|
|
raise ValueError("role must be non-empty")
|
|
if self.count >= self._max_requests:
|
|
raise RequestBudgetExceeded(
|
|
f"request budget of {self._max_requests} exhausted "
|
|
f"before issuing role {role!r}",
|
|
)
|
|
if self.count > 0:
|
|
self._sleeper.sleep(self._min_delay)
|
|
self._roles.append(role)
|
|
return operation()
|
|
|
|
|
|
# --- shared provider-error adapter (Task 9) ---
|
|
|
|
|
|
def issue_provider_request(
|
|
governor: RequestGovernor,
|
|
*,
|
|
role: str,
|
|
operation: Callable[[], tuple[int, str]],
|
|
) -> str:
|
|
"""Run a single provider request through ``governor`` and
|
|
translate the outcome into either a body string or a typed
|
|
:class:`FetchError` subclass.
|
|
|
|
Behaviour:
|
|
|
|
- The wrapped ``operation`` is invoked exactly once through
|
|
:meth:`RequestGovernor.issue`. The governor owns the
|
|
pacing, the role log, and the budget cap.
|
|
- 2xx responses (``200``-``299``) return the body string
|
|
unchanged.
|
|
- ``401`` and ``403`` raise :class:`AuthError`.
|
|
- ``429`` raises :class:`RateLimitError`.
|
|
- Any other non-2xx status raises :class:`ProviderError`.
|
|
- ``TimeoutError``, ``ConnectionError``, and ``OSError`` raised
|
|
by ``operation`` are translated into :class:`ProviderError`
|
|
chained from the original via ``raise ... from exc``.
|
|
- ``RequestBudgetExceeded`` (raised by the governor when the
|
|
cap is reached) propagates unchanged. It is already a
|
|
:class:`FetchError` subclass and remains a
|
|
:class:`RuntimeError`.
|
|
- Arbitrary programmer/parser exceptions (e.g.,
|
|
:class:`ValueError`) propagate unchanged; they are not
|
|
mislabelled as a provider failure.
|
|
|
|
Diagnostic policy: exception messages are limited to the
|
|
numeric HTTP status and the exception class name. They never
|
|
include the response body, the URL, header values, cursor
|
|
values, or any secret value. The ``role`` argument is the
|
|
caller's own label and is intentionally not echoed into the
|
|
exception.
|
|
"""
|
|
try:
|
|
status, body = governor.issue(role, operation)
|
|
except (TimeoutError, ConnectionError, OSError) as exc:
|
|
raise ProviderError("transport error") from exc
|
|
if 200 <= status <= 299:
|
|
return body
|
|
if status in (401, 403):
|
|
raise AuthError(f"http {status}")
|
|
if status == 429:
|
|
raise RateLimitError(f"http {status}")
|
|
raise ProviderError(f"http {status}")
|