expect_warnings(ydl, test_case.get('expected_warnings', []))
def get_tc_filename(tc):
- return ydl.prepare_filename(tc.get('info_dict', {}))
+ return ydl.prepare_filename(dict(tc.get('info_dict', {})))
res_dict = None
from yt_dlp import YoutubeDL
from yt_dlp.compat import compat_shlex_quote
from yt_dlp.postprocessor import (
- ExecAfterDownloadPP,
+ ExecPP,
FFmpegThumbnailsConvertorPP,
MetadataFromFieldPP,
MetadataParserPP,
os.remove(file.format(out))
-class TestExecAfterDownload(unittest.TestCase):
+class TestExec(unittest.TestCase):
def test_parse_cmd(self):
- pp = ExecAfterDownloadPP(YoutubeDL(), '')
+ pp = ExecPP(YoutubeDL(), '')
info = {'filepath': 'file name'}
- quoted_filepath = compat_shlex_quote(info['filepath'])
+ cmd = 'echo %s' % compat_shlex_quote(info['filepath'])
- self.assertEqual(pp.parse_cmd('echo', info), 'echo %s' % quoted_filepath)
- self.assertEqual(pp.parse_cmd('echo.{}', info), 'echo.%s' % quoted_filepath)
- self.assertEqual(pp.parse_cmd('echo "%(filepath)s"', info), 'echo "%s"' % info['filepath'])
+ self.assertEqual(pp.parse_cmd('echo', info), cmd)
+ self.assertEqual(pp.parse_cmd('echo {}', info), cmd)
+ self.assertEqual(pp.parse_cmd('echo %(filepath)q', info), cmd)
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 []
from .postprocessor.embedthumbnail import has_mutagen
from .cookies import SQLITE_AVAILABLE, KEYRING_AVAILABLE
- lib_str = ', '.join(filter(None, (
+ lib_str = ', '.join(sorted(filter(None, (
can_decrypt_frag and 'pycryptodome',
has_websockets and 'websockets',
has_mutagen and 'mutagen',
SQLITE_AVAILABLE and 'sqlite',
KEYRING_AVAILABLE and 'keyring',
- ))) or 'none'
+ )))) or 'none'
self._write_string('[debug] Optional libraries: %s\n' % lib_str)
proxy_map = {}
for k, tmpl in opts.outtmpl.items():
validate_outtmpl(tmpl, '%s output template' % k)
- for tmpl in opts.forceprint:
+ opts.forceprint = opts.forceprint or []
+ for tmpl in opts.forceprint or []:
validate_outtmpl(tmpl, 'print template')
if opts.extractaudio and not opts.keepvideo and opts.format is None:
# Must be after all other before_dl
if opts.exec_before_dl_cmd:
postprocessors.append({
- 'key': 'ExecAfterDownload',
+ 'key': 'Exec',
'exec_cmd': opts.exec_before_dl_cmd,
'when': 'before_dl'
})
# XAttrMetadataPP should be run after post-processors that may change file contents
if opts.xattrs:
postprocessors.append({'key': 'XAttrMetadata'})
- # ExecAfterDownload must be the last PP
+ # Exec must be the last PP
if opts.exec_cmd:
postprocessors.append({
- 'key': 'ExecAfterDownload',
+ 'key': 'Exec',
'exec_cmd': opts.exec_cmd,
# Run this only after the files have been moved to their final locations
'when': 'after_move'
def _write_ytdl_file(self, ctx):
frag_index_stream, _ = sanitize_open(self.ytdl_filename(ctx['filename']), 'w')
- downloader = {
- 'current_fragment': {
- 'index': ctx['fragment_index'],
- },
- }
- if 'extra_state' in ctx:
- downloader['extra_state'] = ctx['extra_state']
- if ctx.get('fragment_count') is not None:
- downloader['fragment_count'] = ctx['fragment_count']
- frag_index_stream.write(json.dumps({'downloader': downloader}))
- frag_index_stream.close()
+ try:
+ downloader = {
+ 'current_fragment': {
+ 'index': ctx['fragment_index'],
+ },
+ }
+ if 'extra_state' in ctx:
+ downloader['extra_state'] = ctx['extra_state']
+ if ctx.get('fragment_count') is not None:
+ downloader['fragment_count'] = ctx['fragment_count']
+ frag_index_stream.write(json.dumps({'downloader': downloader}))
+ finally:
+ frag_index_stream.close()
def _download_fragment(self, ctx, frag_url, info_dict, headers=None, request_data=None):
fragment_filename = '%s-Frag%d' % (ctx['tmpfilename'], ctx['fragment_index'])
from .version import __version__
from .downloader.external import list_external_downloaders
-from .postprocessor.ffmpeg import (
+from .postprocessor import (
FFmpegExtractAudioPP,
FFmpegSubtitlesConvertorPP,
FFmpegThumbnailsConvertorPP,
action='store_true', dest='skip_download', default=False,
help='Do not download the video but write all related files (Alias: --no-download)')
verbosity.add_option(
- '-O', '--print', metavar='TEMPLATE',
- action='callback', dest='forceprint', type='str', default=[],
- callback=_list_from_options_callback, callback_kwargs={'delim': None},
+ '-O', '--print',
+ metavar='TEMPLATE', action='append', dest='forceprint',
help=(
'Quiet, but print the given fields for each video. Simulate unless --no-simulate is used. '
'Either a field name or same syntax as the output template can be used'))
help='Location of the ffmpeg binary; either the path to the binary or its containing directory')
postproc.add_option(
'--exec', metavar='CMD',
- action='callback', dest='exec_cmd', default=[], type='str',
- callback=_list_from_options_callback, callback_kwargs={'delim': None},
+ action='append', dest='exec_cmd',
help=(
'Execute a command on the file after downloading and post-processing. '
'Same syntax as the output template can be used to pass any field as arguments to the command. '
help='Remove any previously defined --exec')
postproc.add_option(
'--exec-before-download', metavar='CMD',
- action='callback', dest='exec_before_dl_cmd', default=[], type='str',
- callback=_list_from_options_callback, callback_kwargs={'delim': None},
+ action='append', dest='exec_before_dl_cmd',
help=(
'Execute a command before the actual download. '
'The syntax is the same as --exec but "filepath" is not available. '
FFmpegVideoRemuxerPP,
)
from .xattrpp import XAttrMetadataPP
-from .execafterdownload import ExecAfterDownloadPP
+from .exec import ExecPP, ExecAfterDownloadPP
from .metadataparser import (
MetadataFromFieldPP,
MetadataFromTitlePP,
__all__ = [
'FFmpegPostProcessor',
'EmbedThumbnailPP',
+ 'ExecPP',
'ExecAfterDownloadPP',
'FFmpegEmbedSubtitlePP',
'FFmpegExtractAudioPP',
)
-class ExecAfterDownloadPP(PostProcessor):
+class ExecPP(PostProcessor):
def __init__(self, downloader, exec_cmd):
- super(ExecAfterDownloadPP, self).__init__(downloader)
+ PostProcessor.__init__(self, downloader)
self.exec_cmd = variadic(exec_cmd)
- @classmethod
- def pp_key(cls):
- return 'Exec'
-
def parse_cmd(self, cmd, info):
tmpl, tmpl_dict = self._downloader.prepare_outtmpl(cmd, info)
if tmpl_dict: # if there are no replacements, tmpl_dict = {}
if retCode != 0:
raise PostProcessingError('Command returned error code %d' % retCode)
return [], info
+
+
+class ExecAfterDownloadPP(ExecPP): # for backward compatibility
+ pass