_configuration_args,
encodeFilename,
PostProcessingError,
+ write_string,
)
-class PostProcessor(object):
+class PostProcessorMetaClass(type):
+ @staticmethod
+ def run_wrapper(func):
+ @functools.wraps(func)
+ def run(self, info, *args, **kwargs):
+ info_copy = self._copy_infodict(info)
+ self._hook_progress({'status': 'started'}, info_copy)
+ ret = func(self, info, *args, **kwargs)
+ if ret is not None:
+ _, info = ret
+ self._hook_progress({'status': 'finished'}, info_copy)
+ return ret
+ return run
+
+ def __new__(cls, name, bases, attrs):
+ if 'run' in attrs:
+ attrs['run'] = cls.run_wrapper(attrs['run'])
+ return type.__new__(cls, name, bases, attrs)
+
+
+class PostProcessor(metaclass=PostProcessorMetaClass):
"""Post Processor class.
PostProcessor objects can be added to downloaders with their
_downloader = None
def __init__(self, downloader=None):
- self._downloader = downloader
+ self._progress_hooks = []
+ self.add_progress_hook(self.report_progress)
+ self.set_downloader(downloader)
self.PP_NAME = self.pp_key()
@classmethod
if self._downloader:
return self._downloader.report_warning(text, *args, **kwargs)
+ def deprecation_warning(self, text):
+ if self._downloader:
+ return self._downloader.deprecation_warning(text)
+ write_string(f'DeprecationWarning: {text}')
+
def report_error(self, text, *args, **kwargs):
+ # Exists only for compatibility. Do not use
if self._downloader:
return self._downloader.report_error(text, *args, **kwargs)
def set_downloader(self, downloader):
"""Sets the downloader for this PP."""
self._downloader = downloader
+ for ph in getattr(downloader, '_postprocessor_hooks', []):
+ self.add_progress_hook(ph)
+
+ def _copy_infodict(self, info_dict):
+ return getattr(self._downloader, '_copy_infodict', dict)(info_dict)
@staticmethod
- def _restrict_to(*, video=True, audio=True, images=True):
+ def _restrict_to(*, video=True, audio=True, images=True, simulated=True):
allowed = {'video': video, 'audio': audio, 'images': images}
def decorator(func):
@functools.wraps(func)
def wrapper(self, info):
+ if not simulated and (self.get_param('simulate') or self.get_param('skip_download')):
+ return [], info
format_type = (
'video' if info.get('vcodec') != 'none'
else 'audio' if info.get('acodec') != 'none'
return _configuration_args(
self.pp_key(), self.get_param('postprocessor_args'), exe, *args, **kwargs)
+ def _hook_progress(self, status, info_dict):
+ if not self._progress_hooks:
+ return
+ status.update({
+ 'info_dict': info_dict,
+ 'postprocessor': self.pp_key(),
+ })
+ for ph in self._progress_hooks:
+ ph(status)
+
+ def add_progress_hook(self, ph):
+ # See YoutubeDl.py (search for postprocessor_hooks) for a description of this interface
+ self._progress_hooks.append(ph)
+
+ def report_progress(self, s):
+ s['_default_template'] = '%(postprocessor)s %(status)s' % s
+
+ progress_dict = s.copy()
+ progress_dict.pop('info_dict')
+ progress_dict = {'info': s['info_dict'], 'progress': progress_dict}
+
+ progress_template = self.get_param('progress_template', {})
+ tmpl = progress_template.get('postprocess')
+ if tmpl:
+ self._downloader.to_stdout(self._downloader.evaluate_outtmpl(tmpl, progress_dict))
+
+ self._downloader.to_console_title(self._downloader.evaluate_outtmpl(
+ progress_template.get('postprocess-title') or 'yt-dlp %(progress._default_template)s',
+ progress_dict))
+
class AudioConversionError(PostProcessingError):
pass