Merge pull request #113 from MiddleRed/main

Add `user_state` endpoint for better detection on account suspension

examples/download_tweet_media.py

@@ -1,6 +1,3 @@
-import asyncio
-import time
-
 from twikit.twikit_async import Client
 
 AUTH_INFO_1 = '...'

twikit/client.py

@@ -16,6 +16,7 @@
 from ._captcha import Capsolver
 from .community import Community, CommunityMember
 from .errors import (
+    AccountLocked,
     AccountSuspended,
     BadRequest,
     CouldNotTweet,
@@ -122,7 +123,7 @@ def request(
             if error_code == 326:
                 # Account unlocking
                 if self.captcha_solver is None:
-                    raise TwitterException(
+                    raise AccountLocked(
                         'Your account is locked. Visit '
                         'https://twitter.com/account/access to unlock it.'
                     )
@@ -151,6 +152,8 @@ def request(
             elif status_code == 408:
                 raise RequestTimeout(message, headers=response.headers)
             elif status_code == 429:
+                if self._get_user_state() == 'suspended':
+                    raise AccountSuspended(message, headers=response.headers)
                 raise TooManyRequests(message, headers=response.headers)
             elif 500 <= status_code < 600:
                 raise ServerError(message, headers=response.headers)
@@ -4936,7 +4939,7 @@ def _get_community_users(
         items = find_dict(response, 'items_results')[0]
         users = []
         for item in items:
-            if not 'result' in item:
+            if 'result' not in item:
                 continue
             if item['result'].get('__typename') != 'User':
                 continue
@@ -5192,3 +5195,10 @@ def _update_subscriptions(
         session.topics -= unsubscribe
 
         return _payload_from_data(response)
+
+    def _get_user_state(self) -> Literal['normal', 'bounced', 'suspended']:
+        response, _ = self.get(
+            Endpoint.USER_STATE,
+            headers=self._base_headers
+        )
+        return response['userState']

twikit/errors.py

@@ -89,6 +89,11 @@ class AccountSuspended(TwitterException):
     Exception raised when the account is suspended.
     """
 
+class AccountLocked(TwitterException):
+    """
+    Exception raised when the account is locked (very likey is Arkose challenge).
+    """
+
 ERROR_CODE_TO_EXCEPTION: dict[int, TwitterException] = {
     187: DuplicateTweet,
     324: InvalidMedia

twikit/tweet.py

@@ -11,7 +11,6 @@
     from httpx import Response
 
     from .client import Client
-    from .community import Community
     from .utils import Result
 
 

twikit/twikit_async/client.py

@@ -13,6 +13,7 @@
 from httpx import Response
 
 from ..errors import (
+    AccountLocked,
     AccountSuspended,
     BadRequest,
     CouldNotTweet,
@@ -124,7 +125,7 @@ async def request(
             if error_code == 326:
                 # Account unlocking
                 if self.captcha_solver is None:
-                    raise TwitterException(
+                    raise AccountLocked(
                         'Your account is locked. Visit '
                         'https://twitter.com/account/access to unlock it.'
                     )
@@ -153,6 +154,8 @@ async def request(
             elif status_code == 408:
                 raise RequestTimeout(message, headers=response.headers)
             elif status_code == 429:
+                if await self._get_user_state() == 'suspended':
+                    raise AccountSuspended(message, headers=response.headers)
                 raise TooManyRequests(message, headers=response.headers)
             elif 500 <= status_code < 600:
                 raise ServerError(message, headers=response.headers)
@@ -4976,7 +4979,7 @@ async def _get_community_users(
         items = find_dict(response, 'items_results')[0]
         users = []
         for item in items:
-            if not 'result' in item:
+            if 'result' not in item:
                 continue
             if item['result'].get('__typename') != 'User':
                 continue
@@ -5236,3 +5239,10 @@ async def _update_subscriptions(
         session.topics -= unsubscribe
 
         return _payload_from_data(response)
+
+    async def _get_user_state(self) -> Literal['normal', 'bounced', 'suspended']:
+        response, _ = await self.get(
+            Endpoint.USER_STATE,
+            headers=self._base_headers
+        )
+        return response['userState']

twikit/utils.py

@@ -300,6 +300,7 @@ class Endpoint:
     REVERSE_GEOCODE = 'https://api.twitter.com/1.1/geo/reverse_geocode.json'
     SEARCH_GEO = 'https://api.twitter.com/1.1/geo/search.json'
     PLACE_BY_ID = 'https://api.twitter.com/1.1/geo/id/{}.json'
+    USER_STATE = 'https://api.twitter.com/help-center/forms/api/prod/user_state.json'
 
 T = TypeVar('T')
 
@@ -474,7 +475,6 @@ def build_tweet_data(raw_data: dict) -> dict:
             'is_quote_status': raw_data.get('is_quote_status'),
             'in_reply_to_status_id_str': raw_data.get('in_reply_to_status_id_str'),
             'retweeted_status_result': raw_data.get('retweeted_status_result'),
-            'is_quote_status': raw_data.get('is_quote_status'),
             'possibly_sensitive': raw_data.get('possibly_sensitive'),
             'possibly_sensitive_editable': raw_data.get('possibly_sensitive_editable'),
             'quote_count': raw_data.get('quote_count'),
@@ -674,13 +674,13 @@ def build_query(text: str, options: SearchOptions) -> str:
     if until := options.get('until'):
         text += f' until:{until}'
 
-    if options.get('positive') == True:
-        text += f' :)'
+    if options.get('positive') is True:
+        text += ' :)'
 
-    if options.get('negative') == True:
-        text += f' :('
+    if options.get('negative') is True:
+        text += ' :('
 
-    if options.get('question') == True:
-        text += f' ?'
+    if options.get('question') is True:
+        text += ' ?'
 
     return text