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
|
{ writers, python3Packages, ... }:
writers.writePython3Bin "async-yt-dlp" {
libraries = with python3Packages; [ yt-dlp ];
flakeIgnore = ["E501"];
} ''
import sys
import os
import yt_dlp
import yt_dlp.options
from collections import namedtuple
import socket
from pathlib import Path
import json
create_parser = yt_dlp.options.create_parser
def parse_patched_options(opts):
patched_parser = create_parser()
patched_parser.defaults.update({
'ignoreerrors': False,
'retries': 0,
'fragment_retries': 0,
'extract_flat': False,
'concat_playlist': 'never',
})
yt_dlp.options.create_parser = lambda: patched_parser
try:
return yt_dlp.parse_options(opts)
finally:
yt_dlp.options.create_parser = create_parser
default_opts = parse_patched_options([]).ydl_opts
def cli_to_api(opts):
opts = parse_patched_options(opts)
urls = opts.urls
opts = opts.ydl_opts
diff = {k: v for k, v in opts.items() if default_opts[k] != v}
if 'postprocessors' in diff:
diff['postprocessors'] = [pp for pp in diff['postprocessors']
if pp not in default_opts['postprocessors']]
return namedtuple('Options', ('params', 'urls'))(diff, urls)
if __name__ == '__main__':
opts = cli_to_api(sys.argv[1:])
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
sock.connect(str(Path(os.environ["XDG_RUNTIME_DIR"]) / "yt-dlp.sock").encode('utf-8'))
with sock.makefile(mode='w', buffering=1, encoding='utf-8') as fh:
json.dump(opts._asdict(), fh)
''
|
