From 5cdeb5369572ac6109bde9904a326b5f42e1faf0 Mon Sep 17 00:00:00 2001 From: d60 Date: Fri, 26 Jul 2024 02:50:12 +0900 Subject: [PATCH] v2.1.0 --- examples/guest.py | 26 +++ twikit/__init__.py | 2 +- twikit/client/client.py | 69 ++++++- twikit/client/gql.py | 91 ++++++--- twikit/client/v11.py | 9 +- twikit/constants.py | 54 ++++++ twikit/guest/__init__.py | 3 + twikit/guest/client.py | 394 +++++++++++++++++++++++++++++++++++++++ twikit/guest/tweet.py | 215 +++++++++++++++++++++ twikit/guest/user.py | 196 +++++++++++++++++++ twikit/notification.py | 4 +- twikit/user.py | 34 ++++ twikit/utils.py | 7 + 13 files changed, 1074 insertions(+), 30 deletions(-) create mode 100644 examples/guest.py create mode 100644 twikit/guest/__init__.py create mode 100644 twikit/guest/client.py create mode 100644 twikit/guest/tweet.py create mode 100644 twikit/guest/user.py diff --git a/examples/guest.py b/examples/guest.py new file mode 100644 index 0000000..4c48444 --- /dev/null +++ b/examples/guest.py @@ -0,0 +1,26 @@ +import asyncio + +from twikit.guest import GuestClient + +client = GuestClient() + + +async def main(): + # Activate the client by generating a guest token. + await client.activate() + + # Get user by screen name + user = await client.get_user_by_screen_name('elonmusk') + print(user) + # Get user by ID + user = await client.get_user_by_id('44196397') + print(user) + + + user_tweets = await client.get_user_tweets('44196397') + print(user_tweets) + + tweet = await client.get_tweet_by_id('1519480761749016577') + print(tweet) + +asyncio.run(main()) diff --git a/twikit/__init__.py b/twikit/__init__.py index b4c28a6..4bf53f0 100644 --- a/twikit/__init__.py +++ b/twikit/__init__.py @@ -7,7 +7,7 @@ A Python library for interacting with the Twitter API. """ -__version__ = '2.0.3' +__version__ = '2.1.0' import asyncio import os diff --git a/twikit/client/client.py b/twikit/client/client.py index 84ef837..b1af0ad 100644 --- a/twikit/client/client.py +++ b/twikit/client/client.py @@ -43,7 +43,7 @@ from ..trend import Location, PlaceTrend, PlaceTrends, Trend from ..tweet import CommunityNote, Poll, ScheduledTweet, Tweet, tweet_from_data from ..user import User -from ..utils import Flow, Result, build_tweet_data, build_user_data, find_dict, httpx_transport_to_url +from ..utils import Flow, Result, build_tweet_data, build_user_data, find_dict, find_entry_by_type, httpx_transport_to_url from .gql import GQLClient from .v11 import V11Client @@ -751,6 +751,73 @@ async def get_similar_tweets(self, tweet_id: str) -> list[Tweet]: return results + async def get_user_highlights_tweets( + self, + user_id: str, + count: int = 20, + cursor: str | None = None + ) -> Result[Tweet]: + """ + Retrieves highlighted tweets from a user's timeline. + + Parameters + ---------- + user_id : :class:`str` + The user ID + count : :class:`int`, default=20 + The number of tweets to retrieve. + + Returns + ------- + Result[:class:`Tweet`] + An instance of the `Result` class containing the highlighted tweets. + + Examples + -------- + >>> result = await client.get_user_highlights_tweets('123456789') + >>> for tweet in result: + ... print(tweet) + + + ... + ... + + >>> more_results = await result.next() # Retrieve more highlighted tweets + >>> for tweet in more_results: + ... print(tweet) + + + ... + ... + """ + response, _ = await self.gql.user_highlights_tweets(user_id, count, cursor) + + instructions = response['data']['user']['result']['timeline']['timeline']['instructions'] + instruction = find_entry_by_type(instructions, 'TimelineAddEntries') + if instruction is None: + return Result.empty() + entries = instruction['entries'] + previous_cursor = None + next_cursor = None + results = [] + + for entry in entries: + entryId = entry['entryId'] + if entryId.startswith('tweet'): + results.append(tweet_from_data(self, entry)) + elif entryId.startswith('cursor-top'): + previous_cursor = entry['content']['value'] + elif entryId.startswith('cursor-bottom'): + next_cursor = entry['content']['value'] + + return Result( + results, + partial(self.get_user_highlights_tweets, user_id, count, next_cursor), + next_cursor, + partial(self.get_user_highlights_tweets, user_id, count, previous_cursor), + previous_cursor + ) + async def upload_media( self, source: str | bytes, diff --git a/twikit/client/gql.py b/twikit/client/gql.py index e806fe2..1ae8386 100644 --- a/twikit/client/gql.py +++ b/twikit/client/gql.py @@ -11,13 +11,18 @@ LIST_FEATURES, NOTE_TWEET_FEATURES, SIMILAR_POSTS_FEATURES, - USER_FEATURES + TWEET_RESULT_BY_REST_ID_FEATURES, + USER_FEATURES, + USER_HIGHLIGHTS_TWEETS_FEATURES ) from ..utils import flatten_params, get_query_id if TYPE_CHECKING: + from ..guest.client import GuestClient from .client import Client + ClientType = Client | GuestClient + class Endpoint: @staticmethod @@ -33,6 +38,7 @@ def url(path): USER_BY_SCREEN_NAME = url('NimuplG1OB7Fd2btCLdBOw/UserByScreenName') USER_BY_REST_ID = url('tD8zKvQzwY3kdx5yz6YmOw/UserByRestId') TWEET_DETAIL = url('U0HTv-bAWTBYylwEMT7x5A/TweetDetail') + TWEET_RESULT_BY_REST_ID = url('Xl5pC_lBk_gcO2ItU39DQw/TweetResultByRestId') FETCH_SCHEDULED_TWEETS = url('ITtjAzvlZni2wWXwf295Qg/FetchScheduledTweets') DELETE_SCHEDULED_TWEET = url('CTOVqej0JBXAZSwkp1US0g/DeleteScheduledTweet') RETWEETERS = url('X-XEqG5qHQSAwmvy00xfyQ/Retweeters') @@ -42,6 +48,7 @@ def url(path): USER_TWEETS_AND_REPLIES = url('vMkJyzx1wdmvOeeNG0n6Wg/UserTweetsAndReplies') USER_MEDIA = url('2tLOJWwGuCTytDrGBg8VwQ/UserMedia') USER_LIKES = url('IohM3gxQHfvWePH5E3KuNA/Likes') + USER_HIGHLIGHTS_TWEETS = url('tHFm_XZc_NNi-CfUThwbNw/UserHighlightsTweets') HOME_TIMELINE = url('-X_hcgQzmHGl29-UXxz4sw/HomeTimeline') HOME_LATEST_TIMELINE = url('U0cdisy7QFIoTfu3-Okw0A/HomeLatestTimeline') FAVORITE_TWEET = url('lI07N6Otwv1PhnEgXILM7A/FavoriteTweet') @@ -92,7 +99,7 @@ def url(path): class GQLClient: - def __init__(self, base: Client) -> None: + def __init__(self, base: ClientType) -> None: self.base = base async def gql_get( @@ -320,14 +327,30 @@ async def user_media(self, user_id, count, cursor): async def user_likes(self, user_id, count, cursor): return await self._get_user_tweets(user_id, count, cursor, Endpoint.USER_LIKES) + async def user_highlights_tweets(self, user_id, count, cursor): + variables = { + 'userId': user_id, + 'count': count, + 'includePromotedContent': True, + 'withVoice': True + } + if cursor is not None: + variables['cursor'] = cursor + return await self.gql_get( + Endpoint.USER_HIGHLIGHTS_TWEETS, + variables, + USER_HIGHLIGHTS_TWEETS_FEATURES, + self.base._base_headers + ) + async def home_timeline(self, count, seen_tweet_ids, cursor): variables = { - "count": count, - "includePromotedContent": True, - "latestControlAvailable": True, - "requestContext": "launch", - "withCommunity": True, - "seenTweetIds": seen_tweet_ids or [] + 'count': count, + 'includePromotedContent': True, + 'latestControlAvailable': True, + 'requestContext': 'launch', + 'withCommunity': True, + 'seenTweetIds': seen_tweet_ids or [] } if cursor is not None: variables['cursor'] = cursor @@ -335,12 +358,12 @@ async def home_timeline(self, count, seen_tweet_ids, cursor): async def home_latest_timeline(self, count, seen_tweet_ids, cursor): variables = { - "count": count, - "includePromotedContent": True, - "latestControlAvailable": True, - "requestContext": "launch", - "withCommunity": True, - "seenTweetIds": seen_tweet_ids or [] + 'count': count, + 'includePromotedContent': True, + 'latestControlAvailable': True, + 'requestContext': 'launch', + 'withCommunity': True, + 'seenTweetIds': seen_tweet_ids or [] } if cursor is not None: variables['cursor'] = cursor @@ -632,16 +655,38 @@ async def moderators_slice_timeline_query(self, community_id, count, cursor): async def community_tweet_search_module_query(self, community_id, query, count, cursor): variables = { - "count": count, - "query": query, - "communityId": community_id, - "includePromotedContent": False, - "withBirdwatchNotes": True, - "withVoice": False, - "isListMemberTargetUserId": "0", - "withCommunity": False, - "withSafetyModeUserFields": True + 'count': count, + 'query': query, + 'communityId': community_id, + 'includePromotedContent': False, + 'withBirdwatchNotes': True, + 'withVoice': False, + 'isListMemberTargetUserId': '0', + 'withCommunity': False, + 'withSafetyModeUserFields': True } if cursor is not None: variables['cursor'] = cursor return await self.gql_get(Endpoint.COMMUNITY_TWEET_SEARCH_MODULE_QUERY, variables, COMMUNITY_TWEETS_FEATURES) + + #################### + # For guest client + #################### + + async def tweet_result_by_rest_id(self, tweet_id): + variables = { + 'tweetId': tweet_id, + 'withCommunity': False, + 'includePromotedContent': False, + 'withVoice': False + } + params = { + 'fieldToggles': { + 'withArticleRichContentState': True, + 'withArticlePlainText': False, + 'withGrokAnalyze': False + } + } + return await self.gql_get( + Endpoint.TWEET_RESULT_BY_REST_ID, variables, TWEET_RESULT_BY_REST_ID_FEATURES, extra_params=params + ) diff --git a/twikit/client/v11.py b/twikit/client/v11.py index fff4623..74a89d2 100644 --- a/twikit/client/v11.py +++ b/twikit/client/v11.py @@ -4,8 +4,11 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: + from ..guest.client import GuestClient from .client import Client + ClientType = Client | GuestClient + class Endpoint: GUEST_ACTIVATE = 'https://api.twitter.com/1.1/guest/activate.json' @@ -45,13 +48,13 @@ class Endpoint: class V11Client: - def __init__(self, base: Client) -> None: + def __init__(self, base: ClientType) -> None: self.base = base async def guest_activate(self): headers = self.base._base_headers - headers.pop('X-Twitter-Active-User') - headers.pop('X-Twitter-Auth-Type') + headers.pop('X-Twitter-Active-User', None) + headers.pop('X-Twitter-Auth-Type', None) return await self.base.post( Endpoint.GUEST_ACTIVATE, headers=headers, diff --git a/twikit/constants.py b/twikit/constants.py index 172c7cb..ee0ab64 100644 --- a/twikit/constants.py +++ b/twikit/constants.py @@ -171,3 +171,57 @@ 'longform_notetweets_inline_media_enabled': True, 'responsive_web_enhance_cards_enabled': False } + +TWEET_RESULT_BY_REST_ID_FEATURES = { + 'creator_subscriptions_tweet_preview_api_enabled': True, + 'communities_web_enable_tweet_community_results_fetch': True, + 'c9s_tweet_anatomy_moderator_badge_enabled': True, + 'articles_preview_enabled': True, + 'tweetypie_unmention_optimization_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, + 'creator_subscriptions_quote_tweet_preview_enabled': False, + 'freedom_of_speech_not_reach_fetch_enabled': True, + 'standardized_nudges_misinfo': True, + 'tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled': True, + 'rweb_video_timestamps_enabled': True, + 'longform_notetweets_rich_text_read_enabled': True, + 'longform_notetweets_inline_media_enabled': True, + 'rweb_tipjar_consumption_enabled': True, + 'responsive_web_graphql_exclude_directive_enabled': True, + 'verified_phone_label_enabled': False, + 'responsive_web_graphql_skip_user_profile_image_extensions_enabled': False, + 'responsive_web_graphql_timeline_navigation_enabled': True, + 'responsive_web_enhance_cards_enabled': False +} + +USER_HIGHLIGHTS_TWEETS_FEATURES = { + 'rweb_tipjar_consumption_enabled': True, + 'responsive_web_graphql_exclude_directive_enabled': True, + '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, + 'communities_web_enable_tweet_community_results_fetch': True, + 'c9s_tweet_anatomy_moderator_badge_enabled': True, + 'articles_preview_enabled': True, + 'tweetypie_unmention_optimization_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, + 'creator_subscriptions_quote_tweet_preview_enabled': False, + 'freedom_of_speech_not_reach_fetch_enabled': True, + 'standardized_nudges_misinfo': True, + 'tweet_with_visibility_results_prefer_gql_limited_actions_policy_enabled': True, + 'rweb_video_timestamps_enabled': True, + 'longform_notetweets_rich_text_read_enabled': True, + 'longform_notetweets_inline_media_enabled': True, + 'responsive_web_enhance_cards_enabled': False +} diff --git a/twikit/guest/__init__.py b/twikit/guest/__init__.py new file mode 100644 index 0000000..9f98287 --- /dev/null +++ b/twikit/guest/__init__.py @@ -0,0 +1,3 @@ +from .client import GuestClient +from .tweet import Tweet +from .user import User diff --git a/twikit/guest/client.py b/twikit/guest/client.py new file mode 100644 index 0000000..a3e2833 --- /dev/null +++ b/twikit/guest/client.py @@ -0,0 +1,394 @@ +from __future__ import annotations + +import json +import warnings +from functools import partial +from typing import Any, Literal + +from httpx import AsyncClient, AsyncHTTPTransport, Response +from httpx._utils import URLPattern + +from ..client.gql import GQLClient +from ..client.v11 import V11Client +from ..constants import TOKEN +from ..errors import ( + BadRequest, + Forbidden, + NotFound, + RequestTimeout, + ServerError, + TooManyRequests, + TwitterException, + Unauthorized +) +from ..utils import Result, find_dict, find_entry_by_type, httpx_transport_to_url +from .tweet import Tweet +from .user import User + + +def tweet_from_data(client: GuestClient, data: dict) -> Tweet: + ':meta private:' + tweet_data_ = find_dict(data, 'result', True) + if not tweet_data_: + return None + tweet_data = tweet_data_[0] + + if tweet_data.get('__typename') == 'TweetTombstone': + return None + if 'tweet' in tweet_data: + tweet_data = tweet_data['tweet'] + if 'core' not in tweet_data: + return None + if 'result' not in tweet_data['core']['user_results']: + return None + if 'legacy' not in tweet_data: + return None + + user_data = tweet_data['core']['user_results']['result'] + return Tweet(client, tweet_data, User(client, user_data)) + + + +class GuestClient: + """ + A client for interacting with the Twitter API as a guest. + This class is used for interacting with the Twitter API + without requiring authentication. + + Parameters + ---------- + language : :class:`str` | None, default=None + The language code to use in API requests. + proxy : :class:`str` | None, default=None + The proxy server URL to use for request + (e.g., 'http://0.0.0.0:0000'). + + Examples + -------- + >>> client = GuestClient() + >>> await client.activate() # Activate the client by generating a guest token. + """ + + def __init__( + self, + language: str | None = None, + proxy: str | None = None, + **kwargs + ) -> None: + if 'proxies' in kwargs: + message = ( + "The 'proxies' argument is now deprecated. Use 'proxy' " + "instead. https://github.com/encode/httpx/pull/2879" + ) + warnings.warn(message) + + self.http = AsyncClient(proxy=proxy, **kwargs) + self.language = language + self.proxy = proxy + + self._token = TOKEN + self._user_agent = ('Mozilla/5.0 (Windows NT 10.0; Win64; x64) ' + 'AppleWebKit/537.36 (KHTML, like Gecko) ' + 'Chrome/122.0.0.0 Safari/537.36') + self._guest_token: str | None = None # set when activate method is called + self.gql = GQLClient(self) + self.v11 = V11Client(self) + + async def request( + self, + method: str, + url: str, + raise_exception: bool = True, + **kwargs + ) -> tuple[dict | Any, Response]: + ':meta private:' + response = await self.http.request(method, url, **kwargs) + + try: + response_data = response.json() + except json.decoder.JSONDecodeError: + response_data = response.text + + status_code = response.status_code + + if status_code >= 400 and raise_exception: + message = f'status: {status_code}, message: "{response.text}"' + if status_code == 400: + raise BadRequest(message, headers=response.headers) + elif status_code == 401: + raise Unauthorized(message, headers=response.headers) + elif status_code == 403: + raise Forbidden(message, headers=response.headers) + elif status_code == 404: + raise NotFound(message, headers=response.headers) + elif status_code == 408: + raise RequestTimeout(message, headers=response.headers) + elif status_code == 429: + raise TooManyRequests(message, headers=response.headers) + elif 500 <= status_code < 600: + raise ServerError(message, headers=response.headers) + else: + raise TwitterException(message, headers=response.headers) + + return response_data, response + + async def get(self, url, **kwargs) -> tuple[dict | Any, Response]: + ':meta private:' + return await self.request('GET', url, **kwargs) + + async def post(self, url, **kwargs) -> tuple[dict | Any, Response]: + ':meta private:' + return await self.request('POST', url, **kwargs) + + @property + def proxy(self) -> str: + ':meta private:' + transport: AsyncHTTPTransport = self.http._mounts.get( + URLPattern('all://') + ) + if transport is None: + return None + if not hasattr(transport._pool, '_proxy_url'): + return None + return httpx_transport_to_url(transport) + + @proxy.setter + def proxy(self, url: str) -> None: + self.http._mounts = { + URLPattern('all://'): AsyncHTTPTransport(proxy=url) + } + + @property + def _base_headers(self) -> dict[str, str]: + """ + Base headers for Twitter API requests. + """ + headers = { + 'authorization': f'Bearer {self._token}', + 'content-type': 'application/json', + 'X-Twitter-Active-User': 'yes', + 'Referer': 'https://twitter.com/', + } + + if self.language is not None: + headers['Accept-Language'] = self.language + headers['X-Twitter-Client-Language'] = self.language + + if self._guest_token is not None: + headers['X-Guest-Token'] = self._guest_token + + return headers + + async def activate(self) -> str: + """ + Activate the client by generating a guest token. + """ + if self._guest_token is not None: + return self._guest_token + response, _ = await self.v11.guest_activate() + self._guest_token = response['guest_token'] + + async def get_user_by_screen_name(self, screen_name: str) -> User: + """ + Retrieves a user object based on the provided screen name. + + Parameters + ---------- + screen_name : :class:`str` + The screen name of the user to retrieve. + + Returns + ------- + :class:`.user.User` + An instance of the `User` class containing user details. + + Examples + -------- + >>> user = await client.get_user_by_screen_name('example_user') + >>> print(user) + + """ + response, _ = await self.gql.user_by_screen_name(screen_name) + return User(self, response['data']['user']['result']) + + async def get_user_by_id(self, user_id: str) -> User: + """ + Retrieves a user object based on the provided user ID. + + Parameters + ---------- + user_id : :class:`str` + The ID of the user to retrieve. + + Returns + ------- + :class:`.user.User` + An instance of the `User` class + + Examples + -------- + >>> user = await client.get_user_by_id('123456789') + >>> print(user) + + """ + response, _ = await self.gql.user_by_rest_id(user_id) + return User(self, response['data']['user']['result']) + + async def get_user_tweets( + self, + user_id: str, + tweet_type: Literal['Tweets'] = 'Tweets', + count: int = 40, + ) -> list[Tweet]: + """ + Fetches tweets from a specific user's timeline. + + Parameters + ---------- + user_id : :class:`str` + The ID of the Twitter user whose tweets to retrieve. + To get the user id from the screen name, you can use + `get_user_by_screen_name` method. + tweet_type : {'Tweets'}, default='Tweets' + The type of tweets to retrieve. + count : :class:`int`, default=40 + The number of tweets to retrieve. + + Returns + ------- + list[:class:`.tweet.Tweet`] + A Result object containing a list of `Tweet` objects. + + Examples + -------- + >>> user_id = '...' + + If you only have the screen name, you can get the user id as follows: + + >>> screen_name = 'example_user' + >>> user = client.get_user_by_screen_name(screen_name) + >>> user_id = user.id + + >>> tweets = await client.get_user_tweets(user_id) + >>> for tweet in tweets: + ... print(tweet) + + + ... + ... + + See Also + -------- + .get_user_by_screen_name + """ + tweet_type = tweet_type.capitalize() + f = { + 'Tweets': self.gql.user_tweets, + }[tweet_type] + response, _ = await f(user_id, count, None) + instructions_ = find_dict(response, 'instructions', True) + if not instructions_: + return [] + instructions = instructions_[0] + items = find_entry_by_type(instructions, 'TimelineAddEntries')['entries'] + results = [] + + for item in items: + entry_id = item['entryId'] + if not entry_id.startswith(('tweet', 'profile-conversation', 'profile-grid')): + continue + tweet = tweet_from_data(self, item) + if tweet is None: + continue + results.append(tweet) + + return results + + async def get_tweet_by_id(self, tweet_id: str) -> Tweet: + """ + Fetches a tweet by tweet ID. + + Parameters + ---------- + tweet_id : :class:`str` + The ID of the tweet. + + Returns + ------- + :class:`.tweet.Tweet` + Tweet object + + Examples + -------- + >>> await client.get_tweet_by_id('123456789') + + """ + response, _ = await self.gql.tweet_result_by_rest_id(tweet_id) + return tweet_from_data(self, response) + + async def get_user_highlights_tweets( + self, + user_id: str, + count: int = 20, + cursor: str | None = None + ) -> Result[Tweet]: + """ + Retrieves highlighted tweets from a user's timeline. + + Parameters + ---------- + user_id : :class:`str` + The user ID + count : :class:`int`, default=20 + The number of tweets to retrieve. + + Returns + ------- + Result[:class:`.tweet.Tweet`] + An instance of the `Result` class containing the highlighted tweets. + + Examples + -------- + >>> result = await client.get_user_highlights_tweets('123456789') + >>> for tweet in result: + ... print(tweet) + + + ... + ... + + >>> more_results = await result.next() # Retrieve more highlighted tweets + >>> for tweet in more_results: + ... print(tweet) + + + ... + ... + """ + response, _ = await self.gql.user_highlights_tweets(user_id, count, cursor) + + instructions = response['data']['user']['result']['timeline']['timeline']['instructions'] + instruction = find_entry_by_type(instructions, 'TimelineAddEntries') + if instruction is None: + return Result.empty() + entries = instruction['entries'] + previous_cursor = None + next_cursor = None + results = [] + + for entry in entries: + entryId = entry['entryId'] + if entryId.startswith('tweet'): + results.append(tweet_from_data(self, entry)) + elif entryId.startswith('cursor-top'): + previous_cursor = entry['content']['value'] + elif entryId.startswith('cursor-bottom'): + next_cursor = entry['content']['value'] + + return Result( + results, + partial(self.get_user_highlights_tweets, user_id, count, next_cursor), + next_cursor, + partial(self.get_user_highlights_tweets, user_id, count, previous_cursor), + previous_cursor + ) diff --git a/twikit/guest/tweet.py b/twikit/guest/tweet.py new file mode 100644 index 0000000..aaa41bf --- /dev/null +++ b/twikit/guest/tweet.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from ..utils import find_dict +from .user import User + +if TYPE_CHECKING: + from .client import GuestClient + + +class Tweet: + """ + Attributes + ---------- + id : :class:`str` + The unique identifier of the tweet. + created_at : :class:`str` + The date and time when the tweet was created. + created_at_datetime : :class:`datetime` + The created_at converted to datetime. + user: :class:`.guest.user.User` + Author of the tweet. + text : :class:`str` + The full text of the tweet. + lang : :class:`str` + The language of the tweet. + in_reply_to : :class:`str` + The tweet ID this tweet is in reply to, if any + is_quote_status : :class:`bool` + Indicates if the tweet is a quote status. + quote : :class:`.guest.tweet.Tweet` | None + The Tweet being quoted (if any) + retweeted_tweet : :class:`.guest.tweet.Tweet` | None + The Tweet being retweeted (if any) + possibly_sensitive : :class:`bool` + Indicates if the tweet content may be sensitive. + possibly_sensitive_editable : :class:`bool` + Indicates if the tweet's sensitivity can be edited. + quote_count : :class:`int` + The count of quotes for the tweet. + media : :class:`list` + A list of media entities associated with the tweet. + reply_count : :class:`int` + The count of replies to the tweet. + favorite_count : :class:`int` + The count of favorites or likes for the tweet. + favorited : :class:`bool` + 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. + place : :class:`.Place` | None + The location associated with the tweet. + editable_until_msecs : :class:`int` + The timestamp until which the tweet is editable. + is_translatable : :class:`bool` + Indicates if the tweet is translatable. + is_edit_eligible : :class:`bool` + Indicates if the tweet is eligible for editing. + edits_remaining : :class:`int` + The remaining number of edits allowed for the tweet. + reply_to: list[:class:`Tweet`] | None + A list of Tweet objects representing the tweets to which to reply. + related_tweets : list[:class:`Tweet`] | None + Related tweets. + hashtags: list[:class:`str`] + Hashtags included in the tweet text. + has_card : :class:`bool` + Indicates if the tweet contains a card. + thumbnail_title : :class:`str` | None + The title of the webpage displayed inside tweet's card. + thumbnail_url : :class:`str` | None + Link to the image displayed in the tweet's card. + urls : :class:`list` + Information about URLs contained in the tweet. + full_text : :class:`str` | None + The full text of the tweet. + """ + + def __init__(self, client: GuestClient, data: dict, user: User = None) -> None: + self._client = client + self._data = data + self.user = user + + self.reply_to: list[Tweet] | None = None + self.related_tweets: list[Tweet] | 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.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._place_data = legacy.get('place') + 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']) + self.quote: Tweet = Tweet(client, quoted_tweet, quoted_user) + else: + self.quote = None + + if legacy.get('retweeted_status_result'): + retweeted_tweet = legacy.pop('retweeted_status_result')['result'] + if 'tweet' in retweeted_tweet: + retweeted_tweet = retweeted_tweet['tweet'] + retweeted_user = User( + client, retweeted_tweet['core']['user_results']['result'] + ) + self.retweeted_tweet: Tweet = Tweet( + client, retweeted_tweet, retweeted_user + ) + else: + self.retweeted_tweet = None + + 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) + if text_list: + self.full_text = text_list[0] + + entity_set = note_tweet_results[0]['result']['entity_set'] + self.urls: list = entity_set.get('urls') + hashtags = entity_set.get('hashtags', []) + else: + self.urls: list = legacy['entities'].get('urls') + hashtags = legacy['entities'].get('hashtags', []) + + self.hashtags: list[str] = [ + i['text'] for i in hashtags + ] + + self.community_note = None + if 'birdwatch_pivot' in data: + community_note_data = data['birdwatch_pivot'] + if 'note' in community_note_data: + self.community_note = { + 'id': community_note_data['note']['rest_id'], + 'text': community_note_data['subtitle']['text'] + } + + if ( + 'card' in data and + 'legacy' in data['card'] and + 'name' in data['card']['legacy'] and + data['card']['legacy']['name'].startswith('poll') + ): + self._poll_data = data['card'] + else: + self._poll_data = None + + self.thumbnail_url = None + self.thumbnail_title = None + self.has_card = 'card' in data + if ( + 'card' in data and + 'legacy' in data['card'] and + 'binding_values' in data['card']['legacy'] + ): + card_data = data['card']['legacy']['binding_values'] + + if isinstance(card_data, list): + binding_values = { + i.get('key'): i.get('value') + for i in card_data + } + + if 'title' in binding_values and 'string_value' in binding_values['title']: + self.thumbnail_title = binding_values['title']['string_value'] + + if ( + 'thumbnail_image_original' in binding_values and + 'image_value' in binding_values['thumbnail_image_original'] and + 'url' in binding_values['thumbnail_image_original']['image_value'] + ): + self.thumbnail_url = binding_values['thumbnail_image_original']['image_value']['url'] + + async def update(self) -> None: + new = await self._client.get_tweet_by_id(self.id) + self.__dict__.update(new.__dict__) + + def __repr__(self) -> str: + return f'' + + def __eq__(self, __value: object) -> bool: + return isinstance(__value, Tweet) and self.id == __value.id + + def __ne__(self, __value: object) -> bool: + return not self == __value \ No newline at end of file diff --git a/twikit/guest/user.py b/twikit/guest/user.py new file mode 100644 index 0000000..eb87385 --- /dev/null +++ b/twikit/guest/user.py @@ -0,0 +1,196 @@ +from __future__ import annotations + +from datetime import datetime +from typing import TYPE_CHECKING, Literal + +from ..utils import Result, timestamp_to_datetime + +if TYPE_CHECKING: + from .client import GuestClient + from .tweet import Tweet + + +class User: + """ + Attributes + ---------- + id : :class:`str` + The unique identifier of the user. + created_at : :class:`str` + The date and time when the user account was created. + name : :class:`str` + The user's name. + screen_name : :class:`str` + The user's screen name. + profile_image_url : :class:`str` + The URL of the user's profile image (HTTPS version). + profile_banner_url : :class:`str` + The URL of the user's profile banner. + url : :class:`str` + The user's URL. + location : :class:`str` + The user's location information. + description : :class:`str` + The user's profile description. + description_urls : :class:`list` + URLs found in the user's profile description. + urls : :class:`list` + URLs associated with the user. + pinned_tweet_ids : :class:`str` + The IDs of tweets that the user has pinned to their profile. + is_blue_verified : :class:`bool` + Indicates if the user is verified with a blue checkmark. + verified : :class:`bool` + Indicates if the user is verified. + possibly_sensitive : :class:`bool` + Indicates if the user's content may be sensitive. + can_media_tag : :class:`bool` + Indicates whether the user can be tagged in media. + want_retweets : :class:`bool` + Indicates if the user wants retweets. + default_profile : :class:`bool` + Indicates if the user has the default profile. + default_profile_image : :class:`bool` + Indicates if the user has the default profile image. + has_custom_timelines : :class:`bool` + Indicates if the user has custom timelines. + followers_count : :class:`int` + The count of followers. + fast_followers_count : :class:`int` + The count of fast followers. + normal_followers_count : :class:`int` + The count of normal followers. + following_count : :class:`int` + The count of users the user is following. + favourites_count : :class:`int` + The count of favorites or likes. + listed_count : :class:`int` + The count of lists the user is a member of. + media_count : :class:`int` + The count of media items associated with the user. + statuses_count : :class:`int` + The count of tweets. + is_translator : :class:`bool` + Indicates if the user is a translator. + translator_type : :class:`str` + The type of translator. + profile_interstitial_type : :class:`str` + The type of profile interstitial. + withheld_in_countries : list[:class:`str`] + Countries where the user's content is withheld. + """ + + def __init__(self, client: GuestClient, data: dict) -> None: + self._client = client + legacy = data['legacy'] + + self.id: str = data['rest_id'] + self.created_at: str = legacy['created_at'] + self.name: str = legacy['name'] + self.screen_name: str = legacy['screen_name'] + self.profile_image_url: str = legacy['profile_image_url_https'] + self.profile_banner_url: str = legacy.get('profile_banner_url') + self.url: str = legacy.get('url') + self.location: str = legacy['location'] + self.description: str = legacy['description'] + self.description_urls: list = legacy['entities']['description']['urls'] + self.urls: list = legacy['entities'].get('url', {}).get('urls') + self.pinned_tweet_ids: list[str] = legacy['pinned_tweet_ids_str'] + self.is_blue_verified: bool = data['is_blue_verified'] + self.verified: bool = legacy['verified'] + self.possibly_sensitive: bool = legacy['possibly_sensitive'] + self.default_profile: bool = legacy['default_profile'] + self.default_profile_image: bool = legacy['default_profile_image'] + self.has_custom_timelines: bool = legacy['has_custom_timelines'] + self.followers_count: int = legacy['followers_count'] + self.fast_followers_count: int = legacy['fast_followers_count'] + self.normal_followers_count: int = legacy['normal_followers_count'] + self.following_count: int = legacy['friends_count'] + self.favourites_count: int = legacy['favourites_count'] + self.listed_count: int = legacy['listed_count'] + self.media_count = legacy['media_count'] + self.statuses_count: int = legacy['statuses_count'] + self.is_translator: bool = legacy['is_translator'] + self.translator_type: str = legacy['translator_type'] + self.withheld_in_countries: list[str] = legacy['withheld_in_countries'] + self.protected: bool = legacy.get('protected', False) + + @property + def created_at_datetime(self) -> datetime: + return timestamp_to_datetime(self.created_at) + + async def get_tweets(self, tweet_type: Literal['Tweets'] = 'Tweets', count: int = 40) -> list[Tweet]: + """ + Retrieves the user's tweets. + + Parameters + ---------- + tweet_type : {'Tweets'}, default='Tweets' + The type of tweets to retrieve. + count : :class:`int`, default=40 + The number of tweets to retrieve. + + Returns + ------- + list[:class:`.tweet.Tweet`] + A list of `Tweet` objects. + + Examples + -------- + >>> user = await client.get_user_by_screen_name('example_user') + >>> tweets = await user.get_tweets() + >>> for tweet in tweets: + ... print(tweet) + + + ... + ... + """ + return await self._client.get_user_tweets(self.id, tweet_type, count) + + async def get_highlights_tweets(self, count: int = 20, cursor: str | None = None) -> Result[Tweet]: + """ + Retrieves highlighted tweets from the user's timeline. + + Parameters + ---------- + count : :class:`int`, default=20 + The number of tweets to retrieve. + + Returns + ------- + Result[:class:`.tweet.Tweet`] + An instance of the `Result` class containing the highlighted tweets. + + Examples + -------- + >>> result = await user.get_highlights_tweets() + >>> for tweet in result: + ... print(tweet) + + + ... + ... + + >>> more_results = await result.next() # Retrieve more highlighted tweets + >>> for tweet in more_results: + ... print(tweet) + + + ... + ... + """ + return await self._client.get_user_highlights_tweets(self.id, count, cursor) + + async def update(self) -> None: + new = await self._client.get_user_by_id(self.id) + self.__dict__.update(new.__dict__) + + def __repr__(self) -> str: + return f'' + + def __eq__(self, __value: object) -> bool: + return isinstance(__value, User) and self.id == __value.id + + def __ne__(self, __value: object) -> bool: + return not self == __value diff --git a/twikit/notification.py b/twikit/notification.py index a8c5281..6f8da27 100644 --- a/twikit/notification.py +++ b/twikit/notification.py @@ -20,9 +20,9 @@ class Notification: Dictionary containing icon data for the notification. message : :class:`str` The message text of the notification. - tweet : :class:`Tweet` + tweet : :class:`.Tweet` The tweet associated with the notification. - from_user : :class:`User` + from_user : :class:`.User` The user who triggered the notification. """ def __init__( diff --git a/twikit/user.py b/twikit/user.py index 84bc20f..9717f32 100644 --- a/twikit/user.py +++ b/twikit/user.py @@ -476,6 +476,40 @@ async def get_dm_history(self, max_id: str = None) -> Result[Message]: """ return await self._client.get_dm_history(self.id, max_id) + async def get_highlights_tweets(self, count: int = 20, cursor: str | None = None) -> Result[Tweet]: + """ + Retrieves highlighted tweets from the user's timeline. + + Parameters + ---------- + count : :class:`int`, default=20 + The number of tweets to retrieve. + + Returns + ------- + Result[:class:`Tweet`] + An instance of the `Result` class containing the highlighted tweets. + + Examples + -------- + >>> result = await user.get_highlights_tweets() + >>> for tweet in result: + ... print(tweet) + + + ... + ... + + >>> more_results = await result.next() # Retrieve more highlighted tweets + >>> for tweet in more_results: + ... print(tweet) + + + ... + ... + """ + return await self._client.get_user_highlights_tweets(self.id, count, cursor) + async def update(self) -> None: new = await self._client.get_user_by_id(self.id) self.__dict__.update(new.__dict__) diff --git a/twikit/utils.py b/twikit/utils.py index 6492fac..11adc9e 100644 --- a/twikit/utils.py +++ b/twikit/utils.py @@ -235,6 +235,13 @@ def b64_to_str(b64: str) -> str: return base64.b64decode(b64).decode() +def find_entry_by_type(entries, type_filter): + for entry in entries: + if entry.get('type') == type_filter: + return entry + return None + + FILTERS = Literal[ 'media', 'retweets',