diff --git a/twikit/__init__.py b/twikit/__init__.py index 8055f394..763a2a98 100644 --- a/twikit/__init__.py +++ b/twikit/__init__.py @@ -7,7 +7,7 @@ A Python library for interacting with the Twitter API. """ -__version__ = '1.7.5' +__version__ = '1.7.6' from ._captcha import Capsolver from .bookmark import BookmarkFolder diff --git a/twikit/client.py b/twikit/client.py index f659a00e..2bed3abd 100644 --- a/twikit/client.py +++ b/twikit/client.py @@ -419,8 +419,7 @@ def login( if flow.task_id == 'LoginTwoFactorAuthChallenge': if totp_secret is None: - print(find_dict( - flow.response, 'secondary_text', find_one=True)[0]['text']) + print(find_dict(flow.response, 'secondary_text', find_one=True)[0]['text']) totp_code = input('>>>') else: totp_code = pyotp.TOTP(totp_secret).now() @@ -434,8 +433,7 @@ def login( }) if flow.task_id == 'LoginAcid': - print(find_dict( - flow.response, 'secondary_text', find_one=True)[0]['text']) + print(find_dict(flow.response, 'secondary_text', find_one=True)[0]['text']) flow.execute_task({ 'subtask_id': 'LoginAcid', @@ -697,9 +695,7 @@ def search_tweet( if next_cursor is None: if product == 'Media': - entries = find_dict( - instructions, 'entries', find_one=True - )[0] + entries = find_dict(instructions, 'entries', find_one=True)[0] next_cursor = entries[-1]['content']['value'] previous_cursor = entries[-2]['content']['value'] else: @@ -1090,12 +1086,8 @@ def create_poll( for i, choice in enumerate(choices, 1): card_data[f'twitter:string:choice{i}_label'] = choice - data = urlencode( - {'card_data': card_data} - ) - headers = self._base_headers | { - 'content-type': 'application/x-www-form-urlencoded' - } + data = urlencode({'card_data': card_data}) + headers = self._base_headers | {'content-type': 'application/x-www-form-urlencoded'} response, _ = self.post( Endpoint.CREATE_CARD, data=data, @@ -1158,8 +1150,7 @@ def create_tweet( media_ids: list[str] | None = None, poll_uri: str | None = None, reply_to: str | None = None, - conversation_control: Literal[ - 'followers', 'verified', 'mentioned'] | None = None, + conversation_control: Literal['followers', 'verified', 'mentioned'] | None = None, attachment_url: str | None = None, community_id: str | None = None, share_with_followers: bool = False, @@ -1314,9 +1305,7 @@ def create_tweet( if not _result: raise_exceptions_from_response(response['errors']) raise CouldNotTweet( - response['errors'][0] - if response['errors'] - else 'Failed to post a tweet.' + response['errors'][0] if response['errors'] else 'Failed to post a tweet.' ) tweet_info = _result[0] @@ -1651,8 +1640,7 @@ def _get_more_replies(self, tweet_id: str, cursor: str) -> Result[Tweet]: if entries[-1]['entryId'].startswith('cursor'): next_cursor = entries[-1]['content']['itemContent']['value'] - _fetch_next_result = partial(self._get_more_replies, - tweet_id, next_cursor) + _fetch_next_result = partial(self._get_more_replies, tweet_id, next_cursor) else: next_cursor = None _fetch_next_result = None @@ -1784,9 +1772,7 @@ def get_scheduled_tweets(self) -> list[ScheduledTweet]: list[:class:`ScheduledTweet`] List of ScheduledTweet objects representing the scheduled tweets. """ - params = flatten_params({ - 'variables': {'ascending': True} - }) + params = flatten_params({'variables': {'ascending': True}}) response, _ = self.get( Endpoint.FETCH_SCHEDULED_TWEETS, params=params, @@ -1810,9 +1796,7 @@ def delete_scheduled_tweet(self, tweet_id: str) -> Response: Response returned from twitter api. """ data = { - 'variables': { - 'scheduled_tweet_id': tweet_id - }, + 'variables': {'scheduled_tweet_id': tweet_id}, 'queryId': get_query_id(Endpoint.DELETE_SCHEDULED_TWEET) } _, response = self.post( @@ -1863,11 +1847,9 @@ def _get_tweet_engagements( return Result( results, - partial(self._get_tweet_engagements, - tweet_id, count, next_cursor, endpoint), + partial(self._get_tweet_engagements, tweet_id, count, next_cursor, endpoint), next_cursor, - partial(self._get_tweet_engagements, - tweet_id, count, previous_cursor, endpoint), + partial(self._get_tweet_engagements, tweet_id, count, previous_cursor, endpoint), previous_cursor ) @@ -1979,9 +1961,7 @@ def get_community_note(self, note_id: str) -> CommunityNote: note_data = response['data']['birdwatch_note_by_rest_id'] if 'data_v1' not in note_data: raise TwitterException(f'Invalid note id: {note_id}') - return CommunityNote( - self, note_data - ) + return CommunityNote(self, note_data) def get_user_tweets( self, @@ -2091,9 +2071,7 @@ def get_user_tweets( for item in items: entry_id = item['entryId'] - if not entry_id.startswith( - ('tweet', 'profile-conversation', 'profile-grid') - ): + if not entry_id.startswith(('tweet', 'profile-conversation', 'profile-grid')): continue if entry_id.startswith('profile-conversation'): @@ -2116,11 +2094,9 @@ def get_user_tweets( return Result( results, - partial(self.get_user_tweets, - user_id, tweet_type, count, next_cursor), + partial(self.get_user_tweets, user_id, tweet_type, count, next_cursor), next_cursor, - partial(self.get_user_tweets, - user_id, tweet_type, count, previous_cursor), + partial(self.get_user_tweets, user_id, tweet_type, count, previous_cursor), previous_cursor ) @@ -2332,8 +2308,7 @@ def get_latest_timeline( return Result( results, - partial(self.get_latest_timeline, - count, seen_tweet_ids, next_cursor), + partial(self.get_latest_timeline, count, seen_tweet_ids, next_cursor), next_cursor ) @@ -2615,8 +2590,7 @@ def get_bookmarks( next_cursor = items[-1]['content']['value'] if folder_id is None: previous_cursor = items[-2]['content']['value'] - fetch_previous_result = partial(self.get_bookmarks, count, - previous_cursor, folder_id) + fetch_previous_result = partial(self.get_bookmarks, count, previous_cursor, folder_id) else: previous_cursor = None fetch_previous_result = None @@ -2690,8 +2664,7 @@ def get_bookmark_folders( headers=self._base_headers ) - slice = find_dict( - response, 'bookmark_collections_slice', find_one=True)[0] + slice = find_dict(response, 'bookmark_collections_slice', find_one=True)[0] results = [] for item in slice['items']: results.append(BookmarkFolder(self, item)) @@ -2744,9 +2717,7 @@ def edit_bookmark_folder( json=data, headers=self._base_headers ) - return BookmarkFolder( - self, response['data']['bookmark_collection_update'] - ) + return BookmarkFolder(self, response['data']['bookmark_collection_update']) def delete_bookmark_folder(self, folder_id: str) -> Response: """ @@ -2762,9 +2733,7 @@ def delete_bookmark_folder(self, folder_id: str) -> Response: :class:`httpx.Response` Response returned from twitter api. """ - variables = { - 'bookmark_collection_id': folder_id - } + variables = {'bookmark_collection_id': folder_id} data = { 'variables': variables, 'queryId': get_query_id(Endpoint.DELETE_BOOKMARK_FOLDER) @@ -2789,9 +2758,7 @@ def create_bookmark_folder(self, name: str) -> BookmarkFolder: :class:`BookmarkFolder` Newly created bookmark folder. """ - variables = { - 'name': name - } + variables = {'name': name} data = { 'variables': variables, 'queryId': get_query_id(Endpoint.CREATE_BOOKMARK_FOLDER) @@ -2801,9 +2768,7 @@ def create_bookmark_folder(self, name: str) -> BookmarkFolder: json=data, headers=self._base_headers ) - return BookmarkFolder( - self, response['data']['bookmark_collection_create'] - ) + return BookmarkFolder(self, response['data']['bookmark_collection_create']) def follow_user(self, user_id: str) -> Response: """ @@ -3015,9 +2980,7 @@ def unmute_user(self, user_id: str) -> Response: def get_trends( self, - category: Literal[ - 'trending', 'for-you', 'news', 'sports', 'entertainment' - ], + category: Literal['trending', 'for-you', 'news', 'sports', 'entertainment'], count: int = 20, retry: bool = True, additional_request_params: dict | None = None @@ -3085,9 +3048,7 @@ def get_trends( return [] # Recall the method again, as the trend information # may not be returned due to a Twitter error. - return self.get_trends( - category, count, retry, additional_request_params - ) + return self.get_trends(category, count, retry, additional_request_params) items = entries[-1]['content']['timelineModule']['items'] @@ -3210,11 +3171,9 @@ def _get_user_friendship_2( return Result( results, - partial(self._get_user_friendship_2, user_id, - count, endpoint, next_cursor), + partial(self._get_user_friendship_2, user_id, count, endpoint, next_cursor), next_cursor, - partial(self._get_user_friendship_2, user_id, - count, endpoint, previous_cursor), + partial(self._get_user_friendship_2, user_id, count, endpoint, previous_cursor), previous_cursor ) @@ -3238,10 +3197,7 @@ def get_user_followers( A list of User objects representing the followers. """ return self._get_user_friendship( - user_id, - count, - Endpoint.FOLLOWERS, - cursor + user_id, count, Endpoint.FOLLOWERS, cursor ) def get_latest_followers( @@ -3253,11 +3209,7 @@ def get_latest_followers( Max count : 200 """ return self._get_user_friendship_2( - user_id, - screen_name, - count, - Endpoint.FOLLOWERS2, - cursor + user_id, screen_name, count, Endpoint.FOLLOWERS2, cursor ) def get_latest_friends( @@ -3269,11 +3221,7 @@ def get_latest_friends( Max count : 200 """ return self._get_user_friendship_2( - user_id, - screen_name, - count, - Endpoint.FOLLOWING2, - cursor + user_id, screen_name, count, Endpoint.FOLLOWING2, cursor ) def get_user_verified_followers( @@ -3295,10 +3243,7 @@ def get_user_verified_followers( A list of User objects representing the verified followers. """ return self._get_user_friendship( - user_id, - count, - Endpoint.BLUE_VERIFIED_FOLLOWERS, - cursor + user_id, count, Endpoint.BLUE_VERIFIED_FOLLOWERS, cursor ) def get_user_followers_you_know( @@ -3320,10 +3265,7 @@ def get_user_followers_you_know( A list of User objects representing the followers you might know. """ return self._get_user_friendship( - user_id, - count, - Endpoint.FOLLOWERS_YOU_KNOW, - cursor + user_id, count, Endpoint.FOLLOWERS_YOU_KNOW, cursor ) def get_user_following( @@ -3345,10 +3287,7 @@ def get_user_following( A list of User objects representing the users being followed. """ return self._get_user_friendship( - user_id, - count, - Endpoint.FOLLOWING, - cursor + user_id, count, Endpoint.FOLLOWING, cursor ) def get_user_subscriptions( @@ -3370,10 +3309,7 @@ def get_user_subscriptions( A list of User objects representing the subscribed users. """ return self._get_user_friendship( - user_id, - count, - Endpoint.SUBSCRIPTIONS, - cursor + user_id, count, Endpoint.SUBSCRIPTIONS, cursor ) def _get_friendship_ids( @@ -3406,8 +3342,7 @@ def _get_friendship_ids( partial(self._get_friendship_ids, user_id, screen_name, count, endpoint, next_cursor), next_cursor, - partial(self._get_friendship_ids, user_id, - screen_name, count, endpoint, previous_cursor), + partial(self._get_friendship_ids, user_id, screen_name, count, endpoint, previous_cursor), previous_cursor ) @@ -3436,11 +3371,7 @@ def get_followers_ids( A Result object containing the IDs of the followers. """ return self._get_friendship_ids( - user_id, - screen_name, - count, - Endpoint.FOLLOWERS_IDS, - cursor + user_id, screen_name, count, Endpoint.FOLLOWERS_IDS, cursor ) def get_friends_ids( @@ -3468,11 +3399,7 @@ def get_friends_ids( A Result object containing the IDs of the friends. """ return self._get_friendship_ids( - user_id, - screen_name, - count, - Endpoint.FRIENDS_IDS, - cursor + user_id, screen_name, count, Endpoint.FRIENDS_IDS, cursor ) def _send_dm( @@ -3514,9 +3441,7 @@ def _get_dm_history( """ Base function to get dm history. """ - params = { - 'context': 'FETCH_DM_CONVERSATION_HISTORY' - } + params = {'context': 'FETCH_DM_CONVERSATION_HISTORY'} if max_id is not None: params['max_id'] = max_id @@ -3809,9 +3734,7 @@ def send_dm_to_group( .upload_media .delete_dm """ - response = self._send_dm( - group_id, text, media_id, reply_to - ) + response = self._send_dm(group_id, text, media_id, reply_to) message_data = find_dict(response, 'message_data', find_one=True)[0] users = list(response['users'].values()) @@ -4119,9 +4042,7 @@ def edit_list( ... 'new name', 'new description', True ... ) """ - variables = { - 'listId': list_id - } + variables = {'listId': list_id} if name is not None: variables['name'] = name if description is not None: @@ -4240,9 +4161,7 @@ def get_lists( ... >>> more_lists = lists.next() # Retrieve more lists """ - variables = { - 'count': count - } + variables = {'count': count} if cursor is not None: variables['cursor'] = cursor params = flatten_params({ @@ -4263,9 +4182,7 @@ def get_lists( lists = [] for list in items[1]: - lists.append( - List(self, list['item']['itemContent']['list']) - ) + lists.append(List(self, list['item']['itemContent']['list'])) next_cursor = entries[-1]['content']['value'] @@ -4339,10 +4256,7 @@ def get_list_tweets( ... ... """ - variables = { - 'listId': list_id, - 'count': count - } + variables = {'listId': list_id, 'count': count} if cursor is not None: variables['cursor'] = cursor params = flatten_params({ @@ -4379,10 +4293,7 @@ def _get_list_users( """ Base function to retrieve the users associated with a list. """ - variables = { - 'listId': list_id, - 'count': count, - } + variables = {'listId': list_id, 'count': count} if cursor is not None: variables['cursor'] = cursor params = flatten_params({ @@ -4408,8 +4319,7 @@ def _get_list_users( return Result( results, - partial(self._get_list_users, - endpoint, list_id, count, next_cursor), + partial(self._get_list_users, endpoint, list_id, count, next_cursor), next_cursor ) @@ -4442,10 +4352,7 @@ def get_list_members( >>> more_members = members.next() # Retrieve more members """ return self._get_list_users( - Endpoint.LIST_MEMBERS, - list_id, - count, - cursor + Endpoint.LIST_MEMBERS, list_id, count, cursor ) def get_list_subscribers( @@ -4477,10 +4384,7 @@ def get_list_subscribers( >>> more_subscribers = members.next() # Retrieve more subscribers """ return self._get_list_users( - Endpoint.LIST_SUBSCRIBERS, - list_id, - count, - cursor + Endpoint.LIST_SUBSCRIBERS, list_id, count, cursor ) def search_list( @@ -4577,9 +4481,7 @@ def get_notifications( 'Mentions': Endpoint.NOTIFICATIONS_MENTIONES }[type] - params = { - 'count': count - } + params = {'count': count} if cursor is not None: params['cursor'] = cursor @@ -4628,8 +4530,7 @@ def get_notifications( if i['entryId'].startswith('cursor-bottom') ] if cursor_bottom_entry: - next_cursor = find_dict( - cursor_bottom_entry[0], 'value', find_one=True)[0] + next_cursor = find_dict(cursor_bottom_entry[0], 'value', find_one=True)[0] else: next_cursor = None @@ -4666,9 +4567,7 @@ def search_community( >>> more_communities = communities.next() # Retrieve more communities """ - variables = { - 'query': query, - } + variables = {'query': query} if cursor is not None: variables['cursor'] = cursor params = flatten_params({ @@ -4689,8 +4588,7 @@ def search_community( if next_cursor is None: fetch_next_result = None else: - fetch_next_result = partial(self.search_community, - query, next_cursor) + fetch_next_result = partial(self.search_community, query, next_cursor) return Result( communities, fetch_next_result, @@ -4817,11 +4715,9 @@ def get_community_tweets( return Result( tweets, - partial(self.get_community_tweets, community_id, - tweet_type, count, next_cursor), + partial(self.get_community_tweets, community_id, tweet_type, count, next_cursor), next_cursor, - partial(self.get_community_tweets, community_id, - tweet_type, count, previous_cursor), + partial(self.get_community_tweets, community_id, tweet_type, count, previous_cursor), previous_cursor ) @@ -4879,9 +4775,7 @@ def get_communities_timeline( community_data = tweet_data['community_results']['result'] community_data['rest_id'] = community_data['id_str'] community = Community(self, community_data) - tweet = Tweet( - self, tweet_data, User(self, user_data) - ) + tweet = Tweet(self, tweet_data, User(self, user_data)) tweet.community = community tweets.append(tweet) @@ -5057,10 +4951,7 @@ def get_community_members( List of retrieved members. """ return self._get_community_users( - Endpoint.COMMUNITY_MEMBERS, - community_id, - count, - cursor + Endpoint.COMMUNITY_MEMBERS, community_id, count, cursor ) def get_community_moderators( @@ -5082,10 +4973,7 @@ def get_community_moderators( List of retrieved moderators. """ return self._get_community_users( - Endpoint.COMMUNITY_MODERATORS, - community_id, - count, - cursor + Endpoint.COMMUNITY_MODERATORS, community_id, count, cursor ) def search_community_tweet( @@ -5149,11 +5037,9 @@ def search_community_tweet( return Result( tweets, - partial(self.search_community_tweet, community_id, - query, count, next_cursor), + partial(self.search_community_tweet, community_id, query, count, next_cursor), next_cursor, - partial(self.search_community_tweet, community_id, - query, count, previous_cursor), + partial(self.search_community_tweet, community_id, query, count, previous_cursor), previous_cursor, ) @@ -5162,9 +5048,7 @@ def _stream(self, topics: set[str]) -> Generator[tuple[str, Payload]]: headers.pop('content-type') params = {'topics': ','.join(topics)} - with self.stream( - 'GET', Endpoint.EVENTS, params=params, timeout=None - ) as response: + with self.stream('GET', Endpoint.EVENTS, params=params, timeout=None) as response: self._remove_duplicate_ct0_cookie() for line in response.iter_lines(): try: @@ -5245,9 +5129,7 @@ def get_streaming_session( """ stream = self._stream(topics) session_id = next(stream)[1].config.session_id - return StreamingSession( - self, session_id, stream, topics, auto_reconnect - ) + return StreamingSession(self, session_id, stream, topics, auto_reconnect) def _update_subscriptions( self, diff --git a/twikit/tweet.py b/twikit/tweet.py index c184a3d1..6a541317 100644 --- a/twikit/tweet.py +++ b/twikit/tweet.py @@ -54,6 +54,8 @@ class Tweet: Indicates if the tweet is favorited. view_count: :class:`int` | None The count of views. + view_count_state : :class:`str` | None + The state of the tweet views. retweet_count : :class:`int` The count of retweets for the tweet. editable_until_msecs : :class:`int` @@ -64,8 +66,6 @@ class Tweet: Indicates if the tweet is eligible for editing. edits_remaining : :class:`int` The remaining number of edits allowed for the tweet. - state : :class:`str` | None - The state of the tweet views. replies: Result[:class:`Tweet`] | None Replies to the tweet. reply_to: list[:class:`Tweet`] | None @@ -99,16 +99,28 @@ def __init__(self, client: Client, data: dict, user: User = None) -> None: self.thread: list[Tweet] | None = None self.id: str = data['rest_id'] - legacy = data['legacy'] self.created_at: str = legacy['created_at'] self.text: str = legacy['full_text'] - self.lang: str = legacy['lang'] self.is_quote_status: bool = legacy['is_quote_status'] - self.in_reply_to: str | None = self._data['legacy'].get( - 'in_reply_to_status_id_str' - ) + self.in_reply_to: str | None = self._data['legacy'].get('in_reply_to_status_id_str') + self.is_quote_status: bool = legacy['is_quote_status'] + self.possibly_sensitive: bool = legacy.get('possibly_sensitive') + self.possibly_sensitive_editable: bool = legacy.get('possibly_sensitive_editable') + self.quote_count: int = legacy['quote_count'] + self.media: list = legacy['entities'].get('media') + self.reply_count: int = legacy['reply_count'] + self.favorite_count: int = legacy['favorite_count'] + self.favorited: bool = legacy['favorited'] + self.retweet_count: int = legacy['retweet_count'] + self.editable_until_msecs: int = data['edit_control'].get('editable_until_msecs') + self.is_translatable: bool = data.get('is_translatable') + self.is_edit_eligible: bool = data['edit_control'].get('is_edit_eligible') + self.edits_remaining: int = data['edit_control'].get('edits_remaining') + self.view_count: str = data['views'].get('count') if 'views' in data else None + self.view_count_state: str = data['views'].get('state') if 'views' in data else None + self.has_community_notes: bool = data.get('has_birdwatch_notes') if data.get('quoted_status_result'): quoted_tweet = data.pop('quoted_status_result')['result'] @@ -135,9 +147,7 @@ def __init__(self, client: Client, data: dict, user: User = None) -> None: else: self.retweeted_tweet = None - note_tweet_results = find_dict( - data, 'note_tweet_results', find_one=True - ) + note_tweet_results = find_dict(data, 'note_tweet_results', find_one=True) self.full_text: str = self.text if note_tweet_results: text_list = find_dict(note_tweet_results, 'text', find_one=True) @@ -155,28 +165,6 @@ def __init__(self, client: Client, data: dict, user: User = None) -> None: i['text'] for i in hashtags ] - self.is_quote_status: bool = legacy['is_quote_status'] - self.possibly_sensitive: bool = legacy.get('possibly_sensitive') - self.possibly_sensitive_editable: bool = legacy.get( - 'possibly_sensitive_editable') - self.quote_count: int = legacy['quote_count'] - self.media: list = legacy['entities'].get('media') - self.reply_count: int = legacy['reply_count'] - self.favorite_count: int = legacy['favorite_count'] - self.favorited: bool = legacy['favorited'] - self.view_count: int = (data['views'].get('count') - if 'views' in data else None) - self.retweet_count: int = legacy['retweet_count'] - self.editable_until_msecs: int = data['edit_control'].get( - 'editable_until_msecs') - self.is_translatable: bool = data.get('is_translatable') - self.is_edit_eligible: bool = data['edit_control'].get( - 'is_edit_eligible') - self.edits_remaining: int = data['edit_control'].get('edits_remaining') - self.state: str = (data['views'].get('state') - if 'views' in data else None) - self.has_community_notes: bool = data.get('has_birdwatch_notes') - self.community_note = None if 'birdwatch_pivot' in data: community_note_data = data['birdwatch_pivot'] @@ -212,10 +200,7 @@ def __init__(self, client: Client, data: dict, user: User = None) -> None: for i in card_data } - if ( - 'title' in binding_values and - 'string_value' in binding_values['title'] - ): + if 'title' in binding_values and 'string_value' in binding_values['title']: self.thumbnail_title = binding_values['title']['string_value'] if ( @@ -599,20 +584,15 @@ def __init__( self.choices = choices - duration_minutes = binding_values['duration_minutes']['string_value'] - self.duration_minutes = int(duration_minutes) - - end = binding_values['end_datetime_utc']['string_value'] + self.duration_minutes = int(binding_values['duration_minutes']['string_value']) + self.end_datetime_utc: str = binding_values['end_datetime_utc']['string_value'] updated = binding_values['last_updated_datetime_utc']['string_value'] - self.end_datetime_utc: str = end self.last_updated_datetime_utc: str = updated - counts_are_final = binding_values['counts_are_final']['boolean_value'] - self.counts_are_final: bool = counts_are_final + self.counts_are_final: bool = binding_values['counts_are_final']['boolean_value'] if 'selected_choice' in binding_values: - selected_choice = binding_values['selected_choice']['string_value'] - self.selected_choice: str = selected_choice + self.selected_choice: str = binding_values['selected_choice']['string_value'] else: self.selected_choice = None diff --git a/twikit/twikit_async/client.py b/twikit/twikit_async/client.py index 92145d9d..ed15b054 100644 --- a/twikit/twikit_async/client.py +++ b/twikit/twikit_async/client.py @@ -435,8 +435,7 @@ async def login( if flow.task_id == 'LoginTwoFactorAuthChallenge': if totp_secret is None: - print(find_dict( - flow.response, 'secondary_text', find_one=True)[0]['text']) + print(find_dict(flow.response, 'secondary_text', find_one=True)[0]['text']) totp_code = input('>>>') else: totp_code = pyotp.TOTP(totp_secret).now() @@ -450,8 +449,7 @@ async def login( }) if flow.task_id == 'LoginAcid': - print(find_dict( - flow.response, 'secondary_text', find_one=True)[0]['text']) + print(find_dict(flow.response, 'secondary_text', find_one=True)[0]['text']) await flow.execute_task({ 'subtask_id': 'LoginAcid', @@ -714,9 +712,7 @@ async def search_tweet( if next_cursor is None: if product == 'Media': - entries = find_dict( - instructions, 'entries', find_one=True - )[0] + entries = find_dict(instructions, 'entries', find_one=True)[0] next_cursor = entries[-1]['content']['value'] previous_cursor = entries[-2]['content']['value'] else: @@ -786,8 +782,7 @@ async def search_user( return Result( results, - partial(self.search_user, - query, count, next_cursor), + partial(self.search_user, query, count, next_cursor), next_cursor ) @@ -960,12 +955,8 @@ async def upload_media( ) } - coro = self.post( - endpoint, - params=params, - headers=headers, - files=files - ) + coro = self.post(endpoint, params=params, + headers=headers, files=files) tasks.append(asyncio.create_task(coro)) chunk_streams.append(chunk_stream) @@ -1119,12 +1110,8 @@ async def create_poll( for i, choice in enumerate(choices, 1): card_data[f'twitter:string:choice{i}_label'] = choice - data = urlencode( - {'card_data': card_data} - ) - headers = self._base_headers | { - 'content-type': 'application/x-www-form-urlencoded' - } + data = urlencode({'card_data': card_data}) + headers = self._base_headers | {'content-type': 'application/x-www-form-urlencoded'} response, _ = await self.post( Endpoint.CREATE_CARD, data=data, @@ -1185,8 +1172,7 @@ async def create_tweet( media_ids: list[str] | None = None, poll_uri: str | None = None, reply_to: str | None = None, - conversation_control: Literal[ - 'followers', 'verified', 'mentioned'] | None = None, + conversation_control: Literal['followers', 'verified', 'mentioned'] | None = None, attachment_url: str | None = None, community_id: str | None = None, share_with_followers: bool = False, @@ -1340,9 +1326,7 @@ async def create_tweet( if not _result: raise_exceptions_from_response(response['errors']) raise CouldNotTweet( - response['errors'][0] - if response['errors'] - else 'Failed to post a tweet.' + response['errors'][0] if response['errors'] else 'Failed to post a tweet.' ) tweet_info = _result[0] @@ -1679,8 +1663,7 @@ async def _get_more_replies( if entries[-1]['entryId'].startswith('cursor'): next_cursor = entries[-1]['content']['itemContent']['value'] - _fetch_next_result = partial(self._get_more_replies, - tweet_id, next_cursor) + _fetch_next_result = partial(self._get_more_replies, tweet_id, next_cursor) else: next_cursor = None _fetch_next_result = None @@ -1815,9 +1798,7 @@ async def get_scheduled_tweets(self) -> list[ScheduledTweet]: List of ScheduledTweet objects representing the scheduled tweets. """ - params = flatten_params({ - 'variables': {'ascending': True} - }) + params = flatten_params({'variables': {'ascending': True}}) response, _ = await self.get( Endpoint.FETCH_SCHEDULED_TWEETS, params=params, @@ -1841,9 +1822,7 @@ async def delete_scheduled_tweet(self, tweet_id: str) -> Response: Response returned from twitter api. """ data = { - 'variables': { - 'scheduled_tweet_id': tweet_id - }, + 'variables': {'scheduled_tweet_id': tweet_id}, 'queryId': get_query_id(Endpoint.DELETE_SCHEDULED_TWEET) } _, response = await self.post( @@ -1894,11 +1873,9 @@ async def _get_tweet_engagements( return Result( results, - partial(self._get_tweet_engagements, - tweet_id, count, next_cursor, endpoint), + partial(self._get_tweet_engagements, tweet_id, count, next_cursor, endpoint), next_cursor, - partial(self._get_tweet_engagements, - tweet_id, count, previous_cursor, endpoint), + partial(self._get_tweet_engagements, tweet_id, count, previous_cursor, endpoint), previous_cursor ) @@ -2012,9 +1989,7 @@ async def get_community_note(self, note_id: str) -> CommunityNote: note_data = response['data']['birdwatch_note_by_rest_id'] if 'data_v1' not in note_data: raise TwitterException(f'Invalid user id: {note_id}') - return CommunityNote( - self, note_data - ) + return CommunityNote(self, note_data) async def get_user_tweets( self, @@ -2125,9 +2100,7 @@ async def get_user_tweets( for item in items: entry_id = item['entryId'] - if not entry_id.startswith( - ('tweet', 'profile-conversation', 'profile-grid') - ): + if not entry_id.startswith(('tweet', 'profile-conversation', 'profile-grid')): continue if entry_id.startswith('profile-conversation'): @@ -2150,11 +2123,9 @@ async def get_user_tweets( return Result( results, - partial(self.get_user_tweets, - user_id, tweet_type, count, next_cursor), + partial(self.get_user_tweets, user_id, tweet_type, count, next_cursor), next_cursor, - partial(self.get_user_tweets, - user_id, tweet_type, count, previous_cursor), + partial(self.get_user_tweets, user_id, tweet_type, count, previous_cursor), previous_cursor ) @@ -2366,8 +2337,7 @@ async def get_latest_timeline( return Result( results, - partial(self.get_latest_timeline, - count, seen_tweet_ids, next_cursor), + partial(self.get_latest_timeline, count, seen_tweet_ids, next_cursor), next_cursor ) @@ -2650,8 +2620,7 @@ async def get_bookmarks( next_cursor = items[-1]['content']['value'] if folder_id is None: previous_cursor = items[-2]['content']['value'] - fetch_previous_result = partial(self.get_bookmarks, count, - previous_cursor, folder_id) + fetch_previous_result = partial(self.get_bookmarks, count, previous_cursor, folder_id) else: previous_cursor = None fetch_previous_result = None @@ -2723,9 +2692,7 @@ async def get_bookmark_folders( headers=self._base_headers ) - slice = find_dict( - response, 'bookmark_collections_slice', find_one=True - )[0] + slice = find_dict(response, 'bookmark_collections_slice', find_one=True)[0] results = [] for item in slice['items']: results.append(BookmarkFolder(self, item)) @@ -2778,9 +2745,7 @@ async def edit_bookmark_folder( json=data, headers=self._base_headers ) - return BookmarkFolder( - self, response['data']['bookmark_collection_update'] - ) + return BookmarkFolder(self, response['data']['bookmark_collection_update']) async def delete_bookmark_folder(self, folder_id: str) -> Response: """ @@ -2796,9 +2761,7 @@ async def delete_bookmark_folder(self, folder_id: str) -> Response: :class:`httpx.Response` Response returned from twitter api. """ - variables = { - 'bookmark_collection_id': folder_id - } + variables = {'bookmark_collection_id': folder_id} data = { 'variables': variables, 'queryId': get_query_id(Endpoint.DELETE_BOOKMARK_FOLDER) @@ -2823,9 +2786,7 @@ async def create_bookmark_folder(self, name: str) -> BookmarkFolder: :class:`BookmarkFolder` Newly created bookmark folder. """ - variables = { - 'name': name - } + variables = {'name': name} data = { 'variables': variables, 'queryId': get_query_id(Endpoint.CREATE_BOOKMARK_FOLDER) @@ -2835,9 +2796,7 @@ async def create_bookmark_folder(self, name: str) -> BookmarkFolder: json=data, headers=self._base_headers ) - return BookmarkFolder( - self, response['data']['bookmark_collection_create'] - ) + return BookmarkFolder(self, response['data']['bookmark_collection_create']) async def follow_user(self, user_id: str) -> Response: """ @@ -3049,9 +3008,7 @@ async def unmute_user(self, user_id: str) -> Response: async def get_trends( self, - category: Literal[ - 'trending', 'for-you', 'news', 'sports', 'entertainment' - ], + category: Literal['trending', 'for-you', 'news', 'sports', 'entertainment'], count: int = 20, retry: bool = True, additional_request_params: dict | None = None @@ -3119,9 +3076,7 @@ async def get_trends( return [] # Recall the method again, as the trend information # may not be returned due to a Twitter error. - return await self.get_trends( - category, count, retry, additional_request_params - ) + return await self.get_trends(category, count, retry, additional_request_params) items = entries[-1]['content']['timelineModule']['items'] @@ -3244,11 +3199,9 @@ async def _get_user_friendship_2( return Result( results, - partial(self._get_user_friendship_2, user_id, - count, endpoint, next_cursor), + partial(self._get_user_friendship_2, user_id, count, endpoint, next_cursor), next_cursor, - partial(self._get_user_friendship_2, user_id, - count, endpoint, previous_cursor), + partial(self._get_user_friendship_2, user_id, count, endpoint, previous_cursor), previous_cursor ) @@ -3271,10 +3224,7 @@ async def get_user_followers( A list of User objects representing the followers. """ return await self._get_user_friendship( - user_id, - count, - Endpoint.FOLLOWERS, - cursor + user_id, count, Endpoint.FOLLOWERS, cursor ) async def get_latest_followers( @@ -3286,11 +3236,7 @@ async def get_latest_followers( Max count : 200 """ return await self._get_user_friendship_2( - user_id, - screen_name, - count, - Endpoint.FOLLOWERS2, - cursor + user_id, screen_name, count, Endpoint.FOLLOWERS2, cursor ) async def get_latest_friends( @@ -3302,11 +3248,7 @@ async def get_latest_friends( Max count : 200 """ return await self._get_user_friendship_2( - user_id, - screen_name, - count, - Endpoint.FOLLOWING2, - cursor + user_id, screen_name, count, Endpoint.FOLLOWING2, cursor ) async def get_user_verified_followers( @@ -3328,10 +3270,7 @@ async def get_user_verified_followers( A list of User objects representing the verified followers. """ return await self._get_user_friendship( - user_id, - count, - Endpoint.BLUE_VERIFIED_FOLLOWERS, - cursor + user_id, count, Endpoint.BLUE_VERIFIED_FOLLOWERS, cursor ) async def get_user_followers_you_know( @@ -3353,10 +3292,7 @@ async def get_user_followers_you_know( A list of User objects representing the followers you might know. """ return await self._get_user_friendship( - user_id, - count, - Endpoint.FOLLOWERS_YOU_KNOW, - cursor + user_id, count, Endpoint.FOLLOWERS_YOU_KNOW, cursor ) async def get_user_following( @@ -3378,10 +3314,7 @@ async def get_user_following( A list of User objects representing the users being followed. """ return await self._get_user_friendship( - user_id, - count, - Endpoint.FOLLOWING, - cursor + user_id, count, Endpoint.FOLLOWING, cursor ) async def get_user_subscriptions( @@ -3403,10 +3336,7 @@ async def get_user_subscriptions( A list of User objects representing the subscribed users. """ return await self._get_user_friendship( - user_id, - count, - Endpoint.SUBSCRIPTIONS, - cursor + user_id, count, Endpoint.SUBSCRIPTIONS, cursor ) async def _get_friendship_ids( @@ -3436,11 +3366,9 @@ async def _get_friendship_ids( return Result( response['ids'], - partial(self._get_friendship_ids, user_id, - screen_name, count, endpoint, next_cursor), + partial(self._get_friendship_ids, user_id, screen_name, count, endpoint, next_cursor), next_cursor, - partial(self._get_friendship_ids, user_id, - screen_name, count, endpoint, previous_cursor), + partial(self._get_friendship_ids, user_id, screen_name, count, endpoint, previous_cursor), previous_cursor ) @@ -3469,11 +3397,7 @@ async def get_followers_ids( A Result object containing the IDs of the followers. """ return await self._get_friendship_ids( - user_id, - screen_name, - count, - Endpoint.FOLLOWERS_IDS, - cursor + user_id, screen_name, count, Endpoint.FOLLOWERS_IDS, cursor ) async def get_friends_ids( @@ -3501,11 +3425,7 @@ async def get_friends_ids( A Result object containing the IDs of the friends. """ return await self._get_friendship_ids( - user_id, - screen_name, - count, - Endpoint.FRIENDS_IDS, - cursor + user_id, screen_name, count, Endpoint.FRIENDS_IDS, cursor ) async def _send_dm( @@ -3547,9 +3467,7 @@ async def _get_dm_history( """ Base function to get dm history. """ - params = { - 'context': 'FETCH_DM_CONVERSATION_HISTORY' - } + params = {'context': 'FETCH_DM_CONVERSATION_HISTORY'} if max_id is not None: params['max_id'] = max_id @@ -3846,9 +3764,7 @@ async def send_dm_to_group( .upload_media .delete_dm """ - response = await self._send_dm( - group_id, text, media_id, reply_to - ) + response = await self._send_dm(group_id, text, media_id, reply_to) message_data = find_dict(response, 'message_data', find_one=True)[0] users = list(response['users'].values()) @@ -4157,9 +4073,7 @@ async def edit_list( ... 'new name', 'new description', True ... ) """ - variables = { - 'listId': list_id - } + variables = {'listId': list_id} if name is not None: variables['name'] = name if description is not None: @@ -4278,9 +4192,7 @@ async def get_lists( ... >>> more_lists = lists.next() # Retrieve more lists """ - variables = { - 'count': count - } + variables = {'count': count} if cursor is not None: variables['cursor'] = cursor params = flatten_params({ @@ -4301,9 +4213,7 @@ async def get_lists( lists = [] for list in items[1]: - lists.append( - List(self, list['item']['itemContent']['list']) - ) + lists.append(List(self, list['item']['itemContent']['list'])) next_cursor = entries[-1]['content']['value'] @@ -4377,10 +4287,7 @@ async def get_list_tweets( ... ... """ - variables = { - 'listId': list_id, - 'count': count - } + variables = {'listId': list_id, 'count': count} if cursor is not None: variables['cursor'] = cursor params = flatten_params({ @@ -4417,10 +4324,7 @@ async def _get_list_users( """ Base function to retrieve the users associated with a list. """ - variables = { - 'listId': list_id, - 'count': count, - } + variables = {'listId': list_id, 'count': count} if cursor is not None: variables['cursor'] = cursor params = flatten_params({ @@ -4444,14 +4348,9 @@ async def _get_list_users( next_cursor = item['content']['value'] break - async def _get_more_users(): - return await self._get_list_users( - endpoint, list_id, count, next_cursor - ) - return Result( results, - _get_more_users, + partial(self._get_list_users, endpoint, list_id, count, next_cursor), next_cursor ) @@ -4484,10 +4383,7 @@ async def get_list_members( >>> more_members = members.next() # Retrieve more members """ return await self._get_list_users( - Endpoint.LIST_MEMBERS, - list_id, - count, - cursor + Endpoint.LIST_MEMBERS, list_id, count, cursor ) async def get_list_subscribers( @@ -4519,10 +4415,7 @@ async def get_list_subscribers( >>> more_subscribers = members.next() # Retrieve more subscribers """ return await self._get_list_users( - Endpoint.LIST_SUBSCRIBERS, - list_id, - count, - cursor + Endpoint.LIST_SUBSCRIBERS, list_id, count, cursor ) async def search_list( @@ -4619,9 +4512,7 @@ async def get_notifications( 'Mentions': Endpoint.NOTIFICATIONS_MENTIONES }[type] - params = { - 'count': count - } + params = {'count': count} if cursor is not None: params['cursor'] = cursor @@ -4670,9 +4561,7 @@ async def get_notifications( if i['entryId'].startswith('cursor-bottom') ] if cursor_bottom_entry: - next_cursor = find_dict( - cursor_bottom_entry[0], 'value', find_one=True - )[0] + next_cursor = find_dict(cursor_bottom_entry[0], 'value', find_one=True)[0] else: next_cursor = None @@ -4733,8 +4622,7 @@ async def search_community( if next_cursor is None: fetch_next_result = None else: - fetch_next_result = partial(self.search_community, - query, next_cursor) + fetch_next_result = partial(self.search_community, query, next_cursor) return Result( communities, fetch_next_result, @@ -4861,11 +4749,9 @@ async def get_community_tweets( return Result( tweets, - partial(self.get_community_tweets, community_id, - tweet_type, count, next_cursor), + partial(self.get_community_tweets, community_id, tweet_type, count, next_cursor), next_cursor, - partial(self.get_community_tweets, community_id, - tweet_type, count, previous_cursor), + partial(self.get_community_tweets, community_id, tweet_type, count, previous_cursor), previous_cursor ) @@ -4922,9 +4808,7 @@ async def get_communities_timeline( community_data = tweet_data['community_results']['result'] community_data['rest_id'] = community_data['id_str'] community = Community(self, community_data) - tweet = Tweet( - self, tweet_data, User(self, user_data) - ) + tweet = Tweet(self, tweet_data, User(self, user_data)) tweet.community = community tweets.append(tweet) @@ -5100,10 +4984,7 @@ async def get_community_members( List of retrieved members. """ return await self._get_community_users( - Endpoint.COMMUNITY_MEMBERS, - community_id, - count, - cursor + Endpoint.COMMUNITY_MEMBERS, community_id, count, cursor ) async def get_community_moderators( @@ -5125,10 +5006,7 @@ async def get_community_moderators( List of retrieved moderators. """ return await self._get_community_users( - Endpoint.COMMUNITY_MODERATORS, - community_id, - count, - cursor + Endpoint.COMMUNITY_MODERATORS, community_id, count, cursor ) async def search_community_tweet( @@ -5192,11 +5070,9 @@ async def search_community_tweet( return Result( tweets, - partial(self.search_community_tweet, community_id, - query, count, next_cursor), + partial(self.search_community_tweet, community_id, query, count, next_cursor), next_cursor, - partial(self.search_community_tweet, community_id, - query, count, previous_cursor), + partial(self.search_community_tweet, community_id, query, count, previous_cursor), previous_cursor, ) @@ -5207,9 +5083,7 @@ async def _stream( headers.pop('content-type') params = {'topics': ','.join(topics)} - async with self.stream( - 'GET', Endpoint.EVENTS, params=params, timeout=None - ) as response: + async with self.stream('GET', Endpoint.EVENTS, params=params, timeout=None) as response: self._remove_duplicate_ct0_cookie() async for line in response.aiter_lines(): try: @@ -5292,9 +5166,7 @@ async def get_streaming_session( """ stream = self._stream(topics) session_id = (await anext(stream))[1].config.session_id - return StreamingSession( - self, session_id, stream, topics, auto_reconnect - ) + return StreamingSession(self, session_id, stream, topics, auto_reconnect) async def _update_subscriptions( self, diff --git a/twikit/twikit_async/tweet.py b/twikit/twikit_async/tweet.py index f8d3efa3..8e3197f2 100644 --- a/twikit/twikit_async/tweet.py +++ b/twikit/twikit_async/tweet.py @@ -54,6 +54,8 @@ class Tweet: Indicates if the tweet is favorited. view_count: :class:`int` | None The count of views. + view_count_state : :class:`str` | None + The state of the tweet views. retweet_count : :class:`int` The count of retweets for the tweet. editable_until_msecs : :class:`int` @@ -64,8 +66,6 @@ class Tweet: Indicates if the tweet is eligible for editing. edits_remaining : :class:`int` The remaining number of edits allowed for the tweet. - state : :class:`str` | None - The state of the tweet views. replies: Result[:class:`Tweet`] | None Replies to the tweet. reply_to: list[:class:`Tweet`] | None @@ -97,25 +97,35 @@ def __init__(self, client: Client, data: dict, user: User = None) -> None: self.thread: list[Tweet] | None = None self.id: str = data['rest_id'] - legacy = data['legacy'] self.created_at: str = legacy['created_at'] self.text: str = legacy['full_text'] - self.lang: str = legacy['lang'] self.is_quote_status: bool = legacy['is_quote_status'] - self.in_reply_to: str | None = self._data['legacy'].get( - 'in_reply_to_status_id_str' - ) + self.in_reply_to: str | None = self._data['legacy'].get('in_reply_to_status_id_str') + self.is_quote_status: bool = legacy['is_quote_status'] + self.possibly_sensitive: bool = legacy.get('possibly_sensitive') + self.possibly_sensitive_editable: bool = legacy.get('possibly_sensitive_editable') + self.quote_count: int = legacy['quote_count'] + self.media: list = legacy['entities'].get('media') + self.reply_count: int = legacy['reply_count'] + self.favorite_count: int = legacy['favorite_count'] + self.favorited: bool = legacy['favorited'] + self.retweet_count: int = legacy['retweet_count'] + self.editable_until_msecs: int = data['edit_control'].get('editable_until_msecs') + self.is_translatable: bool = data.get('is_translatable') + self.is_edit_eligible: bool = data['edit_control'].get('is_edit_eligible') + self.edits_remaining: int = data['edit_control'].get('edits_remaining') + self.view_count: str = data['views'].get('count') if 'views' in data else None + self.view_count_state: str = data['views'].get('state') if 'views' in data else None + self.has_community_notes: bool = data.get('has_birdwatch_notes') if data.get('quoted_status_result'): quoted_tweet = data.pop('quoted_status_result')['result'] if 'tweet' in quoted_tweet: quoted_tweet = quoted_tweet['tweet'] if quoted_tweet.get('__typename') != 'TweetTombstone': - quoted_user = User( - client, quoted_tweet['core']['user_results']['result'] - ) + quoted_user = User(client, quoted_tweet['core']['user_results']['result']) self.quote: Tweet = Tweet(client, quoted_tweet, quoted_user) else: self.quote = None @@ -133,9 +143,7 @@ def __init__(self, client: Client, data: dict, user: User = None) -> None: else: self.retweeted_tweet = None - note_tweet_results = find_dict( - data, 'note_tweet_results', find_one=True - ) + note_tweet_results = find_dict(data, 'note_tweet_results', find_one=True) self.full_text: str = self.text if note_tweet_results: text_list = find_dict(note_tweet_results, 'text', find_one=True) @@ -153,28 +161,6 @@ def __init__(self, client: Client, data: dict, user: User = None) -> None: i['text'] for i in hashtags ] - self.is_quote_status: bool = legacy['is_quote_status'] - self.possibly_sensitive: bool = legacy.get('possibly_sensitive') - self.possibly_sensitive_editable: bool = legacy.get( - 'possibly_sensitive_editable') - self.quote_count: int = legacy['quote_count'] - self.media: list = legacy['entities'].get('media') - self.reply_count: int = legacy['reply_count'] - self.favorite_count: int = legacy['favorite_count'] - self.favorited: bool = legacy['favorited'] - self.view_count: int = (data['views'].get('count') - if 'views' in data else None) - self.retweet_count: int = legacy['retweet_count'] - self.editable_until_msecs: int = data['edit_control'].get( - 'editable_until_msecs') - self.is_translatable: bool = data.get('is_translatable') - self.is_edit_eligible: bool = data['edit_control'].get( - 'is_edit_eligible') - self.edits_remaining: int = data['edit_control'].get('edits_remaining') - self.state: str = (data['views'].get('state') - if 'views' in data else None) - self.has_community_notes: bool = data.get('has_birdwatch_notes') - self.community_note = None if 'birdwatch_pivot' in data: community_note_data = data['birdwatch_pivot'] @@ -210,10 +196,7 @@ def __init__(self, client: Client, data: dict, user: User = None) -> None: for i in card_data } - if ( - 'title' in binding_values and - 'string_value' in binding_values['title'] - ): + if 'title' in binding_values and 'string_value' in binding_values['title']: self.thumbnail_title = binding_values['title']['string_value'] if ( @@ -590,20 +573,15 @@ def __init__( self.choices = choices - duration_minutes = binding_values['duration_minutes']['string_value'] - self.duration_minutes = int(duration_minutes) - - end = binding_values['end_datetime_utc']['string_value'] + self.duration_minutes = int(binding_values['duration_minutes']['string_value']) + self.end_datetime_utc: str = binding_values['end_datetime_utc']['string_value'] updated = binding_values['last_updated_datetime_utc']['string_value'] - self.end_datetime_utc: str = end self.last_updated_datetime_utc: str = updated - counts_are_final = binding_values['counts_are_final']['boolean_value'] - self.counts_are_final: bool = counts_are_final + self.counts_are_final: bool = binding_values['counts_are_final']['boolean_value'] if 'selected_choice' in binding_values: - selected_choice = binding_values['selected_choice']['string_value'] - self.selected_choice: str = selected_choice + self.selected_choice: str = binding_values['selected_choice']['string_value'] else: self.selected_choice = None