jprdonnelly/twitter-api-client
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix csrf conflicts (ct0), added more draft/schedule tweet operations

Browse Source
This commit is contained in:
trevor hobenshield
2023-06-19 09:50:09 -07:00
parent 89a29e68da
commit b36c462ffc
4 changed files with 122 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
readme.md
View File

@@ -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,

20
setup.py
View File

@@ -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,

80
twitter/account.py
View File

@@ -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})

18
twitter/util.py
View File

@@ -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