feat: update twitter details

feat: update twitter details

Author: aturret

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "twitter-api-client-v2"
-version = "0.1.0"
+version = "0.1.1"
 description = "Implementation of X/Twitter v1, v2, and GraphQL APIs."
 requires-python = ">=3.12"
 license = "MIT"

twitter/constants.py

@@ -111,7 +111,7 @@ class Operation:
     UserTweetsAndReplies = {'userId': int}, 'RIWc55YCNyUJ-U3HHGYkdg', 'UserTweetsAndReplies'
     TweetResultByRestId = {'tweetId': int}, 'd6YKjvQ920F-D4Y1PruO-A', 'TweetResultByRestId'
     TweetResultsByRestIds = {'tweetIds': list[int | str]}, 'BWy5aoI-WvwbeSiHUIf2Hw', 'TweetResultsByRestIds'
-    TweetDetail = {'focalTweetId': int}, 'zXaXQgfyR4GxE21uwYQSyA', 'TweetDetail'
+    TweetDetail = {'focalTweetId': int}, 'ooUbmy0T2DmvwfjgARktiQ', 'TweetDetail'
     TweetStats = {'rest_id': int}, 'EvbTkPDT-xQCfupPu0rWMA', 'TweetStats'
     Likes = {'userId': int}, 'nXEl0lfN_XSznVMlprThgQ', 'Likes'
     Followers = {'userId': int}, 'pd8Tt1qUz1YWrICegqZ8cw', 'Followers'
@@ -671,26 +671,56 @@ class Operation:
     'ext': 'mediaStats,highlightedLabel,hasNftAvatar,voiceInfo,birdwatchPivot,superFollowMetadata,unmentionInfo,editControl'
 }
 
-latest_features = {
-        "responsive_web_grok_share_attachment_enabled": True,
-        "responsive_web_grok_show_grok_translated_post": True,
-        "responsive_web_profile_redirect_enabled": True,
-        "responsive_web_grok_analyze_post_followups_enabled": True,
-        "post_ctas_fetch_enabled": True,
-        "rweb_tipjar_consumption_enabled": True,
-        "responsive_web_grok_analysis_button_from_backend": True,
-        "responsive_web_grok_imagine_annotation_enabled": True,
-        "responsive_web_jetfuel_frame": True,
-        "premium_content_api_read_enabled": True,
-        "responsive_web_grok_analyze_button_fetch_trends_enabled": True,
-        "communities_web_enable_tweet_community_results_fetch": True,
-        "profile_label_improvements_pcf_label_in_post_enabled": True,
-        "responsive_web_grok_image_annotation_enabled": True,
-        "responsive_web_grok_community_note_auto_translation_is_enabled": True,
-        "articles_preview_enabled": True,
-        "responsive_web_grok_annotations_enabled": True,
-        "longform_notetweets_inline_media_enabled": True,
-        "responsive_web_edit_tweet_api_enabled": True,
-        "graphql_is_translatable_rweb_tweet_is_translatable_enabled": True,
-        "view_counts_everywhere_api_enabled": True
-    }
+target_features = {
+    "rweb_video_screen_enabled": False,
+    "profile_label_improvements_pcf_label_in_post_enabled": True,
+    "responsive_web_profile_redirect_enabled": False,
+    "rweb_tipjar_consumption_enabled": False,
+    "verified_phone_label_enabled": False,
+    "creator_subscriptions_tweet_preview_api_enabled": True,
+    "responsive_web_graphql_timeline_navigation_enabled": True,
+    "responsive_web_graphql_skip_user_profile_image_extensions_enabled": False,
+    "premium_content_api_read_enabled": False,
+    "communities_web_enable_tweet_community_results_fetch": True,
+    "c9s_tweet_anatomy_moderator_badge_enabled": True,
+    "responsive_web_grok_analyze_button_fetch_trends_enabled": False,
+    "responsive_web_grok_analyze_post_followups_enabled": True,
+    "responsive_web_jetfuel_frame": True,
+    "responsive_web_grok_share_attachment_enabled": True,
+    "responsive_web_grok_annotations_enabled": True,
+    "articles_preview_enabled": True,
+    "responsive_web_edit_tweet_api_enabled": True,
+    "graphql_is_translatable_rweb_tweet_is_translatable_enabled": True,
+    "view_counts_everywhere_api_enabled": True,
+    "longform_notetweets_consumption_enabled": True,
+    "responsive_web_twitter_article_tweet_consumption_enabled": True,
+    "tweet_awards_web_tipping_enabled": False,
+    "responsive_web_grok_show_grok_translated_post": True,
+    "responsive_web_grok_analysis_button_from_backend": True,
+    "post_ctas_fetch_enabled": True,
+    "freedom_of_speech_not_reach_fetch_enabled": True,
+    "standardized_nudges_misinfo": True,
+    "tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled": True,
+    "longform_notetweets_rich_text_read_enabled": True,
+    "longform_notetweets_inline_media_enabled": True,
+    "responsive_web_grok_image_annotation_enabled": True,
+    "responsive_web_grok_imagine_annotation_enabled": True,
+    "responsive_web_grok_community_note_auto_translation_is_enabled": False,
+    "responsive_web_enhance_cards_enabled": False}
+
+target_field_toggles = {
+    "withArticleRichContentState": True,
+    "withArticlePlainText": False,
+    "withGrokAnalyze": False,
+    "withDisallowedReplyControls": False
+}
+
+static_variables = {
+    "with_rux_injections": False,
+    "rankingMode": "Relevance",
+    "includePromotedContent": True,
+    "withCommunity": True,
+    "withQuickPromoteEligibilityTweetFields": True,
+    "withBirdwatchNotes": True,
+    "withVoice": True
+}

