123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109 |
- import httpx
- import api_common
- import api_1_1
- import api_2
- import api_graph
- import config
- import redis_decorator
- import authman
- @redis_decorator.cache(ttl_secs=60*15)
- def get_guest_token():
- return request(api_1_1.guest_token()).json()["guest_token"]
- def gen_guest_token_headers():
- return {"x-guest-token": get_guest_token()}
- auth_man = authman.AuthManager([api_common.gen_authenticated_headers_base(auth) for auth in config.auth], gen_guest_token_headers)
- def add_base_request_params(params):
- params['headers'] = params.get('headers', {}) | api_common.gen_headers_base()
- params['method'] = params.get('method', 'get')
- def request(params):
- add_base_request_params(params)
- resp = httpx.request(**params)
- resp.raise_for_status()
- return resp
- def request_as_guest(params):
- params['headers'] = params.get('headers', {}) | gen_guest_token_headers()
- resp = request(params)
- if resp.headers['x-rate-limit-remaining'] == '1':
- get_guest_token.invalidate_cache()
- return resp
- def request_as_user(params):
- add_base_request_params(params)
- while True:
- params['headers'] = params.get('headers', {}) | auth_man.get()
- resp = httpx.request(**params)
- if resp.status_code == 429:
- auth_man.block_last()
- continue
- resp.raise_for_status()
- return resp
- @redis_decorator.cache(ttl_secs=60*60)
- def get_user(username):
- return request_as_guest(api_graph.user_by_screen_name(username)).json()
- def get_user_id(username):
- user = get_user(username)
- return user['data']['user']['result']['rest_id']
- @redis_decorator.cache(ttl_secs=60*60)
- def get_likes(username, cursor=None):
- return request_as_user(api_graph.likes(get_user_id(username), cursor)).json()
- def get_appropriate_request_func(username):
- user = get_user(username)
- if user['data']['user']['result']['legacy']['possibly_sensitive']:
- return request_as_user
- return request_as_guest
- @redis_decorator.cache(ttl_secs=60*60)
- def get_user_tweets(username, cursor=None):
- request_func = get_appropriate_request_func(username)
- return request_func(api_graph.user_tweets(get_user_id(username), cursor)).json()
- @redis_decorator.cache(ttl_secs=60*60)
- def get_tweet(tweet_id, username, cursor=None):
- request_func = get_appropriate_request_func(username)
- return request_func(api_graph.tweet_detail(tweet_id, cursor)).json()
- @redis_decorator.cache(ttl_secs=60*60)
- def get_favoriters(tweet_id, cursor=None):
- return request_as_user(api_graph.favoriters(tweet_id, cursor)).json()
- @redis_decorator.cache(ttl_secs=60*60)
- def get_retweeters(tweet_id, cursor=None):
- return request_as_user(api_graph.retweeters(tweet_id, cursor)).json()
- @redis_decorator.cache(ttl_secs=60*60)
- def get_followers(user_id, cursor=None):
- return request_as_user(api_graph.followers(user_id, cursor)).json()
- @redis_decorator.cache(ttl_secs=60*60)
- def get_following(user_id, cursor=None):
- return request_as_user(api_graph.following(user_id, cursor)).json()
- @redis_decorator.cache(ttl_secs=60*60)
- def search(query, cursor=None):
- try:
- return {
- 'search': request_as_user(api_graph.search(query, cursor)).json(),
- 'version': 'graphql'
- }
- except httpx.HTTPStatusError:
- #TODO implement cursor
- return {
- 'search': request_as_user(api_2.search(query)).json(),
- 'version': '2'
- }
|