import html.parser
import http.client
import http.cookiejar
-import importlib.util
import inspect
import io
import itertools
self.exc_info = sys.exc_info() # preserve original exception
if isinstance(self.exc_info[1], ExtractorError):
self.exc_info = self.exc_info[1].exc_info
+ super().__init__(self.__msg)
- super().__init__(''.join((
- format_field(ie, None, '[%s] '),
- format_field(video_id, None, '%s: '),
- msg,
- format_field(cause, None, ' (caused by %r)'),
- '' if expected else bug_reports_message())))
+ @property
+ def __msg(self):
+ return ''.join((
+ format_field(self.ie, None, '[%s] '),
+ format_field(self.video_id, None, '%s: '),
+ self.orig_msg,
+ format_field(self.cause, None, ' (caused by %r)'),
+ '' if self.expected else bug_reports_message()))
def format_traceback(self):
return join_nonempty(
self.cause and ''.join(traceback.format_exception(None, self.cause, self.cause.__traceback__)[1:]),
delim='\n') or None
+ def __setattr__(self, name, value):
+ super().__setattr__(name, value)
+ if getattr(self, 'msg', None) and name not in ('msg', 'args'):
+ self.msg = self.__msg or type(self).__name__
+ self.args = (self.msg, ) # Cannot be property
+
class UnsupportedError(ExtractorError):
def __init__(self, url):
('hEvent', ctypes.wintypes.HANDLE),
]
- kernel32 = ctypes.windll.kernel32
+ kernel32 = ctypes.WinDLL('kernel32')
LockFileEx = kernel32.LockFileEx
LockFileEx.argtypes = [
ctypes.wintypes.HANDLE, # hFile
# STDIN should be redirected too. On UNIX-like systems, ffmpeg triggers
# SIGTTOU if yt-dlp is run in the background.
# See https://github.com/ytdl-org/youtube-dl/issues/955#issuecomment-209789656
- stdout, _, _ = Popen.run([encodeArgument(exe)] + args, text=True,
- stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+ stdout, _, ret = Popen.run([encodeArgument(exe)] + args, text=True,
+ stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+ if ret:
+ return None
except OSError:
return False
return stdout
def get_exe_version(exe, args=['--version'],
- version_re=None, unrecognized='present'):
+ version_re=None, unrecognized=('present', 'broken')):
""" Returns the version of the specified executable,
or False if the executable is not present """
+ unrecognized = variadic(unrecognized)
+ assert len(unrecognized) in (1, 2)
out = _get_exe_version_output(exe, args)
- return detect_exe_version(out, version_re, unrecognized) if out else False
+ if out is None:
+ return unrecognized[-1]
+ return out and detect_exe_version(out, version_re, unrecognized[0])
def frange(start=0, stop=None, step=1):
return f'"{i}":' if v.endswith(':') else str(i)
if v in vars:
- return json.dumps(vars[v])
+ try:
+ if not strict:
+ json.loads(vars[v])
+ except json.JSONDecodeError:
+ return json.dumps(vars[v])
+ else:
+ return vars[v]
if not strict:
return f'"{v}"'
if not strict:
code = re.sub(r'new Date\((".+")\)', r'\g<1>', code)
code = re.sub(r'new \w+\((.*?)\)', lambda m: json.dumps(m.group(0)), code)
+ code = re.sub(r'parseInt\([^\d]+(\d+)[^\d]+\)', r'\1', code)
+ code = re.sub(r'\(function\([^)]*\)\s*\{[^}]*\}\s*\)\s*\(\s*(["\'][^)]*["\'])\s*\)', r'\1', code)
return re.sub(rf'''(?sx)
{STRING_RE}|
return q
-POSTPROCESS_WHEN = ('pre_process', 'after_filter', 'before_dl', 'post_process', 'after_move', 'after_video', 'playlist')
+POSTPROCESS_WHEN = ('pre_process', 'after_filter', 'video', 'before_dl', 'post_process', 'after_move', 'after_video', 'playlist')
DEFAULT_OUTTMPL = {
return f'{type(err).__name__}: {err}'
-def mimetype2ext(mt):
- if mt is None:
+def mimetype2ext(mt, default=NO_DEFAULT):
+ if not isinstance(mt, str):
+ if default is not NO_DEFAULT:
+ return default
return None
- mt, _, params = mt.partition(';')
- mt = mt.strip()
-
- FULL_MAP = {
- 'audio/mp4': 'm4a',
- # Per RFC 3003, audio/mpeg can be .mp1, .mp2 or .mp3. Here use .mp3 as
- # it's the most popular one
- 'audio/mpeg': 'mp3',
- 'audio/x-wav': 'wav',
- 'audio/wav': 'wav',
- 'audio/wave': 'wav',
- }
-
- ext = FULL_MAP.get(mt)
- if ext is not None:
- return ext
-
- SUBTYPE_MAP = {
+ MAP = {
+ # video
'3gpp': '3gp',
- 'smptett+xml': 'tt',
- 'ttaf+xml': 'dfxp',
- 'ttml+xml': 'ttml',
+ 'mp2t': 'ts',
+ 'mp4': 'mp4',
+ 'mpeg': 'mpeg',
+ 'mpegurl': 'm3u8',
+ 'quicktime': 'mov',
+ 'webm': 'webm',
+ 'vp9': 'vp9',
'x-flv': 'flv',
+ 'x-m4v': 'm4v',
+ 'x-matroska': 'mkv',
+ 'x-mng': 'mng',
'x-mp4-fragmented': 'mp4',
- 'x-ms-sami': 'sami',
+ 'x-ms-asf': 'asf',
'x-ms-wmv': 'wmv',
- 'mpegurl': 'm3u8',
- 'x-mpegurl': 'm3u8',
- 'vnd.apple.mpegurl': 'm3u8',
+ 'x-msvideo': 'avi',
+
+ # application (streaming playlists)
'dash+xml': 'mpd',
'f4m+xml': 'f4m',
'hds+xml': 'f4m',
+ 'vnd.apple.mpegurl': 'm3u8',
'vnd.ms-sstr+xml': 'ism',
- 'quicktime': 'mov',
- 'mp2t': 'ts',
+ 'x-mpegurl': 'm3u8',
+
+ # audio
+ 'audio/mp4': 'm4a',
+ # Per RFC 3003, audio/mpeg can be .mp1, .mp2 or .mp3.
+ # Using .mp3 as it's the most popular one
+ 'audio/mpeg': 'mp3',
+ 'audio/webm': 'webm',
+ 'audio/x-matroska': 'mka',
+ 'audio/x-mpegurl': 'm3u',
+ 'midi': 'mid',
+ 'ogg': 'ogg',
+ 'wav': 'wav',
+ 'wave': 'wav',
+ 'x-aac': 'aac',
+ 'x-flac': 'flac',
+ 'x-m4a': 'm4a',
+ 'x-realaudio': 'ra',
'x-wav': 'wav',
- 'filmstrip+json': 'fs',
- 'svg+xml': 'svg',
- }
- _, _, subtype = mt.rpartition('/')
- ext = SUBTYPE_MAP.get(subtype.lower())
- if ext is not None:
- return ext
+ # image
+ 'avif': 'avif',
+ 'bmp': 'bmp',
+ 'gif': 'gif',
+ 'jpeg': 'jpg',
+ 'png': 'png',
+ 'svg+xml': 'svg',
+ 'tiff': 'tif',
+ 'vnd.wap.wbmp': 'wbmp',
+ 'webp': 'webp',
+ 'x-icon': 'ico',
+ 'x-jng': 'jng',
+ 'x-ms-bmp': 'bmp',
+
+ # caption
+ 'filmstrip+json': 'fs',
+ 'smptett+xml': 'tt',
+ 'ttaf+xml': 'dfxp',
+ 'ttml+xml': 'ttml',
+ 'x-ms-sami': 'sami',
- SUFFIX_MAP = {
+ # misc
+ 'gzip': 'gz',
'json': 'json',
'xml': 'xml',
'zip': 'zip',
- 'gzip': 'gz',
}
- _, _, suffix = subtype.partition('+')
- ext = SUFFIX_MAP.get(suffix)
- if ext is not None:
- return ext
+ mimetype = mt.partition(';')[0].strip().lower()
+ _, _, subtype = mimetype.rpartition('/')
+ ext = traverse_obj(MAP, mimetype, subtype, subtype.rsplit('+')[-1])
+ if ext:
+ return ext
+ elif default is not NO_DEFAULT:
+ return default
return subtype.replace('+', '.')
hdr = 'HDR10'
elif parts[:2] == ['vp9', '2']:
hdr = 'HDR10'
- elif parts[0] in ('flac', 'mp4a', 'opus', 'vorbis', 'mp3', 'aac',
+ elif parts[0] in ('flac', 'mp4a', 'opus', 'vorbis', 'mp3', 'aac', 'ac-4',
'ac-3', 'ec-3', 'eac3', 'dtsc', 'dtse', 'dtsh', 'dtsl'):
acodec = acodec or full_codec
elif parts[0] in ('stpp', 'wvtt'):
# TODO: All codecs supported by parse_codecs isn't handled here
COMPATIBLE_CODECS = {
'mp4': {
- 'av1', 'hevc', 'avc1', 'mp4a', # fourcc (m3u8, mpd)
+ 'av1', 'hevc', 'avc1', 'mp4a', 'ac-4', # fourcc (m3u8, mpd)
'h264', 'aacl', 'ec-3', # Set in ISM
},
'webm': {
COMPATIBLE_EXTS = (
{'mp3', 'mp4', 'm4a', 'm4p', 'm4b', 'm4r', 'm4v', 'ismv', 'isma', 'mov'},
- {'webm'},
+ {'webm', 'weba'},
)
for ext in preferences or vexts:
current_exts = {ext, *vexts, *aexts}
return 'mkv' if allow_mkv else preferences[-1]
-def urlhandle_detect_ext(url_handle):
+def urlhandle_detect_ext(url_handle, default=NO_DEFAULT):
getheader = url_handle.headers.get
cd = getheader('Content-Disposition')
if e:
return e
- return mimetype2ext(getheader('Content-Type'))
+ meta_ext = getheader('x-amz-meta-name')
+ if meta_ext:
+ e = meta_ext.rpartition('.')[2]
+ if e:
+ return e
+
+ return mimetype2ext(getheader('Content-Type'), default=default)
def encode_data_uri(data, mime_type):
return (isinstance(other, download_range_func)
and self.chapters == other.chapters and self.ranges == other.ranges)
+ def __repr__(self):
+ return f'{type(self).__name__}({self.chapters}, {self.ranges})'
+
def parse_dfxp_time_expr(time_expr):
if not time_expr:
}
+def find_available_port(interface=''):
+ try:
+ with socket.socket() as sock:
+ sock.bind((interface, 0))
+ return sock.getsockname()[1]
+ except OSError:
+ return None
+
+
# Templates for internet shortcut files, which are plain text files.
DOT_URL_LINK_TEMPLATE = '''\
[InternetShortcut]
return os.path.dirname(os.path.abspath(_get_variant_and_executable_path()[1]))
-def load_plugins(name, suffix, namespace):
- classes = {}
- with contextlib.suppress(FileNotFoundError):
- plugins_spec = importlib.util.spec_from_file_location(
- name, os.path.join(get_executable_path(), 'ytdlp_plugins', name, '__init__.py'))
- plugins = importlib.util.module_from_spec(plugins_spec)
- sys.modules[plugins_spec.name] = plugins
- plugins_spec.loader.exec_module(plugins)
- for name in dir(plugins):
- if name in namespace:
- continue
- if not name.endswith(suffix):
- continue
- klass = getattr(plugins, name)
- classes[name] = namespace[name] = klass
- return classes
+def get_user_config_dirs(package_name):
+ # .config (e.g. ~/.config/package_name)
+ xdg_config_home = os.getenv('XDG_CONFIG_HOME') or compat_expanduser('~/.config')
+ yield os.path.join(xdg_config_home, package_name)
+
+ # appdata (%APPDATA%/package_name)
+ appdata_dir = os.getenv('appdata')
+ if appdata_dir:
+ yield os.path.join(appdata_dir, package_name)
+
+ # home (~/.package_name)
+ yield os.path.join(compat_expanduser('~'), f'.{package_name}')
+
+
+def get_system_config_dirs(package_name):
+ # /etc/package_name
+ yield os.path.join('/etc', package_name)
def traverse_obj(
Each of the provided `paths` is tested and the first producing a valid result will be returned.
The next path will also be tested if the path branched but no results could be found.
Supported values for traversal are `Mapping`, `Sequence` and `re.Match`.
- A value of None is treated as the absence of a value.
+ Unhelpful values (`{}`, `None`) are treated as the absence of a value and discarded.
The paths will be wrapped in `variadic`, so that `'key'` is conveniently the same as `('key', )`.
The keys in the path can be one of:
- `None`: Return the current object.
- - `str`/`int`: Return `obj[key]`. For `re.Match, return `obj.group(key)`.
+ - `set`: Requires the only item in the set to be a type or function,
+ like `{type}`/`{func}`. If a `type`, returns only values
+ of this type. If a function, returns `func(obj)`.
+ - `str`/`int`: Return `obj[key]`. For `re.Match`, return `obj.group(key)`.
- `slice`: Branch out and return all values in `obj[key]`.
- `Ellipsis`: Branch out and return a list of all values.
- `tuple`/`list`: Branch out and return a list of all matching values.
- `function`: Branch out and return values filtered by the function.
Read as: `[value for key, value in obj if function(key, value)]`.
For `Sequence`s, `key` is the index of the value.
+ For `re.Match`es, `key` is the group number (0 = full match)
+ as well as additionally any group names, if given.
- `dict` Transform the current object and return a matching dict.
Read as: `{key: traverse_obj(obj, path) for key, path in dct.items()}`.
@params paths Paths which to traverse by.
@param default Value to return if the paths do not match.
+ If the last key in the path is a `dict`, it will apply to each value inside
+ the dict instead, depth first. Try to avoid if using nested `dict` keys.
@param expected_type If a `type`, only accept final values of this type.
If any other callable, try to call the function on each result.
+ If the last key in the path is a `dict`, it will apply to each value inside
+ the dict instead, recursively. This does respect branching paths.
@param get_all If `False`, return the first matching result, otherwise all matching ones.
@param casesense If `False`, consider string dictionary keys as case insensitive.
@param traverse_string Whether to traverse into objects as strings.
If `True`, any non-compatible object will first be
converted into a string and then traversed into.
+ The return value of that path will be a string instead,
+ not respecting any further branching.
@returns The result of the object traversal.
If successful, `get_all=True`, and the path branches at least once,
then a list of results is returned instead.
- A list is always returned if the last path branches and no `default` is given.
+ If no `default` is given and the last path branches, a `list` of results
+ is always returned. If a path ends on a `dict` that result will always be a `dict`.
"""
is_sequence = lambda x: isinstance(x, collections.abc.Sequence) and not isinstance(x, (str, bytes))
casefold = lambda k: k.casefold() if isinstance(k, str) else k
else:
type_test = lambda val: try_call(expected_type or IDENTITY, args=(val,))
- def apply_key(key, obj):
- if obj is None:
- return
+ def apply_key(key, obj, is_last):
+ branching = False
+ result = None
+
+ if obj is None and traverse_string:
+ pass
elif key is None:
- yield obj
+ result = obj
+
+ elif isinstance(key, set):
+ assert len(key) == 1, 'Set should only be used to wrap a single item'
+ item = next(iter(key))
+ if isinstance(item, type):
+ if isinstance(obj, item):
+ result = obj
+ else:
+ result = try_call(item, args=(obj,))
elif isinstance(key, (list, tuple)):
- for branch in key:
- _, result = apply_path(obj, branch)
- yield from result
+ branching = True
+ result = itertools.chain.from_iterable(
+ apply_path(obj, branch, is_last)[0] for branch in key)
elif key is ...:
+ branching = True
if isinstance(obj, collections.abc.Mapping):
- yield from obj.values()
+ result = obj.values()
elif is_sequence(obj):
- yield from obj
+ result = obj
elif isinstance(obj, re.Match):
- yield from obj.groups()
+ result = obj.groups()
elif traverse_string:
- yield from str(obj)
+ branching = False
+ result = str(obj)
+ else:
+ result = ()
elif callable(key):
- if is_sequence(obj):
- iter_obj = enumerate(obj)
- elif isinstance(obj, collections.abc.Mapping):
+ branching = True
+ if isinstance(obj, collections.abc.Mapping):
iter_obj = obj.items()
+ elif is_sequence(obj):
+ iter_obj = enumerate(obj)
elif isinstance(obj, re.Match):
- iter_obj = enumerate((obj.group(), *obj.groups()))
+ iter_obj = itertools.chain(
+ enumerate((obj.group(), *obj.groups())),
+ obj.groupdict().items())
elif traverse_string:
+ branching = False
iter_obj = enumerate(str(obj))
else:
- return
- yield from (v for k, v in iter_obj if try_call(key, args=(k, v)))
+ iter_obj = ()
+
+ result = (v for k, v in iter_obj if try_call(key, args=(k, v)))
+ if not branching: # string traversal
+ result = ''.join(result)
elif isinstance(key, dict):
- iter_obj = ((k, _traverse_obj(obj, v)) for k, v in key.items())
- yield {k: v if v is not None else default for k, v in iter_obj
- if v is not None or default is not NO_DEFAULT}
+ iter_obj = ((k, _traverse_obj(obj, v, False, is_last)) for k, v in key.items())
+ result = {
+ k: v if v is not None else default for k, v in iter_obj
+ if v is not None or default is not NO_DEFAULT
+ } or None
elif isinstance(obj, collections.abc.Mapping):
- yield (obj.get(key) if casesense or (key in obj)
- else next((v for k, v in obj.items() if casefold(k) == key), None))
+ result = (obj.get(key) if casesense or (key in obj) else
+ next((v for k, v in obj.items() if casefold(k) == key), None))
elif isinstance(obj, re.Match):
if isinstance(key, int) or casesense:
with contextlib.suppress(IndexError):
- yield obj.group(key)
- return
+ result = obj.group(key)
- if not isinstance(key, str):
- return
+ elif isinstance(key, str):
+ result = next((v for k, v in obj.groupdict().items() if casefold(k) == key), None)
- yield next((v for k, v in obj.groupdict().items() if casefold(k) == key), None)
+ elif isinstance(key, (int, slice)):
+ if is_sequence(obj):
+ branching = isinstance(key, slice)
+ with contextlib.suppress(IndexError):
+ result = obj[key]
+ elif traverse_string:
+ with contextlib.suppress(IndexError):
+ result = str(obj)[key]
- else:
- if is_user_input:
- key = (int_or_none(key) if ':' not in key
- else slice(*map(int_or_none, key.split(':'))))
+ return branching, result if branching else (result,)
- if not isinstance(key, (int, slice)):
- return
+ def lazy_last(iterable):
+ iterator = iter(iterable)
+ prev = next(iterator, NO_DEFAULT)
+ if prev is NO_DEFAULT:
+ return
- if not is_sequence(obj):
- if not traverse_string:
- return
- obj = str(obj)
+ for item in iterator:
+ yield False, prev
+ prev = item
- with contextlib.suppress(IndexError):
- yield obj[key]
+ yield True, prev
- def apply_path(start_obj, path):
+ def apply_path(start_obj, path, test_type):
objs = (start_obj,)
has_branched = False
- for key in variadic(path):
- if is_user_input and key == ':':
- key = ...
+ key = None
+ for last, key in lazy_last(variadic(path, (str, bytes, dict, set))):
+ if is_user_input and isinstance(key, str):
+ if key == ':':
+ key = ...
+ elif ':' in key:
+ key = slice(*map(int_or_none, key.split(':')))
+ elif int_or_none(key) is not None:
+ key = int(key)
if not casesense and isinstance(key, str):
key = key.casefold()
- if key is ... or isinstance(key, (list, tuple)) or callable(key):
- has_branched = True
+ if __debug__ and callable(key):
+ # Verify function signature
+ inspect.signature(key).bind(None, None)
+
+ new_objs = []
+ for obj in objs:
+ branching, results = apply_key(key, obj, last)
+ has_branched |= branching
+ new_objs.append(results)
- key_func = functools.partial(apply_key, key)
- objs = itertools.chain.from_iterable(map(key_func, objs))
+ objs = itertools.chain.from_iterable(new_objs)
- return has_branched, objs
+ if test_type and not isinstance(key, (dict, list, tuple)):
+ objs = map(type_test, objs)
- def _traverse_obj(obj, path, use_list=True):
- has_branched, results = apply_path(obj, path)
- results = LazyList(x for x in map(type_test, results) if x is not None)
+ return objs, has_branched, isinstance(key, dict)
+ def _traverse_obj(obj, path, allow_empty, test_type):
+ results, has_branched, is_dict = apply_path(obj, path, test_type)
+ results = LazyList(item for item in results if item not in (None, {}))
if get_all and has_branched:
- return results.exhaust() if results or use_list else None
+ if results:
+ return results.exhaust()
+ if allow_empty:
+ return [] if default is NO_DEFAULT else default
+ return None
- return results[0] if results else None
+ return results[0] if results else {} if allow_empty and is_dict else None
for index, path in enumerate(paths, 1):
- use_list = default is NO_DEFAULT and index == len(paths)
- result = _traverse_obj(obj, path, use_list)
+ result = _traverse_obj(obj, path, index == len(paths), True)
if result is not None:
return result
def time_seconds(**kwargs):
- t = datetime.datetime.now(datetime.timezone(datetime.timedelta(**kwargs)))
- return t.timestamp()
+ """
+ Returns TZ-aware time in seconds since the epoch (1970-01-01T00:00:00Z)
+ """
+ return time.time() + datetime.timedelta(**kwargs).total_seconds()
# create a JSON Web Signature (jws) with HS256 algorithm
return False
-def windows_enable_vt_mode(): # TODO: Do this the proper way https://bugs.python.org/issue30075
+def windows_enable_vt_mode():
+ """Ref: https://bugs.python.org/issue30075 """
if get_windows_version() < (10, 0, 10586):
return
- global WINDOWS_VT_MODE
+
+ import ctypes
+ import ctypes.wintypes
+ import msvcrt
+
+ ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x0004
+
+ dll = ctypes.WinDLL('kernel32', use_last_error=False)
+ handle = os.open('CONOUT$', os.O_RDWR)
try:
- Popen.run('', shell=True)
- except Exception:
- return
+ h_out = ctypes.wintypes.HANDLE(msvcrt.get_osfhandle(handle))
+ dw_original_mode = ctypes.wintypes.DWORD()
+ success = dll.GetConsoleMode(h_out, ctypes.byref(dw_original_mode))
+ if not success:
+ raise Exception('GetConsoleMode failed')
+
+ success = dll.SetConsoleMode(h_out, ctypes.wintypes.DWORD(
+ dw_original_mode.value | ENABLE_VIRTUAL_TERMINAL_PROCESSING))
+ if not success:
+ raise Exception('SetConsoleMode failed')
+ finally:
+ os.close(handle)
+ global WINDOWS_VT_MODE
WINDOWS_VT_MODE = True
supports_terminal_sequences.cache_clear()
common_video=('avi', 'flv', 'mkv', 'mov', 'mp4', 'webm'),
video=('3g2', '3gp', 'f4v', 'mk3d', 'divx', 'mpg', 'ogv', 'm4v', 'wmv'),
common_audio=('aiff', 'alac', 'flac', 'm4a', 'mka', 'mp3', 'ogg', 'opus', 'wav'),
- audio=('aac', 'ape', 'asf', 'f4a', 'f4b', 'm4b', 'm4p', 'm4r', 'oga', 'ogx', 'spx', 'vorbis', 'wma'),
+ audio=('aac', 'ape', 'asf', 'f4a', 'f4b', 'm4b', 'm4p', 'm4r', 'oga', 'ogx', 'spx', 'vorbis', 'wma', 'weba'),
thumbnails=('jpg', 'png', 'webp'),
storyboards=('mhtml', ),
subtitles=('srt', 'vtt', 'ass', 'lrc'),
assert left > 3 and right >= 0
if s is None or len(s) <= left + right:
return s
- return f'{s[:left-3]}...{s[-right:]}'
+ return f'{s[:left-3]}...{s[-right:] if right else ""}'
def orderedSet_from_options(options, alias_dict, *, use_regex=False, start=None):
'vcodec': {'type': 'ordered', 'regex': True,
'order': ['av0?1', 'vp0?9.2', 'vp0?9', '[hx]265|he?vc?', '[hx]264|avc', 'vp0?8', 'mp4v|h263', 'theora', '', None, 'none']},
'acodec': {'type': 'ordered', 'regex': True,
- 'order': ['[af]lac', 'wav|aiff', 'opus', 'vorbis|ogg', 'aac', 'mp?4a?', 'mp3', 'e-?a?c-?3', 'ac-?3', 'dts', '', None, 'none']},
+ 'order': ['[af]lac', 'wav|aiff', 'opus', 'vorbis|ogg', 'aac', 'mp?4a?', 'mp3', 'ac-?4', 'e-?a?c-?3', 'ac-?3', 'dts', '', None, 'none']},
'hdr': {'type': 'ordered', 'regex': True, 'field': 'dynamic_range',
'order': ['dv', '(hdr)?12', r'(hdr)?10\+', '(hdr)?10', 'hlg', '', 'sdr', None]},
'proto': {'type': 'ordered', 'regex': True, 'field': 'protocol',
'order': ['(ht|f)tps', '(ht|f)tp$', 'm3u8.*', '.*dash', 'websocket_frag', 'rtmpe?', '', 'mms|rtsp', 'ws|websocket', 'f4']},
'vext': {'type': 'ordered', 'field': 'video_ext',
- 'order': ('mp4', 'webm', 'flv', '', 'none'),
- 'order_free': ('webm', 'mp4', 'flv', '', 'none')},
- 'aext': {'type': 'ordered', 'field': 'audio_ext',
- 'order': ('m4a', 'aac', 'mp3', 'ogg', 'opus', 'webm', '', 'none'),
- 'order_free': ('ogg', 'opus', 'webm', 'mp3', 'm4a', 'aac', '', 'none')},
+ 'order': ('mp4', 'mov', 'webm', 'flv', '', 'none'),
+ 'order_free': ('webm', 'mp4', 'mov', 'flv', '', 'none')},
+ 'aext': {'type': 'ordered', 'regex': True, 'field': 'audio_ext',
+ 'order': ('m4a', 'aac', 'mp3', 'ogg', 'opus', 'web[am]', '', 'none'),
+ 'order_free': ('ogg', 'opus', 'web[am]', 'mp3', 'm4a', 'aac', '', 'none')},
'hidden': {'visible': False, 'forced': True, 'type': 'extractor', 'max': -1000},
'aud_or_vid': {'visible': False, 'forced': True, 'type': 'multiple',
'field': ('vcodec', 'acodec'),
# if format.get('preference') is None and format.get('ext') in ('f4f', 'f4m'): # Not supported?
# format['preference'] = -1000
+ if format.get('preference') is None and format.get('ext') == 'flv' and re.match('[hx]265|he?vc?', format.get('vcodec') or ''):
+ # HEVC-over-FLV is out-of-spec by FLV's original spec
+ # ref. https://trac.ffmpeg.org/ticket/6389
+ # ref. https://github.com/yt-dlp/yt-dlp/pull/5821
+ format['preference'] = -100
+
# Determine missing bitrates
if format.get('tbr') is None:
if format.get('vbr') is not None and format.get('abr') is not None:
# Deprecated
has_certifi = bool(certifi)
has_websockets = bool(websockets)
+
+
+def load_plugins(name, suffix, namespace):
+ from .plugins import load_plugins
+ ret = load_plugins(name, suffix)
+ namespace.update(ret)
+ return ret
projects / yt-dlp.git / blobdiff