ed/Threadwell
1
0 Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

scrub scripts.

Browse Source
This commit is contained in:
ed committed 2026-08-26 21:47:59 -04:00
1 parent 51ba6bd406
commit 2bce73ccf3
19 files changed
+1 -4876
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

scripts/twitter/README.md
-3
View File
@@ -90,11 +90,8 @@ Later emits use that same title for the folder and the frontmatter. Move both si
| `catbox_client.py` | one Catbox upload |
| `fallback_media.py` | fallback activation and restore-origin |
| `frozen.py` | resolve frozen ids against every captured post id |
| `lift_catbox.py` | retired. Use `tw.py fallback` |
| `ocr_pass.py` | Umi HTTP `:1224` / tesseract / Windows OCR |
| `media_embed.py` | `--attach-ocr` only |
| `media_merge.py` | restore derived rows |
| `crt_pass.py` | stock slang only. ShaderGlass is the look reference |
| `do_not_refetch.txt` | frozen id list |
| `tests/` | isolated unit tests |
| `unfinished_bloat/` | leftover probe code. Not the front door |
scripts/twitter/audit_handle_index.py
+1 -2
View File
@@ -79,8 +79,7 @@ def audit_one(handle_dir: Path) -> dict:
def audit_threads_index(threads_root: Path) -> dict:
"""Audit the top-level archive/threads/index.md against real
handle dirs."""
"""Audit the top-level archive/threads/index.md against real handle dirs."""
idx_path = threads_root / "index.md"
real = {
d.name for d in threads_root.iterdir() if d.is_dir() and not d.name.startswith(".")
scripts/twitter/crt_pass.py
-322
View File
@@ -1,322 +0,0 @@
"""Apply a stock RetroArch slang preset to a still via librashader-cli.
Does not patch shaders. Does not invent PixelSize. Your ShaderGlass .sgp files
are the look reference; this only runs the same .slangp with the listed params.
Manual look you are matching:
1. Nearest-upscale the orig (Affinity zoom). 300% => 3x.
2. Overlay ShaderGlass with the .sgp (Cyclon PixelSize 6, Lottes PixelSize 4).
3. Snapshot, then crop.
This script can do step 1 (ffmpeg neighbor) and a stock slang render of the
result. It cannot reproduce ShaderGlass window capture. If the plate is wrong,
do it in ShaderGlass and drop the PNG next to the orig as *_crt.png.
Presets:
lottes crt-lottes-multipass.slangp
LOTTES_CODE_(LOTTES_MULTIPASS 2X) V2.sgp
cyclon crt-Cyclon.slangp
LOTTES_CRT_CODE_(CYCLON 6X).sgp
auto: cyclon if max(w,h) <= --auto-max (default 720), else lottes.
"""
from __future__ import annotations
import argparse
import json
import shutil
import subprocess
import sys
from pathlib import Path
LOTTES_SGP: Path = Path(
r"C:\Users\Ed\Downloads\ShaderGlass-1.2.2-win-x64"
r"\LOTTES_CODE_(LOTTES_MULTIPASS 2X) V2.sgp"
)
CYCLON_SGP: Path = Path(
r"C:\Users\Ed\Downloads\ShaderGlass-1.2.2-win-x64"
r"\LOTTES_CRT_CODE_(CYCLON 6X).sgp"
)
SLANG_LOTTES: Path = Path(
r"C:\Users\Ed\scoop\apps\retroarch\1.22.2\shaders"
r"\shaders_slang\crt\crt-lottes-multipass.slangp"
)
SLANG_CYCLON: Path = Path(
r"C:\Users\Ed\scoop\apps\retroarch\1.22.2\shaders"
r"\shaders_slang\crt\crt-Cyclon.slangp"
)
CLI_DEFAULT: Path = Path(r"C:\Users\Ed\scoop\apps\librashader\0.12.0\librashader-cli.exe")
# Knobs copied from the .sgp files. Not a substitute for PixelSize.
LOTTES_PARAMS: str = (
"hardScan=-19,hardPix=-7,warpX=0,warpY=0,maskDark=1.3,maskLight=1.3,"
"shadowMask=4,brightBoost=1.15,bloomAmount=0,shape=1.9"
)
CYCLON_PARAMS: str = (
"SCANLINE=0.2,INTERLACE=0,M_TYPE=0,SLOT=0,SLOTW=2,Maskl=0.55,Maskh=1,"
"bzl=0,ambient=0,REFLECT=0,WARPX=0,WARPY=0,BR_DEP=0,BRIGHTNESS_=1.05,"
"BLACK=0.01,vig=0"
)
def probe_size(path: Path) -> tuple[int, int]:
"""Return ``(width, height)`` of an image or video via ``ffprobe``."""
raw = subprocess.check_output(
[
"ffprobe",
"-v",
"error",
"-select_streams",
"v:0",
"-show_entries",
"stream=width,height",
"-of",
"json",
str(path),
],
text=True,
)
stream = json.loads(raw)["streams"][0]
return int(stream["width"]), int(stream["height"])
def pick_preset(width: int, height: int, name: str, auto_max: int) -> tuple[str, Path, Path, str]:
"""Resolve the preset name to ``(name, .sgp_path, .slangp_path, params)``; ``auto`` picks cyclon under ``auto_max``."""
if name == "auto":
name = "cyclon" if max(width, height) <= auto_max else "lottes"
if name == "lottes":
return name, LOTTES_SGP, SLANG_LOTTES, LOTTES_PARAMS
if name == "cyclon":
return name, CYCLON_SGP, SLANG_CYCLON, CYCLON_PARAMS
raise SystemExit(f"unknown preset {name}")
def find_cli(explicit: Path | None) -> Path:
"""Locate ``librashader-cli``: explicit path, then PATH, then the scoop-installed default."""
if explicit is not None:
if not explicit.is_file():
raise SystemExit(f"missing --cli {explicit}")
return explicit
which = shutil.which("librashader-cli")
if which:
return Path(which)
if CLI_DEFAULT.is_file():
return CLI_DEFAULT
raise SystemExit(
"librashader-cli not on PATH. Install or pass --cli. "
"Stock binary lives at "
r"C:\Users\Ed\scoop\apps\librashader\0.12.0\librashader-cli.exe"
)
def parse_crop(spec: str) -> str | None:
"""Parse a ``--crop`` value into an ffmpeg ``crop=...`` filter, or ``None`` to skip cropping."""
if spec in ("none", "", "off"):
return None
if spec == "auto":
return "auto"
parts = [int(p) for p in spec.replace(" ", "").split(",")]
if len(parts) != 4:
raise SystemExit("--crop wants auto, none, or L,T,R,B")
left, top, right, bottom = parts
return f"crop=iw-{left + right}:ih-{top + bottom}:{left}:{top}"
def detect_crop(path: Path) -> str | None:
"""Run ``ffmpeg cropdetect`` on ``path`` and return the auto-detected crop filter (or ``None``)."""
proc = subprocess.run(
[
"ffmpeg",
"-hide_banner",
"-i",
str(path),
"-vf",
"cropdetect=limit=24:round=2:reset=0",
"-frames:v",
"1",
"-f",
"null",
"-",
],
capture_output=True,
text=True,
)
crop = None
for line in (proc.stderr or "").splitlines():
if "crop=" in line:
crop = line.rsplit("crop=", 1)[-1].strip()
return f"crop={crop}" if crop else None
def run_ffmpeg(src: Path, dest: Path, vf: str) -> None:
"""Run ``ffmpeg -y`` with the given ``-vf`` filter to render a single still ``src`` -> ``dest``."""
dest.parent.mkdir(parents=True, exist_ok=True)
subprocess.check_call(
[
"ffmpeg",
"-y",
"-hide_banner",
"-loglevel",
"error",
"-i",
str(src),
"-vf",
vf,
"-frames:v",
"1",
"-update",
"1",
str(dest),
]
)
def run_librashader(
cli: Path,
slang: Path,
src: Path,
dest: Path,
*,
runtime: str,
dimensions: str,
params: str,
) -> None:
"""Render ``src`` through ``slang`` via the ``librashader-cli`` binary into ``dest``."""
dest.parent.mkdir(parents=True, exist_ok=True)
dest.parent.mkdir(parents=True, exist_ok=True)
cmd = [
str(cli),
"render",
"-r",
runtime,
"-d",
dimensions,
"--params",
params,
"-p",
str(slang),
"-i",
str(src),
"-o",
str(dest),
]
print(" ".join(cmd))
subprocess.check_call(cmd)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--in", dest="src", type=Path, required=True)
parser.add_argument("--out", dest="dest", type=Path, required=True)
parser.add_argument("--preset", choices=("auto", "lottes", "cyclon"), default="auto")
parser.add_argument(
"--nn",
type=int,
default=1,
help="nearest integer pre-scale (Affinity zoom). 1 = none, 3 = 300%%",
)
parser.add_argument(
"--shader-scale",
default="100%",
help="librashader -d on the (possibly nn-scaled) image. 100%%, 300%%, 600%%, or WIDTHxHEIGHT",
)
parser.add_argument("--runtime", default="d3d11")
parser.add_argument("--auto-max", type=int, default=720)
parser.add_argument("--crop", default="none", help="auto | none | L,T,R,B")
parser.add_argument("--cli", type=Path, default=None)
parser.add_argument("--dry-run", action="store_true")
parser.add_argument(
"--shaderglass",
action="store_true",
help="launch ShaderGlass with the matching .sgp (manual snapshot)",
)
parser.add_argument(
"--nn-only",
action="store_true",
help="only do the neighbor pre-scale; do not call librashader",
)
args = parser.parse_args(argv)
if not args.src.is_file():
raise SystemExit(f"missing {args.src}")
if args.nn < 1:
raise SystemExit("--nn must be >= 1")
width, height = probe_size(args.src)
name, sgp, slang, params = pick_preset(width, height, args.preset, args.auto_max)
print(f"size {width}x{height} preset={name} nn={args.nn} shader-scale={args.shader_scale}")
print(f"sgp {sgp}")
print(f"slang {slang}")
work = args.src
tmp_nn: Path | None = None
if args.nn > 1:
tmp_nn = args.dest.with_name(args.dest.stem + f"_nn{args.nn}.png")
vf = f"scale=iw*{args.nn}:ih*{args.nn}:flags=neighbor,format=rgba"
print("ffmpeg", "-i", work, "-vf", vf, tmp_nn)
if not args.dry_run:
run_ffmpeg(work, tmp_nn, vf)
work = tmp_nn
if args.nn_only:
if tmp_nn is None:
raise SystemExit("--nn-only needs --nn > 1")
if not args.dry_run:
shutil.copy2(tmp_nn, args.dest)
print(f"wrote {args.dest} (nn only)")
return 0
cli = None if args.dry_run else find_cli(args.cli)
if args.dry_run:
print(
"librashader-cli render",
"-r",
args.runtime,
"-d",
args.shader_scale,
"--params",
params,
"-p",
slang,
"-i",
work,
"-o",
args.dest,
)
else:
assert cli is not None
if not slang.is_file():
raise SystemExit(f"missing slang {slang}")
run_librashader(
cli,
slang,
work,
args.dest,
runtime=args.runtime,
dimensions=args.shader_scale,
params=params,
)
crop_spec = parse_crop(args.crop)
if crop_spec == "auto" and not args.dry_run:
crop_filter = detect_crop(args.dest)
if crop_filter:
cropped = args.dest.with_name(args.dest.stem + "_crop.png")
run_ffmpeg(args.dest, cropped, crop_filter)
shutil.move(str(cropped), str(args.dest))
elif crop_spec and crop_spec != "auto" and not args.dry_run:
cropped = args.dest.with_name(args.dest.stem + "_crop.png")
run_ffmpeg(args.dest, cropped, crop_spec)
shutil.move(str(cropped), str(args.dest))
print(f"wrote {args.dest}")
if args.shaderglass:
exe = shutil.which("shaderglass") or shutil.which("ShaderGlass")
if not exe:
raise SystemExit("shaderglass not on PATH")
subprocess.Popen([exe, str(sgp)])
return 0
if __name__ == "__main__":
raise SystemExit(main())
scripts/twitter/lift_catbox.py
-20
View File
@@ -1,20 +0,0 @@
"""Deprecated.
Bulk lift has been retired. Use `tw.py fallback --id <id> --media-id <media-id> --role <role>
--confirm-origin-unavailable` after confirming the origin is unavailable.
"""
from __future__ import annotations
import sys
def main(argv: list[str] | None = None) -> int: # pragma: no cover - deprecated entry
del argv
sys.stderr.write(
"lift_catbox.py is retired; use tw.py fallback with --confirm-origin-unavailable\n"
)
return 2
if __name__ == "__main__":
raise SystemExit(main())
scripts/twitter/unfinished_bloat/fetch_thread.py
-971
View File
@@ -1,971 +0,0 @@
"""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}")
scripts/twitter/unfinished_bloat/tests/fetch_thread_helpers.py
-206
View File
@@ -1,206 +0,0 @@
"""Test helpers for the Twitter full-conversation fetcher.
Owns the canonical test-truth constants that production
``twitter.fetch_thread`` mirrors (the equality is asserted at
import time once production exists). Also owns the offline fakes
used by every test that exercises the fetcher without touching
the network: ``FakeTransport``, ``RecordingSleeper``, and
``FakeTransaction``.
This module deliberately stays a *test-only* module: it must not
import ``twitter.fetch_thread`` (production does not exist yet
when the helpers are first created) and must not read cookies,
open sockets, or sleep for real.
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from pathlib import Path
# --- canonical test-truth constants (production mirrors these) ---
KNOWN_ROOT_ID: str = "1650678968255913985"
KNOWN_TIP_ID: str = "1651282559287042048"
EXPECTED_16: frozenset[str] = frozenset({
"1650678968255913985", # kenpex OP
"1650685805776732160", # abductee_org -> OP
"1650915682136240129", # streleav -> OP
"1651056827839180800", # wadetb -> OP
"1651086935467917312", # ChristerEricson -> OP
"1651168030557077504", # SebAaltonen -> OP
"1651253961524142081", # kenpex -> wadetb
"1651254727160795137", # kenpex -> ChristerEricson
"1651268028795961344", # NOTimothyLottes -> kenpex
"1651282559287042048", # wvo (the input tip)
"1651295755293036544", # ErikHaliewicz -> OP
"1651508233243095043", # npatsiouras -> kenpex Christer branch
"1651510424649744385", # ChristerEricson -> npatsiouras
"1651511316010663936", # ChristerEricson -> self
"1651579576089300992", # JaceCear -> OP
"1651636988267888641", # kenpex -> JaceCear
})
# --- fixture loader ---
FIXTURES_DIR: Path = (
Path(__file__).resolve().parent / "fixtures" / "fetch_thread"
)
def load_fixture(name: str) -> str:
"""Return the raw UTF-8 text of a sanitized fixture by basename.
The caller may pass ``"tip_tweet_detail"`` or
``"tip_tweet_detail.json"``; either form resolves to the same
file under :data:`FIXTURES_DIR`.
"""
filename = name if name.endswith(".json") else f"{name}.json"
return (FIXTURES_DIR / filename).read_text(encoding="utf-8")
def load_json_fixture(name: str) -> object:
"""Parse the JSON fixture with ``name`` (with or without the
``.json`` suffix) and return the deserialised value.
Return type is ``object`` because the wire shape varies: pages,
cursor lists, error envelopes, and homepage HTML fragments all
round-trip through here.
"""
return json.loads(load_fixture(name))
# --- fakes ---
@dataclass
class RecordingSleeper:
"""Records each requested sleep duration without blocking.
The fetcher's :class:`RequestGovernor` calls ``sleep(min_delay)``
after the first request of every role; tests assert the exact
sequence and value via ``self.durations``.
"""
durations: list[float] = field(default_factory=list)
def sleep(self, seconds: float) -> None:
"""Append the requested duration; return immediately."""
self.durations.append(seconds)
@dataclass
class FakeTransport:
"""In-memory transport. Issues no HTTP requests.
The fetcher's production transport implements
``fetch(role, url) -> (status, body)``. The fake mirrors that
signature so production and tests agree on the call surface.
Attributes:
responses: Role-keyed canned responses. Each value is a
``(status_code, body_text)`` pair. Missing keys raise
``AssertionError`` so tests fail loudly when a new role
is introduced without a corresponding fixture.
calls: Per-``fetch`` call record. Each entry is a dict with
``"role"`` and ``"url"`` keys, in call order.
"""
responses: dict[str, tuple[int, str]] = field(default_factory=dict)
calls: list[dict[str, str]] = field(default_factory=list)
_transaction: object | None = None
def set_transaction(self, transaction: object) -> None:
"""Store a fake transaction for later inspection.
The real ``RequestsSessionTransport`` calls this once the
two bootstrap requests succeed. Tests pass a
:class:`FakeTransaction` (or any object) and assert the
order by reading ``self._transaction`` later.
"""
self._transaction = transaction
def fetch(self, role: str, url: str) -> tuple[int, str]:
"""Return the canned response for ``role`` and record the call.
Raises ``AssertionError`` if ``role`` is not keyed in
:attr:`responses`; this is the offline equivalent of a 5xx
when the live provider returns nothing for an unknown role.
"""
if role not in self.responses:
raise AssertionError(
f"FakeTransport: no response registered for role {role!r}"
)
self.calls.append({"role": role, "url": url})
return self.responses[role]
class FakeTransaction:
"""Fake MIT ``XClientTransaction``-shaped object for tests.
The real constructor signature (verified offline) is
``ClientTransaction(home_page_response, ondemand_file_response,
random_keyword=None, random_number=None)``. Tests capture both
arguments verbatim so they can assert the fetcher passed the
parsed ``BeautifulSoup`` and raw JS body to the factory in the
right order. The ``generate_transaction_id`` method is directly
defined on this class (delegating to the module-level
:func:`generate_transaction_id` helper) so the fake exactly
matches the real call surface.
"""
def __init__(
self,
home_soup: object,
js_body: object,
**kwargs: object,
) -> None:
self.home_soup = home_soup
self.js_body = js_body
self.kwargs = kwargs
def generate_transaction_id(
self,
method: str,
path: str,
**kwargs: object,
) -> str:
"""Return a fixed test transaction id."""
return generate_transaction_id(method, path, **kwargs)
_FIXED_TRANSACTION_ID: str = "fixed-test-transaction-id"
def generate_transaction_id(
method: str,
path: str,
**kwargs: object,
) -> str:
"""Return a fixed transaction id for offline tests.
The id value is irrelevant to behaviour; tests assert that the
production code calls ``generate_transaction_id(method, path,
**kwargs)`` and that the result reaches the request header.
"""
return _FIXED_TRANSACTION_ID
def make_fake_transaction(
home_soup: object,
js_body: object,
**kwargs: object,
) -> FakeTransaction:
"""Construct a :class:`FakeTransaction` with the production
call signature so a test's ``transaction_factory`` matches the
real ``ClientTransaction(home_page_response, ondemand_file_response)``
two-positional-arg constructor.
Exposed as a free function (rather than baking the signature
into ``FakeTransaction``) so tests can pass it as a
``transaction_factory`` callable to the fetcher.
"""
return FakeTransaction(home_soup, js_body, **kwargs)
scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/.gitkeep
Vendored
View File
Whitespace-only changes.
scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_1.json
Vendored
-58
View File
@@ -1,58 +0,0 @@
{
"data": {
"threaded_conversation_with_injections_v2": {
"instructions": [
{
"type": "TimelineAddEntries",
"entries": [
{
"entryId": "depth-1-tweet",
"content": {
"entryType": "TimelineTimelineItem",
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1700000000000000001",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "8100000001",
"core": {
"name": "fixture-depth-1-author",
"screen_name": "fixture-depth-1-handle"
}
}
}
},
"legacy": {
"id_str": "1700000000000000001",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:01:00 +0000 2026",
"full_text": "fixture-depth-1-text",
"in_reply_to_status_id_str": "1650678968255913985",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "depth-1-cursor-forward",
"content": {
"entryType": "TimelineTimelineCursor",
"cursorType": "ShowMore",
"value": "fixture-depth-2"
}
}
]
}
]
}
}
}
scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_2.json
Vendored
-58
View File
@@ -1,58 +0,0 @@
{
"data": {
"threaded_conversation_with_injections_v2": {
"instructions": [
{
"type": "TimelineAddEntries",
"entries": [
{
"entryId": "depth-2-tweet",
"content": {
"entryType": "TimelineTimelineItem",
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1700000000000000002",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "8100000002",
"core": {
"name": "fixture-depth-2-author",
"screen_name": "fixture-depth-2-handle"
}
}
}
},
"legacy": {
"id_str": "1700000000000000002",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:02:00 +0000 2026",
"full_text": "fixture-depth-2-text",
"in_reply_to_status_id_str": "1700000000000000001",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "depth-2-cursor-forward",
"content": {
"entryType": "TimelineTimelineCursor",
"cursorType": "Bottom",
"value": "fixture-depth-3"
}
}
]
}
]
}
}
}
scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_3.json
Vendored
-66
View File
@@ -1,66 +0,0 @@
{
"data": {
"threaded_conversation_with_injections_v2": {
"instructions": [
{
"type": "TimelineAddEntries",
"entries": [
{
"entryId": "depth-3-tweet",
"content": {
"entryType": "TimelineTimelineItem",
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1700000000000000003",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "8100000003",
"core": {
"name": "fixture-depth-3-author",
"screen_name": "fixture-depth-3-handle"
}
}
}
},
"legacy": {
"id_str": "1700000000000000003",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:03:00 +0000 2026",
"full_text": "fixture-depth-3-text",
"in_reply_to_status_id_str": "1700000000000000002",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "depth-3-cursor-echo-current",
"content": {
"entryType": "TimelineTimelineCursor",
"cursorType": "Bottom",
"value": "fixture-depth-3"
}
},
{
"entryId": "depth-3-cursor-forward",
"content": {
"entryType": "TimelineTimelineCursor",
"cursorType": "ShowMoreThreads",
"value": "fixture-depth-4"
}
}
]
}
]
}
}
}
scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_4.json
Vendored
-58
View File
@@ -1,58 +0,0 @@
{
"data": {
"threaded_conversation_with_injections_v2": {
"instructions": [
{
"type": "TimelineAddEntries",
"entries": [
{
"entryId": "depth-4-tweet",
"content": {
"entryType": "TimelineTimelineItem",
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1700000000000000004",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "8100000004",
"core": {
"name": "fixture-depth-4-author",
"screen_name": "fixture-depth-4-handle"
}
}
}
},
"legacy": {
"id_str": "1700000000000000004",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:04:00 +0000 2026",
"full_text": "fixture-depth-4-text",
"in_reply_to_status_id_str": "1700000000000000003",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "depth-4-cursor-forward",
"content": {
"entryType": "TimelineTimelineCursor",
"cursorType": "ShowMore",
"value": "fixture-depth-5"
}
}
]
}
]
}
}
}
scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_5.json
Vendored
-58
View File
@@ -1,58 +0,0 @@
{
"data": {
"threaded_conversation_with_injections_v2": {
"instructions": [
{
"type": "TimelineAddEntries",
"entries": [
{
"entryId": "depth-5-tweet",
"content": {
"entryType": "TimelineTimelineItem",
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1700000000000000005",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "8100000005",
"core": {
"name": "fixture-depth-5-author",
"screen_name": "fixture-depth-5-handle"
}
}
}
},
"legacy": {
"id_str": "1700000000000000005",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:05:00 +0000 2026",
"full_text": "fixture-depth-5-text",
"in_reply_to_status_id_str": "1700000000000000004",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "depth-5-cursor-forward",
"content": {
"entryType": "TimelineTimelineCursor",
"cursorType": "Bottom",
"value": "fixture-depth-6"
}
}
]
}
]
}
}
}
scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_6.json
Vendored
-58
View File
@@ -1,58 +0,0 @@
{
"data": {
"threaded_conversation_with_injections_v2": {
"instructions": [
{
"type": "TimelineAddEntries",
"entries": [
{
"entryId": "depth-6-tweet",
"content": {
"entryType": "TimelineTimelineItem",
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1700000000000000006",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "8100000006",
"core": {
"name": "fixture-depth-6-author",
"screen_name": "fixture-depth-6-handle"
}
}
}
},
"legacy": {
"id_str": "1700000000000000006",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:06:00 +0000 2026",
"full_text": "fixture-depth-6-text",
"in_reply_to_status_id_str": "1700000000000000005",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "depth-6-cursor-forward",
"content": {
"entryType": "TimelineTimelineCursor",
"cursorType": "ShowMoreThreads",
"value": "fixture-depth-7"
}
}
]
}
]
}
}
}
scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_7.json
Vendored
-50
View File
@@ -1,50 +0,0 @@
{
"data": {
"threaded_conversation_with_injections_v2": {
"instructions": [
{
"type": "TimelineAddEntries",
"entries": [
{
"entryId": "depth-7-tweet",
"content": {
"entryType": "TimelineTimelineItem",
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1700000000000000007",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "8100000007",
"core": {
"name": "fixture-depth-7-author",
"screen_name": "fixture-depth-7-handle"
}
}
}
},
"legacy": {
"id_str": "1700000000000000007",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:07:00 +0000 2026",
"full_text": "fixture-depth-7-text",
"in_reply_to_status_id_str": "1700000000000000006",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
}
]
}
]
}
}
}
scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_tip_branch.json
Vendored
-89
View File
@@ -1,89 +0,0 @@
{
"data": {
"threaded_conversation_with_injections_v2": {
"instructions": [
{
"type": "TimelineAddEntries",
"entries": [
{
"entryId": "continuation-tweet-lottes",
"content": {
"entryType": "TimelineTimelineItem",
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651268028795961344",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "444444444",
"core": {
"name": "fixture-author-lottes-continuation",
"screen_name": "fixture-handle-lottes-continuation"
}
}
}
},
"legacy": {
"id_str": "1651268028795961344",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:03:00 +0000 2026",
"full_text": "fixture-text-lottes-continuation",
"in_reply_to_status_id_str": "1651253961524142081",
"in_reply_to_user_id_str": "333333333",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "continuation-tweet-tip",
"content": {
"entryType": "TimelineTimelineItem",
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651282559287042048",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "555555555",
"core": {
"name": "fixture-author-tip-continuation",
"screen_name": "fixture-handle-tip-continuation"
}
}
}
},
"legacy": {
"id_str": "1651282559287042048",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:04:00 +0000 2026",
"full_text": "fixture-text-tip-continuation",
"in_reply_to_status_id_str": "1651268028795961344",
"in_reply_to_user_id_str": "444444444",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
}
]
}
]
}
}
}
scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/root_tweet_detail.json
Vendored
-539
View File
@@ -1,539 +0,0 @@
{
"data": {
"threaded_conversation_with_injections_v2": {
"instructions": [
{
"type": "TimelineAddEntries",
"entries": [
{
"entryId": "tweet-root",
"content": {
"entryType": "TimelineTimelineItem",
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1650678968255913985",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "955913985",
"core": {
"name": "fixture-author-root-page",
"screen_name": "fixture-handle-root-page"
}
}
}
},
"legacy": {
"id_str": "1650678968255913985",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-root-page",
"in_reply_to_status_id_str": null,
"quoted_status_id_str": null,
"reply_count": 5,
"retweet_count": 6,
"favorite_count": 7
},
"views": {
"count": 1000
}
}
}
}
}
}
]
},
{
"type": "TimelineAddToModule",
"moduleItems": [
{
"entryId": "mod-01-abductee_org-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1650685805776732160",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "976732160",
"core": {
"name": "fixture-author-abductee_org-page",
"screen_name": "fixture-handle-abductee-org-page"
}
}
}
},
"legacy": {
"id_str": "1650685805776732160",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-abductee-org-page",
"in_reply_to_status_id_str": "1650678968255913985",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-02-streleav-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1650915682136240129",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "936240129",
"core": {
"name": "fixture-author-streleav-page",
"screen_name": "fixture-handle-streleav-page"
}
}
}
},
"legacy": {
"id_str": "1650915682136240129",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-streleav-page",
"in_reply_to_status_id_str": "1650678968255913985",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-03-wadetb-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651056827839180800",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "939180800",
"core": {
"name": "fixture-author-wadetb-page",
"screen_name": "fixture-handle-wadetb-page"
}
}
}
},
"legacy": {
"id_str": "1651056827839180800",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-wadetb-page",
"in_reply_to_status_id_str": "1650678968255913985",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-04-Christer-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651086935467917312",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "967917312",
"core": {
"name": "fixture-author-Christer-page",
"screen_name": "fixture-handle-Christer-page"
}
}
}
},
"legacy": {
"id_str": "1651086935467917312",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-Christer-page",
"in_reply_to_status_id_str": "1650678968255913985",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-05-Seb-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651168030557077504",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "957077504",
"core": {
"name": "fixture-author-Seb-page",
"screen_name": "fixture-handle-Seb-page"
}
}
}
},
"legacy": {
"id_str": "1651168030557077504",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-Seb-page",
"in_reply_to_status_id_str": "1650678968255913985",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-06-kenpex-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651253961524142081",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "924142081",
"core": {
"name": "fixture-author-kenpex-page",
"screen_name": "fixture-handle-kenpex-page"
}
}
}
},
"legacy": {
"id_str": "1651253961524142081",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-kenpex-to-wadetb-page",
"in_reply_to_status_id_str": "1651056827839180800",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-07-kenpex-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651254727160795137",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "960795137",
"core": {
"name": "fixture-author-kenpex-page",
"screen_name": "fixture-handle-kenpex-page-2"
}
}
}
},
"legacy": {
"id_str": "1651254727160795137",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-kenpex-to-Christer-page",
"in_reply_to_status_id_str": "1651086935467917312",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-08-Erik-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651295755293036544",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "993036544",
"core": {
"name": "fixture-author-Erik-page",
"screen_name": "fixture-handle-Erik-page"
}
}
}
},
"legacy": {
"id_str": "1651295755293036544",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-Erik-page",
"in_reply_to_status_id_str": "1650678968255913985",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-09-npatsiouras-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651508233243095043",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "943095043",
"core": {
"name": "fixture-author-npatsiouras-page",
"screen_name": "fixture-handle-npatsiouras-page"
}
}
}
},
"legacy": {
"id_str": "1651508233243095043",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-npatsiouras-page",
"in_reply_to_status_id_str": "1651254727160795137",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-10-Christer-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651510424649744385",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "949744385",
"core": {
"name": "fixture-author-Christer-page",
"screen_name": "fixture-handle-Christer-page-2"
}
}
}
},
"legacy": {
"id_str": "1651510424649744385",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-Christer-to-npatsiouras-page",
"in_reply_to_status_id_str": "1651508233243095043",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-11-Christer-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651511316010663936",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "910663936",
"core": {
"name": "fixture-author-Christer-page",
"screen_name": "fixture-handle-Christer-page-3"
}
}
}
},
"legacy": {
"id_str": "1651511316010663936",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-Christer-self-page",
"in_reply_to_status_id_str": "1651510424649744385",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-12-Jace-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651579576089300992",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "989300992",
"core": {
"name": "fixture-author-Jace-page",
"screen_name": "fixture-handle-Jace-page"
}
}
}
},
"legacy": {
"id_str": "1651579576089300992",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-Jace-page",
"in_reply_to_status_id_str": "1650678968255913985",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-13-kenpex-page",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651636988267888641",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "967888641",
"core": {
"name": "fixture-author-kenpex-page",
"screen_name": "fixture-handle-kenpex-page-3"
}
}
}
},
"legacy": {
"id_str": "1651636988267888641",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-kenpex-to-Jace-page",
"in_reply_to_status_id_str": "1651579576089300992",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
}
]
}
]
}
}
}
scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/tip_tweet_detail.json
Vendored
-277
View File
@@ -1,277 +0,0 @@
{
"data": {
"threaded_conversation_with_injections_v2": {
"instructions": [
{
"type": "TimelineAddEntries",
"entries": [
{
"entryId": "tweet-root",
"content": {
"entryType": "TimelineTimelineItem",
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1650678968255913985",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "111111111",
"core": {
"name": "fixture-author-root",
"screen_name": "fixture-handle-root"
},
"legacy": {
"name": "fixture-author-root-legacy-should-be-ignored",
"screen_name": "fixture-handle-root-legacy-should-be-ignored"
}
}
}
},
"legacy": {
"id_str": "1650678968255913985",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:00:00 +0000 2026",
"full_text": "fixture-text-root-legacy",
"in_reply_to_status_id_str": null,
"in_reply_to_user_id_str": null,
"quoted_status_id_str": null,
"reply_count": 1,
"retweet_count": 2,
"favorite_count": 3
},
"views": {"count": 100}
}
}
}
}
},
{
"entryId": "conversationthread-1",
"content": {
"entryType": "TimelineTimelineModule",
"items": [
{
"entryId": "mod-tweet-wadetb",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651056827839180800",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "222222222",
"core": {
"name": "fixture-author-wadetb",
"screen_name": "fixture-handle-wadetb"
}
}
}
},
"legacy": {
"id_str": "1651056827839180800",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:01:00 +0000 2026",
"full_text": "fixture-text-wadetb-full",
"in_reply_to_status_id_str": "1650678968255913985",
"in_reply_to_user_id_str": "111111111",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0,
"extended_entities": {
"media": [
{
"media_url_https": "https://pbs.twimg.com/media/fixture-photo-wadetb.jpg",
"type": "photo"
}
]
}
}
}
}
}
}
},
{
"entryId": "mod-tweet-kenpex",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651253961524142081",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "333333333",
"core": {
"name": "fixture-author-kenpex",
"screen_name": "fixture-handle-kenpex"
}
}
}
},
"note_tweet": {
"note_tweet_results": {
"result": {
"text": "fixture-note-tweet-text-kenpex-override"
}
}
},
"legacy": {
"id_str": "1651253961524142081",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:02:00 +0000 2026",
"full_text": "fixture-text-kenpex-legacy-should-be-ignored",
"in_reply_to_status_id_str": "1651056827839180800",
"in_reply_to_user_id_str": "222222222",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
}
}
}
}
}
},
{
"entryId": "mod-tweet-lottes",
"itemType": "TimelineTimelineItem",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "TweetWithVisibilityResults",
"tweet": {
"__typename": "Tweet",
"rest_id": "1651268028795961344",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "444444444",
"core": {
"name": "fixture-author-lottes",
"screen_name": "fixture-handle-lottes"
}
}
}
},
"legacy": {
"id_str": "1651268028795961344",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:03:00 +0000 2026",
"full_text": "fixture-text-lottes",
"in_reply_to_status_id_str": "1651253961524142081",
"in_reply_to_user_id_str": "333333333",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0,
"extended_entities": {
"media": [
{
"media_url_https": "https://pbs.twimg.com/media/fixture-photo-lottes-poster.jpg",
"type": "video",
"video_info": {
"variants": [
{
"content_type": "video/mp4",
"url": "https://video.twimg.com/fixture-video-low.mp4",
"bitrate": 320000
},
{
"content_type": "video/mp4",
"url": "https://video.twimg.com/fixture-video-med.mp4",
"bitrate": 832000
},
{
"content_type": "video/mp4",
"url": "https://video.twimg.com/fixture-video-high.mp4",
"bitrate": 1280000
},
{
"content_type": "application/x-mpegURL",
"url": "https://video.twimg.com/fixture-video.m3u8",
"bitrate": 0
}
]
}
}
]
}
}
}
}
}
}
}
},
{
"entryId": "mod-tweet-tip",
"itemType": "TimelineTweet",
"item": {
"itemContent": {
"tweet_results": {
"result": {
"__typename": "Tweet",
"rest_id": "1651282559287042048",
"core": {
"user_results": {
"result": {
"__typename": "User",
"id_str": "555555555",
"core": {
"name": "fixture-author-tip",
"screen_name": "fixture-handle-tip"
}
}
}
},
"legacy": {
"id_str": "1651282559287042048",
"conversation_id_str": "1650678968255913985",
"created_at": "Mon Apr 26 12:04:00 +0000 2026",
"full_text": "fixture-text-tip",
"in_reply_to_status_id_str": "1651268028795961344",
"in_reply_to_user_id_str": "444444444",
"quoted_status_id_str": null,
"reply_count": 0,
"retweet_count": 0,
"favorite_count": 0
},
"views": {"count": "1234"}
}
}
}
}
},
{
"entryId": "mod-cursor-top",
"itemType": "TimelineTimelineCursor",
"item": {
"itemContent": {
"cursorType": "Top",
"value": "fixture-top-cursor-token-should-be-ignored"
}
}
}
]
}
}
]
}
]
}
}
}
scripts/twitter/unfinished_bloat/tests/test_fetch_thread.py
-2025
View File
File diff suppressed because it is too large. Load diff
secrets/README.md
-16
View File
@@ -1,21 +1,5 @@
Local credentials. Gitignored except this file.
| File | Contents |
| --- | --- |
| `twitter_cookies.txt` | Netscape cookies for gallery-dl |
| `credentials.toml` | `[catbox] userhash`, `[backup] id` and `root` |
Shape of `credentials.toml`:
```toml
[catbox]
userhash = ""
[backup]
id = "cozy"
root = ""
```
The scripts read these files. They never print the values.
Do not commit this folder. Do not paste the contents into chat or issues.