import datetime
import errno
import fileinput
+import functools
import io
import itertools
import json
import tokenize
import traceback
import random
+import unicodedata
+from enum import Enum
from string import ascii_letters
-from zipimport import zipimporter
from .compat import (
compat_basestring,
compat_kwargs,
compat_numeric_types,
compat_os_name,
+ compat_pycrypto_AES,
compat_shlex_quote,
compat_str,
compat_tokenize_tokenize,
compat_urllib_error,
compat_urllib_request,
compat_urllib_request_DataHandler,
+ windows_enable_vt_mode,
)
from .cookies import load_cookies
from .utils import (
DEFAULT_OUTTMPL,
determine_ext,
determine_protocol,
- DOT_DESKTOP_LINK_TEMPLATE,
- DOT_URL_LINK_TEMPLATE,
- DOT_WEBLOC_LINK_TEMPLATE,
+ DownloadCancelled,
DownloadError,
encode_compat_str,
encodeFilename,
float_or_none,
format_bytes,
format_field,
- STR_FORMAT_RE_TMPL,
- STR_FORMAT_TYPES,
formatSeconds,
GeoRestrictedError,
HEADRequest,
iri_to_uri,
ISO3166Utils,
LazyList,
+ LINK_TEMPLATES,
locked_file,
make_dir,
make_HTTPS_handler,
MaxDownloadsReached,
network_exceptions,
+ number_of_digits,
orderedSet,
OUTTMPL_TYPES,
PagedList,
parse_filesize,
PerRequestProxyHandler,
platform_name,
+ Popen,
PostProcessingError,
preferredencoding,
prepend_extension,
- process_communicate_or_kill,
register_socks_protocols,
RejectedVideoReached,
render_table,
sanitize_url,
sanitized_Request,
std_headers,
+ STR_FORMAT_RE_TMPL,
+ STR_FORMAT_TYPES,
str_or_none,
strftime_or_none,
subtitles_filename,
+ supports_terminal_sequences,
ThrottledDownload,
to_high_limit_path,
traverse_obj,
YoutubeDLRedirectHandler,
)
from .cache import Cache
+from .minicurses import format_text
from .extractor import (
gen_extractor_classes,
get_info_extractor,
_LAZY_LOADER,
- _PLUGIN_CLASSES
+ _PLUGIN_CLASSES as plugin_extractors
)
from .extractor.openload import PhantomJSwrapper
from .downloader import (
from .downloader.rtmp import rtmpdump_version
from .postprocessor import (
get_postprocessor,
+ EmbedThumbnailPP,
FFmpegFixupDurationPP,
FFmpegFixupM3u8PP,
FFmpegFixupM4aPP,
FFmpegMergerPP,
FFmpegPostProcessor,
MoveFilesAfterDownloadPP,
+ _PLUGIN_CLASSES as plugin_postprocessors
)
+from .update import detect_variant
from .version import __version__
if compat_os_name == 'nt':
(or video) as a single JSON line.
force_write_download_archive: Force writing download archive regardless
of 'skip_download' or 'simulate'.
- simulate: Do not download the video files.
+ simulate: Do not download the video files. If unset (or None),
+ simulate only if listsubtitles, listformats or list_thumbnails is used
format: Video format code. see "FORMAT SELECTION" for more details.
allow_unplayable_formats: Allow unplayable formats to be extracted and downloaded.
ignore_no_formats_error: Ignore "No video formats" error. Usefull for
extracting metadata even if the video is not actually
available for download (experimental)
- format_sort: How to sort the video formats. see "Sorting Formats"
- for more details.
+ format_sort: A list of fields by which to sort the video formats.
+ See "Sorting Formats" for more details.
format_sort_force: Force the given format_sort. see "Sorting Formats"
for more details.
allow_multiple_video_streams: Allow multiple video streams to be merged
allow_multiple_audio_streams: Allow multiple audio streams to be merged
into a single file
check_formats Whether to test if the formats are downloadable.
- Can be True (check all), False (check none)
+ Can be True (check all), False (check none),
+ 'selected' (check selected formats),
or None (check only if requested by extractor)
paths: Dictionary of output paths. The allowed keys are 'home'
'temp' and the keys of OUTTMPL_TYPES (in utils.py)
outtmpl: Dictionary of templates for output names. Allowed keys
are 'default' and the keys of OUTTMPL_TYPES (in utils.py).
- A string a also accepted for backward compatibility
+ For compatibility with youtube-dl, a single string can also be used
outtmpl_na_placeholder: Placeholder for unavailable meta fields.
restrictfilenames: Do not allow "&" and spaces in file names
trim_file_name: Limit length of filename (extension excluded)
windowsfilenames: Force the filenames to be windows compatible
- ignoreerrors: Do not stop on download errors
- (Default True when running yt-dlp,
- but False when directly accessing YoutubeDL class)
+ ignoreerrors: Do not stop on download/postprocessing errors.
+ Can be 'only_download' to ignore only download errors.
+ 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
overwrites: Overwrite all video and metadata files if True,
overwrite only non-video files if None
and don't overwrite any file if False
+ For compatibility with youtube-dl,
+ "nooverwrites" may also be used instead
playliststart: Playlist item to start at.
playlistend: Playlist item to end at.
playlist_items: Specific indices of playlist to download.
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.
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
- writecomments: Extract video comments. This will not be written to disk
+ getcomments: Extract video comments. This will not be written to disk
unless writeinfojson is also given
writeannotations: Write the video annotations to a .annotations.xml file
writethumbnail: Write the thumbnail image to a file
cookiefile: File name where cookies should be read from and dumped to
cookiesfrombrowser: A tuple containing the name of the browser and the profile
name/path from where cookies are loaded.
- Eg: ('chrome', ) or (vivaldi, 'default')
+ Eg: ('chrome', ) or ('vivaldi', 'default')
nocheckcertificate:Do not verify SSL certificates
prefer_insecure: Use HTTP instead of HTTPS to retrieve information.
At the moment, this is only supported by YouTube.
* when: When to run the postprocessor. Can be one of
pre_process|before_dl|post_process|after_move.
Assumed to be 'post_process' if not given
- post_hooks: A list of functions that get called as the final step
+ post_hooks: Deprecated - Register a custom postprocessor instead
+ A list of functions that get called as the final step
for each video file, after all postprocessors have been
called. The filename will be passed as the only argument.
progress_hooks: A list of functions that get called on download
Progress hooks are guaranteed to be called at least once
(with status "finished") if the download is successful.
+ postprocessor_hooks: A list of functions that get called on postprocessing
+ progress, with a dictionary with the entries
+ * status: One of "started", "processing", or "finished".
+ Check this first and ignore unknown values.
+ * postprocessor: Name of the postprocessor
+ * info_dict: The extracted info_dict
+
+ 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.
final_ext: Expected final extension; used to detect when the file was
already downloaded and converted. "merge_output_format" is
use downloader suggested by extractor if None.
compat_opts: Compatibility options. See "Differences in default behavior".
The following options do not work when used through the API:
- filename, abort-on-error, multistreams, no-live-chat,
+ filename, abort-on-error, multistreams, no-live-chat, format-sort
no-clean-infojson, no-playlist-metafiles, no-keep-subs.
Refer __init__.py for their implementation
+ progress_template: Dictionary of templates for progress outputs.
+ Allowed keys are 'download', 'postprocess',
+ 'download-title' (console title) and 'postprocess-title'.
+ The template is mapped on a dictionary with keys 'progress' and 'info'
The following parameters are not used by YoutubeDL itself, they are used by
the downloader (see yt_dlp/downloader/common.py):
nopart, updatetime, buffersize, ratelimit, throttledratelimit, min_filesize,
- max_filesize, test, noresizebuffer, retries, continuedl, noprogress, consoletitle,
- xattr_set_filesize, external_downloader_args, hls_use_mpegts, http_chunk_size.
+ max_filesize, test, noresizebuffer, retries, fragment_retries, continuedl,
+ noprogress, xattr_set_filesize, hls_use_mpegts, http_chunk_size,
+ external_downloader_args.
The following options are used by the post processors:
prefer_ffmpeg: If False, use avconv instead of ffmpeg if both are available,
ffmpeg_location: Location of the ffmpeg/avconv binary; either the path
to the binary or its containing directory.
postprocessor_args: A dictionary of postprocessor/executable keys (in lower case)
- and a list of additional command-line arguments for the
- postprocessor/executable. The dict can also have "PP+EXE" keys
- which are used when the given exe is used by the given PP.
- Use 'default' as the name for arguments to passed to all PP
+ and a list of additional command-line arguments for the
+ postprocessor/executable. The dict can also have "PP+EXE" keys
+ which are used when the given exe is used by the given PP.
+ Use 'default' as the name for arguments to passed to all PP
+ For compatibility with youtube-dl, a single list of args
+ can also be used
The following options are used by the extractors:
extractor_retries: Number of times to retry for known errors
_NUMERIC_FIELDS = set((
'width', 'height', 'tbr', 'abr', 'asr', 'vbr', 'fps', 'filesize', 'filesize_approx',
- 'timestamp', 'upload_year', 'upload_month', 'upload_day',
+ 'timestamp', 'release_timestamp',
'duration', 'view_count', 'like_count', 'dislike_count', 'repost_count',
'average_rating', 'comment_count', 'age_limit',
'start_time', 'end_time',
'chapter_number', 'season_number', 'episode_number',
'track_number', 'disc_number', 'release_year',
- 'playlist_index',
))
+ _format_selection_exts = {
+ 'audio': {'m4a', 'mp3', 'ogg', 'aac'},
+ 'video': {'mp4', 'flv', 'webm', '3gp'},
+ 'storyboards': {'mhtml'},
+ }
+
params = None
- _ies = []
+ _ies = {}
_pps = {'pre_process': [], 'before_dl': [], 'after_move': [], 'post_process': []}
_printed_messages = set()
_first_webpage_request = True
_screen_file = None
def __init__(self, params=None, auto_init=True):
- """Create a FileDownloader object with the given options."""
+ """Create a FileDownloader object with the given options.
+ @param auto_init Whether to load the default extractors and print header (if verbose).
+ Set to 'no_verbose_header' to not print the header
+ """
if params is None:
params = {}
- self._ies = []
+ self._ies = {}
self._ies_instances = {}
self._pps = {'pre_process': [], 'before_dl': [], 'after_move': [], 'post_process': []}
self._printed_messages = set()
self._first_webpage_request = True
self._post_hooks = []
self._progress_hooks = []
+ self._postprocessor_hooks = []
self._download_retcode = 0
self._num_downloads = 0
self._screen_file = [sys.stdout, sys.stderr][params.get('logtostderr', False)]
self._err_file = sys.stderr
- self.params = {
- # Default parameters
- 'nocheckcertificate': False,
- }
- self.params.update(params)
+ self.params = params
self.cache = Cache(self)
+ windows_enable_vt_mode()
+ # FIXME: This will break if we ever print color to stdout
+ self._allow_colors = {
+ 'screen': not self.params.get('no_color') and supports_terminal_sequences(self._screen_file),
+ 'err': not self.params.get('no_color') and supports_terminal_sequences(self._err_file),
+ }
+
if sys.version_info < (3, 6):
self.report_warning(
'Python version %d.%d is not supported! Please update to Python 3.6 or above' % sys.version_info[:2])
+ if self.params.get('allow_unplayable_formats'):
+ self.report_warning(
+ f'You have asked for {self._format_err("UNPLAYABLE", self.Styles.EMPHASIS)} formats to be listed/downloaded. '
+ 'This is a developer option intended for debugging. \n'
+ ' If you experience any issues while using this option, '
+ f'{self._format_err("DO NOT", self.Styles.ERROR)} open a bug report')
+
def check_deprecated(param, option, suggestion):
if self.params.get(param) is not None:
self.report_warning('%s is deprecated. Use %s instead' % (option, suggestion))
check_deprecated('usetitle', '--title', '-o "%(title)s-%(id)s.%(ext)s"')
check_deprecated('useid', '--id', '-o "%(id)s.%(ext)s"')
- for msg in self.params.get('warnings', []):
+ for msg in self.params.get('_warnings', []):
self.report_warning(msg)
- if self.params.get('final_ext'):
- if self.params.get('merge_output_format'):
- self.report_warning('--merge-output-format will be ignored since --remux-video or --recode-video is given')
- self.params['merge_output_format'] = self.params['final_ext']
+ if 'list-formats' in self.params.get('compat_opts', []):
+ self.params['listformats_table'] = False
- if 'overwrites' in self.params and self.params['overwrites'] is None:
- del self.params['overwrites']
+ if 'overwrites' not in self.params and self.params.get('nooverwrites') is not None:
+ # nooverwrites was unnecessarily changed to overwrites
+ # in 0c3d0f51778b153f65c21906031c2e091fcfb641
+ # This ensures compatibility with both keys
+ self.params['overwrites'] = not self.params['nooverwrites']
+ elif self.params.get('overwrites') is None:
+ self.params.pop('overwrites', None)
+ else:
+ self.params['nooverwrites'] = not self.params['overwrites']
if params.get('bidi_workaround', False):
try:
stdout=slave,
stderr=self._err_file)
try:
- self._output_process = subprocess.Popen(
- ['bidiv'] + width_args, **sp_kwargs
- )
+ self._output_process = Popen(['bidiv'] + width_args, **sp_kwargs)
except OSError:
- self._output_process = subprocess.Popen(
- ['fribidi', '-c', 'UTF-8'] + width_args, **sp_kwargs)
+ 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.')
+ 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._setup_opener()
- """Preload the archive, if any is specified"""
- def preload_download_archive(fn):
- if fn is None:
- return False
- self.write_debug('Loading archive file %r\n' % fn)
- try:
- with locked_file(fn, 'r', encoding='utf-8') as archive_file:
- for line in archive_file:
- self.archive.add(line.strip())
- except IOError as ioe:
- if ioe.errno != errno.ENOENT:
- raise
- return False
- return True
-
- self.archive = set()
- preload_download_archive(self.params.get('download_archive'))
-
if auto_init:
- self.print_debug_header()
+ if auto_init != 'no_verbose_header':
+ self.print_debug_header()
self.add_default_info_extractors()
for pp_def_raw in self.params.get('postprocessors', []):
register_socks_protocols()
+ def preload_download_archive(fn):
+ """Preload the archive, if any is specified"""
+ if fn is None:
+ return False
+ 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())
+ except IOError as ioe:
+ if ioe.errno != errno.ENOENT:
+ raise
+ return False
+ return True
+
+ self.archive = set()
+ preload_download_archive(self.params.get('download_archive'))
+
def warn_if_short_id(self, argv):
# short YouTube ID starting with dash?
idxs = [
)
self.report_warning(
'Long argument string detected. '
- 'Use -- to separate parameters and URLs, like this:\n%s\n' %
+ 'Use -- to separate parameters and URLs, like this:\n%s' %
args_to_str(correct_argv))
def add_info_extractor(self, ie):
"""Add an InfoExtractor object to the end of the list."""
- self._ies.append(ie)
+ ie_key = ie.ie_key()
+ self._ies[ie_key] = ie
if not isinstance(ie, type):
- self._ies_instances[ie.ie_key()] = ie
+ 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
self._post_hooks.append(ph)
def add_progress_hook(self, ph):
- """Add the progress hook (currently only for the file downloader)"""
+ """Add the download progress hook"""
self._progress_hooks.append(ph)
+ def add_postprocessor_hook(self, ph):
+ """Add the postprocessing progress hook"""
+ self._postprocessor_hooks.append(ph)
+
def _bidi_workaround(self, message):
if not hasattr(self, '_output_channel'):
return message
def save_console_title(self):
if not self.params.get('consoletitle', False):
return
- if self.params.get('simulate', False):
+ if self.params.get('simulate'):
return
if compat_os_name != 'nt' and 'TERM' in os.environ:
# Save the title on stack
def restore_console_title(self):
if not self.params.get('consoletitle', False):
return
- if self.params.get('simulate', False):
+ if self.params.get('simulate'):
return
if compat_os_name != 'nt' and 'TERM' in os.environ:
# Restore the title from stack
tb = ''.join(tb_data)
if tb:
self.to_stderr(tb)
- if not self.params.get('ignoreerrors', False):
+ if not self.params.get('ignoreerrors'):
if sys.exc_info()[0] and hasattr(sys.exc_info()[1], 'exc_info') and sys.exc_info()[1].exc_info[0]:
exc_info = sys.exc_info()[1].exc_info
else:
self.to_stdout(
message, skip_eol, quiet=self.params.get('quiet', False))
+ class Styles(Enum):
+ HEADERS = 'yellow'
+ EMPHASIS = 'blue'
+ ID = 'green'
+ DELIM = 'blue'
+ ERROR = 'red'
+ WARNING = 'yellow'
+
+ def __format_text(self, out, text, f, fallback=None, *, test_encoding=False):
+ assert out in ('screen', 'err')
+ if test_encoding:
+ original_text = text
+ handle = self._screen_file if out == 'screen' else self._err_file
+ encoding = self.params.get('encoding') or getattr(handle, 'encoding', 'ascii')
+ text = text.encode(encoding, 'ignore').decode(encoding)
+ if fallback is not None and text != original_text:
+ text = fallback
+ if isinstance(f, self.Styles):
+ f = f._value_
+ return format_text(text, f) if self._allow_colors[out] else text if fallback is None else fallback
+
+ def _format_screen(self, *args, **kwargs):
+ return self.__format_text('screen', *args, **kwargs)
+
+ def _format_err(self, *args, **kwargs):
+ return self.__format_text('err', *args, **kwargs)
+
def report_warning(self, message, only_once=False):
'''
Print the message to stderr, it will be prefixed with 'WARNING:'
else:
if self.params.get('no_warnings'):
return
- if not self.params.get('no_color') and self._err_file.isatty() and compat_os_name != 'nt':
- _msg_header = '\033[0;33mWARNING:\033[0m'
- else:
- _msg_header = 'WARNING:'
- warning_message = '%s %s' % (_msg_header, message)
- self.to_stderr(warning_message, only_once)
+ self.to_stderr(f'{self._format_err("WARNING:", self.Styles.WARNING)} {message}', only_once)
def report_error(self, message, tb=None):
'''
Do the same as trouble, but prefixes the message with 'ERROR:', colored
in red if stderr is a tty file.
'''
- if not self.params.get('no_color') and self._err_file.isatty() and compat_os_name != 'nt':
- _msg_header = '\033[0;31mERROR:\033[0m'
- else:
- _msg_header = 'ERROR:'
- error_message = '%s %s' % (_msg_header, message)
- self.trouble(error_message, tb)
+ self.trouble(f'{self._format_err("ERROR:", self.Styles.ERROR)} {message}', tb)
def write_debug(self, message, only_once=False):
'''Log debug message or Print message to stderr'''
except UnicodeEncodeError:
self.to_screen('Deleting existing file')
+ def raise_no_formats(self, info, forced=False):
+ has_drm = info.get('__has_drm')
+ msg = 'This video is DRM protected' if has_drm else 'No video formats found!'
+ expected = self.params.get('ignore_no_formats_error')
+ if forced or not expected:
+ raise ExtractorError(msg, video_id=info['id'], ie=info['extractor'],
+ expected=has_drm or expected)
+ else:
+ self.report_warning(msg)
+
def parse_outtmpl(self):
outtmpl_dict = self.params.get('outtmpl', {})
if not isinstance(outtmpl_dict, dict):
outtmpl_dict = {'default': outtmpl_dict}
+ # Remove spaces in the default template
+ if self.params.get('restrictfilenames'):
+ sanitize = lambda x: x.replace(' - ', ' ').replace(' ', '-')
+ else:
+ sanitize = lambda x: x
outtmpl_dict.update({
- k: v for k, v in DEFAULT_OUTTMPL.items()
- if not outtmpl_dict.get(k)})
+ k: sanitize(v) for k, v in DEFAULT_OUTTMPL.items()
+ if outtmpl_dict.get(k) is None})
for key, val in outtmpl_dict.items():
if isinstance(val, bytes):
self.report_warning(
def validate_outtmpl(cls, outtmpl):
''' @return None or Exception object '''
outtmpl = re.sub(
- STR_FORMAT_RE_TMPL.format('[^)]*', '[ljq]'),
+ STR_FORMAT_RE_TMPL.format('[^)]*', '[ljqBU]'),
lambda mobj: f'{mobj.group(0)[:-1]}s',
cls._outtmpl_expandpath(outtmpl))
try:
except ValueError as err:
return err
- def prepare_outtmpl(self, outtmpl, info_dict, sanitize=None):
- """ Make the template and info_dict suitable for substitution : ydl.outtmpl_escape(outtmpl) % info_dict """
+ @staticmethod
+ def _copy_infodict(info_dict):
info_dict = dict(info_dict)
- na = self.params.get('outtmpl_na_placeholder', 'NA')
+ for key in ('__original_infodict', '__postprocessors'):
+ info_dict.pop(key, None)
+ return info_dict
+ def prepare_outtmpl(self, outtmpl, info_dict, sanitize=None):
+ """ Make the outtmpl and info_dict suitable for substitution: ydl.escape_outtmpl(outtmpl) % info_dict """
+ info_dict.setdefault('epoch', int(time.time())) # keep epoch consistent once set
+
+ info_dict = self._copy_infodict(info_dict)
info_dict['duration_string'] = ( # %(duration>%H-%M-%S)s is wrong if duration > 24hrs
formatSeconds(info_dict['duration'], '-' if sanitize else ':')
if info_dict.get('duration', None) is not None
else None)
- info_dict['epoch'] = int(time.time())
info_dict['autonumber'] = self.params.get('autonumber_start', 1) - 1 + self._num_downloads
if info_dict.get('resolution') is None:
info_dict['resolution'] = self.format_resolution(info_dict, default=None)
- # For fields playlist_index and autonumber convert all occurrences
+ # For fields playlist_index, playlist_autonumber and autonumber convert all occurrences
# of %(field)s to %(field)0Nd for backward compatibility
field_size_compat_map = {
- 'playlist_index': len(str(info_dict.get('_last_playlist_index') or '')),
+ 'playlist_index': number_of_digits(info_dict.get('_last_playlist_index') or 0),
+ 'playlist_autonumber': number_of_digits(info_dict.get('n_entries') or 0),
'autonumber': self.params.get('autonumber_size') or 5,
}
TMPL_DICT = {}
- EXTERNAL_FORMAT_RE = re.compile(STR_FORMAT_RE_TMPL.format('[^)]*', f'[{STR_FORMAT_TYPES}ljq]'))
+ EXTERNAL_FORMAT_RE = re.compile(STR_FORMAT_RE_TMPL.format('[^)]*', f'[{STR_FORMAT_TYPES}ljqBU]'))
MATH_FUNCTIONS = {
'+': float.__add__,
'-': 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+)')
+ FIELD_RE = r'\w*(?:\.(?:\w+|{num}|{num}?(?::{num}?){{1,2}}))*'.format(num=r'(?:-?\d+)')
MATH_FIELD_RE = r'''{field}|{num}'''.format(field=FIELD_RE, num=r'-?\d+(?:.\d+)?')
MATH_OPERATORS_RE = r'(?:%s)' % '|'.join(map(re.escape, MATH_FUNCTIONS.keys()))
INTERNAL_FORMAT_RE = re.compile(r'''(?x)
(?P<fields>{field})
(?P<maths>(?:{math_op}{math_field})*)
(?:>(?P<strf_format>.+?))?
+ (?P<alternate>(?<!\\),[^|)]+)?
(?:\|(?P<default>.*?))?
$'''.format(field=FIELD_RE, math_op=MATH_OPERATORS_RE, math_field=MATH_FIELD_RE))
- get_key = lambda k: traverse_obj(
- info_dict, k.split('.'), is_user_input=True, traverse_string=True)
+ 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 get_value(mdict):
# Object traversal
- value = get_key(mdict['fields'])
+ value = _traverse_infodict(mdict['fields'])
# Negative
if mdict['negate']:
value = float_or_none(value)
item, multiplier = (item[1:], -1) if item[0] == '-' else (item, 1)
offset = float_or_none(item)
if offset is None:
- offset = float_or_none(get_key(item))
+ offset = float_or_none(_traverse_infodict(item))
try:
value = operator(value, multiplier * offset)
except (TypeError, ZeroDivisionError):
operator = None
# Datetime formatting
if mdict['strf_format']:
- value = strftime_or_none(value, mdict['strf_format'])
+ value = strftime_or_none(value, mdict['strf_format'].replace('\\,', ','))
return value
+ na = self.params.get('outtmpl_na_placeholder', 'NA')
+
+ def _dumpjson_default(obj):
+ if isinstance(obj, (set, LazyList)):
+ return list(obj)
+ raise TypeError(f'Object of type {type(obj).__name__} is not JSON serializable')
+
def create_key(outer_mobj):
if not outer_mobj.group('has_key'):
- return f'%{outer_mobj.group(0)}'
-
- prefix = outer_mobj.group('prefix')
+ return outer_mobj.group(0)
key = outer_mobj.group('key')
- original_fmt = fmt = outer_mobj.group('format')
mobj = re.match(INTERNAL_FORMAT_RE, key)
- if mobj is None:
- value, default, mobj = None, na, {'fields': ''}
- else:
+ initial_field = mobj.group('fields').split('.')[-1] if mobj else ''
+ value, default = None, na
+ while mobj:
mobj = mobj.groupdict()
- default = mobj['default'] if mobj['default'] is not None else na
+ default = mobj['default'] if mobj['default'] is not None else default
value = get_value(mobj)
+ if value is None and mobj['alternate']:
+ mobj = re.match(INTERNAL_FORMAT_RE, mobj['alternate'][1:])
+ else:
+ break
+ fmt = outer_mobj.group('format')
if fmt == 's' and value is not None and key in field_size_compat_map.keys():
fmt = '0{:d}d'.format(field_size_compat_map[key])
value = default if value is None else value
str_fmt = f'{fmt[:-1]}s'
- if fmt[-1] == 'l':
- value, fmt = ', '.join(variadic(value)), str_fmt
- elif fmt[-1] == 'j':
- value, fmt = json.dumps(value), str_fmt
- elif fmt[-1] == 'q':
+ if fmt[-1] == 'l': # list
+ delim = '\n' if '#' in (outer_mobj.group('conversion') or '') else ', '
+ value, fmt = delim.join(variadic(value)), str_fmt
+ elif fmt[-1] == 'j': # json
+ value, fmt = json.dumps(value, default=_dumpjson_default), str_fmt
+ elif fmt[-1] == 'q': # quoted
value, fmt = compat_shlex_quote(str(value)), str_fmt
+ elif fmt[-1] == 'B': # bytes
+ value = f'%{str_fmt}'.encode('utf-8') % str(value).encode('utf-8')
+ value, fmt = value.decode('utf-8', 'ignore'), 's'
+ elif fmt[-1] == 'U': # unicode normalized
+ opts = outer_mobj.group('conversion') or ''
+ value, fmt = unicodedata.normalize(
+ # "+" = compatibility equivalence, "#" = NFD
+ 'NF%s%s' % ('K' if '+' in opts else '', 'D' if '#' in opts else 'C'),
+ value), str_fmt
elif fmt[-1] == 'c':
- value = str(value)
- if value is None:
- value, fmt = default, 's'
+ if value:
+ value = str(value)[0]
else:
- value = value[0]
+ fmt = str_fmt
elif fmt[-1] not in 'rs': # numeric
value = float_or_none(value)
if value is None:
# So we convert it to repr first
value, fmt = repr(value), str_fmt
if fmt[-1] in 'csr':
- value = sanitize(mobj['fields'].split('.')[-1], value)
+ value = sanitize(initial_field, value)
- key = '%s\0%s' % (key.replace('%', '%\0'), original_fmt)
+ key = '%s\0%s' % (key.replace('%', '%\0'), outer_mobj.group('format'))
TMPL_DICT[key] = value
- return f'{prefix}%({key}){fmt}'
+ return '{prefix}%({key}){fmt}'.format(key=key, fmt=fmt, prefix=outer_mobj.group('prefix'))
return EXTERNAL_FORMAT_RE.sub(create_key, outtmpl), TMPL_DICT
+ def evaluate_outtmpl(self, outtmpl, info_dict, *args, **kwargs):
+ outtmpl, info_dict = self.prepare_outtmpl(outtmpl, info_dict, *args, **kwargs)
+ return self.escape_outtmpl(outtmpl) % info_dict
+
def _prepare_filename(self, info_dict, tmpl_type='default'):
try:
sanitize = lambda k, v: sanitize_filename(
compat_str(v),
restricted=self.params.get('restrictfilenames'),
is_id=(k == 'id' or k.endswith('_id')))
- outtmpl = self.outtmpl_dict.get(tmpl_type, self.outtmpl_dict['default'])
- outtmpl, template_dict = self.prepare_outtmpl(outtmpl, info_dict, sanitize)
- outtmpl = self.escape_outtmpl(self._outtmpl_expandpath(outtmpl))
- filename = outtmpl % template_dict
+ outtmpl = self._outtmpl_expandpath(self.outtmpl_dict.get(tmpl_type, self.outtmpl_dict['default']))
+ filename = self.evaluate_outtmpl(outtmpl, info_dict, sanitize)
force_ext = OUTTMPL_TYPES.get(tmpl_type)
- if force_ext is not None:
+ if filename and force_ext is not None:
filename = replace_extension(filename, force_ext, info_dict.get('ext'))
# https://github.com/blackjack4494/youtube-dlc/issues/85
"""Generate the output filename."""
filename = self._prepare_filename(info_dict, dir_type or 'default')
+ if not filename and dir_type not in ('', 'temp'):
+ return ''
if warn:
if not self.params.get('paths'):
self.report_warning('--paths is ignored when an outputting to stdout', only_once=True)
elif os.path.isabs(filename):
self.report_warning('--paths is ignored since an absolute path is given in output template', only_once=True)
- self.__prepare_filename_warned = True
if filename == '-' or not filename:
return filename
if age_restricted(info_dict.get('age_limit'), self.params.get('age_limit')):
return 'Skipping "%s" because it is age restricted' % video_title
- if not incomplete:
- match_filter = self.params.get('match_filter')
- if match_filter is not None:
- ret = match_filter(info_dict)
- if ret is not None:
- return ret
+ match_filter = self.params.get('match_filter')
+ if match_filter is not None:
+ 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 not None:
+ return ret
return None
if self.in_download_archive(info_dict):
for key, value in extra_info.items():
info_dict.setdefault(key, value)
- def extract_info(self, url, download=True, ie_key=None, extra_info={},
+ 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.
force_generic_extractor -- force using the generic extractor
"""
+ if extra_info is None:
+ extra_info = {}
+
if not ie_key and force_generic_extractor:
ie_key = 'Generic'
if ie_key:
- ies = [self.get_info_extractor(ie_key)]
+ ies = {ie_key: self._get_info_extractor_class(ie_key)}
else:
ies = self._ies
- for ie in ies:
+ for ie_key, ie in ies.items():
if not ie.suitable(url):
continue
- ie_key = ie.ie_key()
- ie = self.get_info_extractor(ie_key)
if not ie.working():
self.report_warning('The program functionality for this site has been marked as broken, '
'and will probably not work.')
- try:
- temp_id = str_or_none(
- ie.extract_id(url) if callable(getattr(ie, 'extract_id', None))
- else ie._match_id(url))
- except (AssertionError, IndexError, AttributeError):
- temp_id = None
+ 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("[%s] %s: has already been recorded in archive" % (
ie_key, temp_id))
break
- return self.__extract_info(url, ie, download, extra_info, process)
+ return self.__extract_info(url, self.get_info_extractor(ie_key), download, extra_info, process)
else:
self.report_error('no suitable InfoExtractor for URL %s' % url)
- def __handle_extraction_exceptions(func, handle_all_errors=True):
+ def __handle_extraction_exceptions(func):
+ @functools.wraps(func)
def wrapper(self, *args, **kwargs):
try:
return func(self, *args, **kwargs)
self.to_stderr('\r')
self.report_warning('The download speed is below throttle limit. Re-extracting data')
return wrapper(self, *args, **kwargs)
- except (MaxDownloadsReached, ExistingVideoReached, RejectedVideoReached):
+ except (DownloadCancelled, LazyList.IndexError):
raise
except Exception as e:
- if handle_all_errors and self.params.get('ignoreerrors', False):
+ if self.params.get('ignoreerrors'):
self.report_error(error_to_compat_str(e), tb=encode_compat_str(traceback.format_exc()))
else:
raise
'extractor_key': ie.ie_key(),
})
- def process_ie_result(self, ie_result, download=True, extra_info={}):
+ def process_ie_result(self, ie_result, download=True, extra_info=None):
"""
Take the result of the ie(may be modified) and resolve all unresolved
references (URLs, playlist items).
It will also download the videos if 'download'.
Returns the resolved ie_result.
"""
+ if extra_info is None:
+ extra_info = {}
result_type = ie_result.get('_type', 'video')
if result_type in ('url', 'url_transparent'):
if ((extract_flat == 'in_playlist' and 'playlist' in extra_info)
or extract_flat is True):
info_copy = ie_result.copy()
- self.add_extra_info(info_copy, extra_info)
ie = try_get(ie_result.get('ie_key'), self.get_info_extractor)
+ if ie and not ie_result.get('id'):
+ info_copy['id'] = ie.get_temp_id(ie_result['url'])
self.add_default_extra_info(info_copy, ie, ie_result['url'])
+ self.add_extra_info(info_copy, extra_info)
self.__forced_printings(info_copy, self.prepare_filename(info_copy), incomplete=True)
+ if self.params.get('force_write_download_archive', False):
+ self.record_download_archive(info_copy)
return ie_result
if result_type == 'video':
ie_result = self.process_video_result(ie_result, download=download)
additional_urls = (ie_result or {}).get('additional_urls')
if additional_urls:
- # TODO: Improve MetadataFromFieldPP to allow setting a list
+ # TODO: Improve MetadataParserPP to allow setting a list
if isinstance(additional_urls, compat_str):
additional_urls = [additional_urls]
self.to_screen(
'It needs to be updated.' % ie_result.get('extractor'))
def _fixup(r):
- self.add_extra_info(
- r,
- {
- 'extractor': ie_result['extractor'],
- 'webpage_url': ie_result['webpage_url'],
- 'webpage_url_basename': url_basename(ie_result['webpage_url']),
- 'extractor_key': ie_result['extractor_key'],
- }
- )
+ self.add_extra_info(r, {
+ 'extractor': ie_result['extractor'],
+ 'webpage_url': ie_result['webpage_url'],
+ 'webpage_url_basename': url_basename(ie_result['webpage_url']),
+ 'extractor_key': ie_result['extractor_key'],
+ })
return r
ie_result['entries'] = [
self.process_ie_result(_fixup(r), download, extra_info)
msg = (
'Downloading %d videos' if not isinstance(ie_entries, list)
else 'Collected %d videos; downloading %%d of them' % len(ie_entries))
- if not isinstance(ie_entries, (list, PagedList)):
- ie_entries = LazyList(ie_entries)
- def get_entry(i):
- return YoutubeDL.__handle_extraction_exceptions(
- lambda self, i: ie_entries[i - 1],
- False
- )(self, i)
+ if isinstance(ie_entries, list):
+ def get_entry(i):
+ return ie_entries[i - 1]
+ else:
+ if not isinstance(ie_entries, PagedList):
+ ie_entries = LazyList(ie_entries)
+
+ def get_entry(i):
+ return YoutubeDL.__handle_extraction_exceptions(
+ lambda self, i: ie_entries[i - 1]
+ )(self, i)
entries = []
- for i in playlistitems or itertools.count(playliststart):
+ items = playlistitems if playlistitems is not None else itertools.count(playliststart)
+ for i in items:
+ if i == 0:
+ continue
if playlistitems is None and playlistend is not None and playlistend < i:
break
entry = None
# Save playlist_index before re-ordering
entries = [
- ((playlistitems[i - 1] if playlistitems else i), entry)
+ ((playlistitems[i - 1] if playlistitems else i + playliststart - 1), entry)
for i, entry in enumerate(entries, 1)
if entry is not None]
n_entries = len(entries)
playlistitems = list(range(playliststart, playliststart + n_entries))
ie_result['requested_entries'] = playlistitems
- if self.params.get('allow_playlist_files', True):
+ if not self.params.get('simulate') and self.params.get('allow_playlist_files', True):
ie_copy = {
'playlist': playlist,
'playlist_id': ie_result.get('id'),
'playlist_uploader': ie_result.get('uploader'),
'playlist_uploader_id': ie_result.get('uploader_id'),
'playlist_index': 0,
+ 'n_entries': n_entries,
}
ie_copy.update(dict(ie_result))
- if self.params.get('writeinfojson', False):
- infofn = self.prepare_filename(ie_copy, 'pl_infojson')
- if not self._ensure_dir_exists(encodeFilename(infofn)):
- return
- if not self.params.get('overwrites', True) and os.path.exists(encodeFilename(infofn)):
- self.to_screen('[info] Playlist metadata is already present')
- else:
- self.to_screen('[info] Writing playlist metadata as JSON to: ' + infofn)
- try:
- write_json_file(self.sanitize_info(ie_result, self.params.get('clean_infojson', True)), infofn)
- except (OSError, IOError):
- self.report_error('Cannot write playlist metadata to JSON file ' + infofn)
-
+ if self._write_info_json('playlist', ie_result,
+ self.prepare_filename(ie_copy, 'pl_infojson')) is None:
+ return
+ if self._write_description('playlist', ie_result,
+ self.prepare_filename(ie_copy, 'pl_description')) is None:
+ return
# TODO: This should be passed to ThumbnailsConvertor if necessary
- self._write_thumbnails(ie_copy, self.prepare_filename(ie_copy, 'pl_thumbnail'))
-
- if self.params.get('writedescription', False):
- descfn = self.prepare_filename(ie_copy, 'pl_description')
- if not self._ensure_dir_exists(encodeFilename(descfn)):
- return
- if not self.params.get('overwrites', True) and os.path.exists(encodeFilename(descfn)):
- self.to_screen('[info] Playlist description is already present')
- elif ie_result.get('description') is None:
- self.report_warning('There\'s no playlist description to write.')
- else:
- try:
- self.to_screen('[info] Writing playlist description to: ' + descfn)
- with io.open(encodeFilename(descfn), 'w', encoding='utf-8') as descfile:
- descfile.write(ie_result['description'])
- except (OSError, IOError):
- self.report_error('Cannot write playlist description file ' + descfn)
- return
+ self._write_thumbnails('playlist', ie_copy, self.prepare_filename(ie_copy, 'pl_thumbnail'))
if self.params.get('playlistreverse', False):
entries = entries[::-1]
max_failures = self.params.get('skip_playlist_after_errors') or float('inf')
for i, entry_tuple in enumerate(entries, 1):
playlist_index, entry = entry_tuple
- if 'playlist_index' in self.params.get('compat_options', []):
- playlist_index = playlistitems[i - 1] if playlistitems else i
+ if 'playlist-index' in self.params.get('compat_opts', []):
+ playlist_index = playlistitems[i - 1] if playlistitems else i + playliststart - 1
self.to_screen('[download] Downloading video %s of %s' % (i, n_entries))
# This __x_forwarded_for_ip thing is a bit ugly but requires
# minimal changes
return op(actual_value, comparison_value)
return _filter
+ def _check_formats(self, formats):
+ for f in formats:
+ self.to_screen('[info] Testing format %s' % f['format_id'])
+ temp_file = tempfile.NamedTemporaryFile(
+ suffix='.tmp', delete=False,
+ dir=self.get_output_path('temp') or None)
+ temp_file.close()
+ try:
+ success, _ = self.dl(temp_file.name, f, test=True)
+ except (DownloadError, IOError, OSError, ValueError) + network_exceptions:
+ success = False
+ finally:
+ if os.path.exists(temp_file.name):
+ try:
+ os.remove(temp_file.name)
+ except OSError:
+ self.report_warning('Unable to delete temporary file "%s"' % temp_file.name)
+ if success:
+ yield f
+ else:
+ self.to_screen('[info] Unable to download format %s. Skipping...' % f['format_id'])
+
def _default_format_spec(self, info_dict, download=True):
def can_merge():
return merger.available and merger.can_merge()
prefer_best = (
- not self.params.get('simulate', False)
+ not self.params.get('simulate')
and download
and (
not can_merge()
allow_multiple_streams = {'audio': self.params.get('allow_multiple_audio_streams', False),
'video': self.params.get('allow_multiple_video_streams', False)}
- check_formats = self.params.get('check_formats')
+ check_formats = self.params.get('check_formats') == 'selected'
def _parse_filter(tokens):
filter_parts = []
else:
output_ext = 'mkv'
+ filtered = lambda *keys: filter(None, (traverse_obj(fmt, *keys) for fmt in formats_info))
+
new_dict = {
'requested_formats': formats_info,
- 'format': '+'.join(fmt_info.get('format') for fmt_info in formats_info),
- 'format_id': '+'.join(fmt_info.get('format_id') for fmt_info in formats_info),
+ 'format': '+'.join(filtered('format')),
+ 'format_id': '+'.join(filtered('format_id')),
'ext': output_ext,
+ 'protocol': '+'.join(map(determine_protocol, formats_info)),
+ 'language': '+'.join(orderedSet(filtered('language'))),
+ 'format_note': '+'.join(orderedSet(filtered('format_note'))),
+ 'filesize_approx': sum(filtered('filesize', 'filesize_approx')),
+ 'tbr': sum(filtered('tbr', 'vbr', 'abr')),
}
if the_only_video:
'height': the_only_video.get('height'),
'resolution': the_only_video.get('resolution') or self.format_resolution(the_only_video),
'fps': the_only_video.get('fps'),
+ 'dynamic_range': the_only_video.get('dynamic_range'),
'vcodec': the_only_video.get('vcodec'),
'vbr': the_only_video.get('vbr'),
'stretched_ratio': the_only_video.get('stretched_ratio'),
new_dict.update({
'acodec': the_only_audio.get('acodec'),
'abr': the_only_audio.get('abr'),
+ 'asr': the_only_audio.get('asr'),
})
return new_dict
if not check_formats:
yield from formats
return
- for f in formats:
- self.to_screen('[info] Testing format %s' % f['format_id'])
- temp_file = tempfile.NamedTemporaryFile(
- suffix='.tmp', delete=False,
- dir=self.get_output_path('temp') or None)
- temp_file.close()
- try:
- success, _ = self.dl(temp_file.name, f, test=True)
- except (DownloadError, IOError, OSError, ValueError) + network_exceptions:
- success = False
- finally:
- if os.path.exists(temp_file.name):
- try:
- os.remove(temp_file.name)
- except OSError:
- self.report_warning('Unable to delete temporary file "%s"' % temp_file.name)
- if success:
- yield f
- else:
- self.to_screen('[info] Unable to download format %s. Skipping...' % f['format_id'])
+ yield from self._check_formats(formats)
def _build_selector_function(selector):
if isinstance(selector, list): # ,
filter_f = lambda f: _filter_f(f) and (
f.get('vcodec') != 'none' or f.get('acodec') != 'none')
else:
- filter_f = ((lambda f: f.get('ext') == format_spec)
- if format_spec in ['mp4', 'flv', 'webm', '3gp', 'm4a', 'mp3', 'ogg', 'aac', 'wav'] # extension
- else (lambda f: f.get('format_id') == format_spec)) # id
+ if format_spec in self._format_selection_exts['audio']:
+ filter_f = lambda f: f.get('ext') == format_spec and f.get('acodec') != 'none'
+ elif format_spec in self._format_selection_exts['video']:
+ filter_f = lambda f: f.get('ext') == format_spec and f.get('acodec') != 'none' and f.get('vcodec') != 'none'
+ elif format_spec in self._format_selection_exts['storyboards']:
+ filter_f = lambda f: f.get('ext') == format_spec and f.get('acodec') == 'none' and f.get('vcodec') == 'none'
+ else:
+ filter_f = lambda f: f.get('format_id') == format_spec # id
def selector_function(ctx):
formats = list(ctx['formats'])
self.cookiejar.add_cookie_header(pr)
return pr.get_header('Cookie')
+ def _sort_thumbnails(self, thumbnails):
+ thumbnails.sort(key=lambda t: (
+ t.get('preference') if t.get('preference') is not None else -1,
+ t.get('width') if t.get('width') is not None else -1,
+ t.get('height') if t.get('height') is not None else -1,
+ t.get('id') if t.get('id') is not None else '',
+ t.get('url')))
+
def _sanitize_thumbnails(self, info_dict):
thumbnails = info_dict.get('thumbnails')
if thumbnails is None:
thumbnail = info_dict.get('thumbnail')
if thumbnail:
info_dict['thumbnails'] = thumbnails = [{'url': thumbnail}]
- if thumbnails:
- thumbnails.sort(key=lambda t: (
- t.get('preference') if t.get('preference') is not None else -1,
- t.get('width') if t.get('width') is not None else -1,
- t.get('height') if t.get('height') is not None else -1,
- t.get('id') if t.get('id') is not None else '',
- t.get('url')))
-
- def thumbnail_tester():
- if self.params.get('check_formats'):
- test_all = True
- to_screen = lambda msg: self.to_screen(f'[info] {msg}')
- else:
- test_all = False
- to_screen = self.write_debug
-
- def test_thumbnail(t):
- if not test_all and not t.get('_test_url'):
- return True
- to_screen('Testing thumbnail %s' % t['id'])
- try:
- self.urlopen(HEADRequest(t['url']))
- except network_exceptions as err:
- to_screen('Unable to connect to thumbnail %s URL "%s" - %s. Skipping...' % (
- t['id'], t['url'], error_to_compat_str(err)))
- return False
- return True
-
- return test_thumbnail
-
- for i, t in enumerate(thumbnails):
- if t.get('id') is None:
- t['id'] = '%d' % i
- if t.get('width') and t.get('height'):
- t['resolution'] = '%dx%d' % (t['width'], t['height'])
- t['url'] = sanitize_url(t['url'])
+ if not thumbnails:
+ return
- if self.params.get('check_formats') is not False:
- info_dict['thumbnails'] = LazyList(filter(thumbnail_tester(), thumbnails[::-1])).reverse()
- else:
- info_dict['thumbnails'] = thumbnails
+ def check_thumbnails(thumbnails):
+ for t in thumbnails:
+ self.to_screen(f'[info] Testing thumbnail {t["id"]}')
+ try:
+ self.urlopen(HEADRequest(t['url']))
+ except network_exceptions as err:
+ self.to_screen(f'[info] Unable to connect to thumbnail {t["id"]} URL {t["url"]!r} - {err}. Skipping...')
+ continue
+ yield t
+
+ self._sort_thumbnails(thumbnails)
+ for i, t in enumerate(thumbnails):
+ if t.get('id') is None:
+ t['id'] = '%d' % i
+ if t.get('width') and t.get('height'):
+ t['resolution'] = '%dx%d' % (t['width'], t['height'])
+ t['url'] = sanitize_url(t['url'])
+
+ if self.params.get('check_formats') is True:
+ info_dict['thumbnails'] = LazyList(check_thumbnails(thumbnails[::-1])).reverse()
+ else:
+ info_dict['thumbnails'] = thumbnails
def process_video_result(self, info_dict, download=True):
assert info_dict.get('_type', 'video') == 'video'
if 'id' not in info_dict:
raise ExtractorError('Missing "id" field in extractor result')
if 'title' not in info_dict:
- raise ExtractorError('Missing "title" field in extractor result')
+ raise ExtractorError('Missing "title" field in extractor result',
+ video_id=info_dict['id'], ie=info_dict['extractor'])
def report_force_conversion(field, field_not, conversion):
self.report_warning(
if info_dict.get('display_id') is None and 'id' in info_dict:
info_dict['display_id'] = info_dict['id']
+ if info_dict.get('duration') is not None:
+ info_dict['duration_string'] = formatSeconds(info_dict['duration'])
+
for ts_key, date_key in (
('timestamp', 'upload_date'),
('release_timestamp', 'release_date'),
info_dict['requested_subtitles'] = self.process_subtitles(
info_dict['id'], subtitles, automatic_captions)
- # We now pick which formats have to be downloaded
if info_dict.get('formats') is None:
# There's only one format available
formats = [info_dict]
else:
formats = info_dict['formats']
+ info_dict['__has_drm'] = any(f.get('has_drm') for f in formats)
+ if not self.params.get('allow_unplayable_formats'):
+ formats = [f for f in formats if not f.get('has_drm')]
+
if not formats:
- if not self.params.get('ignore_no_formats_error'):
- raise ExtractorError('No video formats found!')
- else:
- self.report_warning('No video formats found!')
+ self.raise_no_formats(info_dict)
def is_wellformed(f):
url = f.get('url')
formats_dict[format_id].append(format)
# Make sure all formats have unique format_id
+ common_exts = set(itertools.chain(*self._format_selection_exts.values()))
for format_id, ambiguous_formats in formats_dict.items():
- if len(ambiguous_formats) > 1:
- for i, format in enumerate(ambiguous_formats):
+ ambigious_id = len(ambiguous_formats) > 1
+ 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=' ({0})'.format(format['format_note']) if format.get('format_note') is not None else '',
+ note=format_field(format, 'format_note', ' (%s)'),
)
- # Automatically determine file extension if missing
- if format.get('ext') is None:
- format['ext'] = determine_ext(format['url']).lower()
- # Automatically determine protocol if missing (useful for format
- # selection purposes)
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'] = 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()
# TODO Central sorting goes here
- if formats and formats[0] is not info_dict:
+ if self.params.get('check_formats') is True:
+ formats = LazyList(self._check_formats(formats[::-1])).reverse()
+
+ if not formats or formats[0] is not info_dict:
# only set the 'formats' fields if the original info_dict list them
# otherwise we end up with a circular reference, the first (and unique)
# element in the 'formats' field in info_dict is info_dict itself,
info_dict, _ = self.pre_process(info_dict)
- list_only = self.params.get('list_thumbnails') or self.params.get('listformats') or self.params.get('listsubtitles')
+ if self.params.get('list_thumbnails'):
+ self.list_thumbnails(info_dict)
+ if self.params.get('listformats'):
+ if not info_dict.get('formats') and not info_dict.get('url'):
+ self.to_screen('%s has no formats' % info_dict['id'])
+ else:
+ self.list_formats(info_dict)
+ if self.params.get('listsubtitles'):
+ if 'automatic_captions' in info_dict:
+ self.list_subtitles(
+ info_dict['id'], automatic_captions, 'automatic captions')
+ self.list_subtitles(info_dict['id'], subtitles, 'subtitles')
+ list_only = self.params.get('simulate') is None and (
+ self.params.get('list_thumbnails') or self.params.get('listformats') or self.params.get('listsubtitles'))
if list_only:
+ # Without this printing, -F --print-json will not work
self.__forced_printings(info_dict, self.prepare_filename(info_dict), incomplete=True)
- if self.params.get('list_thumbnails'):
- self.list_thumbnails(info_dict)
- if self.params.get('listformats'):
- if not info_dict.get('formats'):
- raise ExtractorError('No video formats found', expected=True)
- self.list_formats(info_dict)
- if self.params.get('listsubtitles'):
- if 'automatic_captions' in info_dict:
- self.list_subtitles(
- info_dict['id'], automatic_captions, 'automatic captions')
- self.list_subtitles(info_dict['id'], subtitles, 'subtitles')
return
format_selector = self.format_selector
formats_to_download = list(format_selector(ctx))
if not formats_to_download:
if not self.params.get('ignore_no_formats_error'):
- raise ExtractorError('Requested format is not available', expected=True)
+ raise ExtractorError('Requested format is not available', expected=True,
+ video_id=info_dict['id'], ie=info_dict['extractor'])
else:
self.report_warning('Requested format is not available')
# Process what we can, even without any available formats.
new_info['__original_infodict'] = info_dict
new_info.update(fmt)
self.process_info(new_info)
- # We update the info dict with the best quality format (backwards compatibility)
+ # We update the info dict with the selected best quality format (backwards compatibility)
if formats_to_download:
info_dict.update(formats_to_download[-1])
return info_dict
if self.params.get('allsubtitles', False):
requested_langs = all_sub_langs
elif self.params.get('subtitleslangs', False):
- requested_langs = set()
- for lang in self.params.get('subtitleslangs'):
- if lang == 'all':
- requested_langs.update(all_sub_langs)
+ # 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'):
+ if lang_re == 'all':
+ requested_langs.extend(all_sub_langs)
continue
- discard = lang[0] == '-'
+ discard = lang_re[0] == '-'
if discard:
- lang = lang[1:]
- current_langs = filter(re.compile(lang + '$').match, all_sub_langs)
+ lang_re = lang_re[1:]
+ current_langs = filter(re.compile(lang_re + '$').match, all_sub_langs)
if discard:
for lang in current_langs:
- requested_langs.discard(lang)
+ while lang in requested_langs:
+ requested_langs.remove(lang)
else:
- requested_langs.update(current_langs)
+ requested_langs.extend(current_langs)
+ requested_langs = orderedSet(requested_langs)
elif 'en' in available_subs:
requested_langs = ['en']
else:
requested_langs = [list(all_sub_langs)[0]]
- self.write_debug('Downloading subtitles: %s' % ', '.join(requested_langs))
+ if requested_langs:
+ self.write_debug('Downloading subtitles: %s' % ', '.join(requested_langs))
formats_query = self.params.get('subtitlesformat', 'best')
formats_preference = formats_query.split('/') if formats_query else []
elif 'url' in info_dict:
info_dict['urls'] = info_dict['url'] + info_dict.get('play_path', '')
+ if self.params.get('forceprint') or self.params.get('forcejson'):
+ self.post_extract(info_dict)
for tmpl in self.params.get('forceprint', []):
- if re.match(r'\w+$', tmpl):
+ mobj = re.match(r'\w+(=?)$', tmpl)
+ if mobj and mobj.group(1):
+ tmpl = f'{tmpl[:-1]} = %({tmpl[:-1]})s'
+ elif mobj:
tmpl = '%({})s'.format(tmpl)
- tmpl, info_copy = self.prepare_outtmpl(tmpl, info_dict)
- self.to_stdout(self.escape_outtmpl(tmpl) % info_copy)
+ self.to_stdout(self.evaluate_outtmpl(tmpl, info_dict))
print_mandatory('title')
print_mandatory('id')
print_optional('thumbnail')
print_optional('description')
print_optional('filename')
- if self.params.get('forceduration', False) and info_dict.get('duration') is not None:
+ if self.params.get('forceduration') and info_dict.get('duration') is not None:
self.to_stdout(formatSeconds(info_dict['duration']))
print_mandatory('format')
- if self.params.get('forcejson', False):
- self.post_extract(info_dict)
- self.to_stdout(json.dumps(self.sanitize_info(info_dict), default=repr))
+ if self.params.get('forcejson'):
+ self.to_stdout(json.dumps(self.sanitize_info(info_dict)))
def dl(self, name, info, subtitle=False, test=False):
+ if not info.get('url'):
+ self.raise_no_formats(info, True)
if test:
verbose = self.params.get('verbose')
params = {
'test': True,
- 'quiet': not verbose,
+ 'quiet': self.params.get('quiet') or not verbose,
'verbose': verbose,
'noprogress': not verbose,
'nopart': True,
fd.add_progress_hook(ph)
urls = '", "'.join([f['url'] for f in info.get('requested_formats', [])] or [info['url']])
self.write_debug('Invoking downloader on "%s"' % urls)
- new_info = dict(info)
+
+ new_info = copy.deepcopy(self._copy_infodict(info))
if new_info.get('http_headers') is None:
new_info['http_headers'] = self._calc_headers(new_info)
return fd.download(name, new_info, subtitle)
assert info_dict.get('_type', 'video') == 'video'
- info_dict.setdefault('__postprocessors', [])
-
max_downloads = self.params.get('max_downloads')
if max_downloads is not None:
if self._num_downloads >= int(max_downloads):
# Forced printings
self.__forced_printings(info_dict, full_filename, incomplete=('format' not in info_dict))
- if self.params.get('simulate', False):
+ if self.params.get('simulate'):
if self.params.get('force_write_download_archive', False):
self.record_download_archive(info_dict)
-
# Do nothing else if in simulate mode
return
if full_filename is None:
return
-
if not self._ensure_dir_exists(encodeFilename(full_filename)):
return
if not self._ensure_dir_exists(encodeFilename(temp_filename)):
return
- if self.params.get('writedescription', False):
- descfn = self.prepare_filename(info_dict, 'description')
- if not self._ensure_dir_exists(encodeFilename(descfn)):
- return
- if not self.params.get('overwrites', True) and os.path.exists(encodeFilename(descfn)):
- self.to_screen('[info] Video description is already present')
- elif info_dict.get('description') is None:
- self.report_warning('There\'s no description to write.')
- else:
- try:
- self.to_screen('[info] Writing video description to: ' + descfn)
- with io.open(encodeFilename(descfn), 'w', encoding='utf-8') as descfile:
- descfile.write(info_dict['description'])
- except (OSError, IOError):
- self.report_error('Cannot write description file ' + descfn)
- return
+ if self._write_description('video', info_dict,
+ self.prepare_filename(info_dict, 'description')) is None:
+ return
+
+ sub_files = self._write_subtitles(info_dict, temp_filename)
+ if sub_files is None:
+ return
+ files_to_move.update(dict(sub_files))
+
+ thumb_files = self._write_thumbnails(
+ 'video', info_dict, temp_filename, self.prepare_filename(info_dict, 'thumbnail'))
+ if thumb_files is None:
+ return
+ files_to_move.update(dict(thumb_files))
+
+ infofn = self.prepare_filename(info_dict, 'infojson')
+ _infojson_written = self._write_info_json('video', info_dict, infofn)
+ if _infojson_written:
+ info_dict['__infojson_filename'] = infofn
+ elif _infojson_written is None:
+ return
+ # Note: Annotations are deprecated
+ annofn = None
if self.params.get('writeannotations', False):
annofn = self.prepare_filename(info_dict, 'annotation')
+ if annofn:
if not self._ensure_dir_exists(encodeFilename(annofn)):
return
if not self.params.get('overwrites', True) and os.path.exists(encodeFilename(annofn)):
self.report_error('Cannot write annotations file: ' + annofn)
return
- subtitles_are_requested = any([self.params.get('writesubtitles', False),
- self.params.get('writeautomaticsub')])
-
- if subtitles_are_requested and info_dict.get('requested_subtitles'):
- # subtitles download errors are already managed as troubles in relevant IE
- # that way it will silently go on when used with unsupporting IE
- subtitles = info_dict['requested_subtitles']
- # ie = self.get_info_extractor(info_dict['extractor_key'])
- for sub_lang, sub_info in subtitles.items():
- sub_format = sub_info['ext']
- sub_filename = subtitles_filename(temp_filename, sub_lang, sub_format, info_dict.get('ext'))
- sub_filename_final = subtitles_filename(
- self.prepare_filename(info_dict, 'subtitle'), sub_lang, sub_format, info_dict.get('ext'))
- if not self.params.get('overwrites', True) and os.path.exists(encodeFilename(sub_filename)):
- self.to_screen('[info] Video subtitle %s.%s is already present' % (sub_lang, sub_format))
- sub_info['filepath'] = sub_filename
- files_to_move[sub_filename] = sub_filename_final
- else:
- self.to_screen('[info] Writing video subtitles to: ' + sub_filename)
- if sub_info.get('data') is not None:
- try:
- # Use newline='' to prevent conversion of newline characters
- # See https://github.com/ytdl-org/youtube-dl/issues/10268
- with io.open(encodeFilename(sub_filename), 'w', encoding='utf-8', newline='') as subfile:
- subfile.write(sub_info['data'])
- sub_info['filepath'] = sub_filename
- files_to_move[sub_filename] = sub_filename_final
- except (OSError, IOError):
- self.report_error('Cannot write subtitles file ' + sub_filename)
- return
- else:
- try:
- self.dl(sub_filename, sub_info.copy(), subtitle=True)
- sub_info['filepath'] = sub_filename
- files_to_move[sub_filename] = sub_filename_final
- except (ExtractorError, IOError, OSError, ValueError) + network_exceptions as err:
- self.report_warning('Unable to download subtitle for "%s": %s' %
- (sub_lang, error_to_compat_str(err)))
- continue
-
- if self.params.get('writeinfojson', False):
- infofn = self.prepare_filename(info_dict, 'infojson')
- if not self._ensure_dir_exists(encodeFilename(infofn)):
- return
- if not self.params.get('overwrites', True) and os.path.exists(encodeFilename(infofn)):
- self.to_screen('[info] Video metadata is already present')
- else:
- self.to_screen('[info] Writing video metadata as JSON to: ' + infofn)
- try:
- write_json_file(self.sanitize_info(info_dict, self.params.get('clean_infojson', True)), infofn)
- except (OSError, IOError):
- self.report_error('Cannot write video metadata to JSON file ' + infofn)
- return
- info_dict['__infojson_filename'] = infofn
-
- for thumb_ext in self._write_thumbnails(info_dict, temp_filename):
- thumb_filename_temp = replace_extension(temp_filename, thumb_ext, info_dict.get('ext'))
- thumb_filename = replace_extension(
- self.prepare_filename(info_dict, 'thumbnail'), thumb_ext, info_dict.get('ext'))
- files_to_move[thumb_filename_temp] = thumb_filename
-
# Write internet shortcut files
- url_link = webloc_link = desktop_link = False
- if self.params.get('writelink', False):
- if sys.platform == "darwin": # macOS.
- webloc_link = True
- elif sys.platform.startswith("linux"):
- desktop_link = True
- else: # if sys.platform in ['win32', 'cygwin']:
- url_link = True
- if self.params.get('writeurllink', False):
- url_link = True
- if self.params.get('writewebloclink', False):
- webloc_link = True
- if self.params.get('writedesktoplink', False):
- desktop_link = True
-
- if url_link or webloc_link or desktop_link:
+ def _write_link_file(link_type):
if 'webpage_url' not in info_dict:
self.report_error('Cannot write internet shortcut file because the "webpage_url" field is missing in the media information')
- return
- ascii_url = iri_to_uri(info_dict['webpage_url'])
-
- def _write_link_file(extension, template, newline, embed_filename):
- linkfn = replace_extension(full_filename, extension, info_dict.get('ext'))
+ return False
+ linkfn = replace_extension(self.prepare_filename(info_dict, 'link'), link_type, info_dict.get('ext'))
if self.params.get('overwrites', True) and os.path.exists(encodeFilename(linkfn)):
- self.to_screen('[info] Internet shortcut is already present')
- else:
- try:
- self.to_screen('[info] Writing internet shortcut to: ' + linkfn)
- with io.open(encodeFilename(to_high_limit_path(linkfn)), 'w', encoding='utf-8', newline=newline) as linkfile:
- template_vars = {'url': ascii_url}
- if embed_filename:
- template_vars['filename'] = linkfn[:-(len(extension) + 1)]
- linkfile.write(template % template_vars)
- except (OSError, IOError):
- self.report_error('Cannot write internet shortcut ' + linkfn)
- return False
+ self.to_screen(f'[info] Internet shortcut (.{link_type}) is already present')
+ return True
+ try:
+ self.to_screen(f'[info] Writing internet shortcut (.{link_type}) to: {linkfn}')
+ with io.open(encodeFilename(to_high_limit_path(linkfn)), 'w', encoding='utf-8',
+ newline='\r\n' if link_type == 'url' else '\n') as linkfile:
+ template_vars = {'url': iri_to_uri(info_dict['webpage_url'])}
+ if link_type == 'desktop':
+ template_vars['filename'] = linkfn[:-(len(link_type) + 1)]
+ linkfile.write(LINK_TEMPLATES[link_type] % template_vars)
+ except (OSError, IOError):
+ self.report_error(f'Cannot write internet shortcut {linkfn}')
+ return False
return True
- if url_link:
- if not _write_link_file('url', DOT_URL_LINK_TEMPLATE, '\r\n', embed_filename=False):
- return
- if webloc_link:
- if not _write_link_file('webloc', DOT_WEBLOC_LINK_TEMPLATE, '\n', embed_filename=False):
- return
- if desktop_link:
- if not _write_link_file('desktop', DOT_DESKTOP_LINK_TEMPLATE, '\n', embed_filename=True):
- return
+ write_links = {
+ 'url': self.params.get('writeurllink'),
+ 'webloc': self.params.get('writewebloclink'),
+ 'desktop': self.params.get('writedesktoplink'),
+ }
+ if self.params.get('writelink'):
+ link_type = ('webloc' if sys.platform == 'darwin'
+ else 'desktop' if sys.platform.startswith('linux')
+ else 'url')
+ write_links[link_type] = True
+
+ if any(should_write and not _write_link_file(link_type)
+ for link_type, should_write in write_links.items()):
+ return
try:
info_dict, files_to_move = self.pre_process(info_dict, 'before_dl', files_to_move)
info_dict = self.run_pp(MoveFilesAfterDownloadPP(self, False), info_dict)
else:
# Download
+ info_dict.setdefault('__postprocessors', [])
try:
def existing_file(*filepaths):
os.remove(encodeFilename(file))
return None
- self.report_file_already_downloaded(existing_files[0])
info_dict['ext'] = os.path.splitext(existing_files[0])[1][1:]
return existing_files[0]
requested_formats = info_dict['requested_formats']
old_ext = info_dict['ext']
- if self.params.get('merge_output_format') is None and 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 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
+ # since we dont want any custom PPs to trigger this
+ and any(type(pp) == EmbedThumbnailPP for pp in self._pps['post_process'])):
+ info_dict['ext'] = 'mkv'
+ self.report_warning(
+ 'webm doesn\'t support embedding a thumbnail, mkv will be used')
new_ext = info_dict['ext']
def correct_ext(filename, ext=new_ext):
dl_filename = existing_file(full_filename, temp_filename)
info_dict['__real_download'] = False
- _protocols = set(determine_protocol(f) for f in requested_formats)
- if len(_protocols) == 1: # All requested formats have same protocol
- info_dict['protocol'] = _protocols.pop()
- directly_mergable = FFmpegFD.can_merge_formats(info_dict)
if dl_filename is not None:
- pass
- elif (directly_mergable and get_suitable_downloader(
- info_dict, self.params, to_stdout=(temp_filename == '-')) == FFmpegFD):
+ self.report_file_already_downloaded(dl_filename)
+ elif get_suitable_downloader(info_dict, self.params, to_stdout=temp_filename == '-'):
info_dict['url'] = '\n'.join(f['url'] for f in requested_formats)
success, real_download = self.dl(temp_filename, info_dict)
info_dict['__real_download'] = real_download
'The formats won\'t be merged.')
if temp_filename == '-':
- reason = ('using a downloader other than ffmpeg' if directly_mergable
+ reason = ('using a downloader other than ffmpeg' if FFmpegFD.can_merge_formats(info_dict)
else 'but the formats are incompatible for simultaneous download' if merger.available
else 'but ffmpeg is not installed')
self.report_warning(
'f%s' % f['format_id'], new_info['ext'])
if not self._ensure_dir_exists(fname):
return
+ f['filepath'] = fname
downloaded.append(fname)
partial_success, real_download = self.dl(fname, new_info)
info_dict['__real_download'] = info_dict['__real_download'] or real_download
else:
# Just a single file
dl_filename = existing_file(full_filename, temp_filename)
- if dl_filename is None:
+ if dl_filename is None or dl_filename == temp_filename:
+ # dl_filename == temp_filename could mean that the file was partially downloaded with --no-part.
+ # So we should try to resume the download
success, real_download = self.dl(temp_filename, info_dict)
info_dict['__real_download'] = real_download
+ else:
+ self.report_file_already_downloaded(dl_filename)
dl_filename = dl_filename or temp_filename
info_dict['__finaldir'] = os.path.dirname(os.path.abspath(encodeFilename(full_filename)))
'writing DASH m4a. Only some players support this container',
FFmpegFixupM4aPP)
- downloader = (get_suitable_downloader(info_dict, self.params).__name__
- if 'protocol' in info_dict else None)
- ffmpeg_fixup(downloader == 'HlsFD', 'malformed AAC bitstream detected', FFmpegFixupM3u8PP)
+ downloader = get_suitable_downloader(info_dict, self.params) if 'protocol' in info_dict else None
+ downloader = downloader.__name__ if downloader else None
+ ffmpeg_fixup(info_dict.get('requested_formats') is None and downloader == 'HlsFD',
+ 'malformed AAC bitstream detected', FFmpegFixupM3u8PP)
ffmpeg_fixup(downloader == 'WebSocketFragmentFD', 'malformed timestamps detected', FFmpegFixupTimestampPP)
ffmpeg_fixup(downloader == 'WebSocketFragmentFD', 'malformed duration detected', FFmpegFixupDurationPP)
url, force_generic_extractor=self.params.get('force_generic_extractor', False))
except UnavailableVideoError:
self.report_error('unable to download video')
- except MaxDownloadsReached:
- self.to_screen('[info] Maximum number of downloaded files reached')
- raise
- except ExistingVideoReached:
- self.to_screen('[info] Encountered a file that is already in the archive, stopping due to --break-on-existing')
- raise
- except RejectedVideoReached:
- self.to_screen('[info] Encountered a file that did not match filter, stopping due to --break-on-reject')
+ except DownloadCancelled as e:
+ self.to_screen(f'[info] {e.msg}')
raise
else:
if self.params.get('dump_single_json', False):
self.post_extract(res)
- self.to_stdout(json.dumps(self.filter_requested_info(res), default=repr))
+ self.to_stdout(json.dumps(self.sanitize_info(res)))
return self._download_retcode
@staticmethod
def sanitize_info(info_dict, remove_private_keys=False):
''' Sanitize the infodict for converting to json '''
- remove_keys = ['__original_infodict'] # Always remove this since this may contain a copy of the entire dict
+ if info_dict is None:
+ return info_dict
+ info_dict.setdefault('epoch', int(time.time()))
+ remove_keys = {'__original_infodict'} # Always remove this since this may contain a copy of the entire dict
keep_keys = ['_type'], # Always keep this to facilitate load-info-json
if remove_private_keys:
- remove_keys += ('requested_formats', 'requested_subtitles', 'requested_entries', 'filepath', 'entries', 'original_url')
+ remove_keys |= {
+ 'requested_formats', 'requested_subtitles', 'requested_entries',
+ 'filepath', 'entries', 'original_url', 'playlist_autonumber',
+ }
empty_values = (None, {}, [], set(), tuple())
reject = lambda k, v: k not in keep_keys and (
k.startswith('_') or k in remove_keys or v in empty_values)
else:
- info_dict['epoch'] = int(time.time())
reject = lambda k, v: k in remove_keys
filter_fn = lambda obj: (
list(map(filter_fn, obj)) if isinstance(obj, (LazyList, list, tuple, set))
files_to_delete = []
if '__files_to_move' not in infodict:
infodict['__files_to_move'] = {}
- files_to_delete, infodict = pp.run(infodict)
+ try:
+ files_to_delete, infodict = pp.run(infodict)
+ except PostProcessingError as e:
+ # Must be True and not 'only_download'
+ if self.params.get('ignoreerrors') is True:
+ self.report_error(e)
+ return infodict
+ raise
+
if not files_to_delete:
return infodict
-
if self.params.get('keepvideo', False):
for f in files_to_delete:
infodict['__files_to_move'].setdefault(f, '')
if not url:
return
# Try to find matching extractor for the URL and take its ie_key
- for ie in self._ies:
+ for ie_key, ie in self._ies.items():
if ie.suitable(url):
- extractor = ie.ie_key()
+ extractor = ie_key
break
else:
return
@staticmethod
def format_resolution(format, default='unknown'):
- if format.get('vcodec') == 'none':
- if format.get('acodec') == 'none':
- return 'images'
+ is_images = format.get('vcodec') == 'none' and format.get('acodec') == 'none'
+ if format.get('vcodec') == 'none' and format.get('acodec') != 'none':
return 'audio only'
if format.get('resolution') is not None:
return format['resolution']
res = '%sp' % format['height']
elif format.get('width'):
res = '%dx?' % format['width']
+ elif is_images:
+ return 'images'
else:
- res = default
- return res
+ return default
+ return f'{res} images' if is_images else res
def _format_note(self, fdict):
res = ''
res += '~' + format_bytes(fdict['filesize_approx'])
return res
+ def _list_format_headers(self, *headers):
+ if self.params.get('listformats_table', True) is not False:
+ return [self._format_screen(header, self.Styles.HEADERS) for header in headers]
+ return headers
+
def list_formats(self, info_dict):
formats = info_dict.get('formats', [info_dict])
- new_format = (
- 'list-formats' not in self.params.get('compat_opts', [])
- and self.params.get('listformats_table', True) is not False)
+ new_format = self.params.get('listformats_table', True) is not False
if new_format:
+ tbr_digits = number_of_digits(max(f.get('tbr') or 0 for f in formats))
+ vbr_digits = number_of_digits(max(f.get('vbr') or 0 for f in formats))
+ abr_digits = number_of_digits(max(f.get('abr') or 0 for f in formats))
+ delim = self._format_screen('\u2502', self.Styles.DELIM, '|', test_encoding=True)
table = [
[
- format_field(f, 'format_id'),
+ self._format_screen(format_field(f, 'format_id'), self.Styles.ID),
format_field(f, 'ext'),
self.format_resolution(f),
format_field(f, 'fps', '%d'),
- '|',
+ format_field(f, 'dynamic_range', '%s', ignore=(None, 'SDR')).replace('HDR', ''),
+ delim,
format_field(f, 'filesize', ' %s', func=format_bytes) + format_field(f, 'filesize_approx', '~%s', func=format_bytes),
- format_field(f, 'tbr', '%4dk'),
+ format_field(f, 'tbr', f'%{tbr_digits}dk'),
shorten_protocol_name(f.get('protocol', '').replace("native", "n")),
- '|',
+ delim,
format_field(f, 'vcodec', default='unknown').replace('none', ''),
- format_field(f, 'vbr', '%4dk'),
+ format_field(f, 'vbr', f'%{vbr_digits}dk'),
format_field(f, 'acodec', default='unknown').replace('none', ''),
- format_field(f, 'abr', '%3dk'),
+ format_field(f, 'abr', f'%{abr_digits}dk'),
format_field(f, 'asr', '%5dHz'),
', '.join(filter(None, (
- 'UNSUPPORTED' if f.get('ext') in ('f4f', 'f4m') else '',
+ self._format_screen('UNSUPPORTED', 'light red') if f.get('ext') in ('f4f', 'f4m') else '',
format_field(f, 'language', '[%s]'),
format_field(f, 'format_note'),
format_field(f, 'container', ignore=(None, f.get('ext'))),
))),
] for f in formats if f.get('preference') is None or f['preference'] >= -1000]
- header_line = ['ID', 'EXT', 'RESOLUTION', 'FPS', '|', ' FILESIZE', ' TBR', 'PROTO',
- '|', 'VCODEC', ' VBR', 'ACODEC', ' ABR', ' ASR', 'MORE INFO']
+ header_line = self._list_format_headers(
+ 'ID', 'EXT', 'RESOLUTION', 'FPS', 'HDR', delim, ' FILESIZE', ' TBR', 'PROTO',
+ delim, 'VCODEC', ' VBR', 'ACODEC', ' ABR', ' ASR', 'MORE INFO')
else:
table = [
[
self.to_screen(
'[info] Available formats for %s:' % info_dict['id'])
self.to_stdout(render_table(
- header_line, table, delim=new_format, extraGap=(0 if new_format else 1), hideEmpty=new_format))
+ header_line, table,
+ extraGap=(0 if new_format else 1),
+ hideEmpty=new_format,
+ delim=new_format and self._format_screen('\u2500', self.Styles.DELIM, '-', test_encoding=True)))
def list_thumbnails(self, info_dict):
thumbnails = list(info_dict.get('thumbnails'))
self.to_screen(
'[info] Thumbnails for %s:' % info_dict['id'])
self.to_stdout(render_table(
- ['ID', 'width', 'height', 'URL'],
+ self._list_format_headers('ID', 'Width', 'Height', 'URL'),
[[t['id'], t.get('width', 'unknown'), t.get('height', 'unknown'), t['url']] for t in thumbnails]))
def list_subtitles(self, video_id, subtitles, name='subtitles'):
return [lang, ', '.join(names), ', '.join(exts)]
self.to_stdout(render_table(
- ['Language', 'Name', 'Formats'],
+ self._list_format_headers('Language', 'Name', 'Formats'),
[_row(lang, formats) for lang, formats in subtitles.items()],
hideEmpty=True))
if not self.params.get('verbose'):
return
- if type('') is not compat_str:
- # Python 2.6 on SLES11 SP1 (https://github.com/ytdl-org/youtube-dl/issues/3326)
- self.report_warning(
- 'Your Python is broken! Update to a newer and supported version')
-
- stdout_encoding = getattr(
- sys.stdout, 'encoding', 'missing (%s)' % type(sys.stdout).__name__)
- encoding_str = (
- '[debug] Encodings: locale %s, fs %s, out %s, pref %s\n' % (
- locale.getpreferredencoding(),
- sys.getfilesystemencoding(),
- stdout_encoding,
- self.get_encoding()))
- write_string(encoding_str, encoding=None)
-
- source = (
- '(exe)' if hasattr(sys, 'frozen')
- else '(zip)' if isinstance(globals().get('__loader__'), zipimporter)
- else '(source)' if os.path.basename(sys.argv[0]) == '__main__.py'
- else '')
- self._write_string('[debug] yt-dlp version %s %s\n' % (__version__, source))
- if _LAZY_LOADER:
- self._write_string('[debug] Lazy loading extractors enabled\n')
- if _PLUGIN_CLASSES:
- self._write_string(
- '[debug] Plugin Extractors: %s\n' % [ie.ie_key() for ie in _PLUGIN_CLASSES])
+ def get_encoding(stream):
+ ret = getattr(stream, 'encoding', 'missing (%s)' % type(stream).__name__)
+ if not supports_terminal_sequences(stream):
+ ret += ' (No ANSI)'
+ return ret
+
+ encoding_str = 'Encodings: locale %s, fs %s, out %s, err %s, pref %s' % (
+ locale.getpreferredencoding(),
+ sys.getfilesystemencoding(),
+ get_encoding(self._screen_file), get_encoding(self._err_file),
+ self.get_encoding())
+
+ logger = self.params.get('logger')
+ if logger:
+ write_debug = lambda msg: logger.debug(f'[debug] {msg}')
+ write_debug(encoding_str)
+ else:
+ write_string(f'[debug] {encoding_str}\n', encoding=None)
+ write_debug = lambda msg: self._write_string(f'[debug] {msg}\n')
+
+ source = detect_variant()
+ write_debug('yt-dlp version %s%s' % (__version__, '' if source == 'unknown' else f' ({source})'))
+ 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.get('compat_opts'):
- self._write_string(
- '[debug] Compatibility options: %s\n' % ', '.join(self.params.get('compat_opts')))
+ write_debug('Compatibility options: %s' % ', '.join(self.params.get('compat_opts')))
try:
- sp = subprocess.Popen(
+ sp = Popen(
['git', 'rev-parse', '--short', 'HEAD'],
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
cwd=os.path.dirname(os.path.abspath(__file__)))
- out, err = process_communicate_or_kill(sp)
+ out, err = sp.communicate_or_kill()
out = out.decode().strip()
if re.match('[0-9a-f]+', out):
- self._write_string('[debug] Git HEAD: %s\n' % out)
+ write_debug('Git HEAD: %s' % out)
except Exception:
try:
sys.exc_clear()
return impl_name + ' version %d.%d.%d' % sys.pypy_version_info[:3]
return impl_name
- self._write_string('[debug] Python version %s (%s %s) - %s\n' % (
+ write_debug('Python version %s (%s %s) - %s' % (
platform.python_version(),
python_implementation(),
platform.architecture()[0],
exe_versions['rtmpdump'] = rtmpdump_version()
exe_versions['phantomjs'] = PhantomJSwrapper._version()
exe_str = ', '.join(
- '%s %s' % (exe, v)
- for exe, v in sorted(exe_versions.items())
- if v
- )
- if not exe_str:
- exe_str = 'none'
- self._write_string('[debug] exe versions: %s\n' % exe_str)
+ f'{exe} {v}' for exe, v in sorted(exe_versions.items()) if v
+ ) or 'none'
+ write_debug('exe versions: %s' % exe_str)
+
+ from .downloader.websocket import has_websockets
+ from .postprocessor.embedthumbnail import has_mutagen
+ from .cookies import SQLITE_AVAILABLE, KEYRING_AVAILABLE
+
+ lib_str = ', '.join(sorted(filter(None, (
+ compat_pycrypto_AES and compat_pycrypto_AES.__name__.split('.')[0],
+ has_websockets and 'websockets',
+ has_mutagen and 'mutagen',
+ SQLITE_AVAILABLE and 'sqlite',
+ KEYRING_AVAILABLE and 'keyring',
+ )))) or 'none'
+ write_debug('Optional libraries: %s' % lib_str)
proxy_map = {}
for handler in self._opener.handlers:
if hasattr(handler, 'proxies'):
proxy_map.update(handler.proxies)
- self._write_string('[debug] Proxy map: ' + compat_str(proxy_map) + '\n')
+ write_debug(f'Proxy map: {proxy_map}')
- if self.params.get('call_home', False):
+ # Not implemented
+ if False and self.params.get('call_home'):
ipaddr = self.urlopen('https://yt-dl.org/ip').read().decode('utf-8')
- self._write_string('[debug] Public IP address: %s\n' % ipaddr)
- return
+ write_debug('Public IP address: %s' % ipaddr)
latest_version = self.urlopen(
'https://yt-dl.org/latest/version').read().decode('utf-8')
if version_tuple(latest_version) > version_tuple(__version__):
def _setup_opener(self):
timeout_val = self.params.get('socket_timeout')
- self._socket_timeout = 600 if timeout_val is None else float(timeout_val)
+ self._socket_timeout = 20 if timeout_val is None else float(timeout_val)
opts_cookiesfrombrowser = self.params.get('cookiesfrombrowser')
opts_cookiefile = self.params.get('cookiefile')
encoding = preferredencoding()
return encoding
- def _write_thumbnails(self, info_dict, filename): # return the extensions
+ def _write_info_json(self, label, ie_result, infofn):
+ ''' Write infojson and returns True = written, False = skip, None = error '''
+ if not self.params.get('writeinfojson'):
+ return False
+ elif not infofn:
+ self.write_debug(f'Skipping writing {label} infojson')
+ return False
+ elif not self._ensure_dir_exists(infofn):
+ return None
+ elif not self.params.get('overwrites', True) and os.path.exists(infofn):
+ self.to_screen(f'[info] {label.title()} metadata is already present')
+ else:
+ self.to_screen(f'[info] Writing {label} metadata as JSON to: {infofn}')
+ try:
+ write_json_file(self.sanitize_info(ie_result, self.params.get('clean_infojson', True)), infofn)
+ except (OSError, IOError):
+ self.report_error(f'Cannot write {label} metadata to JSON file {infofn}')
+ return None
+ return True
+
+ def _write_description(self, label, ie_result, descfn):
+ ''' Write description and returns True = written, False = skip, None = error '''
+ if not self.params.get('writedescription'):
+ return False
+ elif not descfn:
+ self.write_debug(f'Skipping writing {label} description')
+ return False
+ elif not self._ensure_dir_exists(descfn):
+ return None
+ 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')
+ return False
+ else:
+ try:
+ self.to_screen(f'[info] Writing {label} description to: {descfn}')
+ with io.open(encodeFilename(descfn), 'w', encoding='utf-8') as descfile:
+ descfile.write(ie_result['description'])
+ except (OSError, IOError):
+ self.report_error(f'Cannot write {label} description file {descfn}')
+ return None
+ return True
+
+ def _write_subtitles(self, info_dict, filename):
+ ''' 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')):
+ # 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
+
+ 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'))
+ sub_filename_final = subtitles_filename(sub_filename_base, sub_lang, sub_format, info_dict.get('ext'))
+ if not self.params.get('overwrites', True) and os.path.exists(sub_filename):
+ self.to_screen(f'[info] Video subtitle {sub_lang}.{sub_format} is already present')
+ sub_info['filepath'] = sub_filename
+ ret.append((sub_filename, sub_filename_final))
+ continue
+
+ self.to_screen(f'[info] Writing video subtitles to: {sub_filename}')
+ if sub_info.get('data') is not None:
+ try:
+ # Use newline='' to prevent conversion of newline characters
+ # See https://github.com/ytdl-org/youtube-dl/issues/10268
+ with io.open(sub_filename, 'w', encoding='utf-8', newline='') as subfile:
+ subfile.write(sub_info['data'])
+ sub_info['filepath'] = sub_filename
+ ret.append((sub_filename, sub_filename_final))
+ continue
+ except (OSError, IOError):
+ self.report_error(f'Cannot write video subtitles file {sub_filename}')
+ return None
+
+ try:
+ sub_copy = sub_info.copy()
+ sub_copy.setdefault('http_headers', info_dict.get('http_headers'))
+ self.dl(sub_filename, sub_copy, subtitle=True)
+ sub_info['filepath'] = sub_filename
+ ret.append((sub_filename, sub_filename_final))
+ except (ExtractorError, IOError, OSError, ValueError) + network_exceptions as err:
+ self.report_warning(f'Unable to download video subtitles for {sub_lang!r}: {err}')
+ continue
+ return ret
+
+ def _write_thumbnails(self, label, info_dict, filename, thumb_filename_base=None):
+ ''' Write thumbnails to file and return list of (thumb_filename, final_thumb_filename) '''
write_all = self.params.get('write_all_thumbnails', False)
- thumbnails = []
+ thumbnails, ret = [], []
if write_all or self.params.get('writethumbnail', False):
thumbnails = info_dict.get('thumbnails') or []
multiple = write_all and len(thumbnails) > 1
- ret = []
+ if thumb_filename_base is None:
+ thumb_filename_base = filename
+ if thumbnails and not thumb_filename_base:
+ self.write_debug(f'Skipping writing {label} thumbnail')
+ return ret
+
for t in thumbnails[::-1]:
- thumb_ext = determine_ext(t['url'], 'jpg')
- suffix = '%s.' % t['id'] if multiple else ''
- thumb_display_id = '%s ' % t['id'] if multiple else ''
- thumb_filename = replace_extension(filename, suffix + thumb_ext, info_dict.get('ext'))
+ thumb_ext = (f'{t["id"]}.' if multiple else '') + determine_ext(t['url'], 'jpg')
+ thumb_display_id = f'{label} thumbnail' + (f' {t["id"]}' if multiple else '')
+ thumb_filename = replace_extension(filename, thumb_ext, info_dict.get('ext'))
+ thumb_filename_final = replace_extension(thumb_filename_base, thumb_ext, info_dict.get('ext'))
- if not self.params.get('overwrites', True) and os.path.exists(encodeFilename(thumb_filename)):
- ret.append(suffix + thumb_ext)
+ if not self.params.get('overwrites', True) and os.path.exists(thumb_filename):
+ ret.append((thumb_filename, thumb_filename_final))
t['filepath'] = thumb_filename
- self.to_screen('[%s] %s: Thumbnail %sis already present' %
- (info_dict['extractor'], info_dict['id'], thumb_display_id))
+ self.to_screen(f'[info] {thumb_display_id.title()} is already present')
else:
- self.to_screen('[%s] %s: Downloading thumbnail %s ...' %
- (info_dict['extractor'], info_dict['id'], thumb_display_id))
+ self.to_screen(f'[info] Downloading {thumb_display_id} ...')
try:
uf = self.urlopen(t['url'])
+ self.to_screen(f'[info] Writing {thumb_display_id} to: {thumb_filename}')
with open(encodeFilename(thumb_filename), 'wb') as thumbf:
shutil.copyfileobj(uf, thumbf)
- ret.append(suffix + thumb_ext)
- self.to_screen('[%s] %s: Writing thumbnail %sto: %s' %
- (info_dict['extractor'], info_dict['id'], thumb_display_id, thumb_filename))
+ ret.append((thumb_filename, thumb_filename_final))
t['filepath'] = thumb_filename
except network_exceptions as err:
- self.report_warning('Unable to download thumbnail "%s": %s' %
- (t['url'], error_to_compat_str(err)))
+ self.report_warning(f'Unable to download {thumb_display_id}: {err}')
if ret and not write_all:
break
return ret
projects / yt-dlp.git / blobdiff