rediculous...

This commit is contained in:
ed committed 2026-08-24 21:54:56 -04:00
1 parent d6b3ebfd04
commit 526816c10f
14 files changed
+4513

No files matched your search

+971
View File
@@ -0,0 +1,971 @@
"""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}")