Skip to content

Added followage information and access function#48

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sharkbound merged 1 commit into from
Jun 16, 2022
Merged

Added followage information and access function #48

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions twitchbot/data/follower.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -8,3 +8,4 @@ class Follower(NamedTuple):
following: str
id: int
name: str
followed_at: str
3 changes: 2 additions & 1 deletion twitchbot/data/user_followers.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -13,7 +13,8 @@ def __init__(self, follower_count: int, following: str, following_id: int, name:
Follower(following=following,
following_id=int(following_id),
id=int(json['from_id']),
name=json['from_name'])
name=json['from_name'],
followed_at=json['followed_at'])
for json in followers
if 'from_id' in json and 'from_name' in json
]
Expand Down
27 changes: 25 additions & 2 deletions twitchbot/util/twitch_api_util.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -9,19 +9,21 @@
from async_timeout import timeout

from ..config import get_client_id, get_oauth, DEFAULT_CLIENT_ID
from ..data import UserFollowers, UserInfo, RateLimit
from ..data import UserFollowers, UserInfo, RateLimit, Follower

__all__ = ('CHANNEL_CHATTERS_URL', 'get_channel_chatters', 'get_stream_data', 'get_url', 'get_user_data', 'get_user_id',
'STREAM_API_URL', 'USER_API_URL', 'get_user_followers', 'USER_FOLLOWERS_API_URL', 'get_headers',
'get_user_info', 'USER_ACCOUNT_AGE_API', 'CHANNEL_INFO_API', 'get_channel_info', 'ChannelInfo',
'get_channel_name_from_user_id', 'OauthTokenInfo', 'get_oauth_token_info', '_check_token', 'post_url')
'get_channel_name_from_user_id', 'OauthTokenInfo', 'get_oauth_token_info', '_check_token', 'post_url', 'USER_FOLLOWAGE_API_URL',
'get_user_followage')

USER_API_URL = 'https://api.twitch.tv/helix/users?login={}'
STREAM_API_URL = 'https://api.twitch.tv/helix/streams?user_login={}'
CHANNEL_CHATTERS_URL = 'https://tmi.twitch.tv/group/user/{}/chatters'
USER_FOLLOWERS_API_URL = 'https://api.twitch.tv/helix/users/follows?to_id={}'
USER_ACCOUNT_AGE_API = 'https://api.twitch.tv/kraken/users/{}'
CHANNEL_INFO_API = 'https://api.twitch.tv/helix/channels?broadcaster_id={}'
USER_FOLLOWAGE_API_URL = 'https://api.twitch.tv/helix/users/follows?to_id={}&from_id={}'

user_id_cache: Dict[str, int] = {}

Expand DownExpand Up@@ -99,6 +101,27 @@ async def get_user_followers(user: str, headers: dict = None) -> UserFollowers:
id=user_id_cache[user],
followers=json['data'])

async def get_user_followage(user: str, follower: str, headers: dict = None) -> Follower:
headers = headers if headers is not None else get_headers()
if not _check_headers_has_auth(headers):
warnings.warn('[GET_USER_FOLLOWAGE] headers for the twitch api request are missing authorization')
return Follower(-1, '', -1, '', '')

user_id = await get_user_id(user, headers)
follower_id = await get_user_id(follower, headers)
_, json = await get_url(USER_FOLLOWAGE_API_URL.format(user_id, follower_id), headers)
# ratelimit = RateLimit.from_headers(_.headers)

# covers invalid user id, or some other API error, such as invalid client-id
if not json or json.get('status', -1) == 400:
return Follower(-1, '', -1, '', '')

return Follower(following=user,
following_id=user_id,
id=follower_id,
name=json['data'][0]['from_name'],
followed_at=json['data'][0]['followed_at'])


async def get_user_data(user: str, headers: dict = None) -> dict:
headers = headers if headers is not None else get_headers()
Expand Down