Repository navigation
Expand file tree
/
Copy pathadultswim.py
More file actions
215 lines (180 loc) · 6.61 KB
/
Copy pathadultswim.py
File metadata and controls
215 lines (180 loc) · 6.61 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
"""
$description American live TV channel and video on-demand service owned by Warner Bros.
$url adultswim.com
$type live, vod
$region various
$notes VODs may be protected by DRM
"""
import re
from urllib.parse import urlparse, urlunparse
from streamlink.logger import getLogger
from streamlink.plugin import Plugin, PluginError, pluginmatcher
from streamlink.plugin.api import validate
from streamlink.stream.hls import HLSStream
from streamlink.utils.parse import parse_json
log = getLogger(__name__)
@pluginmatcher(
re.compile(r"https?://(?:www\.)?adultswim\.com/(streams|videos)(?:/([^/]+))?(?:/([^/]+))?"),
)
class AdultSwim(Plugin):
token_url = "https://token.ngtv.io/token/token_spe"
video_data_url = "https://www.adultswim.com/api/shows/v1/media/{0}/desktop"
app_id_js_url_re = re.compile(
r"""<script src="([^"]*asvp\..*?\.bundle\.js)">""",
)
app_id_re = re.compile(
r'''CDN_TOKEN_APP_ID="(.*?)"''',
)
json_data_re = re.compile(
r"""<script id="__NEXT_DATA__" type="application/json">({.*})</script>""",
)
truncate_url_re = re.compile(r"""(.*)/\w+/?""")
_api_schema = validate.Schema(
{
"media": {
"desktop": {
str: {
"url": validate.url(),
},
},
},
},
validate.get("media"),
validate.get("desktop"),
validate.filter(lambda k, v: k in ["unprotected", "bulkaes"]),
)
_stream_data_schema = validate.Schema(
{
"props": {
"__REDUX_STATE__": {
"streams": [
{
"id": str,
"stream": str,
},
],
},
},
},
validate.get("props"),
validate.get("__REDUX_STATE__"),
validate.get("streams"),
)
_token_schema = validate.Schema(
validate.any(
{"auth": {"token": str}},
{"auth": {"error": {"message": str}}},
),
validate.get("auth"),
)
_video_data_schema = validate.Schema(
{
"props": {
"pageProps": {
"__APOLLO_STATE__": {
str: {
validate.optional("id"): str,
validate.optional("slug"): str,
},
},
},
},
},
validate.get("props"),
validate.get("pageProps"),
validate.get("__APOLLO_STATE__"),
validate.filter(lambda k, v: k.startswith("Video:")),
)
def _get_stream_data(self, streamid):
res = self.session.http.get(self.url)
m = self.json_data_re.search(res.text)
if m and m.group(1):
streams = parse_json(m.group(1), schema=self._stream_data_schema)
else:
raise PluginError("Failed to get json_data")
for stream in streams:
if "id" in stream and streamid == stream["id"] and "stream" in stream:
return stream["stream"]
def _get_video_data(self, slug):
m = self.truncate_url_re.search(self.url)
if m and m.group(1):
log.debug(f"Truncated URL={m.group(1)}")
else:
raise PluginError("Failed to truncate URL")
res = self.session.http.get(m.group(1))
m = self.json_data_re.search(res.text)
if m and m.group(1):
videos = parse_json(m.group(1), schema=self._video_data_schema)
else:
raise PluginError("Failed to get json_data")
for video in videos:
if "slug" in videos[video]:
if slug == videos[video]["slug"] and "id" in videos[video]:
return videos[video]["id"]
def _get_token(self, path):
res = self.session.http.get(self.url)
m = self.app_id_js_url_re.search(res.text)
app_id_js_url = m and m.group(1)
if not app_id_js_url:
raise PluginError("Could not determine app_id_js_url")
log.debug(f"app_id_js_url={app_id_js_url}")
res = self.session.http.get(app_id_js_url)
m = self.app_id_re.search(res.text)
app_id = m and m.group(1)
if not app_id:
raise PluginError("Could not determine app_id")
log.debug(f"app_id={app_id}")
res = self.session.http.get(
self.token_url,
params=dict(
format="json",
appId=app_id,
path=path,
),
)
token_data = self.session.http.json(res, schema=self._token_schema)
if "error" in token_data:
raise PluginError(token_data["error"]["message"])
return token_data["token"]
def _get_streams(self):
url_type, show_name, episode_name = self.match.groups()
if url_type == "streams" and not show_name:
url_type = "live-stream"
elif not show_name:
raise PluginError(f"Missing show_name for url_type: {url_type}")
log.debug(f"URL type={url_type}")
if url_type == "live-stream":
video_id = self._get_stream_data(url_type)
elif url_type == "streams":
video_id = self._get_stream_data(show_name)
elif url_type == "videos":
if show_name is None or episode_name is None:
raise PluginError(
f"Missing show_name or episode_name for url_type: {url_type}",
)
video_id = self._get_video_data(episode_name)
else:
raise PluginError(f"Unrecognised url_type: {url_type}")
if video_id is None:
raise PluginError("Could not find video_id")
log.debug(f"Video ID={video_id}")
res = self.session.http.get(self.video_data_url.format(video_id))
url_data = self.session.http.json(res, schema=self._api_schema)
if "unprotected" in url_data:
url = url_data["unprotected"]["url"]
elif "bulkaes" in url_data:
url_parsed = urlparse(url_data["bulkaes"]["url"])
token = self._get_token(url_parsed.path)
url = urlunparse((
url_parsed.scheme,
url_parsed.netloc,
url_parsed.path,
url_parsed.params,
f"hdnts={token}",
url_parsed.fragment,
))
else:
raise PluginError("Could not find a usable URL in url_data")
log.debug(f"URL={url}")
return HLSStream.parse_variant_playlist(self.session, url)
__plugin__ = AdultSwim