twitch-dl/twitchdl/twitch.py
2020-04-25 20:04:21 +02:00

130 lines
3.1 KiB
Python

"""
Twitch API access.
"""
import requests
from twitchdl import CLIENT_ID
from twitchdl.exceptions import ConsoleError
def authenticated_get(url, params={}, headers={}):
headers['Client-ID'] = CLIENT_ID
response = requests.get(url, params, headers=headers)
if response.status_code == 400:
data = response.json()
raise ConsoleError(data["message"])
response.raise_for_status()
return response
def authenticated_post(url, data=None, json=None, headers={}):
headers['Client-ID'] = CLIENT_ID
response = requests.post(url, data=data, json=json, headers=headers)
if response.status_code == 400:
data = response.json()
raise ConsoleError(data["message"])
response.raise_for_status()
return response
def kraken_get(url, params={}, headers={}):
"""
Add accept header required by kraken API v5.
see: https://discuss.dev.twitch.tv/t/change-in-access-to-deprecated-kraken-twitch-apis/22241
"""
headers["Accept"] = "application/vnd.twitchtv.v5+json"
return authenticated_get(url, params, headers)
def get_user(login):
"""
https://dev.twitch.tv/docs/api/reference/#get-users
"""
response = authenticated_get("https://api.twitch.tv/helix/users", {
"login": login
})
users = response.json()["data"]
return users[0] if users else None
def get_video(video_id):
"""
https://dev.twitch.tv/docs/v5/reference/videos#get-video
"""
url = "https://api.twitch.tv/kraken/videos/{}".format(video_id)
return kraken_get(url).json()
def get_clip(slug):
url = "https://gql.twitch.tv/gql"
query = """
{{
clip(slug: "{}") {{
title
durationSeconds
game {{
name
}}
broadcaster {{
login
displayName
}}
videoQualities {{
frameRate
quality
sourceURL
}}
}}
}}
"""
payload = {"query": query.format(slug)}
data = authenticated_post(url, json=payload).json()
return data["data"]["clip"]
def get_channel_videos(channel_id, limit, offset, sort, type="archive"):
"""
https://dev.twitch.tv/docs/v5/reference/channels#get-channel-videos
"""
url = "https://api.twitch.tv/kraken/channels/{}/videos".format(channel_id)
return kraken_get(url, {
"broadcast_type": type,
"limit": limit,
"offset": offset,
"sort": sort,
}).json()
def get_access_token(video_id):
url = "https://api.twitch.tv/api/vods/{}/access_token".format(video_id)
return authenticated_get(url).json()
def get_playlists(video_id, access_token):
"""
For a given video return a playlist which contains possible video qualities.
"""
url = "http://usher.twitch.tv/vod/{}".format(video_id)
response = requests.get(url, params={
"nauth": access_token['token'],
"nauthsig": access_token['sig'],
"allow_source": "true",
"player": "twitchweb",
})
response.raise_for_status()
return response.content.decode('utf-8')