From c0e66fc416ce9e31d68410172c287cfff9721bf2 Mon Sep 17 00:00:00 2001 From: Ivan Habunek Date: Wed, 3 Apr 2024 08:14:34 +0200 Subject: [PATCH] Add AccessToken typed dict --- twitchdl/commands/download.py | 10 +++++----- twitchdl/twitch.py | 22 ++++++++++++++-------- 2 files changed, 19 insertions(+), 13 deletions(-) diff --git a/twitchdl/commands/download.py b/twitchdl/commands/download.py index 15fab80..94dd6d7 100644 --- a/twitchdl/commands/download.py +++ b/twitchdl/commands/download.py @@ -20,7 +20,7 @@ from twitchdl.entities import DownloadOptions from twitchdl.exceptions import ConsoleError from twitchdl.http import download_all from twitchdl.output import blue, bold, dim, green, print_log, yellow -from twitchdl.twitch import Video +from twitchdl.twitch import Clip, Video def download(ids: list[str], args: DownloadOptions): @@ -203,7 +203,7 @@ def _crete_temp_dir(base_uri: str) -> str: return str(temp_dir) -def _get_clip_url(clip, quality): +def _get_clip_url(clip: Clip, quality: str) -> str: qualities = clip["videoQualities"] # Quality given as an argument @@ -231,7 +231,7 @@ def _get_clip_url(clip, quality): return selected_quality["sourceURL"] -def get_clip_authenticated_url(slug, quality): +def get_clip_authenticated_url(slug: str, quality: str): print_log("Fetching access token...") access_token = twitch.get_clip_access_token(slug) @@ -241,8 +241,8 @@ def get_clip_authenticated_url(slug, quality): url = _get_clip_url(access_token, quality) query = urlencode({ - "sig": access_token["playbackAccessToken"]["signature"], - "token": access_token["playbackAccessToken"]["value"], + "sig": access_token["signature"], + "token": access_token["value"], }) return f"{url}?{query}" diff --git a/twitchdl/twitch.py b/twitchdl/twitch.py index 5ef33c7..21e5206 100644 --- a/twitchdl/twitch.py +++ b/twitchdl/twitch.py @@ -16,6 +16,12 @@ ClipsPeriod = Literal["last_day", "last_week", "last_month", "all_time"] VideosSort = Literal["views", "time"] VideosType = Literal["archive", "highlight", "upload"] + +class AccessToken(TypedDict): + signature: str + value: str + + class User(TypedDict): login: str displayName: str @@ -166,7 +172,7 @@ def get_clip(slug: str) -> Clip | None: return response["data"]["clip"] -def get_clip_access_token(slug: str): +def get_clip_access_token(slug: str) -> AccessToken: query = f""" {{ "operationName": "VideoAccessToken_Clip", @@ -183,7 +189,7 @@ def get_clip_access_token(slug: str): """ response = gql_post(query.strip()) - return response["data"]["clip"] + return response["data"]["clip"]["playbackAccessToken"] def get_channel_clips(channel_id: str, period: ClipsPeriod, limit: int, after: str | None= None): @@ -341,7 +347,7 @@ def channel_videos_generator( return videos["totalCount"], _generator(videos, max_videos) -def get_access_token(video_id: str, auth_token: str | None = None): +def get_access_token(video_id: str, auth_token: str | None = None) -> AccessToken: query = f""" {{ videoPlaybackAccessToken( @@ -360,7 +366,7 @@ def get_access_token(video_id: str, auth_token: str | None = None): headers: dict[str, str] = {} if auth_token is not None: - headers['authorization'] = f'OAuth {auth_token}' + headers["authorization"] = f"OAuth {auth_token}" try: response = gql_query(query, headers=headers) @@ -380,15 +386,15 @@ def get_access_token(video_id: str, auth_token: str | None = None): raise -def get_playlists(video_id, access_token): +def get_playlists(video_id: str, access_token: AccessToken): """ For a given video return a playlist which contains possible video qualities. """ url = f"https://usher.ttvnw.net/vod/{video_id}" response = httpx.get(url, params={ - "nauth": access_token['value'], - "nauthsig": access_token['signature'], + "nauth": access_token["value"], + "nauthsig": access_token["signature"], "allow_audio_only": "true", "allow_source": "true", "player": "twitchweb", @@ -397,7 +403,7 @@ def get_playlists(video_id, access_token): return response.content.decode('utf-8') -def get_game_id(name): +def get_game_id(name: str): query = f""" {{ game(name: "{name.strip()}") {{