Fix for YouTube changes
[youtube-cgi] / youtube.cgi
1 #!/usr/bin/env python3
2
3 import cgi
4 import html.parser
5 import http.cookiejar
6 import json
7 import os
8 import re
9 import shutil
10 import subprocess
11 import sys
12 import time
13 import urllib.error
14 import urllib.parse
15 import urllib.request
16
17
18 MAX_MEMORY_BYTES = 128 * 1024*1024
19 USER_AGENT = "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:15.0) Gecko/20100101 Firefox/15.0.1"
20
21 MIMETYPES = {
22 "video/mp4": "mp4",
23 "video/x-flv": "flv",
24 "video/3gpp": "3gp",
25 }
26
27 QUALITIES = {
28 "hd1080": 5,
29 "hd720": 4,
30 "large": 3,
31 "medium": 2,
32 "small": 1,
33 }
34
35
36 class VideoUnavailable(Exception):
37 pass
38
39 def print_form(url="", msg=""):
40 script_url = "http://%s%s" % (os.environ["HTTP_HOST"], os.environ["REQUEST_URI"])
41 sys.stdout.write("Content-Type: text/html\r\n\r\n")
42 sys.stdout.write("""
43 <!DOCTYPE html>
44 <html>
45 <head>
46 <title>delx.net.au - YouTube Scraper</title>
47 <link rel="stylesheet" type="text/css" href="/style.css">
48 <style type="text/css">
49 input[type="text"] {
50 width: 100%;
51 }
52 .error {
53 color: red;
54 }
55 </style>
56 </head>
57 <body>
58 <h1>delx.net.au - YouTube Scraper</h1>
59 {0}
60 <form action="" method="get">
61 <p>This page will let you easily download YouTube videos to watch offline. It
62 will automatically grab the highest quality version.</p>
63 <div><input type="text" name="url" value="{1}"/></div>
64 <div><input type="submit" value="Download!"/></div>
65 </form>
66 <p>Tip! Use this bookmarklet: <a href="javascript:(function(){window.location='{2}?url='+escape(location);})()">YouTube Download</a>
67 to easily download videos. Right-click the link and add it to bookmarks,
68 then when you're looking at a YouTube page select that bookmark from your
69 browser's bookmarks menu to download the video straight away.</p>
70 </body>
71 </html>
72 """.replace("{0}", msg).replace("{1}", url).replace("{2}", script_url))
73
74 cookiejar = http.cookiejar.CookieJar()
75 urlopener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cookiejar))
76 referrer = ""
77
78 def urlopen(url, offset=None):
79 if url.startswith("//"):
80 url = "https:" + url
81 if not url.startswith("http://") and not url.startswith("https://"):
82 url = "https://www.youtube.com" + url
83
84 global referrer
85 req = urllib.request.Request(url)
86 if not referrer:
87 referrer = url
88 else:
89 req.add_header("Referer", referrer)
90
91 req.add_header("User-Agent", USER_AGENT)
92
93 if offset:
94 req.add_header("Range", "bytes=%d-" % offset)
95
96 res = urlopener.open(req)
97
98 content_range = res.getheader("Content-Range")
99 if content_range:
100 tokens = content_range.split()
101 assert tokens[0] == "bytes"
102 start = int(tokens[1].split("-")[0])
103 assert start == offset
104 return res
105
106 def parse_url(url, parser):
107 f = urlopen(url)
108 parser.feed(f.read().decode("utf-8"))
109 parser.close()
110 f.close()
111
112 def append_to_qs(url, params):
113 r = list(urllib.parse.urlsplit(url))
114 qs = urllib.parse.parse_qs(r[3])
115 qs.update(params)
116 r[3] = urllib.parse.urlencode(qs, True)
117 url = urllib.parse.urlunsplit(r)
118 return url
119
120 def get_player_config(scripts):
121 player_config = None
122 for script in scripts:
123 for line in script.split("\n"):
124 s = "ytplayer.config = {"
125 if s in line:
126 p1 = line.find(s) + len(s) - 1
127 p2 = line.find("};", p1) + 1
128 if p1 >= 0 and p2 > 0:
129 return json.loads(line[p1:p2])
130
131 def extract_js(script):
132 PREFIX = "var _yt_player={};(function(g){var window=this;"
133 SUFFIX = ";})(_yt_player);\n"
134 assert script.startswith(PREFIX)
135 assert script.endswith(SUFFIX)
136
137 return script[len(PREFIX):-len(SUFFIX)]
138
139 def find_func_name(script):
140 FUNC_NAME = R"([a-zA-Z0-9$]+)"
141 FUNC_PARAMS = R"(\([a-zA-Z,\.]+\.s\))"
142 TERMINATOR = R"[,;\)]"
143 PATTERN = FUNC_NAME + FUNC_PARAMS + TERMINATOR
144
145 match = re.search(PATTERN, script)
146 func_name = match.groups()[0]
147 return func_name
148
149 def decode_signature(js_url, signature):
150 f = urlopen(js_url)
151 script = f.read().decode("utf-8")
152 f.close()
153
154 func_name = find_func_name(script)
155
156 params = {
157 "func_name": func_name,
158 "signature": json.dumps(signature),
159 "code": json.dumps(extract_js(script)),
160 }
161 p = subprocess.Popen(
162 "node",
163 shell=True,
164 close_fds=True,
165 stdin=subprocess.PIPE,
166 stdout=subprocess.PIPE
167 )
168 js_decode_script = ("""
169 const vm = require('vm');
170
171 const sandbox = {
172 location: {
173 hash: '',
174 href: '',
175 protocol: 'http:'
176 },
177 history: {
178 pushState: function(){}
179 },
180 document: {},
181 navigator: {
182 userAgent: ''
183 },
184 matchMedia: () => ({matches: () => {}, media: ''}),
185 signature: %(signature)s,
186 transformed_signature: null,
187 g: function(){} // this is _yt_player
188 };
189 sandbox.window = sandbox;
190
191 const code_string = %(code)s + ';';
192 const exec_string = 'transformed_signature = %(func_name)s("", "MARKER", signature);';
193 vm.runInNewContext(code_string + exec_string, sandbox);
194
195 function findSignature(obj) {
196 if (typeof obj !== 'object') {
197 return;
198 }
199 for (const [key, value] of Object.entries(obj)) {
200 if (key === 'MARKER') {
201 return value;
202 }
203 const result = findSignature(value);
204 if (result) {
205 return result;
206 }
207 }
208 }
209 console.log(findSignature(sandbox.transformed_signature));
210 """ % params)
211
212 p.stdin.write(js_decode_script.encode("utf-8"))
213 p.stdin.close()
214
215 transformed_signature = p.stdout.read().decode("utf-8").strip()
216 if p.wait() != 0:
217 raise Exception("js failed to execute: %d" % p.returncode)
218
219 return transformed_signature
220
221 def get_best_video(player_config):
222 url_data_list = player_config["args"]["url_encoded_fmt_stream_map"].split(",")
223 js_url = player_config["assets"]["js"]
224
225 best_url = None
226 best_quality = None
227 best_extension = None
228 for url_data in url_data_list:
229 url_data = urllib.parse.parse_qs(url_data)
230 mimetype = url_data["type"][0].split(";")[0]
231 quality = url_data["quality"][0]
232
233 if "stereo3d" in url_data:
234 continue
235 if quality not in QUALITIES:
236 continue
237 if mimetype not in MIMETYPES:
238 continue
239
240 extension = MIMETYPES[mimetype]
241 quality = QUALITIES.get(quality, -1)
242
243 if best_quality is not None and quality < best_quality:
244 continue
245
246 video_url = url_data["url"][0]
247 if "sig" in url_data:
248 signature = url_data["sig"][0]
249 elif "s" in url_data:
250 signature = decode_signature(js_url, url_data["s"][0])
251 else:
252 signature = None
253
254 if signature:
255 video_url = append_to_qs(video_url, {"signature": signature})
256
257 best_url = video_url
258 best_quality = quality
259 best_extension = extension
260
261 return best_url, best_extension
262
263 def sanitize_filename(filename):
264 return (
265 re.sub("\s+", " ", filename.strip())
266 .replace("\\", "-")
267 .replace("/", "-")
268 .replace("\0", " ")
269 )
270
271 def get_video_url(page):
272 player_config = get_player_config(page.scripts)
273 if not player_config:
274 raise VideoUnavailable(page.unavailable_message or "Could not find video URL")
275
276 video_url, extension = get_best_video(player_config)
277 if not video_url:
278 return None, None
279
280 filename = sanitize_filename(page.title)
281 filename += "." + extension
282
283 return video_url, filename
284
285 class YouTubeVideoPageParser(html.parser.HTMLParser):
286 def __init__(self):
287 super().__init__()
288 self.title = None
289 self.unavailable_message = None
290 self.scripts = []
291
292 def handle_starttag(self, tag, attrs):
293 attrs = dict(attrs)
294 self._handle_title(tag, attrs)
295 self._handle_unavailable_message(tag, attrs)
296 self._handle_script(tag, attrs)
297
298 def handle_endtag(self, tag):
299 self.handle_data = self._ignore_data
300
301 def _ignore_data(self, _):
302 pass
303
304 def _handle_title(self, tag, attrs):
305 if tag == "title":
306 self.handle_data = self._handle_title_data
307
308 def _handle_title_data(self, data):
309 self.title = data.strip()
310
311 def _handle_unavailable_message(self, tag, attrs):
312 if attrs.get("id", None) == "unavailable-message":
313 self.handle_data = self._handle_unavailable_message_data
314
315 def _handle_unavailable_message_data(self, data):
316 self.unavailable_message = data.strip()
317
318 def _handle_script(self, tag, attrs):
319 if tag == "script":
320 self.handle_data = self._handle_script_data
321
322 def _handle_script_data(self, data):
323 if data:
324 self.scripts.append(data)
325
326 def write_video(filename, video_data):
327 quoted_filename = urllib.parse.quote(filename.encode("utf-8"))
328 sys.stdout.buffer.write(
329 b"Content-Disposition: attachment; filename*=UTF-8''{0}\r\n"
330 .replace(b"{0}", quoted_filename.encode("utf-8"))
331 )
332 sys.stdout.buffer.write(
333 b"Content-Length: {0}\r\n"
334 .replace(b"{0}", video_data.getheader("Content-Length").encode("utf-8"))
335 )
336 sys.stdout.buffer.write(b"\r\n")
337 shutil.copyfileobj(video_data, sys.stdout.buffer)
338 video_data.close()
339
340 def cgimain():
341 args = cgi.parse()
342 try:
343 url = args["url"][0]
344 except:
345 print_form(url="http://www.youtube.com/watch?v=FOOBAR")
346 return
347
348 try:
349 page = YouTubeVideoPageParser()
350 parse_url(url, page)
351 video_url, filename = get_video_url(page)
352 video_data = urlopen(video_url)
353 except VideoUnavailable as e:
354 print_form(
355 url=url,
356 msg="<p class='error'>Sorry, there was an error: %s</p>" % cgi.escape(e.args[0])
357 )
358 except Exception as e:
359 print_form(
360 url=url,
361 msg="<p class='error'>Sorry, there was an error. Check your URL?</p>"
362 )
363 return
364
365 write_video(filename, video_data)
366
367 def pp_size(size):
368 suffixes = ["", "KiB", "MiB", "GiB"]
369 for i, suffix in enumerate(suffixes):
370 if size < 1024:
371 break
372 size /= 1024
373 return "%.2f %s" % (size, suffix)
374
375 def copy_with_progress(content_length, infile, outfile):
376 def print_status():
377 rate = 0
378 if now != last_ts:
379 rate = last_bytes_read / (now - last_ts)
380 sys.stdout.write("\33[2K\r")
381 sys.stdout.write("%s / %s (%s/sec)" % (
382 pp_size(bytes_read),
383 pp_size(content_length),
384 pp_size(rate),
385 ))
386 sys.stdout.flush()
387
388 last_ts = 0
389 last_bytes_read = 0
390 bytes_read = 0
391 while True:
392 now = time.time()
393 if now - last_ts > 0.5:
394 print_status()
395 last_ts = now
396 last_bytes_read = 0
397
398 buf = infile.read(32768)
399 if not buf:
400 break
401 outfile.write(buf)
402 last_bytes_read += len(buf)
403 bytes_read += len(buf)
404
405 # Newline at the end
406 print_status()
407 print()
408
409 def main():
410 try:
411 url = sys.argv[1]
412 except:
413 print("Usage: %s http://youtube.com/watch?v=FOOBAR" % sys.argv[0], file=sys.stderr)
414 sys.exit(1)
415
416 page = YouTubeVideoPageParser()
417 parse_url(url, page)
418 video_url, filename = get_video_url(page)
419 print("Downloading", filename)
420
421 outfile = open(filename, "ab")
422 offset = outfile.tell()
423 if offset > 0:
424 print("Resuming download from", pp_size(offset))
425 total_size = None
426
427 while True:
428 try:
429 video_data = urlopen(video_url, offset)
430 except urllib.error.HTTPError as e:
431 if e.code == 416:
432 print("File is complete!")
433 break
434 else:
435 raise
436
437 content_length = int(video_data.getheader("Content-Length"))
438 if total_size is None:
439 total_size = content_length
440
441 try:
442 copy_with_progress(content_length, video_data, outfile)
443 except IOError as e:
444 print()
445
446 video_data.close()
447 if outfile.tell() != total_size:
448 old_offset = offset
449 offset = outfile.tell()
450 if old_offset == offset:
451 time.sleep(1)
452 print("Restarting download from", pp_size(offset))
453 else:
454 break
455
456 outfile.close()
457
458
459 if __name__ == "__main__":
460 if "SCRIPT_NAME" in os.environ:
461 cgimain()
462 else:
463 try:
464 main()
465 except KeyboardInterrupt:
466 print("\nExiting...")
467 sys.exit(1)
468