From 1f3a9ee5351d0908bb1b10476b765493da59346f Mon Sep 17 00:00:00 2001 From: nemo Date: Wed, 29 Apr 2026 21:20:39 +0800 Subject: [PATCH] fix: adapt to Twitter API schema changes (April 2026) (#51) - Migrate user profile fields from `legacy{}` to `core{}` / `avatar{}` / `location{}` (with legacy fallback for older response shapes). Affects `user`, `whoami`, `status`, `followers`, `following`. - Fix `user-posts` empty results: add missing `includePromotedContent` variable and update `instructions` path from `timeline_v2` to `timeline`. - Switch `followers` / `following` to POST (Twitter changed the requirement). - Refresh 16 stale `FALLBACK_QUERY_IDS` from fa0311/twitter-openapi. - Drop `legacy{}` early-return in `parse_user_result`; key existence on `rest_id` so followers/following keep working when Twitter fully drops legacy. - Add unit tests for the new core/avatar/location fallback chain and rest_id/typename guards. --- tests/test_client.py | 85 +++++++++++++++++++++++++++++++++++++++++- twitter_cli/client.py | 30 ++++++++++----- twitter_cli/graphql.py | 32 ++++++++-------- twitter_cli/parser.py | 21 ++++++++--- 4 files changed, 136 insertions(+), 32 deletions(-) diff --git a/tests/test_client.py b/tests/test_client.py index d0a6971..931a6b9 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -208,7 +208,7 @@ class TestBuildGraphqlUrl: def test_searchtimeline_fallback_query_id_regression(self): """Keep SearchTimeline fallback aligned with the live operation after issue #39.""" - assert FALLBACK_QUERY_IDS["SearchTimeline"] == "rkp6b4vtR9u7v3naGoOzUQ" + assert FALLBACK_QUERY_IDS["SearchTimeline"] == "VhUd6vHVmLBcw0uX-6jMLA" # ── _best_chrome_target ────────────────────────────────────────────────── @@ -1230,6 +1230,89 @@ class TestParseUserResult: assert user.tweets_count == 78 assert user.likes_count == 0 + def test_reads_core_avatar_location_when_legacy_absent(self): + """New API shape: name/screen_name/created_at moved to core{}, + profile_image_url to avatar.image_url, location to location.location. + legacy{} may be empty or missing entirely.""" + user = parse_user_result( + { + "rest_id": "user-2", + "core": { + "name": "Bob", + "screen_name": "bob", + "created_at": "Tue Mar 21 17:25:43 +0000 2023", + }, + "avatar": {"image_url": "https://example.com/bob.jpg"}, + "location": {"location": "Earth"}, + "is_blue_verified": True, + } + ) + + assert user is not None + assert user.id == "user-2" + assert user.name == "Bob" + assert user.screen_name == "bob" + assert user.created_at == "Tue Mar 21 17:25:43 +0000 2023" + assert user.profile_image_url == "https://example.com/bob.jpg" + assert user.location == "Earth" + assert user.verified is True + + def test_prefers_core_over_legacy_when_both_present(self): + """During the migration both shapes coexist — core{} should win.""" + user = parse_user_result( + { + "rest_id": "user-3", + "core": {"name": "NewName", "screen_name": "new_handle"}, + "avatar": {"image_url": "https://example.com/new.jpg"}, + "legacy": { + "name": "OldName", + "screen_name": "old_handle", + "profile_image_url_https": "https://example.com/old.jpg", + "description": "old bio", + }, + } + ) + + assert user is not None + assert user.name == "NewName" + assert user.screen_name == "new_handle" + assert user.profile_image_url == "https://example.com/new.jpg" + # bio still comes from legacy — it hasn't migrated + assert user.bio == "old bio" + + def test_falls_back_to_legacy_when_core_missing(self): + """Older response shape with only legacy{} — keep working.""" + user = parse_user_result( + { + "rest_id": "user-4", + "legacy": { + "name": "Carol", + "screen_name": "carol", + "profile_image_url_https": "https://example.com/carol.jpg", + "location": "Mars", + "created_at": "Mon Jan 01 00:00:00 +0000 2020", + }, + } + ) + + assert user is not None + assert user.name == "Carol" + assert user.screen_name == "carol" + assert user.profile_image_url == "https://example.com/carol.jpg" + assert user.location == "Mars" + assert user.created_at == "Mon Jan 01 00:00:00 +0000 2020" + + def test_returns_none_without_rest_id(self): + """No rest_id means no user — drop the row instead of emitting an + empty-id UserProfile.""" + assert parse_user_result({"core": {"name": "Anon"}}) is None + assert parse_user_result({}) is None + + def test_returns_none_for_user_unavailable(self): + assert ( + parse_user_result({"__typename": "UserUnavailable", "rest_id": "x"}) is None + ) + # ── upload_media ───────────────────────────────────────────────────────── diff --git a/twitter_cli/client.py b/twitter_cli/client.py index 4fc36ce..8668a44 100644 --- a/twitter_cli/client.py +++ b/twitter_cli/client.py @@ -280,20 +280,23 @@ class TwitterClient: raise NotFoundError("User @%s not found" % screen_name) legacy = result.get("legacy", {}) + core = result.get("core", {}) + avatar = result.get("avatar", {}) + location_obj = result.get("location", {}) return UserProfile( id=result.get("rest_id", ""), - name=legacy.get("name", ""), - screen_name=legacy.get("screen_name", screen_name), + name=core.get("name") or legacy.get("name", ""), + screen_name=core.get("screen_name") or legacy.get("screen_name", screen_name), bio=legacy.get("description", ""), - location=legacy.get("location", ""), + location=location_obj.get("location") or legacy.get("location", ""), url=_deep_get(legacy, "entities", "url", "urls", 0, "expanded_url") or "", followers_count=_parse_int(legacy.get("followers_count"), 0), following_count=_parse_int(legacy.get("friends_count"), 0), tweets_count=_parse_int(legacy.get("statuses_count"), 0), likes_count=_parse_int(legacy.get("favourites_count"), 0), verified=bool(result.get("is_blue_verified") or legacy.get("verified", False)), - profile_image_url=legacy.get("profile_image_url_https", ""), - created_at=legacy.get("created_at", ""), + profile_image_url=avatar.get("image_url") or legacy.get("profile_image_url_https", ""), + created_at=core.get("created_at") or legacy.get("created_at", ""), ) def fetch_user_tweets(self, user_id, count=20): @@ -302,9 +305,13 @@ class TwitterClient: return self._fetch_timeline( "UserTweets", count, - lambda data: _deep_get(data, "data", "user", "result", "timeline_v2", "timeline", "instructions"), + lambda data: ( + _deep_get(data, "data", "user", "result", "timeline", "timeline", "instructions") + or _deep_get(data, "data", "user", "result", "timeline_v2", "timeline", "instructions") + ), extra_variables={ "userId": user_id, + "includePromotedContent": True, "withQuickPromoteEligibilityTweetFields": True, "withVoice": True, "withV2Timeline": True, @@ -447,6 +454,7 @@ class TwitterClient: return self._fetch_user_list( "Followers", user_id, count, lambda data: _deep_get(data, "data", "user", "result", "timeline", "timeline", "instructions"), + use_post=True, ) def fetch_following(self, user_id, count=20): @@ -455,6 +463,7 @@ class TwitterClient: return self._fetch_user_list( "Following", user_id, count, lambda data: _deep_get(data, "data", "user", "result", "timeline", "timeline", "instructions"), + use_post=True, ) # ── Write operations ───────────────────────────────────────────── @@ -813,8 +822,8 @@ class TwitterClient: return tweets[:count], continuation_cursor return tweets[:count] - def _fetch_user_list(self, operation_name, user_id, count, get_instructions): - # type: (str, str, int, Callable[[Any], Any]) -> List[UserProfile] + def _fetch_user_list(self, operation_name, user_id, count, get_instructions, use_post=False): + # type: (str, str, int, Callable[[Any], Any], bool) -> List[UserProfile] """Generic user list fetcher (for followers/following) with pagination.""" if count <= 0: return [] @@ -835,7 +844,10 @@ class TwitterClient: if cursor: variables["cursor"] = cursor - data = self._graphql_get(operation_name, variables, FEATURES) + if use_post: + data = self._graphql_post(operation_name, variables, FEATURES) + else: + data = self._graphql_get(operation_name, variables, FEATURES) instructions = get_instructions(data) if not instructions: logger.warning("No user list instructions found") diff --git a/twitter_cli/graphql.py b/twitter_cli/graphql.py index d600497..d34ea35 100644 --- a/twitter_cli/graphql.py +++ b/twitter_cli/graphql.py @@ -27,26 +27,26 @@ TWITTER_OPENAPI_URL = ( # ── Fallback (hardcoded) queryIds ──────────────────────────────────────── FALLBACK_QUERY_IDS = { - "HomeTimeline": "L8Lb9oomccM012S7fQ-QKA", - "HomeLatestTimeline": "tzmrSIWxyV4IRRh9nij6TQ", - "UserByScreenName": "IGgvgiOx4QZndDHuD3x9TQ", - "UserTweets": "O0epvwaQPUx-bT9YlqlL6w", - "TweetDetail": "xIYgDwjboktoFeXe_fgacw", - "Likes": "RozQdCp4CilQzrcuU0NY5w", - "SearchTimeline": "rkp6b4vtR9u7v3naGoOzUQ", - "Bookmarks": "uzboyXSHSJrR-mGJqep0TQ", - "ListLatestTweetsTimeline": "fb_6wmHD2dk9D-xYXOQlgw", - "Followers": "Enf9DNUZYiT037aersI5gg", - "Following": "ntIPnH1WMBKW--4Tn1q71A", - "CreateTweet": "zkcFc6F-RKRgWN8HUkJfZg", - "DeleteTweet": "nxpZCY2K-I6QoFHAHeojFQ", + "HomeTimeline": "c-CzHF1LboFilMpsx4ZCrQ", + "HomeLatestTimeline": "BKB7oi212Fi7kQtCBGE4zA", + "UserByScreenName": "1VOOyvKkiI3FMmkeDNxM9A", + "UserTweets": "q6xj5bs0hapm9309hexA_g", + "TweetDetail": "xd_EMdYvB9hfZsZ6Idri0w", + "Likes": "lIDpu_NWL7_VhimGGt0o6A", + "SearchTimeline": "VhUd6vHVmLBcw0uX-6jMLA", + "Bookmarks": "2neUNDqrrFzbLui8yallcQ", + "ListLatestTweetsTimeline": "RlZzktZY_9wJynoepm8ZsA", + "Followers": "IOh4aS6UdGWGJUYTqliQ7Q", + "Following": "zx6e-TLzRkeDO_a7p4b3JQ", + "CreateTweet": "IID9x6WsdMnTlXnzXGq8ng", + "DeleteTweet": "VaenaVgh5q5ih7kvyVjgtg", "FavoriteTweet": "lI07N6Otwv1PhnEgXILM7A", "UnfavoriteTweet": "ZYKSe-w7KEslx3JhSIk5LA", - "CreateRetweet": "mbRO74GrOvSfRcJnlMapnQ", - "DeleteRetweet": "ZyZigVsNiFO6v1dEks1eWg", + "CreateRetweet": "ojPdsZsimiJrUGLR1sjUtA", + "DeleteRetweet": "iQtK4dl5hBmXewYZuEOKVw", "CreateBookmark": "aoDbu3RHznuiSkQ9aNM67Q", "DeleteBookmark": "Wlmlj2-xzyS1GN3a6cj-mQ", - "TweetResultByRestId": "zy39CwTyYhU-_0LP7dljjg", + "TweetResultByRestId": "7xflPyRiUxGVbJd4uWmbfg", "BookmarkFoldersSlice": "i78YDd0Tza-dV4SYs58kRg", "BookmarkFolderTimeline": "hNY7X2xE2N7HVF6Qb_mu6w", } diff --git a/twitter_cli/parser.py b/twitter_cli/parser.py index 6dd6f6f..58378be 100644 --- a/twitter_cli/parser.py +++ b/twitter_cli/parser.py @@ -375,23 +375,32 @@ def parse_user_result(user_data): """Parse a user result object into UserProfile.""" if user_data.get("__typename") == "UserUnavailable": return None + # Twitter API migrated name/screen_name/created_at to core{}, avatar to + # avatar.image_url, and location to location.location. Newer responses + # may omit legacy{} entirely; fall back to legacy for older shapes. legacy = user_data.get("legacy", {}) - if not legacy: + core = user_data.get("core", {}) + avatar = user_data.get("avatar", {}) + location_obj = user_data.get("location", {}) + # Use rest_id presence as the existence signal, not legacy{}, so + # this stays consistent with fetch_user() once Twitter fully drops + # legacy. + if not user_data.get("rest_id"): return None return UserProfile( id=user_data.get("rest_id", ""), - name=legacy.get("name", ""), - screen_name=legacy.get("screen_name", ""), + name=core.get("name") or legacy.get("name", ""), + screen_name=core.get("screen_name") or legacy.get("screen_name", ""), bio=legacy.get("description", ""), - location=legacy.get("location", ""), + location=location_obj.get("location") or legacy.get("location", ""), url=_deep_get(legacy, "entities", "url", "urls", 0, "expanded_url") or "", followers_count=_parse_int(legacy.get("followers_count"), 0), following_count=_parse_int(legacy.get("friends_count"), 0), tweets_count=_parse_int(legacy.get("statuses_count"), 0), likes_count=_parse_int(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", ""), + profile_image_url=avatar.get("image_url") or legacy.get("profile_image_url_https", ""), + created_at=core.get("created_at") or legacy.get("created_at", ""), )