From b36c462ffc2fe62bb5e63c5f4d9a39dfed4b68bb Mon Sep 17 00:00:00 2001 From: trevor hobenshield Date: Mon, 19 Jun 2023 09:50:09 -0700 Subject: [PATCH] fix csrf conflicts (ct0), added more draft/schedule tweet operations --- readme.md | 17 ++++++++++ setup.py | 20 +++++++++++- twitter/account.py | 80 +++++++++++++++++++++++++++++++++++++++++----- twitter/util.py | 18 ++++++++--- 4 files changed, 122 insertions(+), 13 deletions(-) diff --git a/readme.md b/readme.md index b8a859f..1173389 100644 --- a/readme.md +++ b/readme.md @@ -136,8 +136,25 @@ account.dm_delete(conversation_id='123456-789012') # delete (hide) specific DM account.dm_delete(message_id='123456') + +# get all scheduled tweets +scheduled_tweets = account.scheduled_tweets() +# delete a scheduled tweet +account.delete_scheduled_tweet(12345678) +# get all draft tweets +draft_tweets = account.draft_tweets() + +# delete a draft tweet +account.delete_draft_tweet(12345678) + +# delete all scheduled tweets +account.clear_scheduled_tweets() + +# delete all draft tweets +account.clear_draft_tweets() + # example configuration account.update_settings({ "address_book_live_sync_enabled": False, diff --git a/setup.py b/setup.py index 31cc402..7d72576 100644 --- a/setup.py +++ b/setup.py @@ -14,7 +14,7 @@ install_requires = [ setup( name="twitter-api-client", - version="0.9.7", + version="0.9.8", python_requires=">=3.10.10", description="Twitter API", long_description=dedent(''' @@ -156,6 +156,24 @@ setup( # delete (hide) specific DM account.dm_delete(message_id='123456') + # get all scheduled tweets + scheduled_tweets = account.scheduled_tweets() + + # delete a scheduled tweet + account.delete_scheduled_tweet(12345678) + + # get all draft tweets + draft_tweets = account.draft_tweets() + + # delete a draft tweet + account.delete_draft_tweet(12345678) + + # delete all scheduled tweets + account.clear_scheduled_tweets() + + # delete all draft tweets + account.clear_draft_tweets() + # example configuration account.update_settings({ "address_book_live_sync_enabled": False, diff --git a/twitter/account.py b/twitter/account.py index 067bc9a..3d1b9d6 100644 --- a/twitter/account.py +++ b/twitter/account.py @@ -117,6 +117,44 @@ class Account: }, 'semantic_annotation_ids': [], } + + if reply_params := kwargs.get('reply_params', {}): + variables |= reply_params + if quote_params := kwargs.get('quote_params', {}): + variables |= quote_params + if poll_params := kwargs.get('poll_params', {}): + variables |= poll_params + + draft = kwargs.get('draft') + schedule = kwargs.get('schedule') + + if draft or schedule: + variables = { + 'post_tweet_request': { + 'auto_populate_reply_metadata': False, + 'status': text, + 'exclude_reply_user_ids': [], + 'media_ids': [], + }, + } + if media: + for m in media: + media_id = self._upload_media(m['media']) + variables['post_tweet_request']['media_ids'].append(media_id) + if alt := m.get('alt'): + self._add_alt_text(media_id, alt) + + if schedule: + variables['execute_at'] = ( + datetime.strptime(schedule, "%Y-%m-%d %H:%M").timestamp() + if isinstance(schedule, str) + else schedule + ) + return self.gql('POST', Operation.CreateScheduledTweet, variables) + + return self.gql('POST', Operation.CreateDraftTweet, variables) + + # regular tweet if media: for m in media: media_id = self._upload_media(m['media']) @@ -126,12 +164,7 @@ class Account: }) if alt := m.get('alt'): self._add_alt_text(media_id, alt) - if reply_params := kwargs.get('reply_params', {}): - variables |= reply_params - if quote_params := kwargs.get('quote_params', {}): - variables |= quote_params - if poll_params := kwargs.get('poll_params', {}): - variables |= poll_params + return self.gql('POST', Operation.CreateTweet, variables) def schedule_tweet(self, text: str, date: int | str, *, media: list = None) -> dict: @@ -685,7 +718,7 @@ class Account: # delete single message _id, op = Operation.DMMessageDeleteMutation results['message'] = self.session.post( - f'https://twitter.com/i/api/graphql/{_id}/{op}', + f'{self.gql_api}/{_id}/{op}', json={'queryId': _id, 'variables': {'messageId': message_id}}, ).json() return results @@ -703,7 +736,7 @@ class Account: params['variables']['cursor'] = cursor.pop() _id, op = Operation.DmAllSearchSlice r = self.session.get( - f'https://twitter.com/i/api/graphql/{_id}/{op}', + f'{self.gql_api}/{_id}/{op}', params=build_params(params), ) res = r.json() @@ -721,3 +754,34 @@ class Account: res, cursor = get(cursor) data.append(res) return {'query': query, 'data': data} + + def scheduled_tweets(self, ascending: bool = True) -> dict: + variables = {"ascending": ascending} + return self.gql('GET', Operation.FetchScheduledTweets, variables) + + def delete_scheduled_tweet(self, tweet_id: int) -> dict: + """duplicate, same as `unschedule_tweet()`""" + variables = {'scheduled_tweet_id': tweet_id} + return self.gql('POST', Operation.DeleteScheduledTweet, variables) + + def clear_scheduled_tweets(self) -> None: + user_id = int(re.findall('"u=(\d+)"', self.session.cookies.get('twid'))[0]) + drafts = self.gql('GET', Operation.FetchScheduledTweets, {"ascending": True}) + for _id in set(find_key(drafts, 'rest_id')): + if _id != user_id: + self.gql('POST', Operation.DeleteScheduledTweet, {'scheduled_tweet_id': _id}) + + def draft_tweets(self, ascending: bool = True) -> dict: + variables = {"ascending": ascending} + return self.gql('GET', Operation.FetchDraftTweets, variables) + + def delete_draft_tweet(self, tweet_id: int) -> dict: + variables = {'draft_tweet_id': tweet_id} + return self.gql('POST', Operation.DeleteDraftTweet, variables) + + def clear_draft_tweets(self) -> None: + user_id = int(re.findall('"u=(\d+)"', self.session.cookies.get('twid'))[0]) + drafts = self.gql('GET', Operation.FetchDraftTweets, {"ascending": True}) + for _id in set(find_key(drafts, 'rest_id')): + if _id != user_id: + self.gql('POST', Operation.DeleteDraftTweet, {'draft_tweet_id': _id}) diff --git a/twitter/util.py b/twitter/util.py index 7a84892..45802b9 100644 --- a/twitter/util.py +++ b/twitter/util.py @@ -108,14 +108,16 @@ def get_headers(session, **kwargs) -> dict: """ Get the headers required for authenticated requests """ + cookies = session.cookies + cookies.delete('ct0', domain='.twitter.com') headers = kwargs | { 'authorization': 'Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs=1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA', - 'cookie': '; '.join(f'{k}={v}' for k, v in session.cookies.items()), + 'cookie': '; '.join(f'{k}={v}' for k, v in cookies.items()), 'referer': 'https://twitter.com/', 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36', - 'x-csrf-token': session.cookies.get('ct0', ''), - 'x-guest-token': session.cookies.get('guest_token', ''), - 'x-twitter-auth-type': 'OAuth2Session' if session.cookies.get('auth_token') else '', + 'x-csrf-token': cookies.get('ct0', ''), + 'x-guest-token': cookies.get('guest_token', ''), + 'x-twitter-auth-type': 'OAuth2Session' if cookies.get('auth_token') else '', 'x-twitter-active-user': 'yes', 'x-twitter-client-language': 'en', } @@ -209,6 +211,14 @@ def get_ids(data: list | dict, operation: tuple) -> set: expr = ID_MAP[operation[-1]] return {k for k in find_key(data, 'entryId') if re.search(expr, k)} + +def dump(path: str, **kwargs): + fname, data = list(kwargs.items())[0] + out = Path(path) + out.mkdir(exist_ok=True, parents=True) + (out / f'{fname}_{time.time_ns()}.json').write_bytes( + orjson.dumps(data, option=orjson.OPT_INDENT_2 | orjson.OPT_SORT_KEYS)) + # def init_protonmail_session(email: str, password: str) -> protonmail.api.Session: # """ # Create an authenticated Proton Mail session