twitter/scraper.py

@@ -25,6 +25,7 @@
 if platform.system() != 'Windows':
     try:
         import uvloop
+
         asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
     except ImportError as e:
         ...
@@ -59,7 +60,8 @@ def tweets_by_id(self, tweet_ids: list[int | str], **kwargs) -> list[dict]:
         @param kwargs: optional keyword arguments
         @return: list of tweet data as dicts
         """
-        return self._run(Operation.TweetResultByRestId, tweet_ids, **(latest_features | kwargs))
+        kwargs['features'] = target_features
+        return self._run(Operation.TweetResultByRestId, tweet_ids, **kwargs)
 
     def tweets_by_ids(self, tweet_ids: list[int | str], **kwargs) -> list[dict]:
         """
@@ -83,6 +85,9 @@ def tweets_details(self, tweet_ids: list[int], **kwargs) -> list[dict]:
         @param kwargs: optional keyword arguments
         @return: list of tweet data as dicts
         """
+        kwargs.update(static_variables)
+        kwargs['features'] = target_features
+        kwargs['fieldToggles'] = target_field_toggles
         return self._run(Operation.TweetDetail, tweet_ids, **kwargs)
 
     def tweets(self, user_ids: list[int], **kwargs) -> list[dict]:
@@ -244,7 +249,8 @@ def users_by_id(self, user_ids: list[int], **kwargs) -> list[dict]:
         """
         return self._run(Operation.UserByRestId, user_ids, **kwargs)
 
-    def download_media(self, ids: list[int], photos: bool = True, videos: bool = True, cards: bool = True, hq_img_variant: bool = True, video_thumb: bool = False, out: str = 'media',
+    def download_media(self, ids: list[int], photos: bool = True, videos: bool = True, cards: bool = True,
+                       hq_img_variant: bool = True, video_thumb: bool = False, out: str = 'media',
                        metadata_out: str = 'media.json', **kwargs) -> dict:
         """
         Download and extract media metadata from Tweets
@@ -267,7 +273,8 @@ async def process(fns: Generator) -> list:
                 'keepalive_expiry': kwargs.pop('keepalive_expiry', 5.0),
             }
             headers = {'user-agent': random.choice(USER_AGENTS)}
-            async with AsyncClient(limits=Limits(**limits), headers=headers, http2=True, verify=False, timeout=60, follow_redirects=True) as client:
+            async with AsyncClient(limits=Limits(**limits), headers=headers, http2=True, verify=False, timeout=60,
+                                   follow_redirects=True) as client:
                 return await tqdm_asyncio.gather(*(fn(client=client) for fn in fns), desc='Downloading Media')
 
         def download(urls: list[tuple], out: str) -> Generator:
@@ -295,7 +302,8 @@ async def get(client: AsyncClient, url: str):
                 if _id := root.get('rest_id'):
                     date = root.get('legacy', {}).get('created_at', '')
                     uid = root.get('legacy', {}).get('user_id_str', '')
-                    media[_id] = {'date': date, 'uid': uid, 'img': set(), 'video': {'thumb': set(), 'video_info': {}, 'hq': set()}, 'card': []}
+                    media[_id] = {'date': date, 'uid': uid, 'img': set(),
+                                  'video': {'thumb': set(), 'video_info': {}, 'hq': set()}, 'card': []}
                     for _media in (y for x in find_key(root, 'media') for y in x if isinstance(x, list)):
                         if videos:
                             if vinfo := _media.get('video_info'):
@@ -588,10 +596,18 @@ def _run(self, operation: tuple[dict, str, str], queries: set | list[int | str |
 
     async def _query(self, client: AsyncClient, operation: tuple, **kwargs) -> Response:
         keys, qid, name = operation
+
+        features_override = kwargs.pop('features', None)
+        field_toggles = kwargs.pop('fieldToggles', None)
+
         params = {
             'variables': Operation.default_variables | keys | kwargs,
-            'features': Operation.default_features,
+            'features': features_override if features_override else Operation.default_features,
         }
+
+        if field_toggles:
+            params['fieldToggles'] = field_toggles
+
         r = await client.get(f'https://twitter.com/i/api/graphql/{qid}/{name}', params=build_params(params))
 
         try:
@@ -608,7 +624,8 @@ async def _query(self, client: AsyncClient, operation: tuple, **kwargs) -> Respo
     async def _process(self, operation: tuple, queries: list[dict], **kwargs):
         headers = self.session.headers if self.guest else get_headers(self.session)
         cookies = self.session.cookies
-        async with AsyncClient(limits=Limits(max_connections=MAX_ENDPOINT_LIMIT), headers=headers, cookies=cookies, timeout=20) as c:
+        async with AsyncClient(limits=Limits(max_connections=MAX_ENDPOINT_LIMIT), headers=headers, cookies=cookies,
+                               timeout=20) as c:
             tasks = (self._paginate(c, operation, **q, **kwargs) for q in queries)
             if self.pbar:
                 return await tqdm_asyncio.gather(*tasks, desc=operation[-1])