libytdlp-sys 0.6.1

Raw FFI bindings to the libytdlp C ABI
Documentation 
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

import json
import os

import yt_dlp
from yt_dlp import YoutubeDL

from .options import merge_options
from .serialize import dumps


class _Cancelled(Exception):
    pass


class _SilentLogger:
    def debug(self, message):
        pass

    def warning(self, message):
        pass

    def error(self, message):
        pass


def _contains(root, candidate):
    try:
        resolved = os.path.realpath(candidate)
    except (OSError, ValueError):
        return False

    try:
        common = os.path.commonpath((root, resolved))
    except ValueError:
        return False

    return common == root


def _resolve_root(value):
    if not isinstance(value, str):
        raise ValueError("libytdlp.restrict_to must be a string")

    if not value:
        raise ValueError("libytdlp.restrict_to must not be empty")

    if not os.path.isabs(value):
        raise ValueError("libytdlp.restrict_to must be an absolute path")

    return os.path.realpath(value)


def _confine_template_path(root, template):
    if not isinstance(template, str):
        raise ValueError("confined outtmpl must be a string")

    if not template:
        raise ValueError("confined outtmpl must not be empty")

    parent = os.path.dirname(template)

    if not parent:
        parent = "."

    if not _contains(root, parent):
        raise ValueError("outtmpl escapes the restricted root")


def _confine_outtmpl(root, value):
    if isinstance(value, str):
        _confine_template_path(root, value)

        return

    if isinstance(value, dict):
        for variant in value.values():
            _confine_template_path(root, variant)

        return

    raise ValueError("confined outtmpl must be a string or a dict")


def _confine_paths(root, value):
    if not isinstance(value, dict):
        raise ValueError("confined paths must be an object")

    for entry in value.values():
        if not isinstance(entry, str):
            raise ValueError("confined paths entries must be strings")

        if not entry:
            raise ValueError("confined paths entries must not be empty")

        if not _contains(root, entry):
            raise ValueError("paths entry escapes the restricted root")


def _enforce_restriction(staged):
    if "libytdlp" not in staged:
        return

    raw = staged.pop("libytdlp")

    if not isinstance(raw, dict):
        raise ValueError("libytdlp options must be an object")

    root = _resolve_root(raw.get("restrict_to"))

    if "outtmpl" not in staged and "paths" not in staged:
        raise ValueError("restricted downloads need an explicit outtmpl or paths")

    if "outtmpl" in staged:
        _confine_outtmpl(root, staged["outtmpl"])

    if "paths" in staged:
        _confine_paths(root, staged["paths"])


def _params(options, events, cancel_event):
    def hook(progress):
        if cancel_event.is_set():
            raise _Cancelled("cancelled")

        events.append(json.dumps(progress, ensure_ascii=True, separators=(",", ":"), default=repr))

    params = {
        "quiet": True,
        "no_warnings": True,
        "logger": _SilentLogger(),
        "cachedir": False,
    }

    if options:
        if not isinstance(options, dict):
            raise ValueError("options must be a JSON object")

        staged = dict(options)

        _enforce_restriction(staged)

        params.update(staged)

    params["progress_hooks"] = [*params.get("progress_hooks", []), hook]

    return params


def download_json(
    url,
    context_options_json="{}",
    job_options_json="{}",
    events=None,
    cancel_event=None,
):
    if not url or not isinstance(url, str):
        raise ValueError("url must be a non-empty string")

    if events is None:
        raise ValueError("events must be a list")

    if cancel_event is None:
        raise ValueError("cancel_event must be a threading.Event")

    options = merge_options(context_options_json, job_options_json)

    with YoutubeDL(_params(options, events, cancel_event)) as ydl:
        info = ydl.extract_info(url, download=True)

    return dumps(info)