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

update

Browse Source
This commit is contained in:
trevorhobenshield
2023-03-11 10:48:57 -08:00
parent 1aea57a3f0
commit 53b96effe6
9 changed files with 4032 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
readme.md Normal file
View File

@@ -0,0 +1,53 @@
**(Work in Progress)** Complete implementation of the undocumented Twitter API
- Written very quickly, crude, needs refactoring/redesign
```bash
pip install twitter-api-client
```
### Usage
```python
from src.main import *
usr, pwd = ..., ...
session = login(usr, pwd)
# create tweet with images, videos, gifs, and tagged users
r = create_tweet('test 123', session, media=[{'file': 'image.jpeg', 'tagged_users': [123234345456], 'alt': 'some image'}])
r = create_tweet('test 123', session, media=['test.jpg', 'test.png'])
r = create_tweet('test 123', session, media=['test.mp4'])
r = create_tweet('test 123', session)
r = delete_tweet(123, session)
# delete all tweets in account
r = delete_all_tweets(456, session)
r = retweet(1633609779745820675, session)
r = unretweet(1633609779745820675, session)
r = quote('test 123', 'elonmusk', 1633609779745820675, session)
r = comment('test 123', 1633609779745820675, session)
r = unlike_tweet(1633609779745820675, session)
r = like_tweet(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)
# some hidden user attribute?
r = stats(50393960, session)
```

0
src/config/__init__.py Normal file
View File

3159
src/config/operations.py Normal file
View File

File diff suppressed because it is too large Load Diff

75
src/config/update_operations.py Normal file
View File

@@ -0,0 +1,75 @@
import re
import json
import requests
import bs4
from pathlib import Path
def find_api_script(res: requests.Response) -> str:
"""
Find api script
@param res: response from homepage: https://twitter.com
@return: url to api script
"""
for s in bs4.BeautifulSoup(res.text, 'html.parser').select('script'):
if x := re.search('(?<=api:")\w+(?=")', s.text):
key = x.group() + 'a' # wtf?
return f'https://abs.twimg.com/responsive-web/client-web/api.{key}.js'
def get_operations() -> list[dict]:
"""
Get operations and their respective queryId and feature definitions
@return: list of operations
"""
session = requests.Session()
headers = {
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36'
}
script = find_api_script(session.get('https://twitter.com', headers=headers))
r = session.get(script, headers=headers)
res = ''
# find operations and their respective queryId and feature definitions
for x in re.split(r'\d+:e=>{e.exports=', r.text)[1:]:
res = res.replace('}}},{', '}},{')
if x.startswith('{queryId:') and '"use strict"' not in x:
for k in ['queryId', 'operationName', 'operationType', 'metadata', 'featureSwitches']:
x = x.replace(k, f'"{k}"')
res += x
return json.loads(f'[{res[:-2]}]')
def update_operations(path=Path('operations.json')):
"""
Update operations.json with queryId and feature definitions
@param path: path to operations operations file
@return: updated operations
"""
operations = get_operations()
# update operations file
if path.stat().st_size == 0:
# empty file, add all operations
config = {}
for o in operations:
config[o['operationName']] = {
'queryId': o['queryId'],
'variables': {},
'features': {k: True for k in o['metadata']['featureSwitches']}
}
else:
config = json.loads(path.read_text())
# update queryId and features for all operations
for o in operations:
config[o['operationName']]['queryId'] = o['queryId']
config[o['operationName']]['features'] = {k: True for k in o['metadata']['featureSwitches']}
path.write_text(json.dumps(config, indent=2))
return config
def main() -> int:
update_operations()
return 0
if __name__ == '__main__':
exit(main())

0
src/log/__init__.py Normal file
View File

40
src/log/config.py Normal file
View File

