From f4a36f235842d081befc920cbb03f3384cb67a1e Mon Sep 17 00:00:00 2001 From: jackwener Date: Sat, 7 Mar 2026 21:41:33 +0800 Subject: [PATCH] refactor: fix all code review issues Bug fixes (P0): - _extract_cursor: both branches now check cursorType=='Bottom', preventing Top cursor from corrupting pagination state - _api_get: add retry loop with exponential backoff for HTTP 429 and JSON error code 88 rate limits Improvements (P1): - _deep_get: support int keys for list index access - Extract BEARER_TOKEN/USER_AGENT into constants.py (was duplicated in auth.py and client.py) Code quality (P2): - fetch_likes: add override_base_variables=True to avoid sending latestControlAvailable/requestContext which Likes endpoint rejects - CLI: extract _fetch_and_display() helper to dedup favorite/likes - _get_client(): remove useless try/except that re-raised same error - formatter: remove no-op tweets_to_json wrapper - _as_int/_as_float: filter.py now imports from config.py Polish (P3): - _parse_tweet_result: split into _extract_media + _extract_author - CLI module docstring: add likes command - verify_cookies: increase timeout from 3s to 10s --- twitter_cli/auth.py | 18 ++-- twitter_cli/cli.py | 99 ++++++++---------- twitter_cli/client.py | 219 ++++++++++++++++++++++++--------------- twitter_cli/constants.py | 12 +++ twitter_cli/filter.py | 24 +---- twitter_cli/formatter.py | 8 -- 6 files changed, 195 insertions(+), 185 deletions(-) create mode 100644 twitter_cli/constants.py diff --git a/twitter_cli/auth.py b/twitter_cli/auth.py index 8db7c6a..518671a 100644 --- a/twitter_cli/auth.py +++ b/twitter_cli/auth.py @@ -17,13 +17,11 @@ import urllib.error import urllib.request from typing import Dict, Optional +from .constants import BEARER_TOKEN, USER_AGENT + logger = logging.getLogger(__name__) -# Public bearer token (same as in client.py) -_BEARER_TOKEN = ( - "AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs" - "%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA" -) + def load_from_env() -> Optional[Dict[str, str]]: @@ -49,16 +47,12 @@ def verify_cookies(auth_token, ct0): ] headers = { - "Authorization": "Bearer %s" % _BEARER_TOKEN, + "Authorization": "Bearer %s" % BEARER_TOKEN, "Cookie": "auth_token=%s; ct0=%s" % (auth_token, ct0), "X-Csrf-Token": ct0, "X-Twitter-Active-User": "yes", "X-Twitter-Auth-Type": "OAuth2Session", - "User-Agent": ( - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " - "AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/131.0.0.0 Safari/537.36" - ), + "User-Agent": USER_AGENT, } for url in urls: @@ -68,7 +62,7 @@ def verify_cookies(auth_token, ct0): ctx = ssl.create_default_context() try: - with urllib.request.urlopen(req, context=ctx, timeout=3) as resp: + with urllib.request.urlopen(req, context=ctx, timeout=10) as resp: data = json.loads(resp.read().decode("utf-8")) return {"screen_name": data.get("screen_name", "")} except urllib.error.HTTPError as e: diff --git a/twitter_cli/cli.py b/twitter_cli/cli.py index 45f2983..f179285 100644 --- a/twitter_cli/cli.py +++ b/twitter_cli/cli.py @@ -7,6 +7,7 @@ Usage: twitter feed --filter # enable score-based filtering twitter feed --json # JSON output twitter favorite # fetch bookmarks + twitter likes # fetch your liked tweets twitter feed --input tweets.json # load existing data twitter feed --output out.json # save filtered tweets twitter user elonmusk # view user profile @@ -64,10 +65,7 @@ def _get_client(): # type: () -> TwitterClient """Create an authenticated API client.""" console.print("\nšŸ” Getting Twitter cookies...") - try: - cookies = get_cookies() - except RuntimeError as exc: - raise RuntimeError(str(exc)) + cookies = get_cookies() return TwitterClient(cookies["auth_token"], cookies["ct0"]) @@ -103,6 +101,35 @@ def cli(verbose): _setup_logging(verbose) +def _fetch_and_display(fetch_fn, label, emoji, max_count, as_json, output_file, do_filter): + # type: (Callable[..., Any], str, str, Optional[int], bool, Optional[str], bool) -> None + """Common fetch-filter-display logic for timeline-like commands.""" + config = load_config() + try: + fetch_count = _resolve_fetch_count(max_count, config.get("fetch", {}).get("count", 50)) + console.print("%s Fetching %s (%d tweets)...\n" % (emoji, label, fetch_count)) + start = time.time() + tweets = fetch_fn(fetch_count) + elapsed = time.time() - start + console.print("āœ… Fetched %d %s in %.1fs\n" % (len(tweets), label, elapsed)) + except RuntimeError as exc: + console.print("[red]āŒ %s[/red]" % exc) + sys.exit(1) + + filtered = _apply_filter(tweets, do_filter, config) + + if output_file: + Path(output_file).write_text(tweets_to_json(filtered), encoding="utf-8") + console.print("šŸ’¾ Saved to %s\n" % output_file) + + if as_json: + click.echo(tweets_to_json(filtered)) + return + + print_tweet_table(filtered, console, title="%s %s ā€” %d tweets" % (emoji, label, len(filtered))) + console.print() + + @cli.command() @click.option( "--type", @@ -166,31 +193,11 @@ def feed(feed_type, max_count, as_json, input_file, output_file, do_filter): def favorite(max_count, as_json, output_file, do_filter): # type: (Optional[int], bool, Optional[str], bool) -> None """Fetch bookmarked (favorite) tweets.""" - config = load_config() - try: - fetch_count = _resolve_fetch_count(max_count, config.get("fetch", {}).get("count", 50)) - client = _get_client() - console.print("šŸ”– Fetching favorites (%d tweets)...\n" % fetch_count) - start = time.time() - tweets = client.fetch_bookmarks(fetch_count) - elapsed = time.time() - start - console.print("āœ… Fetched %d favorites in %.1fs\n" % (len(tweets), elapsed)) - except RuntimeError as exc: - console.print("[red]āŒ %s[/red]" % exc) - sys.exit(1) - - filtered = _apply_filter(tweets, do_filter, config) - - if output_file: - Path(output_file).write_text(tweets_to_json(filtered), encoding="utf-8") - console.print("šŸ’¾ Saved to %s\n" % output_file) - - if as_json: - click.echo(tweets_to_json(filtered)) - return - - print_tweet_table(filtered, console, title="šŸ”– Favorites ā€” %d tweets" % len(filtered)) - console.print() + client = _get_client() + _fetch_and_display( + lambda count: client.fetch_bookmarks(count), + "favorites", "šŸ”–", max_count, as_json, output_file, do_filter, + ) @cli.command() @@ -201,33 +208,13 @@ def favorite(max_count, as_json, output_file, do_filter): def likes(max_count, as_json, output_file, do_filter): # type: (Optional[int], bool, Optional[str], bool) -> None """Fetch your liked tweets (only available for the authenticated user).""" - config = load_config() - try: - fetch_count = _resolve_fetch_count(max_count, config.get("fetch", {}).get("count", 50)) - client = _get_client() - console.print("šŸ‘¤ Resolving your user ID...") - user_id = client.fetch_current_user_id() - console.print("ā¤ļø Fetching likes (%d tweets)...\n" % fetch_count) - start = time.time() - tweets = client.fetch_likes(user_id, fetch_count) - elapsed = time.time() - start - console.print("āœ… Fetched %d likes in %.1fs\n" % (len(tweets), elapsed)) - except RuntimeError as exc: - console.print("[red]āŒ %s[/red]" % exc) - sys.exit(1) - - filtered = _apply_filter(tweets, do_filter, config) - - if output_file: - Path(output_file).write_text(tweets_to_json(filtered), encoding="utf-8") - console.print("šŸ’¾ Saved to %s\n" % output_file) - - if as_json: - click.echo(tweets_to_json(filtered)) - return - - print_tweet_table(filtered, console, title="ā¤ļø Likes ā€” %d tweets" % len(filtered)) - console.print() + client = _get_client() + console.print("šŸ‘¤ Resolving your user ID...") + user_id = client.fetch_current_user_id() + _fetch_and_display( + lambda count: client.fetch_likes(user_id, count), + "likes", "ā¤ļø", max_count, as_json, output_file, do_filter, + ) @cli.command() diff --git a/twitter_cli/client.py b/twitter_cli/client.py index bdb13cd..100d793 100644 --- a/twitter_cli/client.py +++ b/twitter_cli/client.py @@ -7,19 +7,18 @@ import logging import math import re import ssl +import time import urllib.error import urllib.parse import urllib.request +from .constants import BEARER_TOKEN, USER_AGENT from .models import Author, Metrics, Tweet, TweetMedia, UserProfile logger = logging.getLogger(__name__) -BEARER_TOKEN = ( - "AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs" - "%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA" -) + FALLBACK_QUERY_IDS = { "HomeTimeline": "c-CzHF1LboFilMpsx4ZCrQ", @@ -70,11 +69,7 @@ FEATURES = { "responsive_web_enhance_cards_enabled": False, } -USER_AGENT = ( - "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " - "AppleWebKit/537.36 (KHTML, like Gecko) " - "Chrome/131.0.0.0 Safari/537.36" -) + _cached_query_ids = {} # type: Dict[str, str] _bundles_scanned = False @@ -333,11 +328,18 @@ class TwitterClient: "withBirdwatchNotes": False, "withVoice": True, }, + override_base_variables=True, ) - def _fetch_timeline(self, operation_name, count, get_instructions, extra_variables=None): - # type: (str, int, Callable[[Any], Any], Optional[Dict[str, Any]]) -> List[Tweet] - """Generic timeline fetcher with pagination and deduplication.""" + def _fetch_timeline(self, operation_name, count, get_instructions, extra_variables=None, override_base_variables=False): + # type: (str, int, Callable[[Any], Any], Optional[Dict[str, Any]], bool) -> List[Tweet] + """Generic timeline fetcher with pagination and deduplication. + + Args: + override_base_variables: If True, use only extra_variables + count/cursor + instead of the default timeline base variables. Needed for + endpoints like Likes that reject unknown variables. + """ if count <= 0: return [] @@ -349,12 +351,15 @@ class TwitterClient: while len(tweets) < count and attempts < max_attempts: attempts += 1 - variables = { - "count": min(count - len(tweets) + 5, 40), - "includePromotedContent": False, - "latestControlAvailable": True, - "requestContext": "launch", - } # type: Dict[str, Any] + if override_base_variables: + variables = {"count": min(count - len(tweets) + 5, 40)} # type: Dict[str, Any] + else: + variables = { + "count": min(count - len(tweets) + 5, 40), + "includePromotedContent": False, + "latestControlAvailable": True, + "requestContext": "launch", + } # type: Dict[str, Any] if extra_variables: variables.update(extra_variables) if cursor: @@ -412,31 +417,55 @@ class TwitterClient: def _api_get(self, url): # type: (str) -> Dict[str, Any] - """Make authenticated GET request to Twitter API.""" + """Make authenticated GET request to Twitter API with retry on rate limits.""" headers = self._build_headers() - request = urllib.request.Request(url) - for key, value in headers.items(): - request.add_header(key, value) + max_retries = 3 + base_delay = 5.0 - try: - with urllib.request.urlopen(request, context=_create_ssl_context(), timeout=30) as response: - payload = response.read().decode("utf-8") - except urllib.error.HTTPError as exc: - body = exc.read().decode("utf-8", errors="replace") - message = "Twitter API error %d: %s" % (exc.code, body[:500]) - raise TwitterAPIError(exc.code, message) - except urllib.error.URLError as exc: - raise TwitterAPIError(0, "Twitter API network error: %s" % exc.reason) + for attempt in range(max_retries + 1): + request = urllib.request.Request(url) + for key, value in headers.items(): + request.add_header(key, value) - try: - parsed = json.loads(payload) - except json.JSONDecodeError: - raise TwitterAPIError(0, "Twitter API returned invalid JSON") + try: + with urllib.request.urlopen(request, context=_create_ssl_context(), timeout=30) as response: + payload = response.read().decode("utf-8") + except urllib.error.HTTPError as exc: + if exc.code == 429 and attempt < max_retries: + wait = base_delay * (2 ** attempt) + logger.warning( + "Rate limited (429), retrying in %.1fs (attempt %d/%d)", + wait, attempt + 1, max_retries, + ) + time.sleep(wait) + continue + body = exc.read().decode("utf-8", errors="replace") + message = "Twitter API error %d: %s" % (exc.code, body[:500]) + raise TwitterAPIError(exc.code, message) + except urllib.error.URLError as exc: + raise TwitterAPIError(0, "Twitter API network error: %s" % exc.reason) - if isinstance(parsed, dict) and parsed.get("errors"): - message = parsed["errors"][0].get("message", "Unknown error") - raise TwitterAPIError(0, "Twitter API returned errors: %s" % message) - return parsed + try: + parsed = json.loads(payload) + except json.JSONDecodeError: + raise TwitterAPIError(0, "Twitter API returned invalid JSON") + + if isinstance(parsed, dict) and parsed.get("errors"): + err_msg = parsed["errors"][0].get("message", "Unknown error") + # Rate limit can also surface as a JSON error (code 88) + err_code = parsed["errors"][0].get("code", 0) + if err_code == 88 and attempt < max_retries: + wait = base_delay * (2 ** attempt) + logger.warning( + "Rate limited (code 88), retrying in %.1fs (attempt %d/%d)", + wait, attempt + 1, max_retries, + ) + time.sleep(wait) + continue + raise TwitterAPIError(0, "Twitter API returned errors: %s" % err_msg) + return parsed + + raise TwitterAPIError(429, "Rate limited after %d retries" % max_retries) def _parse_timeline_response(self, data, get_instructions): # type: (Any, Callable[[Any], Any]) -> Tuple[List[Tweet], Optional[str]] @@ -516,44 +545,12 @@ class TwitterClient: actual_user = _deep_get(rt_core, "user_results", "result") or {} actual_user_legacy = actual_user.get("legacy", {}) - media = [] # type: List[TweetMedia] - for media_item in _deep_get(actual_legacy, "extended_entities", "media") or []: - media_type = media_item.get("type", "") - if media_type == "photo": - media.append( - TweetMedia( - type="photo", - url=media_item.get("media_url_https", ""), - width=_deep_get(media_item, "original_info", "width"), - height=_deep_get(media_item, "original_info", "height"), - ) - ) - elif media_type in {"video", "animated_gif"}: - variants = media_item.get("video_info", {}).get("variants", []) - mp4_variants = [item for item in variants if item.get("content_type") == "video/mp4"] - mp4_variants.sort(key=lambda item: item.get("bitrate", 0), reverse=True) - media.append( - TweetMedia( - type=media_type, - url=mp4_variants[0]["url"] if mp4_variants else media_item.get("media_url_https", ""), - width=_deep_get(media_item, "original_info", "width"), - height=_deep_get(media_item, "original_info", "height"), - ) - ) - + media = _extract_media(actual_legacy) urls = [item.get("expanded_url", "") for item in _deep_get(actual_legacy, "entities", "urls") or []] quoted = _deep_get(actual_data, "quoted_status_result", "result") quoted_tweet = self._parse_tweet_result(quoted, depth=depth + 1) if isinstance(quoted, dict) else None + author = _extract_author(actual_user, actual_user_legacy) - actual_user_core = actual_user.get("core", {}) - user_name = actual_user_core.get("name") or actual_user_legacy.get("name") or actual_user.get("name", "Unknown") - user_screen_name = ( - actual_user_core.get("screen_name") - or actual_user_legacy.get("screen_name") - or actual_user.get("screen_name", "unknown") - ) - user_profile_image = actual_user.get("avatar", {}).get("image_url") or actual_user_legacy.get("profile_image_url_https", "") - user_verified = bool(actual_user.get("is_blue_verified") or actual_user_legacy.get("verified", False)) retweeted_by = None # type: Optional[str] if is_retweet: retweeted_by = user_core.get("screen_name") or user_legacy.get("screen_name", "unknown") @@ -561,13 +558,7 @@ class TwitterClient: return Tweet( id=actual_data.get("rest_id", ""), text=actual_legacy.get("full_text", ""), - author=Author( - id=actual_user.get("rest_id", ""), - name=user_name, - screen_name=user_screen_name, - profile_image_url=user_profile_image, - verified=user_verified, - ), + author=author, metrics=Metrics( likes=_to_int(actual_legacy.get("favorite_count"), 0), retweets=_to_int(actual_legacy.get("retweet_count"), 0), @@ -586,23 +577,79 @@ class TwitterClient: ) +def _extract_media(legacy): + # type: (Dict[str, Any]) -> List[TweetMedia] + """Extract media items from tweet legacy data.""" + media = [] # type: List[TweetMedia] + for media_item in _deep_get(legacy, "extended_entities", "media") or []: + media_type = media_item.get("type", "") + if media_type == "photo": + media.append( + TweetMedia( + type="photo", + url=media_item.get("media_url_https", ""), + width=_deep_get(media_item, "original_info", "width"), + height=_deep_get(media_item, "original_info", "height"), + ) + ) + elif media_type in {"video", "animated_gif"}: + variants = media_item.get("video_info", {}).get("variants", []) + mp4_variants = [v for v in variants if v.get("content_type") == "video/mp4"] + mp4_variants.sort(key=lambda v: v.get("bitrate", 0), reverse=True) + media.append( + TweetMedia( + type=media_type, + url=mp4_variants[0]["url"] if mp4_variants else media_item.get("media_url_https", ""), + width=_deep_get(media_item, "original_info", "width"), + height=_deep_get(media_item, "original_info", "height"), + ) + ) + return media + + +def _extract_author(user_data, user_legacy): + # type: (Dict[str, Any], Dict[str, Any]) -> Author + """Extract Author from user result data.""" + user_core = user_data.get("core", {}) + return Author( + id=user_data.get("rest_id", ""), + name=user_core.get("name") or user_legacy.get("name") or user_data.get("name", "Unknown"), + screen_name=( + user_core.get("screen_name") + or user_legacy.get("screen_name") + or user_data.get("screen_name", "unknown") + ), + profile_image_url=( + user_data.get("avatar", {}).get("image_url") + or user_legacy.get("profile_image_url_https", "") + ), + verified=bool(user_data.get("is_blue_verified") or user_legacy.get("verified", False)), + ) + + def _deep_get(data, *keys): - # type: (Any, *str) -> Any - """Safely get nested dict values.""" + # type: (Any, *Any) -> Any + """Safely get nested dict/list values. Supports int keys for list access.""" current = data for key in keys: - if not isinstance(current, dict): + if isinstance(key, int): + if isinstance(current, list) and 0 <= key < len(current): + current = current[key] + else: + return None + elif isinstance(current, dict): + current = current.get(key) + else: return None - current = current.get(key) return current def _extract_cursor(content): # type: (Dict[str, Any]) -> Optional[str] - """Extract pagination cursor from timeline content.""" + """Extract Bottom pagination cursor from timeline content.""" if content.get("cursorType") == "Bottom": return content.get("value") - if content.get("entryType") == "TimelineTimelineCursor": + if content.get("entryType") == "TimelineTimelineCursor" and content.get("cursorType") == "Bottom": return content.get("value") return None diff --git a/twitter_cli/constants.py b/twitter_cli/constants.py new file mode 100644 index 0000000..f624feb --- /dev/null +++ b/twitter_cli/constants.py @@ -0,0 +1,12 @@ +"""Shared constants for twitter-cli.""" + +BEARER_TOKEN = ( + "AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs" + "%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA" +) + +USER_AGENT = ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/131.0.0.0 Safari/537.36" +) diff --git a/twitter_cli/filter.py b/twitter_cli/filter.py index 94e74f7..94d338f 100644 --- a/twitter_cli/filter.py +++ b/twitter_cli/filter.py @@ -8,12 +8,8 @@ from __future__ import annotations from dataclasses import replace import math -from typing import Mapping - - -# Type alias for filter weights dict -FilterWeights = Mapping[str, float] +from .config import _as_float, _as_int DEFAULT_WEIGHTS = { "likes": 1.0, @@ -95,21 +91,3 @@ def _build_weights(raw_weights): for key, default_value in DEFAULT_WEIGHTS.items(): merged[key] = _as_float(raw_weights.get(key), default_value) return merged - - -def _as_int(value, default): - # type: (Any, int) -> int - """Best-effort int conversion.""" - try: - return int(value) - except (TypeError, ValueError): - return default - - -def _as_float(value, default): - # type: (Any, float) -> float - """Best-effort float conversion.""" - try: - return float(value) - except (TypeError, ValueError): - return default diff --git a/twitter_cli/formatter.py b/twitter_cli/formatter.py index d40fe13..b3316f1 100644 --- a/twitter_cli/formatter.py +++ b/twitter_cli/formatter.py @@ -7,8 +7,6 @@ from rich.console import Console from rich.panel import Panel from rich.table import Table -from .serialization import tweets_to_json as _tweets_to_json - def format_number(n): # type: (int) -> str @@ -162,12 +160,6 @@ def print_filter_stats(original_count, filtered, console=None): ) -def tweets_to_json(tweets): - # type: (List[Tweet]) -> str - """Export tweets as JSON string.""" - return _tweets_to_json(tweets) - - def print_user_profile(user, console=None): # type: (UserProfile, Optional[Console]) -> None """Print user profile as a rich panel."""