import traceback
import unicodedata
import urllib.request
-from string import ascii_letters
+from string import Formatter, ascii_letters
from .cache import Cache
from .compat import compat_os_name, compat_shlex_quote
from .downloader import FFmpegFD, get_suitable_downloader, shorten_protocol_name
from .downloader.rtmp import rtmpdump_version
from .extractor import gen_extractor_classes, get_info_extractor
+from .extractor.common import UnsupportedURLIE
from .extractor.openload import PhantomJSwrapper
from .minicurses import format_text
-from .postprocessor import _PLUGIN_CLASSES as plugin_postprocessors
+from .plugins import directories as plugin_directories
+from .postprocessor import _PLUGIN_CLASSES as plugin_pps
from .postprocessor import (
EmbedThumbnailPP,
FFmpegFixupDuplicateMoovPP,
get_postprocessor,
)
from .postprocessor.ffmpeg import resolve_mapping as resolve_recode_mapping
-from .update import detect_variant
+from .update import REPOSITORY, current_git_head, detect_variant
from .utils import (
DEFAULT_OUTTMPL,
IDENTITY,
EntryNotInPlaylist,
ExistingVideoReached,
ExtractorError,
+ FormatSorter,
GeoRestrictedError,
HEADRequest,
ISO3166Utils,
args_to_str,
bug_reports_message,
date_from_str,
+ deprecation_warning,
determine_ext,
determine_protocol,
encode_compat_str,
format_decimal_suffix,
format_field,
formatSeconds,
+ get_compatible_ext,
get_domain,
int_or_none,
iri_to_uri,
+ is_path_like,
join_nonempty,
locked_file,
+ make_archive_id,
make_dir,
make_HTTPS_handler,
merge_headers,
network_exceptions,
number_of_digits,
orderedSet,
+ orderedSet_from_options,
parse_filesize,
preferredencoding,
prepend_extension,
timetuple_from_msec,
to_high_limit_path,
traverse_obj,
+ try_call,
try_get,
url_basename,
variadic,
write_json_file,
write_string,
)
-from .version import RELEASE_GIT_HEAD, __version__
+from .version import CHANNEL, RELEASE_GIT_HEAD, VARIANT, __version__
if compat_os_name == 'nt':
import ctypes
Default is 'only_download' for CLI, but False for API
skip_playlist_after_errors: Number of allowed failures until the rest of
the playlist is skipped
- force_generic_extractor: Force downloader to use the generic extractor
+ allowed_extractors: List of regexes to match against extractor names that are allowed
overwrites: Overwrite all video and metadata files if True,
overwrite only non-video files if None
and don't overwrite any file if False
matchtitle: Download only matching titles.
rejecttitle: Reject downloads for matching titles.
logger: Log messages to a logging.Logger instance.
- logtostderr: Log messages to stderr instead of stdout.
- consoletitle: Display progress in console window's titlebar.
+ logtostderr: Print everything to stderr instead of stdout.
+ consoletitle: Display progress in console window's titlebar.
writedescription: Write the video description to a .description file
writeinfojson: Write the video description to a .info.json file
clean_infojson: Remove private fields from the infojson
subtitleslangs: List of languages of the subtitles to download (can be regex).
The list may contain "all" to refer to all the available
subtitles. The language can be prefixed with a "-" to
- exclude it from the requested languages. Eg: ['all', '-live_chat']
+ exclude it from the requested languages, e.g. ['all', '-live_chat']
keepvideo: Keep the video file after post-processing
daterange: A DateRange object, download only if the upload_date is in the range.
skip_download: Skip the actual download of the video file
downloaded.
Videos without view count information are always
downloaded. None for no limit.
- download_archive: File name of a file where all downloads are recorded.
- Videos already present in the file are not downloaded
- again.
+ download_archive: A set, or the name of a file where all downloads are recorded.
+ Videos already present in the file are not downloaded again.
break_on_existing: Stop the download process after attempting to download a
file that is in the archive.
- break_on_reject: Stop the download process when encountering a video that
- has been filtered out.
break_per_url: Whether break_on_reject and break_on_existing
should act on each input URL as opposed to for the entire queue
cookiefile: File name or text stream from where cookies should be read and dumped to
cookiesfrombrowser: A tuple containing the name of the browser, the profile
- name/pathfrom where cookies are loaded, and the name of the
- keyring. Eg: ('chrome', ) or ('vivaldi', 'default', 'BASICTEXT')
+ name/path from where cookies are loaded, the name of the keyring,
+ and the container name, e.g. ('chrome', ) or
+ ('vivaldi', 'default', 'BASICTEXT') or ('firefox', 'default', None, 'Meta')
legacyserverconnect: Explicitly allow HTTPS connection to servers that do not
support RFC 5746 secure renegotiation
nocheckcertificate: Do not verify SSL certificates
If not provided and the key is encrypted, yt-dlp will ask interactively
prefer_insecure: Use HTTP instead of HTTPS to retrieve information.
(Only supported by some extractors)
+ enable_file_urls: Enable file:// URLs. This is disabled by default for security reasons.
http_headers: A dictionary of custom headers to be used for all requests
proxy: URL of the proxy server to use
geo_verification_proxy: URL of the proxy to use for IP address verification
Progress hooks are guaranteed to be called at least twice
(with status "started" and "finished") if the processing is successful.
- merge_output_format: Extension to use when merging formats.
+ merge_output_format: "/" separated list of extensions to use when merging formats.
final_ext: Expected final extension; used to detect when the file was
already downloaded and converted
fixup: Automatically correct known faults of the file.
- If it returns None, the video is downloaded.
- If it returns utils.NO_DEFAULT, the user is interactively
asked whether to download the video.
+ - Raise utils.DownloadCancelled(msg) to abort remaining
+ downloads when a video is rejected.
match_filter_func in utils.py is one example for this.
no_color: Do not emit color codes in output.
geo_bypass: Bypass geographic restriction via faking X-Forwarded-For
* index: Section number (Optional)
force_keyframes_at_cuts: Re-encode the video when downloading ranges to get precise cuts
noprogress: Do not print the progress bar
+ live_from_start: Whether to download livestreams videos from the start
The following parameters are not used by YoutubeDL itself, they are used by
the downloader (see yt_dlp/downloader/common.py):
discontinuities such as ad breaks (default: False)
extractor_args: A dictionary of arguments to be passed to the extractors.
See "EXTRACTOR ARGUMENTS" for details.
- Eg: {'youtube': {'skip': ['dash', 'hls']}}
+ E.g. {'youtube': {'skip': ['dash', 'hls']}}
mark_watched: Mark videos watched (even with --simulate). Only for YouTube
The following options are deprecated and may be removed in the future:
+ break_on_reject: Stop the download process when encountering a video that
+ has been filtered out.
+ - `raise DownloadCancelled(msg)` in match_filter instead
+ force_generic_extractor: Force downloader to use the generic extractor
+ - Use allowed_extractors = ['generic', 'default']
playliststart: - Use playlist_items
Playlist item to start at.
playlistend: - Use playlist_items
"""
_NUMERIC_FIELDS = {
- 'width', 'height', 'tbr', 'abr', 'asr', 'vbr', 'fps', 'filesize', 'filesize_approx',
+ 'width', 'height', 'asr', 'audio_channels', 'fps',
+ 'tbr', 'abr', 'vbr', 'filesize', 'filesize_approx',
'timestamp', 'release_timestamp',
'duration', 'view_count', 'like_count', 'dislike_count', 'repost_count',
'average_rating', 'comment_count', 'age_limit',
_format_fields = {
# NB: Keep in sync with the docstring of extractor/common.py
'url', 'manifest_url', 'manifest_stream_number', 'ext', 'format', 'format_id', 'format_note',
- 'width', 'height', 'resolution', 'dynamic_range', 'tbr', 'abr', 'acodec', 'asr',
- 'vbr', 'fps', 'vcodec', 'container', 'filesize', 'filesize_approx',
+ 'width', 'height', 'aspect_ratio', 'resolution', 'dynamic_range', 'tbr', 'abr', 'acodec', 'asr', 'audio_channels',
+ 'vbr', 'fps', 'vcodec', 'container', 'filesize', 'filesize_approx', 'rows', 'columns',
'player_url', 'protocol', 'fragment_base_url', 'fragments', 'is_from_start',
'preference', 'language', 'language_preference', 'quality', 'source_preference',
- 'http_headers', 'stretched_ratio', 'no_resume', 'has_drm', 'downloader_options',
+ 'http_headers', 'stretched_ratio', 'no_resume', 'has_drm', 'extra_param_to_segment_url', 'hls_aes', 'downloader_options',
'page_url', 'app', 'play_path', 'tc_url', 'flash_version', 'rtmp_live', 'rtmp_conn', 'rtmp_protocol', 'rtmp_real_time'
}
_format_selection_exts = {
self._playlist_urls = set()
self.cache = Cache(self)
- windows_enable_vt_mode()
stdout = sys.stderr if self.params.get('logtostderr') else sys.stdout
self._out_files = Namespace(
out=stdout,
console=None if compat_os_name == 'nt' else next(
filter(supports_terminal_sequences, (sys.stderr, sys.stdout)), None)
)
+
+ try:
+ windows_enable_vt_mode()
+ except Exception as e:
+ self.write_debug(f'Failed to enable VT mode: {e}')
+
self._allow_colors = Namespace(**{
type_: not self.params.get('no_color') and supports_terminal_sequences(stream)
for type_, stream in self._out_files.items_ if type_ != 'console'
'\n You will no longer receive updates on this version')
if current_version < MIN_SUPPORTED:
msg = 'Python version %d.%d is no longer supported'
- self.deprecation_warning(
+ self.deprecated_feature(
f'{msg}! Please update to Python %d.%d or above' % (*current_version, *MIN_RECOMMENDED))
if self.params.get('allow_unplayable_formats'):
' If you experience any issues while using this option, '
f'{self._format_err("DO NOT", self.Styles.ERROR)} open a bug report')
+ if self.params.get('bidi_workaround', False):
+ try:
+ import pty
+ master, slave = pty.openpty()
+ width = shutil.get_terminal_size().columns
+ width_args = [] if width is None else ['-w', str(width)]
+ sp_kwargs = {'stdin': subprocess.PIPE, 'stdout': slave, 'stderr': self._out_files.error}
+ try:
+ self._output_process = Popen(['bidiv'] + width_args, **sp_kwargs)
+ except OSError:
+ self._output_process = Popen(['fribidi', '-c', 'UTF-8'] + width_args, **sp_kwargs)
+ self._output_channel = os.fdopen(master, 'rb')
+ except OSError as ose:
+ if ose.errno == errno.ENOENT:
+ self.report_warning(
+ 'Could not find fribidi executable, ignoring --bidi-workaround. '
+ 'Make sure that fribidi is an executable file in one of the directories in your $PATH.')
+ else:
+ raise
+
+ self.params['compat_opts'] = set(self.params.get('compat_opts', ()))
+ if auto_init and auto_init != 'no_verbose_header':
+ self.print_debug_header()
+
def check_deprecated(param, option, suggestion):
if self.params.get(param) is not None:
self.report_warning(f'{option} is deprecated. Use {suggestion} instead')
for msg in self.params.get('_warnings', []):
self.report_warning(msg)
for msg in self.params.get('_deprecation_warnings', []):
- self.deprecation_warning(msg)
+ self.deprecated_feature(msg)
- self.params['compat_opts'] = set(self.params.get('compat_opts', ()))
if 'list-formats' in self.params['compat_opts']:
self.params['listformats_table'] = False
else:
self.params['nooverwrites'] = not self.params['overwrites']
+ if self.params.get('simulate') is None and any((
+ self.params.get('list_thumbnails'),
+ self.params.get('listformats'),
+ self.params.get('listsubtitles'),
+ )):
+ self.params['simulate'] = 'list_only'
+
self.params.setdefault('forceprint', {})
self.params.setdefault('print_to_file', {})
if not isinstance(params['forceprint'], dict):
self.params['forceprint'] = {'video': params['forceprint']}
- if self.params.get('bidi_workaround', False):
- try:
- import pty
- master, slave = pty.openpty()
- width = shutil.get_terminal_size().columns
- width_args = [] if width is None else ['-w', str(width)]
- sp_kwargs = {'stdin': subprocess.PIPE, 'stdout': slave, 'stderr': self._out_files.error}
- try:
- self._output_process = Popen(['bidiv'] + width_args, **sp_kwargs)
- except OSError:
- self._output_process = Popen(['fribidi', '-c', 'UTF-8'] + width_args, **sp_kwargs)
- self._output_channel = os.fdopen(master, 'rb')
- except OSError as ose:
- if ose.errno == errno.ENOENT:
- self.report_warning(
- 'Could not find fribidi executable, ignoring --bidi-workaround. '
- 'Make sure that fribidi is an executable file in one of the directories in your $PATH.')
- else:
- raise
-
if auto_init:
- if auto_init != 'no_verbose_header':
- self.print_debug_header()
self.add_default_info_extractors()
if (sys.platform != 'win32'
def preload_download_archive(fn):
"""Preload the archive, if any is specified"""
+ archive = set()
if fn is None:
- return False
+ return archive
+ elif not is_path_like(fn):
+ return fn
+
self.write_debug(f'Loading archive file {fn!r}')
try:
with locked_file(fn, 'r', encoding='utf-8') as archive_file:
for line in archive_file:
- self.archive.add(line.strip())
+ archive.add(line.strip())
except OSError as ioe:
if ioe.errno != errno.ENOENT:
raise
- return False
- return True
+ return archive
- self.archive = set()
- preload_download_archive(self.params.get('download_archive'))
+ self.archive = preload_download_archive(self.params.get('download_archive'))
def warn_if_short_id(self, argv):
# short YouTube ID starting with dash?
self._ies_instances[ie_key] = ie
ie.set_downloader(self)
- def _get_info_extractor_class(self, ie_key):
- ie = self._ies.get(ie_key)
- if ie is None:
- ie = get_info_extractor(ie_key)
- self.add_info_extractor(ie)
- return ie
-
def get_info_extractor(self, ie_key):
"""
Get an instance of an IE with name ie_key, it will try to get one from
"""
Add the InfoExtractors returned by gen_extractors to the end of the list
"""
- for ie in gen_extractor_classes():
- self.add_info_extractor(ie)
+ all_ies = {ie.IE_NAME.lower(): ie for ie in gen_extractor_classes()}
+ all_ies['end'] = UnsupportedURLIE()
+ try:
+ ie_names = orderedSet_from_options(
+ self.params.get('allowed_extractors', ['default']), {
+ 'all': list(all_ies),
+ 'default': [name for name, ie in all_ies.items() if ie._ENABLED],
+ }, use_regex=True)
+ except re.error as e:
+ raise ValueError(f'Wrong regex for allowed_extractors: {e.pattern}')
+ for name in ie_names:
+ self.add_info_extractor(all_ies[name])
+ self.write_debug(f'Loaded {len(ie_names)} extractors')
def add_post_processor(self, pp, when='post_process'):
"""Add a PostProcessor object to the end of the chain."""
def to_stdout(self, message, skip_eol=False, quiet=None):
"""Print message to stdout"""
if quiet is not None:
- self.deprecation_warning('"YoutubeDL.to_stdout" no longer accepts the argument quiet. Use "YoutubeDL.to_screen" instead')
+ self.deprecation_warning('"YoutubeDL.to_stdout" no longer accepts the argument quiet. '
+ 'Use "YoutubeDL.to_screen" instead')
if skip_eol is not False:
- self.deprecation_warning('"YoutubeDL.to_stdout" no longer accepts the argument skip_eol. Use "YoutubeDL.to_screen" instead')
+ self.deprecation_warning('"YoutubeDL.to_stdout" no longer accepts the argument skip_eol. '
+ 'Use "YoutubeDL.to_screen" instead')
self._write_string(f'{self._bidi_workaround(message)}\n', self._out_files.out)
- def to_screen(self, message, skip_eol=False, quiet=None):
+ def to_screen(self, message, skip_eol=False, quiet=None, only_once=False):
"""Print message to screen if not in quiet mode"""
if self.params.get('logger'):
self.params['logger'].debug(message)
return
self._write_string(
'%s%s' % (self._bidi_workaround(message), ('' if skip_eol else '\n')),
- self._out_files.screen)
+ self._out_files.screen, only_once=only_once)
def to_stderr(self, message, only_once=False):
"""Print message to stderr"""
return
self.to_stderr(f'{self._format_err("WARNING:", self.Styles.WARNING)} {message}', only_once)
- def deprecation_warning(self, message):
+ def deprecation_warning(self, message, *, stacklevel=0):
+ deprecation_warning(
+ message, stacklevel=stacklevel + 1, printer=self.report_error, is_error=False)
+
+ def deprecated_feature(self, message):
if self.params.get('logger') is not None:
- self.params['logger'].warning(f'DeprecationWarning: {message}')
- else:
- self.to_stderr(f'{self._format_err("DeprecationWarning:", self.Styles.ERROR)} {message}', True)
+ self.params['logger'].warning(f'Deprecated Feature: {message}')
+ self.to_stderr(f'{self._format_err("Deprecated Feature:", self.Styles.ERROR)} {message}', True)
def report_error(self, message, *args, **kwargs):
'''
def get_output_path(self, dir_type='', filename=None):
paths = self.params.get('paths', {})
- assert isinstance(paths, dict)
+ assert isinstance(paths, dict), '"paths" parameter must be a dictionary'
path = os.path.join(
expand_path(paths.get('home', '').strip()),
expand_path(paths.get(dir_type, '').strip()) if dir_type else '',
# correspondingly that is not what we want since we need to keep
# '%%' intact for template dict substitution step. Working around
# with boundary-alike separator hack.
- sep = ''.join([random.choice(ascii_letters) for _ in range(32)])
+ sep = ''.join(random.choices(ascii_letters, k=32))
outtmpl = outtmpl.replace('%%', f'%{sep}%').replace('$$', f'${sep}$')
# outtmpl should be expand_path'ed before template dict substitution
# because meta fields may contain env variables we don't want to
- # be expanded. For example, for outtmpl "%(title)s.%(ext)s" and
+ # be expanded. E.g. for outtmpl "%(title)s.%(ext)s" and
# title "Hello $PATH", we don't want `$PATH` to be expanded.
return expand_path(outtmpl).replace(sep, '')
'-': float.__sub__,
}
# Field is of the form key1.key2...
- # where keys (except first) can be string, int or slice
- FIELD_RE = r'\w*(?:\.(?:\w+|{num}|{num}?(?::{num}?){{1,2}}))*'.format(num=r'(?:-?\d+)')
+ # where keys (except first) can be string, int, slice or "{field, ...}"
+ FIELD_INNER_RE = r'(?:\w+|%(num)s|%(num)s?(?::%(num)s?){1,2})' % {'num': r'(?:-?\d+)'}
+ FIELD_RE = r'\w*(?:\.(?:%(inner)s|{%(field)s(?:,%(field)s)*}))*' % {
+ 'inner': FIELD_INNER_RE,
+ 'field': rf'\w*(?:\.{FIELD_INNER_RE})*'
+ }
MATH_FIELD_RE = rf'(?:{FIELD_RE}|-?{NUMBER_RE})'
MATH_OPERATORS_RE = r'(?:%s)' % '|'.join(map(re.escape, MATH_FUNCTIONS.keys()))
- INTERNAL_FORMAT_RE = re.compile(rf'''(?x)
+ INTERNAL_FORMAT_RE = re.compile(rf'''(?xs)
(?P<negate>-)?
(?P<fields>{FIELD_RE})
(?P<maths>(?:{MATH_OPERATORS_RE}{MATH_FIELD_RE})*)
(?:\|(?P<default>.*?))?
)$''')
- def _traverse_infodict(k):
- k = k.split('.')
- if k[0] == '':
- k.pop(0)
- return traverse_obj(info_dict, k, is_user_input=True, traverse_string=True)
+ def _traverse_infodict(fields):
+ fields = [f for x in re.split(r'\.({.+?})\.?', fields)
+ for f in ([x] if x.startswith('{') else x.split('.'))]
+ for i in (0, -1):
+ if fields and not fields[i]:
+ fields.pop(i)
+
+ for i, f in enumerate(fields):
+ if not f.startswith('{'):
+ continue
+ assert f.endswith('}'), f'No closing brace for {f} in {fields}'
+ fields[i] = {k: k.split('.') for k in f[1:-1].split(',')}
+
+ return traverse_obj(info_dict, fields, is_user_input=True, traverse_string=True)
def get_value(mdict):
# Object traversal
return list(obj)
return repr(obj)
+ class _ReplacementFormatter(Formatter):
+ def get_field(self, field_name, args, kwargs):
+ if field_name.isdigit():
+ return args[0], -1
+ raise ValueError('Unsupported field')
+
+ replacement_formatter = _ReplacementFormatter()
+
def create_key(outer_mobj):
if not outer_mobj.group('has_key'):
return outer_mobj.group(0)
if fmt == 's' and value is not None and key in field_size_compat_map.keys():
fmt = f'0{field_size_compat_map[key]:d}d'
- value = default if value is None else value if replacement is None else replacement
+ if value is None:
+ value = default
+ elif replacement is not None:
+ try:
+ value = replacement_formatter.format(replacement, value)
+ except ValueError:
+ value = na
flags = outer_mobj.group('conversion') or ''
str_fmt = f'{fmt[:-1]}s'
delim = '\n' if '#' in flags else ', '
value, fmt = delim.join(map(str, variadic(value, allowed_types=(str, bytes)))), str_fmt
elif fmt[-1] == 'j': # json
- value, fmt = json.dumps(value, default=_dumpjson_default, indent=4 if '#' in flags else None), str_fmt
+ value, fmt = json.dumps(
+ value, default=_dumpjson_default,
+ indent=4 if '#' in flags else None, ensure_ascii='+' not in flags), str_fmt
elif fmt[-1] == 'h': # html
- value, fmt = escapeHTML(value), str_fmt
+ value, fmt = escapeHTML(str(value)), str_fmt
elif fmt[-1] == 'q': # quoted
value = map(str, variadic(value) if '#' in flags else [value])
value, fmt = ' '.join(map(compat_shlex_quote, value)), str_fmt
return self.get_output_path(dir_type, filename)
def _match_entry(self, info_dict, incomplete=False, silent=False):
- """ Returns None if the file should be downloaded """
+ """Returns None if the file should be downloaded"""
+ _type = info_dict.get('_type', 'video')
+ assert incomplete or _type == 'video', 'Only video result can be considered complete'
video_title = info_dict.get('title', info_dict.get('id', 'entry'))
def check_filter():
+ if _type in ('playlist', 'multi_video'):
+ return
+ elif _type in ('url', 'url_transparent') and not try_call(
+ lambda: self.get_info_extractor(info_dict['ie_key']).is_single_video(info_dict['url'])):
+ return
+
if 'title' in info_dict:
# This can happen when we're just evaluating the playlist
title = info_dict['title']
if rejecttitle:
if re.search(rejecttitle, title, re.IGNORECASE):
return '"' + title + '" title matched reject pattern "' + rejecttitle + '"'
+
date = info_dict.get('upload_date')
if date is not None:
dateRange = self.params.get('daterange', DateRange())
return 'Skipping "%s" because it is age restricted' % video_title
match_filter = self.params.get('match_filter')
- if match_filter is not None:
+ if match_filter is None:
+ return None
+
+ cancelled = None
+ try:
try:
ret = match_filter(info_dict, incomplete=incomplete)
except TypeError:
# For backward compatibility
ret = None if incomplete else match_filter(info_dict)
- if ret is NO_DEFAULT:
- while True:
- filename = self._format_screen(self.prepare_filename(info_dict), self.Styles.FILENAME)
- reply = input(self._format_screen(
- f'Download "{filename}"? (Y/n): ', self.Styles.EMPHASIS)).lower().strip()
- if reply in {'y', ''}:
- return None
- elif reply == 'n':
- return f'Skipping {video_title}'
- elif ret is not None:
- return ret
- return None
+ except DownloadCancelled as err:
+ if err.msg is not NO_DEFAULT:
+ raise
+ ret, cancelled = err.msg, err
+
+ if ret is NO_DEFAULT:
+ while True:
+ filename = self._format_screen(self.prepare_filename(info_dict), self.Styles.FILENAME)
+ reply = input(self._format_screen(
+ f'Download "{filename}"? (Y/n): ', self.Styles.EMPHASIS)).lower().strip()
+ if reply in {'y', ''}:
+ return None
+ elif reply == 'n':
+ if cancelled:
+ raise type(cancelled)(f'Skipping {video_title}')
+ return f'Skipping {video_title}'
+ return ret
if self.in_download_archive(info_dict):
reason = '%s has already been recorded in the archive' % video_title
break_opt, break_err = 'break_on_existing', ExistingVideoReached
else:
- reason = check_filter()
- break_opt, break_err = 'break_on_reject', RejectedVideoReached
+ try:
+ reason = check_filter()
+ except DownloadCancelled as e:
+ reason, break_opt, break_err = e.msg, 'match_filter', type(e)
+ else:
+ break_opt, break_err = 'break_on_reject', RejectedVideoReached
if reason is not None:
if not silent:
self.to_screen('[download] ' + reason)
def extract_info(self, url, download=True, ie_key=None, extra_info=None,
process=True, force_generic_extractor=False):
"""
- Return a list with a dictionary for each video extracted.
+ Extract and return the information dictionary of the URL
Arguments:
- url -- URL to extract
+ @param url URL to extract
Keyword arguments:
- download -- whether to download videos during extraction
- ie_key -- extractor key hint
- extra_info -- dictionary containing the extra values to add to each result
- process -- whether to resolve all unresolved references (URLs, playlist items),
- must be True for download to work.
- force_generic_extractor -- force using the generic extractor
+ @param download Whether to download videos
+ @param process Whether to resolve all unresolved references (URLs, playlist items).
+ Must be True for download to work
+ @param ie_key Use only the extractor with this key
+
+ @param extra_info Dictionary containing the extra values to add to the info (For internal use only)
+ @force_generic_extractor Force using the generic extractor (Deprecated; use ie_key='Generic')
"""
if extra_info is None:
ie_key = 'Generic'
if ie_key:
- ies = {ie_key: self._get_info_extractor_class(ie_key)}
+ ies = {ie_key: self._ies[ie_key]} if ie_key in self._ies else {}
else:
ies = self._ies
- for ie_key, ie in ies.items():
+ for key, ie in ies.items():
if not ie.suitable(url):
continue
'and will probably not work.')
temp_id = ie.get_temp_id(url)
- if temp_id is not None and self.in_download_archive({'id': temp_id, 'ie_key': ie_key}):
- self.to_screen(f'[{ie_key}] {temp_id}: has already been recorded in the archive')
+ if temp_id is not None and self.in_download_archive({'id': temp_id, 'ie_key': key}):
+ self.to_screen(f'[{key}] {temp_id}: has already been recorded in the archive')
if self.params.get('break_on_existing', False):
raise ExistingVideoReached()
break
- return self.__extract_info(url, self.get_info_extractor(ie_key), download, extra_info, process)
+ return self.__extract_info(url, self.get_info_extractor(key), download, extra_info, process)
else:
- self.report_error('no suitable InfoExtractor for URL %s' % url)
+ extractors_restricted = self.params.get('allowed_extractors') not in (None, ['default'])
+ self.report_error(f'No suitable extractor{format_field(ie_key, None, " (%s)")} found for URL {url}',
+ tb=False if extractors_restricted else None)
def _handle_extraction_exceptions(func):
@functools.wraps(func)
result_type = ie_result.get('_type', 'video')
if result_type in ('url', 'url_transparent'):
- ie_result['url'] = sanitize_url(ie_result['url'])
- if ie_result.get('original_url'):
- extra_info.setdefault('original_url', ie_result['original_url'])
+ ie_result['url'] = sanitize_url(
+ ie_result['url'], scheme='http' if self.params.get('prefer_insecure') else 'https')
+ if ie_result.get('original_url') and not extra_info.get('original_url'):
+ extra_info = {'original_url': ie_result['original_url'], **extra_info}
extract_flat = self.params.get('extract_flat', False)
if ((extract_flat == 'in_playlist' and 'playlist' in extra_info)
self.add_default_extra_info(info_copy, ie, ie_result['url'])
self.add_extra_info(info_copy, extra_info)
info_copy, _ = self.pre_process(info_copy)
- self.__forced_printings(info_copy, self.prepare_filename(info_copy), incomplete=True)
+ self._fill_common_fields(info_copy, False)
+ self.__forced_printings(info_copy)
self._raise_pending_errors(info_copy)
if self.params.get('force_write_download_archive', False):
self.record_download_archive(info_copy)
elif result_type in ('playlist', 'multi_video'):
# Protect from infinite recursion due to recursively nested playlists
# (see https://github.com/ytdl-org/youtube-dl/issues/27833)
- webpage_url = ie_result['webpage_url']
- if webpage_url in self._playlist_urls:
+ webpage_url = ie_result.get('webpage_url') # Playlists maynot have webpage_url
+ if webpage_url and webpage_url in self._playlist_urls:
self.to_screen(
'[download] Skipping already downloaded playlist: %s'
% ie_result.get('title') or ie_result.get('id'))
}
if strict:
return info
+ if ie_result.get('webpage_url'):
+ info.update({
+ 'webpage_url': ie_result['webpage_url'],
+ 'webpage_url_basename': url_basename(ie_result['webpage_url']),
+ 'webpage_url_domain': get_domain(ie_result['webpage_url']),
+ })
return {
**info,
'playlist_index': 0,
- '__last_playlist_index': max(ie_result['requested_entries'] or (0, 0)),
+ '__last_playlist_index': max(ie_result.get('requested_entries') or (0, 0)),
'extractor': ie_result['extractor'],
- 'webpage_url': ie_result['webpage_url'],
- 'webpage_url_basename': url_basename(ie_result['webpage_url']),
- 'webpage_url_domain': get_domain(ie_result['webpage_url']),
'extractor_key': ie_result['extractor_key'],
}
# Better to do this after potentially exhausting entries
ie_result['playlist_count'] = all_entries.get_full_count()
- ie_copy = collections.ChainMap(
- ie_result, self._playlist_infodict(ie_result, n_entries=int_or_none(n_entries)))
+ extra = self._playlist_infodict(ie_result, n_entries=int_or_none(n_entries))
+ ie_copy = collections.ChainMap(ie_result, extra)
_infojson_written = False
write_playlist_files = self.params.get('allow_playlist_files', True)
elif self.params.get('playlistrandom'):
random.shuffle(entries)
- self.to_screen(f'[{ie_result["extractor"]}] Playlist {title}: Downloading {n_entries} videos'
+ self.to_screen(f'[{ie_result["extractor"]}] Playlist {title}: Downloading {n_entries} items'
f'{format_field(ie_result, "playlist_count", " of %s")}')
keep_resolved_entries = self.params.get('extract_flat') != 'discard'
if not lazy and 'playlist-index' in self.params.get('compat_opts', []):
playlist_index = ie_result['requested_entries'][i]
- extra = {
+ entry_copy = collections.ChainMap(entry, {
**common_info,
'n_entries': int_or_none(n_entries),
'playlist_index': playlist_index,
'playlist_autonumber': i + 1,
- }
+ })
- if self._match_entry(collections.ChainMap(entry, extra), incomplete=True) is not None:
+ if self._match_entry(entry_copy, incomplete=True) is not None:
+ # For compatabilty with youtube-dl. See https://github.com/yt-dlp/yt-dlp/issues/4369
+ resolved_entries[i] = (playlist_index, NO_DEFAULT)
continue
- self.to_screen('[download] Downloading video %s of %s' % (
+ self.to_screen('[download] Downloading item %s of %s' % (
self._format_screen(i + 1, self.Styles.ID), self._format_screen(n_entries, self.Styles.EMPHASIS)))
- entry_result = self.__process_iterable_entry(entry, download, extra)
+ entry_result = self.__process_iterable_entry(entry, download, collections.ChainMap({
+ 'playlist_index': playlist_index,
+ 'playlist_autonumber': i + 1,
+ }, extra))
if not entry_result:
failures += 1
if failures >= max_failures:
resolved_entries[i] = (playlist_index, entry_result)
# Update with processed data
- ie_result['requested_entries'], ie_result['entries'] = tuple(zip(*resolved_entries)) or ([], [])
+ ie_result['entries'] = [e for _, e in resolved_entries if e is not NO_DEFAULT]
+ ie_result['requested_entries'] = [i for i, e in resolved_entries if e is not NO_DEFAULT]
+ if ie_result['requested_entries'] == try_call(lambda: list(range(1, ie_result['playlist_count'] + 1))):
+ # Do not set for full playlist
+ ie_result.pop('requested_entries')
# Write the updated info to json
if _infojson_written is True and self._write_info_json(
'!=': operator.ne,
}
operator_rex = re.compile(r'''(?x)\s*
- (?P<key>width|height|tbr|abr|vbr|asr|filesize|filesize_approx|fps)\s*
+ (?P<key>[\w.-]+)\s*
(?P<op>%s)(?P<none_inclusive>\s*\?)?\s*
(?P<value>[0-9.]+(?:[kKmMgGtTpPeEzZyY]i?[Bb]?)?)\s*
''' % '|'.join(map(re.escape, OPERATORS.keys())))
filter_parts.append(string)
def _remove_unused_ops(tokens):
- # Remove operators that we don't use and join them with the surrounding strings
- # for example: 'mp4' '-' 'baseline' '-' '16x9' is converted to 'mp4-baseline-16x9'
+ # Remove operators that we don't use and join them with the surrounding strings.
+ # E.g. 'mp4' '-' 'baseline' '-' '16x9' is converted to 'mp4-baseline-16x9'
ALLOWED_OPS = ('/', '+', ',', '(', ')')
last_string, last_start, last_end, last_line = None, None, None, None
for type, string, start, end, line in tokens:
the_only_video = video_fmts[0] if len(video_fmts) == 1 else None
the_only_audio = audio_fmts[0] if len(audio_fmts) == 1 else None
- output_ext = self.params.get('merge_output_format')
- if not output_ext:
- if the_only_video:
- output_ext = the_only_video['ext']
- elif the_only_audio and not video_fmts:
- output_ext = the_only_audio['ext']
- else:
- output_ext = 'mkv'
+ output_ext = get_compatible_ext(
+ vcodecs=[f.get('vcodec') for f in video_fmts],
+ acodecs=[f.get('acodec') for f in audio_fmts],
+ vexts=[f['ext'] for f in video_fmts],
+ aexts=[f['ext'] for f in audio_fmts],
+ preferences=(try_call(lambda: self.params['merge_output_format'].split('/'))
+ or self.params.get('prefer_free_formats') and ('webm', 'mkv')))
filtered = lambda *keys: filter(None, (traverse_obj(fmt, *keys) for fmt in formats_info))
'vcodec': the_only_video.get('vcodec'),
'vbr': the_only_video.get('vbr'),
'stretched_ratio': the_only_video.get('stretched_ratio'),
+ 'aspect_ratio': the_only_video.get('aspect_ratio'),
})
if the_only_audio:
'acodec': the_only_audio.get('acodec'),
'abr': the_only_audio.get('abr'),
'asr': the_only_audio.get('asr'),
+ 'audio_channels': the_only_audio.get('audio_channels')
})
return new_dict
else:
info_dict['thumbnails'] = thumbnails
- def _fill_common_fields(self, info_dict, is_video=True):
+ def _fill_common_fields(self, info_dict, final=True):
# TODO: move sanitization here
- if is_video:
- # playlists are allowed to lack "title"
- title = info_dict.get('title', NO_DEFAULT)
- if title is NO_DEFAULT:
- raise ExtractorError('Missing "title" field in extractor result',
- video_id=info_dict['id'], ie=info_dict['extractor'])
- info_dict['fulltitle'] = title
+ if final:
+ title = info_dict['fulltitle'] = info_dict.get('title')
if not title:
if title == '':
self.write_debug('Extractor gave empty title. Creating a generic title')
for key in live_keys:
if info_dict.get(key) is None:
info_dict[key] = (live_status == key)
+ if live_status == 'post_live':
+ info_dict['was_live'] = True
# Auto generate title fields corresponding to the *_number fields when missing
# in order to always have clean titles. This is very common for TV series.
for field in ('chapter', 'season', 'episode'):
- if info_dict.get('%s_number' % field) is not None and not info_dict.get(field):
+ if final and info_dict.get('%s_number' % field) is not None and not info_dict.get(field):
info_dict[field] = '%s %d' % (field.capitalize(), info_dict['%s_number' % field])
def _raise_pending_errors(self, info):
if err:
self.report_error(err, tb=False)
+ def sort_formats(self, info_dict):
+ formats = self._get_formats(info_dict)
+ formats.sort(key=FormatSorter(
+ self, info_dict.get('_format_sort_fields') or []).calculate_preference)
+
def process_video_result(self, info_dict, download=True):
assert info_dict.get('_type', 'video') == 'video'
self._num_videos += 1
info_dict['requested_subtitles'] = self.process_subtitles(
info_dict['id'], subtitles, automatic_captions)
- if info_dict.get('formats') is None:
- # There's only one format available
- formats = [info_dict]
- else:
- formats = info_dict['formats']
+ formats = self._get_formats(info_dict)
+
+ # Backward compatibility with InfoExtractor._sort_formats
+ field_preference = (formats or [{}])[0].pop('__sort_fields', None)
+ if field_preference:
+ info_dict['_format_sort_fields'] = field_preference
# or None ensures --clean-infojson removes it
info_dict['_has_drm'] = any(f.get('has_drm') for f in formats) or None
if not self.params.get('allow_unplayable_formats'):
formats = [f for f in formats if not f.get('has_drm')]
- if info_dict['_has_drm'] and all(
- f.get('acodec') == f.get('vcodec') == 'none' for f in formats):
- self.report_warning(
- 'This video is DRM protected and only images are available for download. '
- 'Use --list-formats to see them')
+
+ if formats and all(f.get('acodec') == f.get('vcodec') == 'none' for f in formats):
+ self.report_warning(
+ f'{"This video is DRM protected and " if info_dict["_has_drm"] else ""}'
+ 'only images are available for download. Use --list-formats to see them'.capitalize())
get_from_start = not info_dict.get('is_live') or bool(self.params.get('live_from_start'))
if not get_from_start:
'--live-from-start is passed, but there are no formats that can be downloaded from the start. '
'If you want to download from the current time, use --no-live-from-start'))
- if not formats:
- self.raise_no_formats(info_dict)
-
def is_wellformed(f):
url = f.get('url')
if not url:
return True
# Filter out malformed formats for better extraction robustness
- formats = list(filter(is_wellformed, formats))
+ formats = list(filter(is_wellformed, formats or []))
- formats_dict = {}
+ if not formats:
+ self.raise_no_formats(info_dict)
- # We check that all the formats have the format and format_id fields
- for i, format in enumerate(formats):
+ for format in formats:
sanitize_string_field(format, 'format_id')
sanitize_numeric_fields(format)
format['url'] = sanitize_url(format['url'])
+ if format.get('ext') is None:
+ format['ext'] = determine_ext(format['url']).lower()
+ if format.get('protocol') is None:
+ format['protocol'] = determine_protocol(format)
+ if format.get('resolution') is None:
+ format['resolution'] = self.format_resolution(format, default=None)
+ if format.get('dynamic_range') is None and format.get('vcodec') != 'none':
+ format['dynamic_range'] = 'SDR'
+ if format.get('aspect_ratio') is None:
+ format['aspect_ratio'] = try_call(lambda: round(format['width'] / format['height'], 2))
+ if (info_dict.get('duration') and format.get('tbr')
+ and not format.get('filesize') and not format.get('filesize_approx')):
+ format['filesize_approx'] = int(info_dict['duration'] * format['tbr'] * (1024 / 8))
+ format['http_headers'] = self._calc_headers(collections.ChainMap(format, info_dict))
+
+ # This is copied to http_headers by the above _calc_headers and can now be removed
+ if '__x_forwarded_for_ip' in info_dict:
+ del info_dict['__x_forwarded_for_ip']
+
+ self.sort_formats({
+ 'formats': formats,
+ '_format_sort_fields': info_dict.get('_format_sort_fields')
+ })
+
+ # Sanitize and group by format_id
+ formats_dict = {}
+ for i, format in enumerate(formats):
if not format.get('format_id'):
format['format_id'] = str(i)
else:
# Sanitize format_id from characters used in format selector expression
format['format_id'] = re.sub(r'[\s,/+\[\]()]', '_', format['format_id'])
- format_id = format['format_id']
- if format_id not in formats_dict:
- formats_dict[format_id] = []
- formats_dict[format_id].append(format)
+ formats_dict.setdefault(format['format_id'], []).append(format)
# Make sure all formats have unique format_id
common_exts = set(itertools.chain(*self._format_selection_exts.values()))
for i, format in enumerate(ambiguous_formats):
if ambigious_id:
format['format_id'] = '%s-%d' % (format_id, i)
- if format.get('ext') is None:
- format['ext'] = determine_ext(format['url']).lower()
# Ensure there is no conflict between id and ext in format selection
# See https://github.com/yt-dlp/yt-dlp/issues/1282
if format['format_id'] != format['ext'] and format['format_id'] in common_exts:
format['format_id'] = 'f%s' % format['format_id']
- for i, format in enumerate(formats):
- if format.get('format') is None:
- format['format'] = '{id} - {res}{note}'.format(
- id=format['format_id'],
- res=self.format_resolution(format),
- note=format_field(format, 'format_note', ' (%s)'),
- )
- if format.get('protocol') is None:
- format['protocol'] = determine_protocol(format)
- if format.get('resolution') is None:
- format['resolution'] = self.format_resolution(format, default=None)
- if format.get('dynamic_range') is None and format.get('vcodec') != 'none':
- format['dynamic_range'] = 'SDR'
- if (info_dict.get('duration') and format.get('tbr')
- and not format.get('filesize') and not format.get('filesize_approx')):
- format['filesize_approx'] = int(info_dict['duration'] * format['tbr'] * (1024 / 8))
-
- # Add HTTP headers, so that external programs can use them from the
- # json output
- full_format_info = info_dict.copy()
- full_format_info.update(format)
- format['http_headers'] = self._calc_headers(full_format_info)
- # Remove private housekeeping stuff
- if '__x_forwarded_for_ip' in info_dict:
- del info_dict['__x_forwarded_for_ip']
+ if format.get('format') is None:
+ format['format'] = '{id} - {res}{note}'.format(
+ id=format['format_id'],
+ res=self.format_resolution(format),
+ note=format_field(format, 'format_note', ' (%s)'),
+ )
if self.params.get('check_formats') is True:
formats = LazyList(self._check_formats(formats[::-1]), reverse=True)
info_dict, _ = self.pre_process(info_dict, 'after_filter')
# The pre-processors may have modified the formats
- formats = info_dict.get('formats', [info_dict])
+ formats = self._get_formats(info_dict)
- list_only = self.params.get('simulate') is None and (
- self.params.get('list_thumbnails') or self.params.get('listformats') or self.params.get('listsubtitles'))
+ list_only = self.params.get('simulate') == 'list_only'
interactive_format_selection = not list_only and self.format_selector == '-'
if self.params.get('list_thumbnails'):
self.list_thumbnails(info_dict)
self.list_formats(info_dict)
if list_only:
# Without this printing, -F --print-json will not work
- self.__forced_printings(info_dict, self.prepare_filename(info_dict), incomplete=True)
+ self.__forced_printings(info_dict)
return info_dict
format_selector = self.format_selector
# Process what we can, even without any available formats.
formats_to_download = [{}]
- requested_ranges = self.params.get('download_ranges')
- if requested_ranges:
- requested_ranges = tuple(requested_ranges(info_dict, self))
-
+ requested_ranges = tuple(self.params.get('download_ranges', lambda *_: [{}])(info_dict, self))
best_format, downloaded_formats = formats_to_download[-1], []
if download:
- if best_format:
+ if best_format and requested_ranges:
def to_screen(*msg):
self.to_screen(f'[info] {info_dict["id"]}: {" ".join(", ".join(variadic(m)) for m in msg)}')
to_screen(f'Downloading {len(formats_to_download)} format(s):',
(f['format_id'] for f in formats_to_download))
- if requested_ranges:
+ if requested_ranges != ({}, ):
to_screen(f'Downloading {len(requested_ranges)} time ranges:',
- (f'{int(c["start_time"])}-{int(c["end_time"])}' for c in requested_ranges))
+ (f'{c["start_time"]:.1f}-{c["end_time"]:.1f}' for c in requested_ranges))
max_downloads_reached = False
- for fmt, chapter in itertools.product(formats_to_download, requested_ranges or [{}]):
+ for fmt, chapter in itertools.product(formats_to_download, requested_ranges):
new_info = self._copy_infodict(info_dict)
new_info.update(fmt)
offset, duration = info_dict.get('section_start') or 0, info_dict.get('duration') or float('inf')
+ end_time = offset + min(chapter.get('end_time', duration), duration)
if chapter or offset:
new_info.update({
'section_start': offset + chapter.get('start_time', 0),
- 'section_end': offset + min(chapter.get('end_time', duration), duration),
+ # duration may not be accurate. So allow deviations <1sec
+ 'section_end': end_time if end_time <= offset + duration + 1 else None,
'section_title': chapter.get('title'),
'section_number': chapter.get('index'),
})
if lang not in available_subs:
available_subs[lang] = cap_info
- if (not self.params.get('writesubtitles') and not
- self.params.get('writeautomaticsub') or not
- available_subs):
+ if not available_subs or (
+ not self.params.get('writesubtitles')
+ and not self.params.get('writeautomaticsub')):
return None
all_sub_langs = tuple(available_subs.keys())
if self.params.get('allsubtitles', False):
requested_langs = all_sub_langs
elif self.params.get('subtitleslangs', False):
- # A list is used so that the order of languages will be the same as
- # given in subtitleslangs. See https://github.com/yt-dlp/yt-dlp/issues/1041
- requested_langs = []
- for lang_re in self.params.get('subtitleslangs'):
- discard = lang_re[0] == '-'
- if discard:
- lang_re = lang_re[1:]
- if lang_re == 'all':
- if discard:
- requested_langs = []
- else:
- requested_langs.extend(all_sub_langs)
- continue
- current_langs = filter(re.compile(lang_re + '$').match, all_sub_langs)
- if discard:
- for lang in current_langs:
- while lang in requested_langs:
- requested_langs.remove(lang)
- else:
- requested_langs.extend(current_langs)
- requested_langs = orderedSet(requested_langs)
- elif normal_sub_langs:
- requested_langs = ['en'] if 'en' in normal_sub_langs else normal_sub_langs[:1]
+ try:
+ requested_langs = orderedSet_from_options(
+ self.params.get('subtitleslangs'), {'all': all_sub_langs}, use_regex=True)
+ except re.error as e:
+ raise ValueError(f'Wrong regex for subtitlelangs: {e.pattern}')
else:
- requested_langs = ['en'] if 'en' in all_sub_langs else all_sub_langs[:1]
+ requested_langs = LazyList(itertools.chain(
+ ['en'] if 'en' in normal_sub_langs else [],
+ filter(lambda f: f.startswith('en'), normal_sub_langs),
+ ['en'] if 'en' in all_sub_langs else [],
+ filter(lambda f: f.startswith('en'), all_sub_langs),
+ normal_sub_langs, all_sub_langs,
+ ))[:1]
if requested_langs:
- self.write_debug('Downloading subtitles: %s' % ', '.join(requested_langs))
+ self.to_screen(f'[info] {video_id}: Downloading subtitles: {", ".join(requested_langs)}')
formats_query = self.params.get('subtitlesformat', 'best')
formats_preference = formats_query.split('/') if formats_query else []
if info_dict is None:
return
info_copy = info_dict.copy()
+ info_copy.setdefault('filename', self.prepare_filename(info_dict))
+ if info_dict.get('requested_formats') is not None:
+ # For RTMP URLs, also include the playpath
+ info_copy['urls'] = '\n'.join(f['url'] + f.get('play_path', '') for f in info_dict['requested_formats'])
+ elif info_dict.get('url'):
+ info_copy['urls'] = info_dict['url'] + info_dict.get('play_path', '')
info_copy['formats_table'] = self.render_formats_table(info_dict)
info_copy['thumbnails_table'] = self.render_thumbnails_table(info_dict)
info_copy['subtitles_table'] = self.render_subtitles_table(info_dict.get('id'), info_dict.get('subtitles'))
info_copy['automatic_captions_table'] = self.render_subtitles_table(info_dict.get('id'), info_dict.get('automatic_captions'))
def format_tmpl(tmpl):
- mobj = re.match(r'\w+(=?)$', tmpl)
- if mobj and mobj.group(1):
- return f'{tmpl[:-1]} = %({tmpl[:-1]})r'
- elif mobj:
- return f'%({tmpl})s'
- return tmpl
+ mobj = re.fullmatch(r'([\w.:,]|-\d|(?P<dict>{([\w.:,]|-\d)+}))+=?', tmpl)
+ if not mobj:
+ return tmpl
+
+ fmt = '%({})s'
+ if tmpl.startswith('{'):
+ tmpl = f'.{tmpl}'
+ if tmpl.endswith('='):
+ tmpl, fmt = tmpl[:-1], '{0} = %({0})#j'
+ return '\n'.join(map(fmt.format, [tmpl] if mobj.group('dict') else tmpl.split(',')))
for tmpl in self.params['forceprint'].get(key, []):
self.to_stdout(self.evaluate_outtmpl(format_tmpl(tmpl), info_copy))
tmpl = format_tmpl(tmpl)
self.to_screen(f'[info] Writing {tmpl!r} to: {filename}')
if self._ensure_dir_exists(filename):
- with open(filename, 'a', encoding='utf-8') as f:
- f.write(self.evaluate_outtmpl(tmpl, info_copy) + '\n')
+ with open(filename, 'a', encoding='utf-8', newline='') as f:
+ f.write(self.evaluate_outtmpl(tmpl, info_copy) + os.linesep)
- def __forced_printings(self, info_dict, filename, incomplete):
+ return info_copy
+
+ def __forced_printings(self, info_dict, filename=None, incomplete=True):
def print_mandatory(field, actual_field=None):
if actual_field is None:
actual_field = field
and info_dict.get(field) is not None):
self.to_stdout(info_dict[field])
- info_dict = info_dict.copy()
- if filename is not None:
- info_dict['filename'] = filename
- if info_dict.get('requested_formats') is not None:
- # For RTMP URLs, also include the playpath
- info_dict['urls'] = '\n'.join(f['url'] + f.get('play_path', '') for f in info_dict['requested_formats'])
- elif info_dict.get('url'):
- info_dict['urls'] = info_dict['url'] + info_dict.get('play_path', '')
-
if (self.params.get('forcejson')
or self.params['forceprint'].get('video')
or self.params['print_to_file'].get('video')):
self.post_extract(info_dict)
- self._forceprint('video', info_dict)
+
+ if filename:
+ info_dict['filename'] = filename
+ info_dict = self._forceprint('video', info_dict)
print_mandatory('title')
print_mandatory('id')
if 'format' not in info_dict and 'ext' in info_dict:
info_dict['format'] = info_dict['ext']
- # This is mostly just for backward compatibility of process_info
- # As a side-effect, this allows for format-specific filters
if self._match_entry(info_dict) is not None:
info_dict['__write_download_archive'] = 'ignore'
return
# Does nothing under normal operation - for backward compatibility of process_info
self.post_extract(info_dict)
+
+ def replace_info_dict(new_info):
+ nonlocal info_dict
+ if new_info == info_dict:
+ return
+ info_dict.clear()
+ info_dict.update(new_info)
+
+ new_info, _ = self.pre_process(info_dict, 'video')
+ replace_info_dict(new_info)
self._num_downloads += 1
# info_dict['_filename'] needs to be set for backward compatibility
for link_type, should_write in write_links.items()):
return
- def replace_info_dict(new_info):
- nonlocal info_dict
- if new_info == info_dict:
- return
- info_dict.clear()
- info_dict.update(new_info)
-
new_info, files_to_move = self.pre_process(info_dict, 'before_dl', files_to_move)
replace_info_dict(new_info)
fd, success = None, True
if info_dict.get('protocol') or info_dict.get('url'):
fd = get_suitable_downloader(info_dict, self.params, to_stdout=temp_filename == '-')
- if fd is not FFmpegFD and (
+ if fd is not FFmpegFD and 'no-direct-merge' not in self.params['compat_opts'] and (
info_dict.get('section_start') or info_dict.get('section_end')):
msg = ('This format cannot be partially downloaded' if FFmpegFD.available()
else 'You have requested downloading the video partially, but ffmpeg is not installed')
return
if info_dict.get('requested_formats') is not None:
-
- def compatible_formats(formats):
- # TODO: some formats actually allow this (mkv, webm, ogg, mp4), but not all of them.
- video_formats = [format for format in formats if format.get('vcodec') != 'none']
- audio_formats = [format for format in formats if format.get('acodec') != 'none']
- if len(video_formats) > 2 or len(audio_formats) > 2:
- return False
-
- # Check extension
- exts = {format.get('ext') for format in formats}
- COMPATIBLE_EXTS = (
- {'mp3', 'mp4', 'm4a', 'm4p', 'm4b', 'm4r', 'm4v', 'ismv', 'isma'},
- {'webm'},
- )
- for ext_sets in COMPATIBLE_EXTS:
- if ext_sets.issuperset(exts):
- return True
- # TODO: Check acodec/vcodec
- return False
-
requested_formats = info_dict['requested_formats']
old_ext = info_dict['ext']
if self.params.get('merge_output_format') is None:
- if not compatible_formats(requested_formats):
- info_dict['ext'] = 'mkv'
- self.report_warning(
- 'Requested formats are incompatible for merge and will be merged into mkv')
if (info_dict['ext'] == 'webm'
and info_dict.get('thumbnails')
# check with type instead of pp_key, __name__, or isinstance
or info_dict.get('is_live') and self.params.get('hls_use_mpegts') is None,
'Possible MPEG-TS in MP4 container or malformed AAC timestamps',
FFmpegFixupM3u8PP)
- ffmpeg_fixup(info_dict.get('is_live') and downloader == 'DashSegmentsFD',
+ ffmpeg_fixup(info_dict.get('is_live') and downloader == 'dashsegments',
'Possible duplicate MOOV atoms', FFmpegFixupDuplicateMoovPP)
ffmpeg_fixup(downloader == 'web_socket_fragment', 'Malformed timestamps detected', FFmpegFixupTimestampPP)
self.to_screen(f'[info] {e}')
if not self.params.get('break_per_url'):
raise
+ self._num_downloads = 0
else:
if self.params.get('dump_single_json', False):
self.post_extract(res)
[info_filename], mode='r',
openhook=fileinput.hook_encoded('utf-8'))) as f:
# FileInput doesn't have a read method, we can't call json.load
- info = self.sanitize_info(json.loads('\n'.join(f)), self.params.get('clean_infojson', True))
- try:
- self.__download_wrapper(self.process_ie_result)(info, download=True)
- except (DownloadError, EntryNotInPlaylist, ReExtractInfo) as e:
- if not isinstance(e, EntryNotInPlaylist):
- self.to_stderr('\r')
- webpage_url = info.get('webpage_url')
- if webpage_url is not None:
+ infos = [self.sanitize_info(info, self.params.get('clean_infojson', True))
+ for info in variadic(json.loads('\n'.join(f)))]
+ for info in infos:
+ try:
+ self.__download_wrapper(self.process_ie_result)(info, download=True)
+ except (DownloadError, EntryNotInPlaylist, ReExtractInfo) as e:
+ if not isinstance(e, EntryNotInPlaylist):
+ self.to_stderr('\r')
+ webpage_url = info.get('webpage_url')
+ if webpage_url is None:
+ raise
self.report_warning(f'The info failed to download: {e}; trying with URL {webpage_url}')
- return self.download([webpage_url])
- else:
- raise
+ self.download([webpage_url])
return self._download_retcode
@staticmethod
return info_dict
info_dict.setdefault('epoch', int(time.time()))
info_dict.setdefault('_type', 'video')
+ info_dict.setdefault('_version', {
+ 'version': __version__,
+ 'current_git_head': current_git_head(),
+ 'release_git_head': RELEASE_GIT_HEAD,
+ 'repository': REPOSITORY,
+ })
if remove_private_keys:
reject = lambda k, v: v is None or k.startswith('__') or k in {
'requested_downloads', 'requested_formats', 'requested_subtitles', 'requested_entries',
'entries', 'filepath', '_filename', 'infojson_filename', 'original_url', 'playlist_autonumber',
+ '_format_sort_fields',
}
else:
reject = lambda k, v: False
return infodict
def run_all_pps(self, key, info, *, additional_pps=None):
- self._forceprint(key, info)
for pp in (additional_pps or []) + self._pps[key]:
info = self.run_pp(pp, info)
+ if key != 'video':
+ self._forceprint(key, info)
return info
def pre_process(self, ie_info, key='pre_process', files_to_move=None):
break
else:
return
- return f'{extractor.lower()} {video_id}'
+ return make_archive_id(extractor, video_id)
def in_download_archive(self, info_dict):
- fn = self.params.get('download_archive')
- if fn is None:
+ if not self.archive:
return False
- vid_id = self._make_archive_id(info_dict)
- if not vid_id:
- return False # Incomplete video information
-
- return vid_id in self.archive
+ vid_ids = [self._make_archive_id(info_dict)]
+ vid_ids.extend(info_dict.get('_old_archive_ids') or [])
+ return any(id_ in self.archive for id_ in vid_ids)
def record_download_archive(self, info_dict):
fn = self.params.get('download_archive')
return
vid_id = self._make_archive_id(info_dict)
assert vid_id
+
self.write_debug(f'Adding to archive: {vid_id}')
- with locked_file(fn, 'a', encoding='utf-8') as archive_file:
- archive_file.write(vid_id + '\n')
+ if is_path_like(fn):
+ with locked_file(fn, 'a', encoding='utf-8') as archive_file:
+ archive_file.write(vid_id + '\n')
self.archive.add(vid_id)
@staticmethod
res += '~' + format_bytes(fdict['filesize_approx'])
return res
- def render_formats_table(self, info_dict):
- if not info_dict.get('formats') and not info_dict.get('url'):
- return None
+ def _get_formats(self, info_dict):
+ if info_dict.get('formats') is None:
+ if info_dict.get('url') and info_dict.get('_type', 'video') == 'video':
+ return [info_dict]
+ return []
+ return info_dict['formats']
- formats = info_dict.get('formats', [info_dict])
+ def render_formats_table(self, info_dict):
+ formats = self._get_formats(info_dict)
+ if not formats:
+ return
if not self.params.get('listformats_table', True) is not False:
table = [
[
format_field(f, 'ext'),
self.format_resolution(f),
self._format_note(f)
- ] for f in formats if f.get('preference') is None or f['preference'] >= -1000]
+ ] for f in formats if (f.get('preference') or 0) >= -1000]
return render_table(['format code', 'extension', 'resolution', 'note'], table, extra_gap=1)
def simplified_codec(f, field):
format_field(f, func=self.format_resolution, ignore=('audio only', 'images')),
format_field(f, 'fps', '\t%d', func=round),
format_field(f, 'dynamic_range', '%s', ignore=(None, 'SDR')).replace('HDR', ''),
+ format_field(f, 'audio_channels', '\t%s'),
delim,
format_field(f, 'filesize', ' \t%s', func=format_bytes) + format_field(f, 'filesize_approx', '~\t%s', func=format_bytes),
format_field(f, 'tbr', '\t%dk', func=round),
format_field(f, 'asr', '\t%s', func=format_decimal_suffix),
join_nonempty(
self._format_out('UNSUPPORTED', 'light red') if f.get('ext') in ('f4f', 'f4m') else None,
+ self._format_out('DRM', 'light red') if f.get('has_drm') else None,
format_field(f, 'language', '[%s]'),
join_nonempty(format_field(f, 'format_note'),
format_field(f, 'container', ignore=(None, f.get('ext'))),
delim=' '),
] for f in formats if f.get('preference') is None or f['preference'] >= -1000]
header_line = self._list_format_headers(
- 'ID', 'EXT', 'RESOLUTION', '\tFPS', 'HDR', delim, '\tFILESIZE', '\tTBR', 'PROTO',
+ 'ID', 'EXT', 'RESOLUTION', '\tFPS', 'HDR', 'CH', delim, '\tFILESIZE', '\tTBR', 'PROTO',
delim, 'VCODEC', '\tVBR', 'ACODEC', '\tABR', '\tASR', 'MORE INFO')
return render_table(
return None
return render_table(
self._list_format_headers('ID', 'Width', 'Height', 'URL'),
- [[t.get('id'), t.get('width', 'unknown'), t.get('height', 'unknown'), t['url']] for t in thumbnails])
+ [[t.get('id'), t.get('width') or 'unknown', t.get('height') or 'unknown', t['url']] for t in thumbnails])
def render_subtitles_table(self, video_id, subtitles):
def _row(lang, formats):
if not self.params.get('verbose'):
return
+ from . import _IN_CLI # Must be delayed import
+
# These imports can be slow. So import them only as needed
from .extractor.extractors import _LAZY_LOADER
- from .extractor.extractors import _PLUGIN_CLASSES as plugin_extractors
+ from .extractor.extractors import (
+ _PLUGIN_CLASSES as plugin_ies,
+ _PLUGIN_OVERRIDES as plugin_ie_overrides
+ )
def get_encoding(stream):
ret = str(getattr(stream, 'encoding', 'missing (%s)' % type(stream).__name__))
write_debug = lambda msg: self._write_string(f'[debug] {msg}\n')
source = detect_variant()
+ if VARIANT not in (None, 'pip'):
+ source += '*'
+ klass = type(self)
write_debug(join_nonempty(
- 'yt-dlp version', __version__,
- f'[{RELEASE_GIT_HEAD}]' if RELEASE_GIT_HEAD else '',
+ f'{"yt-dlp" if REPOSITORY == "yt-dlp/yt-dlp" else REPOSITORY} version',
+ f'{CHANNEL}@{__version__}',
+ f'[{RELEASE_GIT_HEAD[:9]}]' if RELEASE_GIT_HEAD else '',
'' if source == 'unknown' else f'({source})',
+ '' if _IN_CLI else 'API' if klass == YoutubeDL else f'API:{self.__module__}.{klass.__qualname__}',
delim=' '))
+
+ if not _IN_CLI:
+ write_debug(f'params: {self.params}')
+
if not _LAZY_LOADER:
if os.environ.get('YTDLP_NO_LAZY_EXTRACTORS'):
write_debug('Lazy loading extractors is forcibly disabled')
else:
write_debug('Lazy loading extractors is disabled')
- if plugin_extractors or plugin_postprocessors:
- write_debug('Plugins: %s' % [
- '%s%s' % (klass.__name__, '' if klass.__name__ == name else f' as {name}')
- for name, klass in itertools.chain(plugin_extractors.items(), plugin_postprocessors.items())])
if self.params['compat_opts']:
write_debug('Compatibility options: %s' % ', '.join(self.params['compat_opts']))
- if source == 'source':
- try:
- stdout, _, _ = Popen.run(
- ['git', 'rev-parse', '--short', 'HEAD'],
- text=True, cwd=os.path.dirname(os.path.abspath(__file__)),
- stdout=subprocess.PIPE, stderr=subprocess.PIPE)
- if re.fullmatch('[0-9a-f]+', stdout.strip()):
- write_debug(f'Git HEAD: {stdout.strip()}')
- except Exception:
- with contextlib.suppress(Exception):
- sys.exc_clear()
-
+ if current_git_head():
+ write_debug(f'Git HEAD: {current_git_head()}')
write_debug(system_identifier())
exe_versions, ffmpeg_features = FFmpegPostProcessor.get_versions_and_features(self)
proxy_map.update(handler.proxies)
write_debug(f'Proxy map: {proxy_map}')
+ for plugin_type, plugins in {'Extractor': plugin_ies, 'Post-Processor': plugin_pps}.items():
+ display_list = ['%s%s' % (
+ klass.__name__, '' if klass.__name__ == name else f' as {name}')
+ for name, klass in plugins.items()]
+ if plugin_type == 'Extractor':
+ display_list.extend(f'{plugins[-1].IE_NAME.partition("+")[2]} ({parent.__name__})'
+ for parent, plugins in plugin_ie_overrides.items())
+ if not display_list:
+ continue
+ write_debug(f'{plugin_type} Plugins: {", ".join(sorted(display_list))}')
+
+ plugin_dirs = plugin_directories()
+ if plugin_dirs:
+ write_debug(f'Plugin directories: {plugin_dirs}')
+
# Not implemented
if False and self.params.get('call_home'):
ipaddr = self.urlopen('https://yt-dl.org/ip').read().decode()
# https://github.com/ytdl-org/youtube-dl/issues/8227)
file_handler = urllib.request.FileHandler()
- def file_open(*args, **kwargs):
- raise urllib.error.URLError('file:// scheme is explicitly disabled in yt-dlp for security reasons')
- file_handler.file_open = file_open
+ if not self.params.get('enable_file_urls'):
+ def file_open(*args, **kwargs):
+ raise urllib.error.URLError(
+ 'file:// URLs are explicitly disabled in yt-dlp for security reasons. '
+ 'Use --enable-file-urls to enable at your own risk.')
+ file_handler.file_open = file_open
opener = urllib.request.build_opener(
proxy_handler, https_handler, cookie_processor, ydlh, redirect_handler, data_handler, file_handler)
elif not self.params.get('overwrites', True) and os.path.exists(descfn):
self.to_screen(f'[info] {label.title()} description is already present')
elif ie_result.get('description') is None:
- self.report_warning(f'There\'s no {label} description to write')
+ self.to_screen(f'[info] There\'s no {label} description to write')
return False
else:
try:
''' Write subtitles to file and return list of (sub_filename, final_sub_filename); or None if error'''
ret = []
subtitles = info_dict.get('requested_subtitles')
- if not subtitles or not (self.params.get('writesubtitles') or self.params.get('writeautomaticsub')):
+ if not (self.params.get('writesubtitles') or self.params.get('writeautomaticsub')):
# subtitles download errors are already managed as troubles in relevant IE
# that way it will silently go on when used with unsupporting IE
return ret
-
+ elif not subtitles:
+ self.to_screen('[info] There\'s no subtitles for the requested languages')
+ return ret
sub_filename_base = self.prepare_filename(info_dict, 'subtitle')
if not sub_filename_base:
self.to_screen('[info] Skipping writing video subtitles')
return ret
+
for sub_lang, sub_info in subtitles.items():
sub_format = sub_info['ext']
sub_filename = subtitles_filename(filename, sub_lang, sub_format, info_dict.get('ext'))
thumbnails, ret = [], []
if write_all or self.params.get('writethumbnail', False):
thumbnails = info_dict.get('thumbnails') or []
+ if not thumbnails:
+ self.to_screen(f'[info] There\'s no {label} thumbnails to download')
+ return ret
multiple = write_all and len(thumbnails) > 1
if thumb_filename_base is None:
projects / yt-dlp.git / blobdiff