@@ -0,0 +1,40 @@
log_config = {
"version": 1,
"formatters": {
"simple": {
"format": "%(asctime)s.%(msecs)03d %(levelname)s: %(message)s"
}
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"level": "DEBUG",
"formatter": "simple",
"stream": "ext://sys.stdout"
},
"file": {
"class": "logging.FileHandler",
"level": "DEBUG",
"formatter": "simple",
"filename": "log/debug.log",
"encoding": "utf8",
"mode": "a"
}
},
"loggers": {
"myLogger": {
"level": "DEBUG",
"handlers": [
"console"
],
"propagate": "no"
}
},
"root": {
"level": "DEBUG",
"handlers": [
"console",
"file"
]
}
}

89
src/login.py Normal file
View File

@@ -0,0 +1,89 @@
import sys
from requests import Session
def update_token(session: Session, key: str, url: str, payload: dict) -> Session:
headers = {
"authorization": 'Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA',
"content-type": "application/json",
"user-agent": 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36',
"x-guest-token": session.tokens['guest_token'],
"x-csrf-token": session.cookies.get("ct0"),
"x-twitter-auth-type": "OAuth2Session" if session.cookies.get("auth_token") else '',
"x-twitter-active-user": "yes",
"x-twitter-client-language": 'en',
}
r = session.post(url, headers=headers, json=payload).json()
status = f'\u001b[32mSUCCESS' if r.get('guest_token') or r.get('flow_token') else f'\u001b[31mFAILED'
print(f'{status}\u001b[0m {sys._getframe(1).f_code.co_name}') # check response data
session.tokens[key] = r[key]
return session
def init_guest_token(session: Session) -> Session:
return update_token(session, 'guest_token', 'https://api.twitter.com/1.1/guest/activate.json', {})
def flow_start(session: Session) -> Session:
return update_token(session, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json?flow_name=login', {
"input_flow_data": {
"flow_context": {"debug_overrides": {}, "start_location": {"location": "splash_screen"}}
}, "subtask_versions": {}
})
def flow_instrumentation(session: Session) -> Session:
return update_token(session, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', {
"flow_token": session.tokens['flow_token'],
"subtask_inputs": [{
"subtask_id": "LoginJsInstrumentationSubtask",
"js_instrumentation": {"response": "{}", "link": "next_link"}
}],
})
def flow_username(session: Session) -> Session:
return update_token(session, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', {
"flow_token": session.tokens['flow_token'],
"subtask_inputs": [{
"subtask_id": "LoginEnterUserIdentifierSSO",
"settings_list": {
"setting_responses": [{
"key": "user_identifier",
"response_data": {"text_data": {"result": session.username}}
}], "link": "next_link"}}],
})
def flow_password(session: Session) -> Session:
return update_token(session, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', {
"flow_token": session.tokens['flow_token'],
"subtask_inputs": [{
"subtask_id": "LoginEnterPassword",
"enter_password": {"password": session.password, "link": "next_link"}}]
})
def flow_duplication_check(session: Session) -> Session:
return update_token(session, 'flow_token', 'https://api.twitter.com/1.1/onboarding/task.json', {
"flow_token": session.tokens['flow_token'],
"subtask_inputs": [{
"subtask_id": "AccountDuplicationCheck",
"check_logged_in_account": {"link": "AccountDuplicationCheck_false"},
}],
})
def execute_login_flow(session: Session) -> Session:
session = init_guest_token(session)
for fn in [flow_start, flow_instrumentation, flow_username, flow_password, flow_duplication_check]:
session = fn(session)
return session
def login(username: str, password: str) -> Session:
session = Session()
session.username = username
session.password = password
session.tokens = {'guest_token': None, 'flow_token': None}
return execute_login_flow(session)

590
src/main.py Normal file
View File

