Repository: yt-dlp/yt-dlp
File: yt_dlp/postprocessor/common.py (Python)

Code before:
        if not self._downloader:
            return

        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_screen(
                self._downloader.evaluate_outtmpl(tmpl, progress_dict), quiet=False)

        self._downloader.to_console_title(self._downloader.evaluate_outtmpl(
            progress_template.get('postprocess-title') or 'yt-dlp %(progress._default_template)s',
            progress_dict))

        percent = s.get('_percent')
        if s['status'] not in ('downloading', 'error', 'finished') or percent is None:
            return
        # Emit ConEmu progress codes: https://conemu.github.io/en/AnsiEscapeCodes.html#ConEmu_specific_OSC
        if s['status'] == 'finished':
            self._downloader._send_console_code('\033]9;4;3;0\007')
            return
        state = 1 if s['status'] == 'downloading' else 2
        self._downloader._send_console_code(f'\033]9;4;{state};{int(percent)}\007')

    def _retry_download(self, err, count, retries):
        # While this is not an extractor, it behaves similar to one and
        # so obey extractor_retries and "--retry-sleep extractor"
        RetryManager.report_retry(err, count, retries, info=self.to_screen, warn=self.report_warning,
                                  sleep_func=self.get_param('retry_sleep_functions', {}).get('extractor'))

    def _download_json(self, url, *, expected_http_errors=(404,)):
        self.write_debug(f'{self.PP_NAME} query: {url}')
        for retry in RetryManager(self.get_param('extractor_retries', 3), self._retry_download):
            try:
                rsp = self._downloader.urlopen(Request(url))
            except network_exceptions as e:
                if isinstance(e, HTTPError) and e.status in expected_http_errors:
                    return None
                retry.error = PostProcessingError(f'Unable to communicate with {self.PP_NAME} API: {e}')
                continue
        return json.loads(rsp.read().decode(rsp.headers.get_param('charset') or 'utf-8'))


class AudioConversionError(PostProcessingError):  # Deprecated
    pass


Review comment (refactor):
I don't like that this code is duplicated. Wouldn't it be better to pass `percent` and `state` to `to_console_title` instead?

Code after:
        s['_default_template'] = '%(postprocessor)s %(status)s' % s  # noqa: UP031
        if not self._downloader:
            return

        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_screen(
                self._downloader.evaluate_outtmpl(tmpl, progress_dict), quiet=False)

        self._downloader.to_console_title(self._downloader.evaluate_outtmpl(
            progress_template.get('postprocess-title') or 'yt-dlp %(progress._default_template)s',
            progress_dict), _ProgressState.from_dict(s), s.get('_percent'))

    def _retry_download(self, err, count, retries):
        # While this is not an extractor, it behaves similar to one and
        # so obey extractor_retries and "--retry-sleep extractor"
        RetryManager.report_retry(err, count, retries, info=self.to_screen, warn=self.report_warning,
                                  sleep_func=self.get_param('retry_sleep_functions', {}).get('extractor'))

    def _download_json(self, url, *, expected_http_errors=(404,)):
        self.write_debug(f'{self.PP_NAME} query: {url}')
        for retry in RetryManager(self.get_param('extractor_retries', 3), self._retry_download):
            try:
                rsp = self._downloader.urlopen(Request(url))
            except network_exceptions as e:
                if isinstance(e, HTTPError) and e.status in expected_http_errors:
                    return None
                retry.error = PostProcessingError(f'Unable to communicate with {self.PP_NAME} API: {e}')
                continue
        return json.loads(rsp.read().decode(rsp.headers.get_param('charset') or 'utf-8'))


class AudioConversionError(PostProcessingError):  # Deprecated
    pass