Popen,
PostProcessingError,
_get_exe_version_output,
+ deprecation_warning,
detect_exe_version,
determine_ext,
dfxp2srt,
traverse_obj,
variadic,
write_json_file,
- write_string,
)
EXT_TO_OUT_FORMATS = {
'ts': 'mpegts',
'wma': 'asf',
'wmv': 'asf',
+ 'weba': 'webm',
'vtt': 'webvtt',
}
ACODECS = {
def create_mapping_re(supported):
- return re.compile(r'{0}(?:/{0})*$'.format(r'(?:\s*\w+\s*>)?\s*(?:%s)\s*' % '|'.join(supported)))
+ return re.compile(r'{0}(?:/{0})*$'.format(r'(?:\s*\w+\s*>)?\s*(?:{})\s*'.format('|'.join(supported))))
def resolve_mapping(source, mapping):
f'ffmpeg-location {location} does not exist! Continuing without ffmpeg', only_once=True)
return {}
elif os.path.isdir(location):
- dirname, basename = location, None
+ dirname, basename, filename = location, None, None
else:
- basename = os.path.splitext(os.path.basename(location))[0]
- basename = next((p for p in programs if basename.startswith(p)), 'ffmpeg')
+ filename = os.path.basename(location)
+ basename = next((p for p in programs if p in filename), 'ffmpeg')
dirname = os.path.dirname(os.path.abspath(location))
- if basename in self._ffmpeg_to_avconv.keys():
+ if basename in self._ffmpeg_to_avconv:
self._prefer_ffmpeg = True
paths = {p: os.path.join(dirname, p) for p in programs}
+ if basename and basename in filename:
+ for p in programs:
+ path = os.path.join(dirname, filename.replace(basename, p))
+ if os.path.exists(path):
+ paths[p] = path
if basename:
paths[basename] = location
return paths
path = self._paths.get(prog)
if path in self._version_cache:
return self._version_cache[path], self._features_cache.get(path, {})
- out = _get_exe_version_output(path, ['-bsfs'], to_screen=self.write_debug)
+ out = _get_exe_version_output(path, ['-bsfs'])
ver = detect_exe_version(out) if out else False
if ver:
regexs = [
@functools.cached_property
def basename(self):
- self._version # run property
+ _ = self._version # run property
return self.basename
@functools.cached_property
def probe_basename(self):
- self._probe_version # run property
+ _ = self._probe_version # run property
return self.probe_basename
def _get_version(self, kind):
else:
self.probe_basename = basename
if basename == self._ffmpeg_to_avconv[kind]:
- self.deprecation_warning(
- f'Support for {self._ffmpeg_to_avconv[kind]} is deprecated and may be removed in a future version. Use {kind} instead')
+ self.deprecated_feature(f'Support for {self._ffmpeg_to_avconv[kind]} is deprecated and '
+ f'may be removed in a future version. Use {kind} instead')
return version
@functools.cached_property
None)
return num, len(streams)
+ def _fixup_chapters(self, info):
+ last_chapter = traverse_obj(info, ('chapters', -1))
+ if last_chapter and not last_chapter.get('end_time'):
+ last_chapter['end_time'] = self._get_real_video_duration(info['filepath'])
+
def _get_real_video_duration(self, filepath, fatal=True):
try:
duration = float_or_none(
cmd += [encodeArgument('-loglevel'), encodeArgument('repeat+info')]
def make_args(file, args, name, number):
- keys = ['_%s%d' % (name, number), '_%s' % name]
+ keys = [f'_{name}{number}', f'_{name}']
if name == 'o':
args += ['-movflags', '+faststart']
if number == 1:
make_args(path, list(opts), arg_type, i + 1)
for i, (path, opts) in enumerate(path_opts) if path)
- self.write_debug('ffmpeg command line: %s' % shell_quote(cmd))
+ self.write_debug(f'ffmpeg command line: {shell_quote(cmd)}')
_, stderr, returncode = Popen.run(
cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE)
if returncode not in variadic(expected_retcodes):
"""
concat_file = f'{out_file}.concat'
self.write_debug(f'Writing concat spec to {concat_file}')
- with open(concat_file, 'wt', encoding='utf-8') as f:
+ with open(concat_file, 'w', encoding='utf-8') as f:
f.writelines(self._concat_spec(in_files, concat_opts))
out_flags = list(self.stream_copy_opts(ext=determine_ext(out_file)))
class FFmpegExtractAudioPP(FFmpegPostProcessor):
- COMMON_AUDIO_EXTS = MEDIA_EXTENSIONS.common_audio + ('wma', )
+ COMMON_AUDIO_EXTS = (*MEDIA_EXTENSIONS.common_audio, 'wma')
SUPPORTED_EXTS = tuple(ACODECS.keys())
FORMAT_RE = create_mapping_re(('best', *SUPPORTED_EXTS))
acodec_opts = []
else:
acodec_opts = ['-acodec', codec]
- opts = ['-vn'] + acodec_opts + more_opts
+ opts = ['-vn', *acodec_opts, *more_opts]
try:
FFmpegPostProcessor.run_ffmpeg(self, path, out_path, opts)
except FFmpegPostProcessorError as err:
if acodec != 'copy':
more_opts = self._quality_args(acodec)
- # not os.path.splitext, since the latter does not work on unicode in all setups
- temp_path = new_path = f'{path.rpartition(".")[0]}.{extension}'
+ temp_path = new_path = replace_extension(path, extension, information['ext'])
if new_path == path:
if acodec == 'copy':
temp_path = prepend_extension(path, 'temp')
if (self._nopostoverwrites and os.path.exists(encodeFilename(new_path))
and os.path.exists(encodeFilename(orig_path))):
- self.to_screen('Post-process file %s exists, skipping' % new_path)
+ self.to_screen(f'Post-process file {new_path} exists, skipping')
return [], information
self.to_screen(f'Destination: {new_path}')
class FFmpegVideoConvertorPP(FFmpegPostProcessor):
- SUPPORTED_EXTS = (*MEDIA_EXTENSIONS.common_video, *sorted(MEDIA_EXTENSIONS.common_audio + ('aac', 'vorbis')))
+ SUPPORTED_EXTS = (
+ *sorted((*MEDIA_EXTENSIONS.common_video, 'gif')),
+ *sorted((*MEDIA_EXTENSIONS.common_audio, 'aac', 'vorbis')),
+ )
FORMAT_RE = create_mapping_re(SUPPORTED_EXTS)
_ACTION = 'converting'
if not sub_langs:
return [], info
- input_files = [filename] + sub_filenames
+ input_files = [filename, *sub_filenames]
opts = [
*self.stream_copy_opts(ext=info['ext']),
'-map', '-0:s',
]
for i, (lang, name) in enumerate(zip(sub_langs, sub_names)):
- opts.extend(['-map', '%d:0' % (i + 1)])
+ opts.extend(['-map', f'{i + 1}:0'])
lang_code = ISO639Utils.short2long(lang) or lang
- opts.extend(['-metadata:s:s:%d' % i, 'language=%s' % lang_code])
+ opts.extend([f'-metadata:s:s:{i}', f'language={lang_code}'])
if name:
- opts.extend(['-metadata:s:s:%d' % i, 'handler_name=%s' % name,
- '-metadata:s:s:%d' % i, 'title=%s' % name])
+ opts.extend([f'-metadata:s:s:{i}', f'handler_name={name}',
+ f'-metadata:s:s:{i}', f'title={name}'])
temp_filename = prepend_extension(filename, 'temp')
- self.to_screen('Embedding subtitles in "%s"' % filename)
+ self.to_screen(f'Embedding subtitles in "{filename}"')
self.run_ffmpeg_multiple_files(input_files, temp_filename, opts)
os.replace(temp_filename, filename)
@PostProcessor._restrict_to(images=False)
def run(self, info):
+ self._fixup_chapters(info)
filename, metadata_filename = info['filepath'], None
files_to_delete, options = [], []
if self._add_chapters and info.get('chapters'):
return [], info
temp_filename = prepend_extension(filename, 'temp')
- self.to_screen('Adding metadata to "%s"' % filename)
+ self.to_screen(f'Adding metadata to "{filename}"')
self.run_ffmpeg_multiple_files(
(filename, metadata_filename), temp_filename,
itertools.chain(self._options(info['ext']), *options))
@staticmethod
def _get_chapter_opts(chapters, metadata_filename):
- with open(metadata_filename, 'wt', encoding='utf-8') as f:
+ with open(metadata_filename, 'w', encoding='utf-8') as f:
def ffmpeg_escape(text):
return re.sub(r'([\\=;#\n])', r'\\\1', text)
metadata_file_content += 'END=%d\n' % (chapter['end_time'] * 1000)
chapter_title = chapter.get('title')
if chapter_title:
- metadata_file_content += 'title=%s\n' % ffmpeg_escape(chapter_title)
+ metadata_file_content += f'title={ffmpeg_escape(chapter_title)}\n'
f.write(metadata_file_content)
yield ('-map_metadata', '1')
def add(meta_list, info_list=None):
value = next((
- str(info[key]) for key in [f'{meta_prefix}_'] + list(variadic(info_list or meta_list))
+ info[key] for key in [f'{meta_prefix}_', *variadic(info_list or meta_list)]
if info.get(key) is not None), None)
if value not in ('', None):
+ value = ', '.join(map(str, variadic(value)))
value = value.replace('\0', '') # nul character cannot be passed in command line
metadata['common'].update({meta_f: value for meta_f in variadic(meta_list)})
add(('description', 'synopsis'), 'description')
add(('purl', 'comment'), 'webpage_url')
add('track', 'track_number')
- add('artist', ('artist', 'creator', 'uploader', 'uploader_id'))
- add('genre')
+ add('artist', ('artist', 'artists', 'creator', 'creators', 'uploader', 'uploader_id'))
+ add('composer', ('composer', 'composers'))
+ add('genre', ('genre', 'genres'))
add('album')
- add('album_artist')
+ add('album_artist', ('album_artist', 'album_artists'))
add('disc', 'disc_number')
add('show', 'series')
add('season_number')
yield ('-metadata', f'{name}={value}')
stream_idx = 0
- for fmt in info.get('requested_formats') or []:
+ for fmt in info.get('requested_formats') or [info]:
stream_count = 2 if 'none' not in (fmt.get('vcodec'), fmt.get('acodec')) else 1
lang = ISO639Utils.short2long(fmt.get('language') or '') or fmt.get('language')
for i in range(stream_idx, stream_idx + stream_count):
old_stream, new_stream = self.get_stream_number(info['filepath'], ('tags', 'mimetype'), 'application/json')
if old_stream is not None:
- yield ('-map', '-0:%d' % old_stream)
+ yield ('-map', f'-0:{old_stream}')
new_stream -= 1
yield (
- '-attach', infofn,
+ '-attach', self._ffmpeg_filename_argument(infofn),
f'-metadata:s:{new_stream}', 'mimetype=application/json',
f'-metadata:s:{new_stream}', 'filename=info.json',
)
args.extend([f'-bsf:a:{audio_streams}', 'aac_adtstoasc'])
audio_streams += 1
if fmt.get('vcodec') != 'none':
- args.extend(['-map', '%u:v:0' % (i)])
- self.to_screen('Merging formats into "%s"' % filename)
+ args.extend(['-map', f'{i}:v:0'])
+ self.to_screen(f'Merging formats into "{filename}"')
self.run_ffmpeg_multiple_files(info['__files_to_merge'], temp_filename, args)
os.rename(encodeFilename(temp_filename), encodeFilename(filename))
return info['__files_to_merge'], info
required_version = '10-0'
if is_outdated_version(
self._versions[self.basename], required_version):
- warning = ('Your copy of %s is outdated and unable to properly mux separate video and audio files, '
+ warning = (f'Your copy of {self.basename} is outdated and unable to properly mux separate video and audio files, '
'yt-dlp will download single file media. '
- 'Update %s to version %s or newer to fix this.') % (
- self.basename, self.basename, required_version)
+ f'Update {self.basename} to version {required_version} or newer to fix this.')
self.report_warning(warning)
return False
return True
stretched_ratio = info.get('stretched_ratio')
if stretched_ratio not in (None, 1):
self._fixup('Fixing aspect ratio', info['filepath'], [
- *self.stream_copy_opts(), '-aspect', '%f' % stretched_ratio])
+ *self.stream_copy_opts(), '-aspect', f'{stretched_ratio:f}'])
return [], info
@PostProcessor._restrict_to(images=False)
def run(self, info):
if all(self._needs_fixup(info)):
+ args = ['-f', 'mp4']
+ if self.get_audio_codec(info['filepath']) == 'aac':
+ args.extend(['-bsf:a', 'aac_adtstoasc'])
self._fixup('Fixing MPEG-TS in MP4 container', info['filepath'], [
- *self.stream_copy_opts(), '-f', 'mp4', '-bsf:a', 'aac_adtstoasc'])
+ *self.stream_copy_opts(), *args])
return [], info
opts = ['-vf', 'setpts=PTS-STARTPTS']
else:
opts = ['-c', 'copy', '-bsf', 'setts=ts=TS-STARTPTS']
- self._fixup('Fixing frame timestamp', info['filepath'], opts + [*self.stream_copy_opts(False), '-ss', self.trim])
+ self._fixup('Fixing frame timestamp', info['filepath'], [*opts, *self.stream_copy_opts(False), '-ss', self.trim])
return [], info
continue
ext = sub['ext']
if ext == new_ext:
- self.to_screen('Subtitle file for %s is already in the requested format' % new_ext)
+ self.to_screen(f'Subtitle file for {new_ext} is already in the requested format')
continue
elif ext == 'json':
self.to_screen(
with open(dfxp_file, 'rb') as f:
srt_data = dfxp2srt(f.read())
- with open(srt_file, 'wt', encoding='utf-8') as f:
+ with open(srt_file, 'w', encoding='utf-8') as f:
f.write(srt_data)
old_file = srt_file
@PostProcessor._restrict_to(images=False)
def run(self, info):
+ self._fixup_chapters(info)
chapters = info.get('chapters') or []
if not chapters:
self.to_screen('Chapter information is unavailable')
in_file = info['filepath']
if self._force_keyframes and len(chapters) > 1:
in_file = self.force_keyframes(in_file, (c['start_time'] for c in chapters))
- self.to_screen('Splitting video by chapters; %d chapters found' % len(chapters))
+ self.to_screen(f'Splitting video by chapters; {len(chapters)} chapters found')
for idx, chapter in enumerate(chapters):
destination, opts = self._ffmpeg_args_for_chapter(idx + 1, chapter, info)
self.real_run_ffmpeg([(in_file, opts)], [(destination, self.stream_copy_opts())])
@classmethod
def is_webp(cls, path):
- write_string(f'DeprecationWarning: {cls.__module__}.{cls.__name__}.is_webp is deprecated')
+ deprecation_warning(f'{cls.__module__}.{cls.__name__}.is_webp is deprecated')
return imghdr.what(path) == 'webp'
def fixup_webp(self, info, idx=-1):
_, thumbnail_ext = os.path.splitext(thumbnail_filename)
if thumbnail_ext:
if thumbnail_ext.lower() != '.webp' and imghdr.what(thumbnail_filename) == 'webp':
- self.to_screen('Correcting thumbnail "%s" extension to webp' % thumbnail_filename)
+ self.to_screen(f'Correcting thumbnail "{thumbnail_filename}" extension to webp')
webp_filename = replace_extension(thumbnail_filename, 'webp')
os.replace(thumbnail_filename, webp_filename)
info['thumbnails'][idx]['filepath'] = webp_filename
@staticmethod
def _options(target_ext):
+ yield from ('-update', '1')
if target_ext == 'jpg':
- return ['-bsf:v', 'mjpeg2jpeg']
- return []
+ yield from ('-bsf:v', 'mjpeg2jpeg')
def convert_thumbnail(self, thumbnail_filename, target_ext):
thumbnail_conv_filename = replace_extension(thumbnail_filename, target_ext)
_, source_ext = os.path.splitext(thumbnail_filename)
self.real_run_ffmpeg(
[(thumbnail_filename, [] if source_ext == '.gif' else ['-f', 'image2', '-pattern_type', 'none'])],
- [(thumbnail_conv_filename.replace('%', '%%'), self._options(target_ext))])
+ [(thumbnail_conv_filename, self._options(target_ext))])
return thumbnail_conv_filename
def run(self, info):
continue
has_thumbnail = True
self.fixup_webp(info, idx)
+ original_thumbnail = thumbnail_dict['filepath'] # Path can change during fixup
thumbnail_ext = os.path.splitext(original_thumbnail)[1][1:].lower()
if thumbnail_ext == 'jpeg':
thumbnail_ext = 'jpg'
projects / yt-dlp.git / blobdiff