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

added lists/pins/topics operations

Browse Source
This commit is contained in:
trevorhobenshield
2023-03-16 17:49:42 -07:00
parent a10b50df94
commit 3a634caa23
5 changed files with 406 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
readme.md
View File

@@ -20,57 +20,68 @@ usr, pwd = ..., ...
session = login(usr, pwd)
r = create_poll('test poll', ['hello', 'world', 'foo', 'bar'], 10080, session)
create_poll('test poll', ['hello', 'world', 'foo', 'bar'], 10080, session)
# DM 1 user
r = dm('hello world', [123], session, filename='test.png')
dm('hello world', [123], session, filename='test.png')
# DM group of users
r = dm('foo bar', [123, 456, 789], session, filename='test.mp4')
dm('foo bar', [123, 456, 789], session, filename='test.mp4')
# create tweet with images, videos, gifs, and tagged users
r = tweet('test 123', session, media=[{'file': 'image.jpeg', 'tagged_users': [123234345456], 'alt': 'some image'}])
r = tweet('test 123', session, media=['test.jpg', 'test.png'])
r = tweet('test 123', session, media=['test.mp4'])
r = tweet('test 123', session)
# tweets
tweet('test 123', session)
tweet('test 123', session, media=['test.jpg', 'test.png'])
tweet('test 123', session, media=['test.mp4'])
tweet('test 123', session, media=[{'file': 'image.jpeg', 'tagged_users': [123234345456], 'alt': 'some image'}])
untweet(123, session)
retweet(1633609779745820675, session)
unretweet(1633609779745820675, session)
quote('test 123', 'elonmusk', 1633609779745820675, session)
comment('test 123', 1633609779745820675, session)
like(1633609779745820675, session)
unlike(1633609779745820675, session)
bookmark(1633609779745820675, session)
unbookmark(1633609779745820675, session)
pin(1635479755364651008, session)
unpin(1635479755364651008, session)
# delete tweet
r = untweet(123, session)
# users
follow(50393960, session)
unfollow(50393960, session)
mute(50393960, session)
unmute(50393960, session)
enable_notifications(50393960, session)
disable_notifications(50393960, session)
block(50393960, session)
unblock(50393960, session)
r = retweet(1633609779745820675, session)
r = unretweet(1633609779745820675, session)
# other
stats(50393960, session)
r = quote('test 123', 'elonmusk', 1633609779745820675, session)
r = comment('test 123', 1633609779745820675, session)
r = unlike(1633609779745820675, session)
r = like(1633609779745820675, session)
r = follow(50393960, session)
r = unfollow(50393960, session)
r = mute(50393960, session)
r = unmute(50393960, session)
r = enable_notifications(50393960, session)
r = disable_notifications(50393960, session)
r = block(50393960, session)
r = unblock(50393960, session)
r = bookmark(1633609779745820675, session)
r = unbookmark(1633609779745820675, session)
r = pin(1635479755364651008, session)
r = unpin(1635479755364651008, session)
r = stats(50393960, session)
# update profile
# user profile
update_profile_image('profile.jpg', session)
update_profile_banner('banner.jpg', session)
update_profile_info(session, name='Foo Bar', description='Test 123', location='Victoria, BC')
# topics
follow_topic(session, 123)
unfollow_topic(session, 123)
# lists
create_list(session, 'My List', 'description of my list', private=False)
update_list(session, 456, 'My Updated List', 'some updated description', private=False)
update_list_banner(session, 456, 'test.jpg')
delete_list_banner(session, 456)
add_list_member(session, 456, 678)
remove_list_member(session, 456, 678)
delete_list(session, 456)
pin_list(session, 456)
unpin_list(session, 456)
# refresh all pinned lists in this order
update_pinned_lists(session, [456, 678, 789])
# unpin all lists
update_pinned_lists(session, [])
```

94
setup.py
View File

@@ -14,7 +14,7 @@ if sys.platform != 'win32':
setup(
name="twitter-api-client",
version="0.2.9",
version="0.3.0",
description="Twitter API",
long_description=dedent('''
## The Undocumented Twitter API
@@ -31,57 +31,69 @@ setup(
session = login(usr, pwd)
r = create_poll('test poll', ['hello', 'world', 'foo', 'bar'], 10080, session)
create_poll('test poll', ['hello', 'world', 'foo', 'bar'], 10080, session)
# DM 1 user
r = dm('hello world', [123], session, filename='test.png')
dm('hello world', [123], session, filename='test.png')
# DM group of users
r = dm('foo bar', [123, 456, 789], session, filename='test.mp4')
dm('foo bar', [123, 456, 789], session, filename='test.mp4')
# create tweet with images, videos, gifs, and tagged users
r = tweet('test 123', session, media=[{'file': 'image.jpeg', 'tagged_users': [123234345456], 'alt': 'some image'}])
r = tweet('test 123', session, media=['test.jpg', 'test.png'])
r = tweet('test 123', session, media=['test.mp4'])
r = tweet('test 123', session)
# tweets
tweet('test 123', session)
tweet('test 123', session, media=['test.jpg', 'test.png'])
tweet('test 123', session, media=['test.mp4'])
tweet('test 123', session, media=[{'file': 'image.jpeg', 'tagged_users': [123234345456], 'alt': 'some image'}])
untweet(123, session)
retweet(1633609779745820675, session)
unretweet(1633609779745820675, session)
quote('test 123', 'elonmusk', 1633609779745820675, session)
comment('test 123', 1633609779745820675, session)
like(1633609779745820675, session)
unlike(1633609779745820675, session)
bookmark(1633609779745820675, session)
unbookmark(1633609779745820675, session)
pin(1635479755364651008, session)
unpin(1635479755364651008, session)
# delete tweet
r = untweet(123, session)
# users
follow(50393960, session)
unfollow(50393960, session)
mute(50393960, session)
unmute(50393960, session)
enable_notifications(50393960, session)
disable_notifications(50393960, session)
block(50393960, session)
unblock(50393960, session)
r = retweet(1633609779745820675, session)
r = unretweet(1633609779745820675, session)
# other
stats(50393960, session)
r = quote('test 123', 'elonmusk', 1633609779745820675, session)
r = comment('test 123', 1633609779745820675, session)
r = unlike(1633609779745820675, session)
r = like(1633609779745820675, session)
r = follow(50393960, session)
r = unfollow(50393960, session)
r = mute(50393960, session)
r = unmute(50393960, session)
r = enable_notifications(50393960, session)
r = disable_notifications(50393960, session)
r = block(50393960, session)
r = unblock(50393960, session)
r = bookmark(1633609779745820675, session)
r = unbookmark(1633609779745820675, session)
r = pin(1635479755364651008, session)
r = unpin(1635479755364651008, session)
r = stats(50393960, session)
# update profile
# user profile
update_profile_image('profile.jpg', session)
update_profile_banner('banner.jpg', session)
update_profile_info(session, name='Foo Bar', description='Test 123', location='Victoria, BC')
# topics
follow_topic(session, 123)
unfollow_topic(session, 123)
# lists
create_list(session, 'My List', 'description of my list', private=False)
update_list(session, 456, 'My Updated List', 'some updated description', private=False)
update_list_banner(session, 456, 'test.jpg')
delete_list_banner(session, 456)
add_list_member(session, 456, 678)
remove_list_member(session, 456, 678)
delete_list(session, 456)
pin_list(session, 456)
unpin_list(session, 456)
# refresh all pinned lists in this order
update_pinned_lists(session, [456, 678, 789])
# unpin all lists
update_pinned_lists(session, [])
```
### Scraping

