Add AccessToken typed dict

This commit is contained in:
Ivan Habunek 2024-04-03 08:14:34 +02:00
parent a9facd46ac
commit c0e66fc416
No known key found for this signature in database
GPG Key ID: F5F0623FF5EBCB3D
2 changed files with 19 additions and 13 deletions

View File

@ -20,7 +20,7 @@ from twitchdl.entities import DownloadOptions
from twitchdl.exceptions import ConsoleError from twitchdl.exceptions import ConsoleError
from twitchdl.http import download_all from twitchdl.http import download_all
from twitchdl.output import blue, bold, dim, green, print_log, yellow 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): def download(ids: list[str], args: DownloadOptions):
@ -203,7 +203,7 @@ def _crete_temp_dir(base_uri: str) -> str:
return str(temp_dir) return str(temp_dir)
def _get_clip_url(clip, quality): def _get_clip_url(clip: Clip, quality: str) -> str:
qualities = clip["videoQualities"] qualities = clip["videoQualities"]
# Quality given as an argument # Quality given as an argument
@ -231,7 +231,7 @@ def _get_clip_url(clip, quality):
return selected_quality["sourceURL"] 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...") print_log("Fetching access token...")
access_token = twitch.get_clip_access_token(slug) 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) url = _get_clip_url(access_token, quality)
query = urlencode({ query = urlencode({
"sig": access_token["playbackAccessToken"]["signature"], "sig": access_token["signature"],
"token": access_token["playbackAccessToken"]["value"], "token": access_token["value"],
}) })
return f"{url}?{query}" return f"{url}?{query}"

View File

@ -16,6 +16,12 @@ ClipsPeriod = Literal["last_day", "last_week", "last_month", "all_time"]
VideosSort = Literal["views", "time"] VideosSort = Literal["views", "time"]
VideosType = Literal["archive", "highlight", "upload"] VideosType = Literal["archive", "highlight", "upload"]
class AccessToken(TypedDict):
signature: str
value: str
class User(TypedDict): class User(TypedDict):
login: str login: str
displayName: str displayName: str
@ -166,7 +172,7 @@ def get_clip(slug: str) -> Clip | None:
return response["data"]["clip"] return response["data"]["clip"]
def get_clip_access_token(slug: str): def get_clip_access_token(slug: str) -> AccessToken:
query = f""" query = f"""
{{ {{
"operationName": "VideoAccessToken_Clip", "operationName": "VideoAccessToken_Clip",
@ -183,7 +189,7 @@ def get_clip_access_token(slug: str):
""" """
response = gql_post(query.strip()) 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): 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) 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""" query = f"""
{{ {{
videoPlaybackAccessToken( videoPlaybackAccessToken(
@ -360,7 +366,7 @@ def get_access_token(video_id: str, auth_token: str | None = None):
headers: dict[str, str] = {} headers: dict[str, str] = {}
if auth_token is not None: if auth_token is not None:
headers['authorization'] = f'OAuth {auth_token}' headers["authorization"] = f"OAuth {auth_token}"
try: try:
response = gql_query(query, headers=headers) response = gql_query(query, headers=headers)
@ -380,15 +386,15 @@ def get_access_token(video_id: str, auth_token: str | None = None):
raise 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. For a given video return a playlist which contains possible video qualities.
""" """
url = f"https://usher.ttvnw.net/vod/{video_id}" url = f"https://usher.ttvnw.net/vod/{video_id}"
response = httpx.get(url, params={ response = httpx.get(url, params={
"nauth": access_token['value'], "nauth": access_token["value"],
"nauthsig": access_token['signature'], "nauthsig": access_token["signature"],
"allow_audio_only": "true", "allow_audio_only": "true",
"allow_source": "true", "allow_source": "true",
"player": "twitchweb", "player": "twitchweb",
@ -397,7 +403,7 @@ def get_playlists(video_id, access_token):
return response.content.decode('utf-8') return response.content.decode('utf-8')
def get_game_id(name): def get_game_id(name: str):
query = f""" query = f"""
{{ {{
game(name: "{name.strip()}") {{ game(name: "{name.strip()}") {{