@@ -0,0 +1,590 @@
import sys
import asyncio
import logging.config
from pathlib import Path
from urllib.parse import urlencode
from enum import Enum, auto
from functools import wraps, partial
import aiohttp
import ujson
from .login import login, Session
from .utils import find_key
from .log.config import log_config
from .config.operations import operations
logging.config.dictConfig(log_config)
logger = logging.getLogger(__name__)
MAX_IMAGE_FILE_SIZE = 5_242_880
CHUNK_SIZE = 8192
BOLD = '\u001b[1m'
SUCCESS = '\u001b[32m'
WARN = '\u001b[31m'
RESET = '\u001b[0m'
MEDIA = {
'.mp4': {
'type': 'video/mp4',
'category': 'tweet_video'
},
'.mov': {
'type': 'video/quicktime',
'category': 'tweet_video'
},
'.png': {
'type': 'image/png',
'category': 'tweet_image'
},
'.jpg': {
'type': 'image/jpeg',
'category': 'tweet_image'
},
'.jpeg': {
'type': 'image/jpeg',
'category': 'tweet_image'
},
}
class Operation(Enum):
CreateTweet = auto()
CreateScheduledTweet = auto()
DeleteTweet = auto()
UserTweets = auto()
FavoriteTweet = auto()
UnfavoriteTweet = auto()
CreateRetweet = auto()
DeleteRetweet = auto()
TweetStats = auto()
def log(fn=None, *, level: int = logging.DEBUG, info: list = None):
if fn is None:
return partial(log, level=level, info=info)
@wraps(fn)
def wrapper(*args, **kwargs):
r = fn(*args, **kwargs)
try:
if 200 <= r.status_code < 300:
message = f'[{SUCCESS}SUCCESS{RESET}] ({BOLD}{fn.__name__}{RESET})'
for k in info:
if callable(k):
logger.log(level, f'{message}: {k(r)}')
else:
attr = getattr(r, k)
v = attr() if callable(attr) else attr
d = {f"{k}": v}
logger.log(level, f'{message}: {d}')
else:
logger.log(level, f'[{WARN}ERROR{RESET}] ({fn.__name__}) {r.status_code} {r.text}')
except Exception as e:
logger.log(level, f'[{WARN}FAILED{RESET}] ({fn.__name__}) {e}')
return r
return wrapper
def _get_headers(session: Session) -> dict:
return {
'authorization': 'Bearer AAAAAAAAAAAAAAAAAAAAANRILgAAAAAAnNwIzUejRCOuH5E6I8xnZz4puTs%3D1Zv7ttfk8LF81IUq16cHjhLTvJu4FA33AGWWjCpTnA',
'accept-encoding': 'gzip, deflate, br',
'cookie': '; '.join(f'{k}={v}' for k, v in session.cookies.items()),
'referer': 'https://twitter.com/',
'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36',
'x-csrf-token': session.cookies.get('ct0'),
}
async def get_status(media_id: str, auth_session: Session, check_after_secs: int = 1):
url = 'https://upload.twitter.com/i/media/upload.json'
headers = _get_headers(auth_session)
params = {'command': 'STATUS', 'media_id': media_id}
while 1:
await asyncio.sleep(check_after_secs)
async with aiohttp.ClientSession(headers=headers) as s:
async with s.get(url, params=params) as r:
data = await r.json()
info = data['processing_info']
state = info['state']
if state == 'succeeded':
logger.debug(f'{media_id}: {SUCCESS}processing complete{RESET}')
return data
if state == 'in_progress':
progress = info["progress_percent"]
check_after_secs = info.get('check_after_secs', check_after_secs)
logger.debug(f'{media_id}: upload {progress = }%')
else:
logger.debug(f'{media_id}: upload {state = }')
async def upload_media(fname: str, auth_session: Session):
url = 'https://upload.twitter.com/i/media/upload.json'
headers = _get_headers(auth_session)
conn = aiohttp.TCPConnector(limit=0, ssl=False, ttl_dns_cache=69)
async with aiohttp.ClientSession(headers=headers, connector=conn) as s:
file = Path(fname)
total_bytes = file.stat().st_size
params = {
'command': 'INIT',
'total_bytes': total_bytes,
'media_type': MEDIA[file.suffix]['type'],
'media_category': MEDIA[file.suffix]['category']
}
async with s.post(url, headers=headers, params=params) as r:
info = await r.json()
logger.debug(f'{info = }')
media_id = info['media_id']
with open(fname, 'rb') as f:
i = 0
while chunk := f.read(MAX_IMAGE_FILE_SIZE): # todo: arbitrary max size for now
with aiohttp.MultipartWriter('form-data') as mpw:
part = mpw.append(chunk)
part.set_content_disposition('form-data', name='media', filename='blob')
s.cookie_jar.update_cookies(auth_session.cookies) # csrf cookie/header update
r = await s.post(
url,
data=mpw,
headers=headers,
params={'command': 'APPEND', 'media_id': media_id, 'segment_index': i}
)
logger.debug(f'{r.status = }')
i += 1
async with s.post(url, headers=headers,
params={'command': 'FINALIZE', 'media_id': media_id, 'allow_async': 'true'}) as r:
res = await r.json()
logger.debug(f'{res = }')
if processing_info := res.get('processing_info', {}):
state = processing_info.get('state')
if state == 'pending':
logger.debug(f'{media_id}: {state}')
return await get_status(media_id, auth_session, processing_info.get('check_after_secs', 1))
logger.debug(f'{media_id}: {SUCCESS}upload complete{RESET}')
return res
@log(level=logging.DEBUG, info=['status_code'])
def add_alt_text(text: str, media_id: int, session: Session):
params = {"media_id": media_id, "alt_text": {"text": text}}
url = 'https://api.twitter.com/1.1/media/metadata/create.json'
r = session.post(url, headers=_get_headers(session), json=params)
return r
@log(level=logging.DEBUG, info=['status_code', 'json'])
def like_tweet(tweet_id: int, session: Session):
operation = Operation.FavoriteTweet.name
qid = operations[operation]['queryId']
params = operations[operation]
params['variables']['tweet_id'] = tweet_id
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=_get_headers(session), json=params)
logger.debug(f'{tweet_id = }')
return r
@log(level=logging.DEBUG, info=['status_code', 'json'])
def unlike_tweet(tweet_id: int, session: Session):
operation = Operation.UnfavoriteTweet.name
qid = operations[operation]['queryId']
params = operations[operation]
params['variables']['tweet_id'] = tweet_id
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=_get_headers(session), json=params)
logger.debug(f'{tweet_id = }')
return r
@log(level=logging.DEBUG, info=['status_code', 'json'])
def create_tweet(text: str, session: Session, media: list[dict | str] = None, **kwargs):
operation = Operation.CreateTweet.name
qid = operations[operation]['queryId']
params = operations[operation]
params['variables']['tweet_text'] = text
if media:
for m in media:
if isinstance(m, dict):
media_info = asyncio.run(upload_media(m['file'], session))
params['variables']['media']['media_entities'].append({
'media_id': media_info['media_id'],
'tagged_users': m.get('tagged_users', [])
})
if alt := m.get('alt'):
add_alt_text(alt, media_info['media_id'], session)
# for convenience, so we can just pass list of strings
elif isinstance(m, str):
media_info = asyncio.run(upload_media(m, session))
params['variables']['media']['media_entities'].append({
'media_id': media_info['media_id'],
'tagged_users': []
})
if reply_params := kwargs.get('reply_params', {}):
params['variables'] |= reply_params
if quote_params := kwargs.get('quote_params', {}):
params['variables'] |= quote_params
if poll_params := kwargs.get('poll_params', {}):
params['variables'] |= poll_params
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=_get_headers(session), json=params)
return r
def comment(text: str, tweet_id: int, session: Session, media: list[dict | str] = None):
params = {"reply": {"in_reply_to_tweet_id": tweet_id, "exclude_reply_user_ids": []}}
return create_tweet(text, session, media, reply_params=params)
def quote(text: str, screen_name: str, tweet_id: int, session: Session, media: list[dict | str] = None):
""" no unquote operation, just DeleteTweet"""
params = {"attachment_url": f"https://twitter.com/{screen_name}/status/{tweet_id}"}
return create_tweet(text, session, media, quote_params=params)
def delete_tweet(tweet_id: int, session: Session):
operation = Operation.DeleteTweet.name
qid = operations[operation]['queryId']
params = operations[operation]
params['variables']['tweet_id'] = tweet_id
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=_get_headers(session), json=params)
if 200 <= r.status_code < 300:
logger.debug(f'{WARN}DELETE{RESET} tweet: {tweet_id}')
return r.json()
def delete_all_tweets(user_id: int, session: Session):
tweets = get_tweets(user_id, session)
ids = set(map(int, find_key(find_key(tweets, 'tweet_results'), 'rest_id'))) - {user_id}
[delete_tweet(_id, session) for _id in ids]
def retweet(tweet_id: int, session: Session):
operation = Operation.CreateRetweet.name
qid = operations[operation]['queryId']
params = operations[operation]
params['variables']['tweet_id'] = tweet_id
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=_get_headers(session), json=params)
if 200 <= r.status_code < 300:
logger.debug(f'{SUCCESS}RETWEET{RESET} tweet: {tweet_id}')
return r.json()
def unretweet(tweet_id: int, session: Session):
operation = Operation.DeleteRetweet.name
qid = operations[operation]['queryId']
params = operations[operation]
params['variables']['source_tweet_id'] = tweet_id
url = f"https://api.twitter.com/graphql/{qid}/{operation}"
r = session.post(url, headers=_get_headers(session), json=params)
if 200 <= r.status_code < 300:
logger.debug(f'{SUCCESS}UNRETWEET{RESET} tweet: {tweet_id}')
return r.json()
def get_tweets(user_id: int, session: Session):
operation = Operation.UserTweets.name
qid = operations[operation]['queryId']
params = operations[operation]
params['variables']['userId'] = user_id
query = build_query(params)
url = f"https://api.twitter.com/graphql/{qid}/{operation}?{query}"
r = session.get(url, headers=_get_headers(session))
return r.json()
@log(level=logging.DEBUG, info=['status_code', lambda r: r.json()['id']])
def follow(user_id: int, session: Session):
settings = {
"user_id": user_id,
"include_profile_interstitial_type": "1",
"include_blocking": "1",
"include_blocked_by": "1",
"include_followed_by": "1",
"include_want_retweets": "1",
"include_mute_edge": "1",
"include_can_dm": "1",
"include_can_media_tag": "1",
"include_ext_has_nft_avatar": "1",
"include_ext_is_blue_verified": "1",
"include_ext_verified_type": "1",
"skip_status": "1",
}
headers = _get_headers(session)
headers['content-type'] = 'application/x-www-form-urlencoded'
url = 'https://api.twitter.com/1.1/friendships/create.json'
r = session.post(url, headers=headers, data=urlencode(settings))
return r
@log(level=logging.DEBUG, info=['status_code', lambda r: r.json()['id']])
def unfollow(user_id: int, session: Session):
_name = sys._getframe().f_code.co_name
settings = {
"user_id": user_id,
"include_profile_interstitial_type": "1",
"include_blocking": "1",
"include_blocked_by": "1",
"include_followed_by": "1",
"include_want_retweets": "1",
"include_mute_edge": "1",
"include_can_dm": "1",
"include_can_media_tag": "1",
"include_ext_has_nft_avatar": "1",
"include_ext_is_blue_verified": "1",
"include_ext_verified_type": "1",
"skip_status": "1",
}
headers = _get_headers(session)
headers['content-type'] = 'application/x-www-form-urlencoded'
url = 'https://api.twitter.com/1.1/friendships/destroy.json'
r = session.post(url, headers=headers, data=urlencode(settings))
return r
def mute(user_id: int, session: Session):
_name = sys._getframe().f_code.co_name
settings = {
'user_id': user_id
}
try:
headers = _get_headers(session)
headers['content-type'] = 'application/x-www-form-urlencoded'
url = 'https://api.twitter.com/1.1/mutes/users/create.json'
r = session.post(url, headers=headers, data=urlencode(settings))
if 200 <= r.status_code < 300:
data = r.json()
logger.debug(f'{SUCCESS}MUTE{RESET}: {data["id"]}')
return data
except Exception as e:
logger.debug(f'[FAILED] {_name}: {e}')
def unmute(user_id: int, session: Session):
_name = sys._getframe().f_code.co_name
settings = {
'user_id': user_id
}
try:
headers = _get_headers(session)
headers['content-type'] = 'application/x-www-form-urlencoded'
url = 'https://api.twitter.com/1.1/mutes/users/destroy.json'
r = session.post(url, headers=headers, data=urlencode(settings))
if 200 <= r.status_code < 300:
data = r.json()
logger.debug(f'{WARN}UNMUTE{RESET}: {data["id"]}')
return data
except Exception as e:
logger.debug(f'[FAILED] {_name}: {e}')
def enable_notifications(user_id: int, session: Session):
_name = sys._getframe().f_code.co_name
settings = {
"id": user_id,
"device": "true",
"cursor": "-1",
"include_profile_interstitial_type": "1",
"include_blocking": "1",
"include_blocked_by": "1",
"include_followed_by": "1",
"include_want_retweets": "1",
"include_mute_edge": "1",
"include_can_dm": "1",
"include_can_media_tag": "1",
"include_ext_has_nft_avatar": "1",
"include_ext_is_blue_verified": "1",
"include_ext_verified_type": "1",
"skip_status": "1",
}
try:
headers = _get_headers(session)
headers['content-type'] = 'application/x-www-form-urlencoded'
url = 'https://api.twitter.com/1.1/friendships/update.json'
r = session.post(url, headers=headers, data=urlencode(settings))
if 200 <= r.status_code < 300:
data = r.json()
twid = data["relationship"]["target"]["id"]
logger.debug(f'{SUCCESS}ENABLE NOTIFICATIONS{RESET}: {twid}')
return data
except Exception as e:
logger.debug(f'[FAILED] {_name}: {e}')
def disable_notifications(user_id: int, session: Session):
_name = sys._getframe().f_code.co_name
settings = {
"id": user_id,
"device": "false",
"cursor": "-1",
"include_profile_interstitial_type": "1",
"include_blocking": "1",
"include_blocked_by": "1",
"include_followed_by": "1",
"include_want_retweets": "1",
"include_mute_edge": "1",
"include_can_dm": "1",
"include_can_media_tag": "1",
"include_ext_has_nft_avatar": "1",
"include_ext_is_blue_verified": "1",
"include_ext_verified_type": "1",
"skip_status": "1",
}
try:
headers = _get_headers(session)
headers['content-type'] = 'application/x-www-form-urlencoded'
url = 'https://api.twitter.com/1.1/friendships/update.json'
r = session.post(url, headers=headers, data=urlencode(settings))
if 200 <= r.status_code < 300:
data = r.json()
twid = data["relationship"]["target"]["id"]
logger.debug(f'{WARN}DISABLE NOTIFICATIONS{RESET}: {twid}')
return data
except Exception as e:
logger.debug(f'[FAILED] {_name}: {e}')
def block(user_id: int, session: Session):
_name = sys._getframe().f_code.co_name
settings = {
'user_id': user_id
}
try:
headers = _get_headers(session)
headers['content-type'] = 'application/x-www-form-urlencoded'
url = 'https://api.twitter.com/1.1/blocks/create.json'
r = session.post(url, headers=headers, data=urlencode(settings))
if 200 <= r.status_code < 300:
data = r.json()
logger.debug(f'{SUCCESS}BLOCK{RESET}: {data["id"]}')
return data
except Exception as e:
logger.debug(f'[FAILED] {_name}: {e}')
def unblock(user_id: int, session: Session):
_name = sys._getframe().f_code.co_name
settings = {
'user_id': user_id
}
try:
headers = _get_headers(session)
headers['content-type'] = 'application/x-www-form-urlencoded'
url = 'https://api.twitter.com/1.1/blocks/destroy.json'
r = session.post(url, headers=headers, data=urlencode(settings))
if 200 <= r.status_code < 300:
data = r.json()
logger.debug(f'{WARN}UNBLOCK{RESET}: {data["id"]}')
return data
except Exception as e:
logger.debug(f'[FAILED] {_name}: {e}')
def update_search_settings(session: Session, **kwargs):
_name = sys._getframe().f_code.co_name
try:
if kwargs.get('incognito'):
kwargs.pop('incognito')
settings = {
"optInFiltering": False,
"optInBlocking": True,
}
else:
settings = {}
settings |= kwargs
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=settings,
)
if r.status_code == 200:
logger.debug(f'[SUCCESS] {_name}: {settings}')
return settings
except Exception as e:
logger.debug(f'[FAILED] {_name}: {e}')
def update_content_settings(session: Session, **kwargs):
"""
Update content settings
@param session: authenticated session
@param kwargs: settings to enable/disable
@return: updated settings
"""
_name = sys._getframe().f_code.co_name
try:
if kwargs.get('incognito'):
kwargs.pop('incognito')
settings = {
'include_mention_filter': True,
'include_nsfw_user_flag': True,
'include_nsfw_admin_flag': True,
'include_ranked_timeline': True,
'include_alt_text_compose': True,
'display_sensitive_media': True,
'protected': True,
'discoverable_by_email': False,
'discoverable_by_mobile_phone': False,
'allow_dms_from': 'following', ## {'all'}
'dm_quality_filter': 'enabled', ## {'disabled'}
'dm_receipt_setting': 'all_disabled', ## {'all_enabled'}
'allow_media_tagging': 'none', ## {'all', 'following'}
'nsfw_user': False,
'geo_enabled': False, ## add location information to your tweets
'allow_ads_personalization': False,
'allow_logged_out_device_personalization': False,
'allow_sharing_data_for_third_party_personalization': False,
'allow_location_history_personalization': False,
}
else:
settings = {}
settings |= kwargs
headers = _get_headers(session=session)
headers['content-type'] = 'application/x-www-form-urlencoded'
r = session.post(
url='https://api.twitter.com/1.1/account/settings.json',
headers=headers,
data=urlencode(settings), # case-insensitive, E.g. can be 'TRUE', True, 'true', etc.
)
if r.status_code == 200:
logger.debug(f'[SUCCESS] {_name}: {settings}')
return settings
except Exception as e:
logger.debug(f'[FAILED] {_name}: {e}')
def build_query(params):
return '&'.join(f'{k}={ujson.dumps(v)}' for k, v in params.items())
def stats(rest_id: int, session: Session):
"""private endpoint?"""
operation = Operation.TweetStats.name
qid = operations[operation]['queryId']
params = operations[operation]
params['variables']['rest_id'] = rest_id
query = build_query(params)
url = f"https://api.twitter.com/graphql/{qid}/{operation}?{query}"
r = session.get(url, headers=_get_headers(session))
return r.json()
def main() -> int:
...
return 0
if __name__ == '__main__':
exit(main())

26
src/utils.py Normal file
View File

@@ -0,0 +1,26 @@
def find_key(obj: dict | list[dict], key: str) -> list:
"""
Find all values of a given key within a nested dict or list of dicts
@param obj: dictionary or list of dictionaries
@param key: key to search for
@return: list of values
"""
def helper(obj: any, key: str, L: list) -> list:
if not obj:
return L
if isinstance(obj, list):
for e in obj:
L.extend(helper(e, key, []))
return L
if isinstance(obj, dict) and obj.get(key):
L.append(obj[key])
if isinstance(obj, dict) and obj:
for k in obj:
L.extend(helper(obj[k], key, []))
return L
return helper(obj, key, [])