75
twitter/config/operations.py
View File

@@ -1541,7 +1541,11 @@ operations = {
},
"ListAddMember": {
"queryId": "P8tyfv2_0HzofrB5f6_ugw",
"variables": {},
"variables": {
"listId": None,
"userId": None,
"withSuperFollowsUserFields": False,
},
"features": {
"responsive_web_twitter_blue_verified_badge_is_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
@@ -1552,7 +1556,10 @@ operations = {
},
"DeleteListBanner": {
"queryId": "-bOKetDVCMl20qXn7YDXIA",
"variables": {},
"variables": {
"listId": None,
"withSuperFollowsUserFields": False,
},
"features": {
"responsive_web_twitter_blue_verified_badge_is_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
@@ -1563,7 +1570,11 @@ operations = {
},
"EditListBanner": {
"queryId": "Uk0ZwKSMYng56aQdeJD1yw",
"variables": {},
"variables": {
"listId": None,
"mediaId": None,
"withSuperFollowsUserFields": False,
},
"features": {
"responsive_web_twitter_blue_verified_badge_is_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
@@ -1585,7 +1596,12 @@ operations = {
},
"CreateList": {
"queryId": "hQAsnViq2BrMLbPuQ9umDA",
"variables": {},
"variables": {
"isPrivate": None, # True/False
"name": None,
"description": None,
"withSuperFollowsUserFields": True, # True/False
},
"features": {
"responsive_web_twitter_blue_verified_badge_is_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
@@ -1621,7 +1637,9 @@ operations = {
},
"DeleteList": {
"queryId": "UnN9Th1BDbeLjpgjGSpL3Q",
"variables": {},
"variables": {
"listId": None,
},
"features": {}
},
"FetchDraftTweets": {
@@ -1761,7 +1779,10 @@ operations = {
},
"ListPinOne": {
"queryId": "PdFLmbN9FAT3kxuYphbO6A",
"variables": {},
"variables": {
"listId": None,
"withSuperFollowsUserFields": True,
},
"features": {
"responsive_web_twitter_blue_verified_badge_is_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
@@ -1826,7 +1847,11 @@ operations = {
},
"ListRemoveMember": {
"queryId": "DBZowzFN492FFkBPBptCwg",
"variables": {},
"variables": {
"listId": None,
"userId": None,
"withSuperFollowsUserFields": False,
},
"features": {
"responsive_web_twitter_blue_verified_badge_is_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
@@ -1878,7 +1903,10 @@ operations = {
},
"ListUnpinOne": {
"queryId": "oVn3dJ4Q1HDvq-UYT8AUdg",
"variables": {},
"variables": {
"listId": None,
"withSuperFollowsUserFields": True,
},
"features": {
"responsive_web_twitter_blue_verified_badge_is_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
@@ -1900,7 +1928,13 @@ operations = {
},
"UpdateList": {
"queryId": "4dCEFWtxEbhnSLcJdJ6PNg",
"variables": {},
"variables": {
"listId": None,
"isPrivate": None, # True/False
"description": None,
"name": None,
"withSuperFollowsUserFields": True, # True/False
},
"features": {
"responsive_web_twitter_blue_verified_badge_is_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
@@ -1936,7 +1970,14 @@ operations = {
},
"ListsManagementPageTimeline": {
"queryId": "-xpH2IARz6JqT0nMkHt3KA",
"variables": {},
"variables": {
"count": 1000,
"withDownvotePerspective": False,
"withReactionsMetadata": False,
"withReactionsPerspective": False,
"withSuperFollowsTweetFields": False,
"withSuperFollowsUserFields": False,
},
"features": {
"responsive_web_twitter_blue_verified_badge_is_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
@@ -1961,7 +2002,10 @@ operations = {
},
"ListsPinMany": {
"queryId": "2X4Vqu6XLneR-XZnGK5MAw",
"variables": {},
"variables": {
"listIds": [],
"withSuperFollowsUserFields": True,
},
"features": {
"responsive_web_twitter_blue_verified_badge_is_enabled": True,
"responsive_web_graphql_exclude_directive_enabled": True,
@@ -2320,7 +2364,9 @@ operations = {
},
"TopicFollow": {
"queryId": "ElqSLWFmsPL4NlZI5e1Grg",
"variables": {},
"variables": {
"topicId": None
},
"features": {}
},
"TopicLandingPage": {
@@ -2390,7 +2436,9 @@ operations = {
},
"TopicUnfollow": {
"queryId": "srwjU6JM_ZKTj_QMfUGNcw",
"variables": {},
"variables": {
"topicId": None
},
"features": {}
},
"TopicsManagementPage": {
@@ -3160,6 +3208,7 @@ operations = {
"responsive_web_graphql_timeline_navigation_enabled": True
}
},
# not included in operations discovery through `update_operations.py`
"useSendMessageMutation": {
"queryId": "MaxK2PKX1F9Z-9SwqwavTw",
"variables": {

265
twitter/main.py
View File

@@ -24,12 +24,14 @@ from .utils import get_headers, build_query
try:
if get_ipython().__class__.__name__ == 'ZMQInteractiveShell':
import nest_asyncio
nest_asyncio.apply()
except:
...
if sys.platform != 'win32':
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
else:
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
@@ -39,6 +41,7 @@ logger = logging.getLogger(__name__)
class Operation(Enum):
# tweet
CreateTweet = auto()
CreateScheduledTweet = auto()
DeleteTweet = auto()
@@ -47,12 +50,29 @@ class Operation(Enum):
UnfavoriteTweet = auto()
CreateRetweet = auto()
DeleteRetweet = auto()
# bookmark
CreateBookmark = auto()
DeleteBookmark = auto()
BookmarksAllDelete = auto()
TweetStats = auto()
# topic
TopicFollow = auto()
TopicUnfollow = auto()
# list
ListsManagementPageTimeline = auto()
CreateList = auto()
DeleteList = auto()
EditListBanner = auto()
DeleteListBanner = auto()
ListAddMember = auto()
ListRemoveMember = auto()
ListsPinMany = auto()
ListPinOne = auto()
ListUnpinOne = auto()
UpdateList = auto()
# DM
useSendMessageMutation = auto()
# other
TweetStats = auto()
def log(fn=None, *, level: int = logging.DEBUG, info: list = None) -> callable:
@@ -322,30 +342,6 @@ def unbookmark(_id: int, session: Session) -> Response:
# return graphql_request(0, Operation.BookmarksAllDelete.name, 0, session)
@log(info=['text'])
def update_search_settings(session: Session, **kwargs) -> Response:
twid = int(session.cookies.get_dict()['twid'].split('=')[-1].strip('"'))
headers = get_headers(session=session)
r = session.post(
url=f'https://api.twitter.com/1.1/strato/column/User/{twid}/search/searchSafety',
headers=headers,
json=kwargs,
)
return r
@log(info=['json'])
def update_content_settings(session: Session, **kwargs) -> Response:
"""
Update content settings
@param session: authenticated session
@param kwargs: settings to enable/disable
@return: updated settings
"""
return api_request(kwargs, 'account/settings.json', session)
@log(info=['json'])
def stats(rest_id: int, session: Session) -> Response:
"""private endpoint?"""
@@ -433,3 +429,222 @@ def pin(tweet_id: int, session: Session) -> Response:
def unpin(tweet_id: int, session: Session) -> Response:
settings = {'tweet_mode': 'extended', 'id': tweet_id}
return api_request(settings, 'account/unpin_tweet.json', session)
@log(info=['text'])
def update_search_settings(session: Session, **kwargs) -> Response:
twid = int(session.cookies.get_dict()['twid'].split('=')[-1].strip('"'))
headers = get_headers(session=session)
r = session.post(
url=f'https://api.twitter.com/1.1/strato/column/User/{twid}/search/searchSafety',
headers=headers,
json=kwargs,
)
return r
@log(info=['json'])
def update_content_settings(session: Session, **kwargs) -> Response:
"""
Update content settings
@param session: authenticated session
@param kwargs: settings to enable/disable
@return: updated settings
"""
return api_request(kwargs, 'account/settings.json', session)
@log(info=['json'])
def remove_interests(session: Session, *args):
url = 'https://api.twitter.com/1.1/account/personalization/twitter_interests.json'
r = session.get(url, headers=get_headers(session))
current_interests = r.json()['interested_in']
if args == 'all':
disabled_interests = [x['id'] for x in current_interests]
else:
disabled_interests = [x['id'] for x in current_interests if x['display_name'] in args]
payload = {
"preferences": {
"interest_preferences": {
"disabled_interests": disabled_interests,
"disabled_partner_interests": []
}
}
}
url = 'https://api.twitter.com/1.1/account/personalization/p13n_preferences.json'
r = session.post(url, headers=get_headers(session), json=payload)
return r
@log(info=['json'])
def __get_lists(session: Session) -> Response:
operation = Operation.ListsManagementPageTimeline.name
params = deepcopy(operations[operation])
qid = params['queryId']
query = build_query(params)
url = f"https://api.twitter.com/graphql/{qid}/{operation}?{query}"
r = session.get(url, headers=get_headers(session))
return r
@log(info=['json'])
def create_list(session: Session, name: str, description: str, private: bool) -> Response:
operation = Operation.CreateList.name
payload = deepcopy(operations[operation])
qid = payload['queryId']
payload['variables'] |= {
"isPrivate": private,
"name": name,
"description": description,
}
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=get_headers(session), json=payload)
return r
@log(info=['json'])
def update_list(session: Session, list_id: int, name: str, description: str, private: bool) -> Response:
operation = Operation.UpdateList.name
payload = deepcopy(operations[operation])
qid = payload['queryId']
payload['variables'] |= {
"listId": list_id,
"isPrivate": private,
"name": name,
"description": description,
}
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=get_headers(session), json=payload)
return r
@log(info=['json'])
def update_pinned_lists(session: Session, list_ids: list[int]) -> Response:
"""
Update pinned lists
Reset all pinned lists and pin all specified lists in the order they are provided.
@param session: authenticated session
@param list_ids: list of list ids to pin
@return: response
"""
operation = Operation.ListsPinMany.name
payload = deepcopy(operations[operation])
qid = payload['queryId']
payload['variables']['listIds'] = list_ids
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=get_headers(session), json=payload)
return r
@log(info=['json'])
def pin_list(session: Session, list_id: int) -> Response:
operation = Operation.ListPinOne.name
payload = deepcopy(operations[operation])
qid = payload['queryId']
payload['variables']['listId'] = list_id
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=get_headers(session), json=payload)
return r
@log(info=['json'])
def unpin_list(session: Session, list_id: int) -> Response:
operation = Operation.ListUnpinOne.name
payload = deepcopy(operations[operation])
qid = payload['queryId']
payload['variables']['listId'] = list_id
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=get_headers(session), json=payload)
return r
@log(info=['json'])
def add_list_member(session: Session, list_id: int, user_id: int) -> Response:
operation = Operation.ListAddMember.name
payload = deepcopy(operations[operation])
qid = payload['queryId']
payload['variables'] |= {
"listId": list_id,
"userId": user_id,
}
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=get_headers(session), json=payload)
return r
@log(info=['json'])
def remove_list_member(session: Session, list_id: int, user_id: int) -> Response:
operation = Operation.ListRemoveMember.name
payload = deepcopy(operations[operation])
qid = payload['queryId']
payload['variables'] |= {
"listId": list_id,
"userId": user_id,
}
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=get_headers(session), json=payload)
return r
@log(info=['json'])
def delete_list(session: Session, list_id: int) -> Response:
operation = Operation.DeleteList.name
payload = deepcopy(operations[operation])
qid = payload['queryId']
payload['variables']['listId'] = list_id
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=get_headers(session), json=payload)
return r
@log(info=['json'])
def update_list_banner(session: Session, list_id: int, filename: str) -> Response:
media_id = upload_media(filename, session)
operation = Operation.EditListBanner.name
payload = deepcopy(operations[operation])
qid = payload['queryId']
payload['variables'] |= {
'listId': list_id,
'mediaId': media_id,
}
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=get_headers(session), json=payload)
return r
@log(info=['json'])
def delete_list_banner(session: Session, list_id: int) -> Response:
operation = Operation.DeleteListBanner.name
payload = deepcopy(operations[operation])
qid = payload['queryId']
payload['variables']['listId'] = list_id
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=get_headers(session), json=payload)
return r
@log(info=['json'])
def follow_topic(session: Session, topic_id: int) -> Response:
operation = Operation.TopicFollow.name
payload = deepcopy(operations[operation])
qid = payload['queryId']
# Topic operations are the only ones which require the id to be a string, weird
payload['variables']['topicId'] = str(topic_id)
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=get_headers(session), json=payload)
return r
@log(info=['json'])
def unfollow_topic(session: Session, topic_id: int) -> Response:
operation = Operation.TopicUnfollow.name
payload = deepcopy(operations[operation])
qid = payload['queryId']
# Topic operations are the only ones which require the id to be a string, weird
payload['variables']['topicId'] = str(topic_id)
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=get_headers(session), json=payload)
return r

2
twitter/scrape.py
View File

@@ -204,7 +204,7 @@ async def paginate(session: ClientSession, operation: any, key: str, data: dict,
query = build_query(params)
url = f"https://api.twitter.com/graphql/{qid}/{operation}?{query}"
# update csrf header - must be an easier way
# update csrf header - must be an easier way without importing yarl
if k := session.cookie_jar.__dict__['_cookies'].get('twitter.com'):
if cookie := re.search('(?<=ct0\=)\w+(?=;)', str(k)):
session.headers.update({"x-csrf-token": cookie.group()})