--- /dev/null
+# Copyright (c) 2023 Jakub Czajka <jakub@ekhem.eu.org>
+# License: GPL-3.0 or later.
+
+import argparse
+import os
+import re
+import tempfile
+import urllib.parse
+import uuid
+import yt_dlp
+
+from wsgiref.simple_server import make_server
+
+FORMAT_OPTS = {
+ 'video': {
+ 'mime': 'video/mp4',
+ 'extension': 'mp4',
+ 'ydl_opts': {
+ 'format': 'mp4',
+ }
+ },
+ 'audio': {
+ 'mime': 'audio/ogg',
+ 'extension': 'ogg',
+ 'ydl_opts': {
+ 'format': '251',
+ 'postprocessors': [{
+ 'key': 'FFmpegExtractAudio',
+ }],
+ 'postprocessor_args': [
+ '-ar', '16000'
+ ],
+ 'prefer_ffmpeg': True
+ }
+ }
+}
+
+VIDEO_PARAMETERS_FORM = '''
+<!DOCTYPE html>
+<html>
+<head>
+ <title>YT Download</title>
+ <style>
+ body {{
+ border: ridge 5px;
+ max-width: max-content;
+ }}
+ .container {{
+ background-color: PowderBlue;
+ padding: 7px;
+ }}
+ form {{
+ display: grid;
+ grid-gap: 5px;
+ grid-template-columns: 1fr 4fr;
+ }}
+ #format {{
+ justify-self: end;
+ width: 50%;
+ }}
+ </style>
+</head>
+<body>
+ <div class="container">
+ <form>
+ <label for="url">URL:</label>
+ <input type="text" id="url" name="url">
+
+ <label for="format">Format:</label>
+ <input list="formats" id="format" name="format">
+
+ <datalist id="formats">
+ {0}
+ </datalist>
+
+ <input type="submit">
+ </form>
+ </div>
+</body>
+</html>
+'''
+
+YOUTUBE_URL_RE = '(https:\/\/)?(www\.)?youtu(\.be|be\.com)\/watch\?v=([\w\-_]+)'
+
+def build_form_for_video_parameters():
+ formats_as_html = ['<option value="{}">'.format(f) for f in FORMAT_OPTS]
+ return VIDEO_PARAMETERS_FORM.format(''.join(formats_as_html))
+
+def parse(query_string):
+ parameters = urllib.parse.parse_qs(query_string)
+ if not 'url' in parameters or not 'format' in parameters:
+ return None, None
+ return parameters['url'][0], parameters['format'][0]
+
+def set_temporary_output_file(ydl_opts, extension):
+ without_ext = tempfile.gettempdir() + '/' + str(uuid.uuid4())
+ output_file = without_ext + '.' + extension
+ if 'postprocessor_args' in ydl_opts:
+ ydl_opts['outtmpl'] = { 'default': without_ext }
+ ydl_opts['postprocessor_args'].append(output_file)
+ else:
+ ydl_opts['outtmpl'] = { 'default': output_file }
+ return output_file
+
+def wrap_file_for_serving(environ, file_path):
+ UPLOAD_BLOCK_SIZE = 1024
+ f = open(file_path, 'rb')
+ if 'wsgi.file_wrapper' in environ:
+ return environ['wsgi.file_wrapper'](f, UPLOAD_BLOCK_SIZE)
+ return iter(lambda: f.read(UPLOAD_BLOCK_SIZE), '')
+
+def handle_request(environ, start_response):
+ if not environ['QUERY_STRING']:
+ start_response('200 OK', [('Content-type', 'text/html')])
+ return [str.encode(build_form_for_video_parameters())]
+
+ video_url, output_format = parse(environ['QUERY_STRING'])
+ if not video_url:
+ print('Error: One or more empty parameters.')
+ start_response('200 OK', [('Content-type', 'text/plain')])
+ return [b'Error: One or more empty parameters.']
+
+ if not re.fullmatch(YOUTUBE_URL_RE, video_url):
+ print(f'{video_url} is not a valid YouTube URL.')
+ start_response('200 OK', [('Content-type', 'text/plain')])
+ return [str.encode(f'\'{video_url}\' is not a valid YouTube URL.')]
+
+ if not output_format in FORMAT_OPTS:
+ print(f'{output_format} is not a valid output format.')
+ start_response('200 OK', [('Content-type', 'text/plain')])
+ return [str.encode(f'\'{output_format}\' is not a valid output format.')]
+
+ ydl_opts = FORMAT_OPTS[output_format]['ydl_opts']
+ tmp_file = set_temporary_output_file(ydl_opts,
+ FORMAT_OPTS[output_format]['extension'])
+
+ print(f'Downloading {video_url}...')
+ with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+ error_code = ydl.download([video_url])
+ if error_code:
+ print(f'Download failed with code {error_code}')
+ start_response('200 OK', [('Content-type', 'text/plain')])
+ return [str.encode(f'Download failed with code {error_code}')]
+ print('Done')
+
+ print(f'Sending {video_url}...')
+ start_response('200 OK', [
+ ('Content-type', FORMAT_OPTS[output_format]['mime']),
+ ('Content-Length', str(os.path.getsize(tmp_file))),
+ ('Content-Disposition', f'attachment; filename="{tmp_file}"')])
+ return wrap_file_for_serving(environ, tmp_file)
+
+if __name__ == '__main__':
+ parser = argparse.ArgumentParser(prog='yt-dlp-server', description='HTTP '
+ 'server for downloading from YouTube. Exposes a GET / endpoint with a '
+ 'HTML form for obtaining download parameters.')
+ parser.add_argument('-a', dest='host', default='localhost',
+ help='default=localhost')
+ parser.add_argument('-p', dest='port', type=int, required=True)
+
+ args = parser.parse_args()
+
+ with make_server(args.host, args.port, handle_request) as httpd:
+ print('Serving on {0}:{1}...'.format(args.host, args.port))
+ try:
+ httpd.serve_forever()
+ except KeyboardInterrupt:
+ print('Shutting down.')
+ httpd.server_close()
projects / ydlpd.git / commitdiff