PostProcessingError,
preferredencoding,
prepend_extension,
+ random_uuidv4,
register_socks_protocols,
render_table,
replace_extension,
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')
+
def _parse_filter(tokens):
filter_parts = []
for type, string, start, _, _ in tokens:
return new_dict
+ def _check_formats(formats):
+ for f in formats:
+ self.to_screen('[info] Testing format %s' % f['format_id'])
+ paths = self.params.get('paths', {})
+ temp_file = os.path.join(
+ expand_path(paths.get('home', '').strip()),
+ expand_path(paths.get('temp', '').strip()),
+ 'ytdl.%s.f%s.check-format' % (random_uuidv4(), f['format_id']))
+ dl, _ = self.dl(temp_file, f, test=True)
+ if os.path.exists(temp_file):
+ os.remove(temp_file)
+ if dl:
+ yield f
+ else:
+ self.to_screen('[info] Unable to download format %s. Skipping...' % f['format_id'])
+
def _build_selector_function(selector):
if isinstance(selector, list): # ,
fs = [_build_selector_function(s) for s in selector]
if format_spec == 'all':
def selector_function(ctx):
formats = list(ctx['formats'])
- if formats:
- for f in formats:
- yield f
+ if check_formats:
+ formats = _check_formats(formats)
+ for f in formats:
+ yield f
elif format_spec == 'mergeall':
def selector_function(ctx):
- formats = list(ctx['formats'])
+ formats = list(_check_formats(ctx['formats']))
if not formats:
return
merged_format = formats[-1]
yield merged_format
else:
- format_fallback = False
+ format_fallback, format_reverse, format_idx = False, True, 1
mobj = re.match(
r'(?P<bw>best|worst|b|w)(?P<type>video|audio|v|a)?(?P<mod>\*)?(?:\.(?P<n>[1-9]\d*))?$',
format_spec)
if mobj is not None:
format_idx = int_or_none(mobj.group('n'), default=1)
- format_idx = format_idx - 1 if mobj.group('bw')[0] == 'w' else -format_idx
+ format_reverse = mobj.group('bw')[0] == 'b'
format_type = (mobj.group('type') or [None])[0]
not_format_type = {'v': 'a', 'a': 'v'}.get(format_type)
format_modified = mobj.group('mod') is not None
if not format_modified # b, w
else None) # b*, w*
else:
- format_idx = -1
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 not formats:
return
matches = list(filter(filter_f, formats)) if filter_f is not None else formats
- n = len(matches)
- if -n <= format_idx < n:
- yield matches[format_idx]
- elif format_fallback and ctx['incomplete_formats']:
+ if format_fallback and ctx['incomplete_formats'] and not matches:
# for extractors with incomplete formats (audio only (soundcloud)
# or video only (imgur)) best/worst will fallback to
# best/worst {video,audio}-only format
- n = len(formats)
- if -n <= format_idx < n:
- yield formats[format_idx]
+ matches = formats
+ if format_reverse:
+ matches = matches[::-1]
+ if check_formats:
+ matches = list(itertools.islice(_check_formats(matches), format_idx))
+ n = len(matches)
+ if -n <= format_idx - 1 < n:
+ yield matches[format_idx - 1]
elif selector.type == MERGE: # +
selector_1, selector_2 = map(_build_selector_function, selector.selector)
self.post_extract(info_dict)
self.to_stdout(json.dumps(info_dict, default=repr))
+ def dl(self, name, info, subtitle=False, test=False):
+
+ if test:
+ verbose = self.params.get('verbose')
+ params = {
+ 'test': True,
+ 'quiet': not verbose,
+ 'verbose': verbose,
+ 'noprogress': not verbose,
+ 'nopart': True,
+ 'skip_unavailable_fragments': False,
+ 'keep_fragments': False,
+ 'overwrites': True,
+ '_no_ytdl_file': True,
+ }
+ else:
+ params = self.params
+ fd = get_suitable_downloader(info, params)(self, params)
+ if not test:
+ for ph in self._progress_hooks:
+ fd.add_progress_hook(ph)
+ if self.params.get('verbose'):
+ self.to_screen('[debug] Invoking downloader on %r' % info.get('url'))
+ new_info = dict(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)
+
def process_info(self, info_dict):
"""Process a single resolved IE result."""
self.report_error('Cannot write annotations file: ' + annofn)
return
- def dl(name, info, subtitle=False):
- fd = get_suitable_downloader(info, self.params)(self, self.params)
- for ph in self._progress_hooks:
- fd.add_progress_hook(ph)
- if self.params.get('verbose'):
- self.to_screen('[debug] Invoking downloader on %r' % info.get('url'))
- new_info = dict(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)
-
subtitles_are_requested = any([self.params.get('writesubtitles', False),
self.params.get('writeautomaticsub')])
return
else:
try:
- dl(sub_filename, sub_info.copy(), subtitle=True)
+ self.dl(sub_filename, sub_info.copy(), subtitle=True)
sub_info['filepath'] = sub_filename
files_to_move[sub_filename] = sub_filename_final
except tuple([ExtractorError, IOError, OSError, ValueError] + network_exceptions) as err:
if not self._ensure_dir_exists(fname):
return
downloaded.append(fname)
- partial_success, real_download = dl(fname, new_info)
+ partial_success, real_download = self.dl(fname, new_info)
info_dict['__real_download'] = info_dict['__real_download'] or real_download
success = success and partial_success
if merger.available and not self.params.get('allow_unplayable_formats'):
# Just a single file
dl_filename = existing_file(full_filename, temp_filename)
if dl_filename is None:
- success, real_download = dl(temp_filename, info_dict)
+ success, real_download = self.dl(temp_filename, info_dict)
info_dict['__real_download'] = real_download
dl_filename = dl_filename or temp_filename
projects / yt-dlp.git / commitdiff
