From 80499384c9b98b49aef762c1e99092712785bae5 Mon Sep 17 00:00:00 2001 From: jackwener Date: Sat, 7 Mar 2026 20:07:10 +0800 Subject: [PATCH] feat: add all remaining read/write endpoints Read commands: - twitter tweet : view tweet detail + replies - twitter list : fetch list timeline - twitter followers : list user followers - twitter following : list user following Write commands: - twitter post : create tweet (with --reply-to) - twitter delete : delete tweet - twitter like/unlike : manage likes - twitter rt/unrt : manage retweets - twitter bookmark-add/bookmark-rm : manage bookmarks Infrastructure: - _graphql_post + _api_post for write operations - _fetch_user_list + _parse_user_result for user lists - _deep_get now supports list index access - _build_headers supports POST method for transaction ID --- README.md | 79 ++++++++++-- twitter_cli/cli.py | 272 +++++++++++++++++++++++++++++++++++++++- twitter_cli/client.py | 285 ++++++++++++++++++++++++++++++++++++++++-- 3 files changed, 616 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index d82e4a8..03183c1 100644 --- a/README.md +++ b/README.md @@ -17,12 +17,24 @@ A terminal-first CLI for Twitter/X: read timelines, bookmarks, and user profiles ### Features +**Read:** - Timeline: fetch `for-you` and `following` feeds - Bookmarks: list saved tweets from your account - Search: find tweets by keyword with Top/Latest/Photos/Videos tabs -- User lookup: fetch user profile, tweets, and likes -- JSON output: export feed/bookmarks/user tweets for scripting +- Tweet detail: view a tweet and its replies +- List timeline: fetch tweets from a Twitter List +- User lookup: fetch user profile, tweets, likes, followers, and following +- JSON output: export any data for scripting - Optional scoring filter: rank tweets by engagement weights + +**Write:** +- Post: create new tweets and replies +- Delete: remove your own tweets +- Like / Unlike: manage tweet likes +- Retweet / Unretweet: manage retweets +- Bookmark: add/remove bookmarks + +**Auth:** - Cookie auth: use browser cookies or environment variables ### Installation @@ -73,10 +85,30 @@ twitter search "Claude Code" twitter search "AI agent" -t Latest --max 50 twitter search "机器学习" --json +# Tweet detail (view tweet + replies) +twitter tweet 1234567890 +twitter tweet https://x.com/user/status/1234567890 + +# List timeline +twitter list 1539453138322673664 + # User twitter user elonmusk twitter user-posts elonmusk --max 20 twitter likes elonmusk --max 30 +twitter followers elonmusk --max 50 +twitter following elonmusk --max 50 + +# Write operations +twitter post "Hello from twitter-cli!" +twitter post "reply text" --reply-to 1234567890 +twitter delete 1234567890 +twitter like 1234567890 +twitter unlike 1234567890 +twitter rt 1234567890 +twitter unrt 1234567890 +twitter bookmark-add 1234567890 +twitter bookmark-rm 1234567890 ``` ### Authentication @@ -203,11 +235,22 @@ After installation, OpenClaw can call `twitter-cli` commands directly. ### 功能概览 +**读取:** - 时间线读取：支持 `for-you` 和 `following` - 收藏读取：查看账号书签推文 - 搜索：按关键词搜索推文，支持 Top/Latest/Photos/Videos -- 用户查询：查看用户资料、推文和点赞 +- 推文详情：查看推文及其回复 +- 列表时间线：获取 Twitter List 的推文 +- 用户查询：查看用户资料、推文、点赞、粉丝和关注 - JSON 输出：便于脚本处理 + +**写入:** +- 发推：发布新推文和回复 +- 删除：删除自己的推文 +- 点赞 / 取消点赞 +- 转推 / 取消转推 +- 书签管理：添加/移除书签 + - 可选筛选：按 engagement score 排序 - Cookie 认证：支持环境变量和浏览器自动提取 @@ -216,21 +259,14 @@ After installation, OpenClaw can call `twitter-cli` commands directly. ```bash # 推荐：uv tool uv tool install twitter-cli - -# 其次：pipx -pipx install twitter-cli ``` -### 常用命令 +### 使用指南 ```bash -# 首页推荐流 +# 时间线 twitter feed - -# Following 流 twitter feed -t following - -# 开启筛选（默认不开启） twitter feed --filter # 收藏 @@ -240,10 +276,29 @@ twitter favorite twitter search "Claude Code" twitter search "AI agent" -t Latest --max 50 +# 推文详情 +twitter tweet 1234567890 + +# 列表时间线 +twitter list 1539453138322673664 + # 用户 twitter user elonmusk twitter user-posts elonmusk --max 20 twitter likes elonmusk --max 30 +twitter followers elonmusk +twitter following elonmusk + +# 写操作 +twitter post "你好，世界！" +twitter post "回复内容" --reply-to 1234567890 +twitter delete 1234567890 +twitter like 1234567890 +twitter unlike 1234567890 +twitter rt 1234567890 +twitter unrt 1234567890 +twitter bookmark-add 1234567890 +twitter bookmark-rm 1234567890 ``` ### 认证说明 diff --git a/twitter_cli/cli.py b/twitter_cli/cli.py index 55673ee..ec6f2f7 100644 --- a/twitter_cli/cli.py +++ b/twitter_cli/cli.py @@ -28,7 +28,13 @@ from .auth import get_cookies from .client import TwitterClient from .config import load_config from .filter import filter_tweets -from .formatter import print_filter_stats, print_tweet_table, print_user_profile +from .formatter import ( + print_filter_stats, + print_tweet_detail, + print_tweet_table, + print_user_profile, + print_user_table, +) from .serialization import tweets_from_json, tweets_to_json @@ -320,5 +326,269 @@ def likes(screen_name, max_count, as_json, do_filter): console.print() +@cli.command() +@click.argument("tweet_id") +@click.option("--max", "-n", "max_count", type=int, default=20, help="Max replies to fetch.") +@click.option("--json", "as_json", is_flag=True, help="Output as JSON.") +def tweet(tweet_id, max_count, as_json): + # type: (str, int, bool) -> None + """View a tweet and its replies. TWEET_ID is the numeric tweet ID or full URL.""" + # Extract tweet ID from URL if given + tweet_id = tweet_id.strip().rstrip("/").split("/")[-1] + config = load_config() + try: + client = _get_client(config) + console.print("🐦 Fetching tweet %s...\n" % tweet_id) + start = time.time() + tweets = client.fetch_tweet_detail(tweet_id, max_count) + elapsed = time.time() - start + console.print("✅ Fetched %d tweets in %.1fs\n" % (len(tweets), elapsed)) + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + if as_json: + click.echo(tweets_to_json(tweets)) + return + + if tweets: + print_tweet_detail(tweets[0], console) + if len(tweets) > 1: + console.print("\n💬 Replies:") + print_tweet_table(tweets[1:], console, title="💬 Replies — %d" % (len(tweets) - 1)) + console.print() + + +@cli.command(name="list") +@click.argument("list_id") +@click.option("--max", "-n", "max_count", type=int, default=20, help="Max tweets to fetch.") +@click.option("--json", "as_json", is_flag=True, help="Output as JSON.") +@click.option("--filter", "do_filter", is_flag=True, help="Enable score-based filtering.") +def list_timeline(list_id, max_count, as_json, do_filter): + # type: (str, int, bool, bool) -> None + """Fetch tweets from a Twitter List. LIST_ID is the numeric list ID.""" + config = load_config() + try: + fetch_count = _resolve_fetch_count(max_count, 20) + client = _get_client(config) + console.print("📋 Fetching list %s (%d tweets)...\n" % (list_id, fetch_count)) + start = time.time() + tweets = client.fetch_list_timeline(list_id, fetch_count) + elapsed = time.time() - start + console.print("✅ Fetched %d tweets 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 as_json: + click.echo(tweets_to_json(filtered)) + return + + print_tweet_table(filtered, console, title="📋 List — %d tweets" % len(filtered)) + console.print() + + +@cli.command() +@click.argument("screen_name") +@click.option("--max", "-n", "max_count", type=int, default=20, help="Max users to fetch.") +@click.option("--json", "as_json", is_flag=True, help="Output as JSON.") +def followers(screen_name, max_count, as_json): + # type: (str, int, bool) -> None + """List followers of a user. SCREEN_NAME is the @handle (without @).""" + screen_name = screen_name.lstrip("@") + config = load_config() + try: + client = _get_client(config) + console.print("👤 Fetching @%s's profile..." % screen_name) + profile = client.fetch_user(screen_name) + console.print("👥 Fetching followers (%d)...\n" % max_count) + start = time.time() + users = client.fetch_followers(profile.id, max_count) + elapsed = time.time() - start + console.print("✅ Fetched %d followers in %.1fs\n" % (len(users), elapsed)) + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + if as_json: + import json + click.echo(json.dumps([{"id": u.id, "name": u.name, "screen_name": u.screen_name, + "bio": u.bio, "followers": u.followers_count, + "following": u.following_count} for u in users], indent=2, ensure_ascii=False)) + return + + print_user_table(users, console, title="👥 @%s followers — %d" % (screen_name, len(users))) + console.print() + + +@cli.command() +@click.argument("screen_name") +@click.option("--max", "-n", "max_count", type=int, default=20, help="Max users to fetch.") +@click.option("--json", "as_json", is_flag=True, help="Output as JSON.") +def following(screen_name, max_count, as_json): + # type: (str, int, bool) -> None + """List accounts a user is following. SCREEN_NAME is the @handle (without @).""" + screen_name = screen_name.lstrip("@") + config = load_config() + try: + client = _get_client(config) + console.print("👤 Fetching @%s's profile..." % screen_name) + profile = client.fetch_user(screen_name) + console.print("👥 Fetching following (%d)...\n" % max_count) + start = time.time() + users = client.fetch_following(profile.id, max_count) + elapsed = time.time() - start + console.print("✅ Fetched %d following in %.1fs\n" % (len(users), elapsed)) + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + if as_json: + import json + click.echo(json.dumps([{"id": u.id, "name": u.name, "screen_name": u.screen_name, + "bio": u.bio, "followers": u.followers_count, + "following": u.following_count} for u in users], indent=2, ensure_ascii=False)) + return + + print_user_table(users, console, title="👥 @%s following — %d" % (screen_name, len(users))) + console.print() + + +# ── Write commands ────────────────────────────────────────────────────── + +@cli.command() +@click.argument("text") +@click.option("--reply-to", "-r", default=None, help="Reply to this tweet ID.") +def post(text, reply_to): + # type: (str, Optional[str]) -> None + """Post a new tweet. TEXT is the tweet content.""" + config = load_config() + try: + client = _get_client(config) + action = "Replying to %s" % reply_to if reply_to else "Posting tweet" + console.print("✏️ %s..." % action) + tweet_id = client.create_tweet(text, reply_to_id=reply_to) + console.print("[green]✅ Tweet posted![/green]") + console.print("🔗 https://x.com/i/status/%s" % tweet_id) + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + +@cli.command(name="delete") +@click.argument("tweet_id") +@click.confirmation_option(prompt="Are you sure you want to delete this tweet?") +def delete_tweet(tweet_id): + # type: (str,) -> None + """Delete a tweet. TWEET_ID is the numeric tweet ID.""" + config = load_config() + try: + client = _get_client(config) + console.print("🗑️ Deleting tweet %s..." % tweet_id) + client.delete_tweet(tweet_id) + console.print("[green]✅ Tweet deleted.[/green]") + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + +@cli.command() +@click.argument("tweet_id") +def like(tweet_id): + # type: (str,) -> None + """Like a tweet. TWEET_ID is the numeric tweet ID.""" + config = load_config() + try: + client = _get_client(config) + console.print("❤️ Liking tweet %s..." % tweet_id) + client.like_tweet(tweet_id) + console.print("[green]✅ Liked![/green]") + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + +@cli.command() +@click.argument("tweet_id") +def unlike(tweet_id): + # type: (str,) -> None + """Unlike a tweet. TWEET_ID is the numeric tweet ID.""" + config = load_config() + try: + client = _get_client(config) + console.print("💔 Unliking tweet %s..." % tweet_id) + client.unlike_tweet(tweet_id) + console.print("[green]✅ Unliked.[/green]") + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + +@cli.command() +@click.argument("tweet_id") +def rt(tweet_id): + # type: (str,) -> None + """Retweet a tweet. TWEET_ID is the numeric tweet ID.""" + config = load_config() + try: + client = _get_client(config) + console.print("🔄 Retweeting %s..." % tweet_id) + client.retweet(tweet_id) + console.print("[green]✅ Retweeted![/green]") + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + +@cli.command() +@click.argument("tweet_id") +def unrt(tweet_id): + # type: (str,) -> None + """Undo a retweet. TWEET_ID is the numeric tweet ID.""" + config = load_config() + try: + client = _get_client(config) + console.print("🔄 Undoing retweet %s..." % tweet_id) + client.unretweet(tweet_id) + console.print("[green]✅ Retweet undone.[/green]") + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + +@cli.command(name="bookmark-add") +@click.argument("tweet_id") +def bookmark_add(tweet_id): + # type: (str,) -> None + """Bookmark a tweet. TWEET_ID is the numeric tweet ID.""" + config = load_config() + try: + client = _get_client(config) + console.print("🔖 Bookmarking tweet %s..." % tweet_id) + client.bookmark_tweet(tweet_id) + console.print("[green]✅ Bookmarked![/green]") + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + +@cli.command(name="bookmark-rm") +@click.argument("tweet_id") +def bookmark_rm(tweet_id): + # type: (str,) -> None + """Remove a tweet from bookmarks. TWEET_ID is the numeric tweet ID.""" + config = load_config() + try: + client = _get_client(config) + console.print("🔖 Removing bookmark %s..." % tweet_id) + client.unbookmark_tweet(tweet_id) + console.print("[green]✅ Bookmark removed.[/green]") + except RuntimeError as exc: + console.print("[red]❌ %s[/red]" % exc) + sys.exit(1) + + if __name__ == "__main__": cli() diff --git a/twitter_cli/client.py b/twitter_cli/client.py index 24dbb5e..5784698 100644 --- a/twitter_cli/client.py +++ b/twitter_cli/client.py @@ -32,6 +32,7 @@ BEARER_TOKEN = ( ) FALLBACK_QUERY_IDS = { + # Read operations "HomeTimeline": "c-CzHF1LboFilMpsx4ZCrQ", "HomeLatestTimeline": "BKB7oi212Fi7kQtCBGE4zA", "Bookmarks": "VFdMm9iVZxlU6hD86gfW_A", @@ -39,6 +40,19 @@ FALLBACK_QUERY_IDS = { "UserTweets": "E3opETHurmVJflFsUBVuUQ", "SearchTimeline": "nWemVnGJ6A5eQAR5-oQeAg", "Likes": "lIDpu_NWL7_VhimGGt0o6A", + "TweetDetail": "xd_EMdYvB9hfZsZ6Idri0w", + "ListLatestTweetsTimeline": "RlZzktZY_9wJynoepm8ZsA", + "Followers": "IOh4aS6UdGWGJUYTqliQ7Q", + "Following": "zx6e-TLzRkeDO_a7p4b3JQ", + # Write operations + "CreateTweet": "IID9x6WsdMnTlXnzXGq8ng", + "DeleteTweet": "VaenaVgh5q5ih7kvyVjgtg", + "FavoriteTweet": "lI07N6Otwv1PhnEgXILM7A", + "UnfavoriteTweet": "ZYKSe-w7KEslx3JhSIk5LA", + "CreateRetweet": "ojPdsZsimiJrUGLR1sjUtA", + "DeleteRetweet": "iQtK4dl5hBmXewYZuEOKVw", + "CreateBookmark": "aoDbu3RHznuiSkQ9aNM67Q", + "DeleteBookmark": "Wlmlj2-xzyS1GN3a6cj-mQ", } TWITTER_OPENAPI_URL = ( @@ -370,6 +384,119 @@ class TwitterClient: override_base_variables=True, ) + def fetch_tweet_detail(self, tweet_id, count=20): + # type: (str, int) -> List[Tweet] + """Fetch a tweet and its conversation thread (replies).""" + return self._fetch_timeline( + "TweetDetail", + count, + lambda data: _deep_get(data, "data", "tweetResult", "result", "timeline", "instructions") + or _deep_get(data, "data", "threaded_conversation_with_injections_v2", "instructions"), + extra_variables={ + "focalTweetId": tweet_id, + "referrer": "tweet", + "with_rux_injections": False, + "rankingMode": "Relevance", + "withCommunity": True, + "withQuickPromoteEligibilityTweetFields": True, + "withBirdwatchNotes": True, + "withVoice": True, + }, + override_base_variables=True, + ) + + def fetch_list_timeline(self, list_id, count=20): + # type: (str, int) -> List[Tweet] + """Fetch tweets from a Twitter List.""" + return self._fetch_timeline( + "ListLatestTweetsTimeline", + count, + lambda data: _deep_get(data, "data", "list", "tweets_timeline", "timeline", "instructions"), + extra_variables={"listId": list_id}, + override_base_variables=True, + ) + + def fetch_followers(self, user_id, count=20): + # type: (str, int) -> List[UserProfile] + """Fetch followers of a user.""" + return self._fetch_user_list( + "Followers", user_id, count, + lambda data: _deep_get(data, "data", "user", "result", "timeline", "timeline", "instructions"), + ) + + def fetch_following(self, user_id, count=20): + # type: (str, int) -> List[UserProfile] + """Fetch users that a user is following.""" + return self._fetch_user_list( + "Following", user_id, count, + lambda data: _deep_get(data, "data", "user", "result", "timeline", "timeline", "instructions"), + ) + + # ── Write operations ──────────────────────────────────────────────── + + def create_tweet(self, text, reply_to_id=None): + # type: (str, Optional[str]) -> str + """Post a new tweet. Returns the new tweet ID.""" + variables = { + "tweet_text": text, + "media": {"media_entities": [], "possibly_sensitive": False}, + "semantic_annotation_ids": [], + "dark_request": False, + } # type: Dict[str, Any] + if reply_to_id: + variables["reply"] = { + "in_reply_to_tweet_id": reply_to_id, + "exclude_reply_user_ids": [], + } + data = self._graphql_post("CreateTweet", variables, FEATURES) + result = _deep_get(data, "data", "create_tweet", "tweet_results", "result") + if result: + return result.get("rest_id", "") + raise RuntimeError("Failed to create tweet") + + def delete_tweet(self, tweet_id): + # type: (str) -> bool + """Delete a tweet. Returns True on success.""" + variables = {"tweet_id": tweet_id, "dark_request": False} + self._graphql_post("DeleteTweet", variables) + return True + + def like_tweet(self, tweet_id): + # type: (str) -> bool + """Like a tweet. Returns True on success.""" + self._graphql_post("FavoriteTweet", {"tweet_id": tweet_id}) + return True + + def unlike_tweet(self, tweet_id): + # type: (str) -> bool + """Unlike a tweet. Returns True on success.""" + self._graphql_post("UnfavoriteTweet", {"tweet_id": tweet_id, "dark_request": False}) + return True + + def retweet(self, tweet_id): + # type: (str) -> bool + """Retweet a tweet. Returns True on success.""" + self._graphql_post("CreateRetweet", {"tweet_id": tweet_id, "dark_request": False}) + return True + + def unretweet(self, tweet_id): + # type: (str) -> bool + """Undo a retweet. Returns True on success.""" + self._graphql_post("DeleteRetweet", {"source_tweet_id": tweet_id, "dark_request": False}) + return True + + def bookmark_tweet(self, tweet_id): + # type: (str) -> bool + """Bookmark a tweet. Returns True on success.""" + self._graphql_post("CreateBookmark", {"tweet_id": tweet_id}) + return True + + def unbookmark_tweet(self, tweet_id): + # type: (str) -> bool + """Remove a tweet from bookmarks. Returns True on success.""" + self._graphql_post("DeleteBookmark", {"tweet_id": tweet_id}) + return True + 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. @@ -466,8 +593,8 @@ class TwitterClient: except Exception as exc: logger.warning("Failed to init ClientTransaction: %s", exc) - def _build_headers(self, url=""): - # type: (str) -> Dict[str, str] + def _build_headers(self, url="", method="GET"): + # type: (str, str) -> Dict[str, str] """Build shared headers for authenticated API calls.""" headers = { "Authorization": "Bearer %s" % BEARER_TOKEN, @@ -487,7 +614,7 @@ class TwitterClient: try: path = urllib.parse.urlparse(url).path tid = self._client_transaction.generate_transaction_id( - method="GET", path=path, + method=method, path=path, ) headers["X-Client-Transaction-Id"] = tid except Exception as exc: @@ -546,6 +673,144 @@ class TwitterClient: # Should not be reached, but just in case raise TwitterAPIError(429, "Rate limited after %d retries" % self._max_retries) + def _graphql_post(self, operation_name, variables, features=None): + # type: (str, Dict[str, Any], Optional[Dict[str, Any]]) -> Dict[str, Any] + """Issue GraphQL POST request.""" + query_id = _resolve_query_id(operation_name, prefer_fallback=True) + url = "https://x.com/i/api/graphql/%s/%s" % (query_id, operation_name) + body = {"variables": variables, "queryId": query_id} + if features: + body["features"] = features + return self._api_post(url, body) + + def _api_post(self, url, body): + # type: (str, Dict[str, Any]) -> Dict[str, Any] + """Make authenticated POST request to Twitter API.""" + self._ensure_client_transaction() + headers = self._build_headers(url=url, method="POST") + data = json.dumps(body).encode("utf-8") + + for attempt in range(self._max_retries + 1): + request = urllib.request.Request(url, data=data, method="POST") + for key, value in headers.items(): + request.add_header(key, value) + + 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 < self._max_retries: + wait = self._retry_base_delay * (2 ** attempt) + logger.warning( + "Rate limited (429), retrying in %.1fs (attempt %d/%d)", + wait, attempt + 1, self._max_retries, + ) + time.sleep(wait) + continue + body_text = exc.read().decode("utf-8", errors="replace") + message = "Twitter API error %d: %s" % (exc.code, body_text[:500]) + raise TwitterAPIError(exc.code, message) + except urllib.error.URLError as exc: + raise TwitterAPIError(0, "Twitter API network error: %s" % exc.reason) + + 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") + raise TwitterAPIError(0, "Twitter API returned errors: %s" % err_msg) + return parsed + + raise TwitterAPIError(429, "Rate limited after %d retries" % self._max_retries) + + def _fetch_user_list(self, operation_name, user_id, count, get_instructions): + # type: (str, str, int, Callable[[Any], Any]) -> List[UserProfile] + """Generic user list fetcher (for followers/following) with pagination.""" + if count <= 0: + return [] + count = min(count, self._max_count) + users = [] # type: List[UserProfile] + seen_ids = set() # type: Set[str] + cursor = None # type: Optional[str] + attempts = 0 + max_attempts = int(math.ceil(count / 20.0)) + 2 + + while len(users) < count and attempts < max_attempts: + attempts += 1 + variables = { + "userId": user_id, + "count": min(count - len(users) + 5, 40), + "includePromotedContent": False, + } # type: Dict[str, Any] + if cursor: + variables["cursor"] = cursor + + data = self._graphql_get(operation_name, variables, FEATURES) + instructions = get_instructions(data) + if not instructions: + logger.warning("No user list instructions found") + break + + new_users = [] # type: List[UserProfile] + next_cursor = None # type: Optional[str] + for instruction in instructions: + entries = instruction.get("entries", []) + for entry in entries: + content = entry.get("content", {}) + entry_type = content.get("entryType", "") + + if entry_type == "TimelineTimelineItem": + item = content.get("itemContent", {}) + user_results = _deep_get(item, "user_results", "result") + if user_results: + user = self._parse_user_result(user_results) + if user: + new_users.append(user) + elif entry_type == "TimelineTimelineCursor": + if content.get("cursorType") == "Bottom": + next_cursor = content.get("value") + + for user in new_users: + if user.id and user.id not in seen_ids: + seen_ids.add(user.id) + users.append(user) + + if not next_cursor or not new_users: + break + cursor = next_cursor + + if len(users) < count and self._request_delay > 0: + time.sleep(self._request_delay) + + return users[:count] + + @staticmethod + def _parse_user_result(user_data): + # type: (Dict[str, Any]) -> Optional[UserProfile] + """Parse a user result object into UserProfile.""" + if user_data.get("__typename") == "UserUnavailable": + return None + legacy = user_data.get("legacy", {}) + if not legacy: + return None + return UserProfile( + id=user_data.get("rest_id", ""), + name=legacy.get("name", ""), + screen_name=legacy.get("screen_name", ""), + bio=legacy.get("description", ""), + location=legacy.get("location", ""), + url=_deep_get(legacy, "entities", "url", "urls", 0, "expanded_url") or "", + followers_count=legacy.get("followers_count", 0), + following_count=legacy.get("friends_count", 0), + tweets_count=legacy.get("statuses_count", 0), + likes_count=legacy.get("favourites_count", 0), + verified=user_data.get("is_blue_verified", False) or legacy.get("verified", False), + profile_image_url=legacy.get("profile_image_url_https", ""), + created_at=legacy.get("created_at", ""), + ) + def _parse_timeline_response(self, data, get_instructions): # type: (Any, Callable[[Any], Any]) -> Tuple[List[Tweet], Optional[str]] """Parse timeline GraphQL response into tweets and next cursor.""" @@ -695,13 +960,19 @@ class TwitterClient: 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