code:
from __future__ import unicode_literals

import configparser
import hashlib
import itertools
import json
import optparse
import os
import re
import shlex
import time
import urllib
from collections import defaultdict
from typing import IO, BinaryIO, Dict, Optional, Set
from xml.dom.minidom import parseString

from ._attachments import ShowsProgress
from .exec import ExecPP
from ..YoutubeDL import YoutubeDL
from ..compat import compat_HTTPError
from ..downloader import FileDownloader
from ..options import instantiate_parser
from ..longname import escaped_basename, escaped_remove
from ..utils import (
    PostProcessingError,
    ThrottledDownload,
    int_or_none,
    network_exceptions,
    sanitized_Request,
    time_millis,
    traverse_obj,
    variadic,
)

def _dict_from_options_callback(
        option, opt_str, value, parser,
        allowed_keys=r'[\w-]+', delimiter=':', default_key=None, process=None, multiple_keys=True,
        process_key=str.lower, append=False):

    out_dict = dict(getattr(parser.values, option.dest))
    multiple_args = not isinstance(value, str)
    if multiple_keys:
        allowed_keys = r'(%s)(,(%s))*' % (allowed_keys, allowed_keys)
    mobj = re.match(
        r'(?i)(?P<keys>%s)%s(?P<val>.*)$' % (allowed_keys, delimiter),
        value[0] if multiple_args else value)
    if mobj is not None:
        keys, val = mobj.group('keys').split(','), mobj.group('val')
        if multiple_args:
            val = [val, *value[1:]]
    elif default_key is not None:
        keys, val = [default_key], value
    else:
        raise optparse.OptionValueError(
            'wrong %s formatting; it should be %s, not "%s"' % (opt_str, option.metavar, value))
    try:
        keys = map(process_key, keys) if process_key else keys
        val = process(val) if process else val
    except Exception as err:
        raise optparse.OptionValueError(f'wrong {opt_str} formatting; {err}')
    for key in keys:
        out_dict[key] = out_dict.get(key, []) + [val] if append else val
    setattr(parser.values, option.dest, out_dict)

iaup_options = instantiate_parser()
iaup_options.set_usage('[OPTIONS] IDENTIFIER FILE [FILE...]')

iaup_options.add_option(
    '-q', '--quiet',
    action='store_true', dest='quiet',
    help='Runs without progress. Isolated from --quiet option from the outside')
iaup_options.add_option(
    '-d', '--debug',
    action='store_true', dest='debug',
    help='Runs in debug mode')

iaup_options.add_option(
    '-c', '--config',
    action='store_true', dest='config',
    help='Path to ia.ini. Defaults to where ia command searches')
iaup_options.add_option(
    '-u', '--username',
    dest='username', metavar='USERNAME',
    help='Login to Internet Archive with this account ID')
iaup_options.add_option(
    '-p', '--password',
    dest='password', metavar='PASSWORD',
    help='Account password for Internet Archive. If this option is left out, login will not be done')

iaup_options.add_option(
    '-r', '--remote-name',
    metavar='PATH', dest='remote_name',
    help='Path to remote directory or filename')
iaup_options.add_option(
    '-m', '--metadata',
    metavar='K:V', dest='metadata', default={},
    action='callback', callback=_dict_from_options_callback,
    callback_kwargs={'multiple_keys': False, 'process_key': None},
    help='Metadata to add')
iaup_options.add_option(
    '-H', '--header', '--headers',
    metavar='K:V', dest='headers', default={},
    action='callback', callback=_dict_from_options_callback,
    callback_kwargs={'multiple_keys': False, 'process_key': None},
    help='Header to add')
iaup_options.add_option(
    '-D', '--derive',
    action='store_true', dest='derive', default=False,
    help='Enables "derive" task on IA. derive task is not enabled by default')
iaup_options.add_option(
    '-n', '--no-derive',
    action='store_false', dest='derive', default=False,
    help='Disables "derive" task on IA.')
iaup_options.add_option(
    '-R', '--retries',
    metavar='RETRIES', dest='retries', default=10000,
    help='Number of retries on SlowDown or connection being disconnected.')
iaup_options.add_option(
    '-t', '--throttled-rate',
    metavar='RATE', dest='throttle',
    help=('Same as the option with same name on yt-dlp, but for upload in this time. '
          'Upload failure caused by this option will count for -R retries.'))
iaup_options.add_option(
    '-D', '--delete',
    action='store_true', dest='delete', default=False,
    help='Deletes files after all files are successfully uploaded.')
iaup_options.add_option(
    '-C', '--conflict-resolve',
    metavar='KIND:BEHAVIOR', dest='conflict_resolve', default={},
    action='callback', callback=_dict_from_options_callback,
    help=('Specifies how to avoid/torelate errors while uploading. '
          'Allowed values for KIND are: size_overflow, no_perm. '
          'Allowed values for BEHAVIOR are: rename_ident, error, skip.'))

class Skip(PostProcessingError):
    pass

class InternetArchiveUploadPP(ExecPP):
    # memo
    #  This item total number of bytes(666) is over the per item size limit of 1099511627776. Please contact info@archive.org for help fitting your data into the archive.
    def __init__(self, downloader, exec_cmd):
        super().__init__(downloader, exec_cmd)
        self.login_cache: Dict[str, Dict] = {}
        self.conflict_cache: Dict[str, Set[str]] = {}

    def run(self, info):
        for tmpl in self.exec_cmd:
            cmd = self.parse_cmd(tmpl, info)
            self.to_screen('Processing: %s' % cmd)
            parsed = iaup_options.parse_args(shlex.split(cmd))
            self.parse_inputs(parsed)
            errors = self.process(parsed)
            if errors:
                raise PostProcessingError('There were error(s) while uploading:\n' + '\n'.join(map(str, errors)))
        return [], info

    def parse_inputs(self, opts):
        if opts.throttle is not None:
            numeric_limit = FileDownloader.parse_bytes(opts.throttle)
            if numeric_limit is None:
                self.report_warning('invalid rate limit specified')
            opts.throttle = numeric_limit

    def process(self, parsed):
        opts, args = parsed
        if not args:
            raise PostProcessingError('You have to specify identifier and files to upload.')
        ident = args[0]
        files = args[1:]
        if not files:
            raise PostProcessingError('You have specified nothing to upload!')

        # login
        cred = self.login_cache.get(ident)
        if not cred:
            if opts.username and opts.password:
                self.to_screen(f'Logging in with username and password for identifier {ident}')
                cred = self.login(opts.username, opts.password)
            else:
                self.write_debug(f'Loading config file for identifier {ident}')
                cred = self.parse_config_file(opts.config)
            self.login_cache[ident] = cred

        # create plans for upload
        rp = opts.remote_name
        if rp:
            if rp.endswith('/'):
                # upload files to a remote directory
                plans = [(rp + escaped_basename(x), x) for x in files]
            elif not rp.endswith('/') and len(files) == 1:
                # upload a file with a exact filename
                plans = [(rp, files[0])]
            else:
                raise PostProcessingError('Conflict: You have requested uploading multiple files to one remote file.')
        else:
            # no --remote-name, upload to root
            plans = [(escaped_basename(x), x) for x in files]

        # have we encountered any conflicts previously?
        try:
            ident = self.conflict_trial(opts, ident, pre=True)
        except Skip as skip:
            self.report_warning(skip)
            return []

        # generate header value for auth
        s3_ak = traverse_obj(cred, ('s3', 'access'))
        s3_sk = traverse_obj(cred, ('s3', 'secret'))
        if not s3_ak:
            raise PostProcessingError('Access Key for IAS3 is missing. You should run "ia configure"')
        if not s3_sk:
            raise PostProcessingError('Secret Key for IAS3 is missing. You should run "ia configure"')

        # upload files according to the plan
        errors = []
        for idx, (remotename, filename) in enumerate(plans):
            headers = dict(opts.headers)
            self.generate_headers(headers, opts.metadata, queue_derive=opts.derive)
            headers['Authorization'] = f'LOW {s3_ak}:{s3_sk}'
            self.write_debug(f'=== {filename} -> {remotename}')
            with open(filename, 'rb') as r:
                try:
                    self.real_upload(ident, remotename, r, headers, idx + 1, len(plans), opts.quiet, opts.retries, opts.throttle)
                except PostProcessingError as ex:
                    self.report_warning(ex)
                    errors.append(ex)

        if not errors and opts.delete:
            for _, lf in plans:
                self.write_debug(f'Removing {lf}')
                try:
                    escaped_remove(lf)
                except (IOError, OSError) as ex:
                    self.report_warning(ex)
        return errors

    def real_upload(self, ident, remotename, r, headers, index, total_plans, quiet, retries, throttle=None):
        fsize = os.stat(r.fileno()).st_size
        headers['Content-MD5'] = self.md5(r)
        headers['Content-Length'] = str(fsize)
        true_body = (QuietByteIO if quiet else ProgressByteIO)(self._downloader, r, fsize, throttle)
        true_body._PROGRESS_LABEL = f'{index}/{total_plans}'

        for retry in self.repeat_iter(retries):
            try:
                self.to_screen(f'[{index}/{total_plans}] Uploading {remotename}', prefix=False)
                true_body.start()
                # who wants to use plain HTTP nowadays?
                resp, body = self.do_put_request(f'https://s3.us.archive.org/{ident}/{remotename}', headers, true_body)
                resp_code = resp.code
                self.write_debug(body)
                # TODO: check error message and resolve conflicts
                if resp_code == 200:
                    # that's okay to return without messages
                    return
                else:
                    warn_body = self.get_s3_xml_text(body) or f'Unknown status code: {resp_code}'
                    if resp_code == 503:
                        warn_body = 'S3 is overloaded'
                    retry_remain = self.repeat_remaining(retries, retry)
                    if retry_remain == 0:
                        raise PostProcessingError(f'[{index}/{total_plans}] {warn_body}. Retry exhausted.')
                    else:
                        waittime = 30 - 29 * (1.1**(-retry))
                        self.report_warning(f'[{index}/{total_plans}] {warn_body}. Sleeping for {waittime} seconds. {retry_remain} retries left.')
                        time.sleep(waittime)
                        continue
            except PostProcessingError as ex:
                waittime = 30 - 29 * (1.1**(-retry))
                self.report_warning(f'[{index}/{total_plans}] {ex}. Sleeping for {waittime} seconds. {retry_remain} retries left.')
                time.sleep(waittime)
                continue
            finally:
                # no need to call close() as it's caller's responsiblility to close streams
                true_body.end()
        raise PostProcessingError(f'[{index}/{total_plans}] Retry exhausted')

    @staticmethod
    def md5(file_object):
        file_object.seek(0, os.SEEK_SET)
        m = hashlib.md5()
        while True:
            data = file_object.read(8192)
            if not data:
                break
            m.update(data)
        file_object.seek(0, os.SEEK_SET)
        return m.hexdigest()

    @staticmethod
    def repeat_iter(retries):
        if retries < 1 or retries == float('inf'):
            return itertools.count(0)
        else:
            return range(retries + 1)

    @staticmethod
    def repeat_remaining(retries, count):
        if retries < 1 or retries == float('inf'):
            return float('inf')
        else:
            return retries - count

    def get_s3_xml_text(xml_str):
        if not xml_str:
            return ''

        def _get_tag_text(tag_name, xml_obj):
            text = ''
            elements = xml_obj.getElementsByTagName(tag_name)
            for e in elements:
                for node in e.childNodes:
                    if node.nodeType == node.TEXT_NODE:
                        text += node.data
            return text

        try:
            p = parseString(xml_str)
            _msg = _get_tag_text('Message', p)
            _resource = _get_tag_text('Resource', p)
            # Avoid weird Resource text that contains PUT method.
            if _resource and "'PUT" not in _resource:
                return f'{_msg} - {_resource.strip()}'
            else:
                return _msg
        except KeyboardInterrupt:
            raise
        except BaseException:
            return str(xml_str)

    def generate_headers(self, headers, metadata, file_metadata=None, queue_derive=True):
        metadata = dict() if metadata is None else metadata
        file_metadata = dict() if file_metadata is None else file_metadata

        if not metadata.get('scanner'):
            from ..version import __version__
            scanner = 'ytdl-patched InternetArchiveUploadPP {0}'.format(__version__)
            metadata['scanner'] = scanner
        prepared_metadata = metadata or {}
        prepared_file_metadata = file_metadata or {}

        headers['x-archive-auto-make-bucket'] = '1'
        if 'x-archive-queue-derive' not in headers:
            if queue_derive is False:
                headers['x-archive-queue-derive'] = '0'
            else:
                headers['x-archive-queue-derive'] = '1'

        def needs_quote(s):
            try:
                s.encode('ascii')
            except (UnicodeDecodeError, UnicodeEncodeError):
                return True
            return re.search(r'\s', s) is not None

        def _prepare_metadata_headers(prepared_metadata, meta_type='meta'):
            for meta_key, meta_value in prepared_metadata.items():
                # Encode arrays into JSON strings because Archive.org does not
                # yet support complex metadata structures in
                # <identifier>_meta.xml.
                if isinstance(meta_value, dict):
                    meta_value = json.dumps(meta_value)
                # Convert the metadata value into a list if it is not already
                # iterable.
                meta_value = variadic(meta_value)
                # Convert metadata items into HTTP headers and add to
                # ``headers`` dict.
                for i, value in enumerate(meta_value):
                    if not value:
                        continue
                    header_key = 'x-archive-{0}{1:02d}-{2}'.format(meta_type, i, meta_key)
                    if (isinstance(value, str) and needs_quote(value)):
                        value = 'uri({0})'.format(urllib.parse.quote(value))
                    # because rfc822 http headers disallow _ in names, IA-S3 will
                    # translate two hyphens in a row (--) into an underscore (_).
                    header_key = header_key.replace('_', '--')
                    headers[header_key] = value

        # Parse the prepared metadata into HTTP headers,
        # and add them to the ``headers`` dict.
        _prepare_metadata_headers(prepared_metadata)
        _prepare_metadata_headers(prepared_file_metadata, meta_type='filemeta')

    def conflict_trial(self, opts, ident, pre=False):
        history = self.conflict_cache.get(ident)
        # TODO: test against size overflow beforehand here
        if not history:
            return ident
        cr = opts.conflict_resolve or {}
        resolve_methods = set(filter(None, (cr.get(cc) for cc in history)))
        if not resolve_methods:
            # no method for resolving/avoiding conflict is specified
            return ident
        if 'rename_ident' in resolve_methods:
            # split the last number and others, and increment the number specified
            m = re.fullmatch(r'(.+)(?:_(\d+))?', ident)
            assert m
            name, num = m.groups()
            num = int_or_none(num, default=1) + 1
            ident = f'{name}_{num}'
            # as we've changed the identifier, put into trial again
            return self.conflict_trial(opts, ident, pre)
        elif 'skip' in resolve_methods:
            # literally.
            raise Skip(f'Skipping because of any of errors here: {" ".join(history)}')
        elif 'error' in resolve_methods:
            # literally, again
            if pre:
                raise PostProcessingError(f'Refusing upload to identifier {ident}; {" ".join(history)}')
            else:
                raise PostProcessingError(f'Aborted uploading to identifier {ident}; {" ".join(history)}')
        return ident

    def login(self, email, password):
        j = self._get_json(
            'https://archive.org/services/xauthn/?op=login',
            {'email': email, 'password': password})
        if not j.get('success'):
            msg = traverse_obj(j, ('values', 'reason'), 'error')
            msg = {
                'account_not_found': 'Account not found, check your email and try again.',
                'account_bad_password': 'Incorrect password, try again.',
            }.get(msg) or f'Authentication failed: {msg}'
            raise PostProcessingError(msg)
        return j['values']

    def parse_config_file(self, config_file=None):
        config = configparser.RawConfigParser()

        if not config_file:
            candidates = []
            if os.environ.get('IA_CONFIG_FILE'):
                candidates.append(os.environ['IA_CONFIG_FILE'])
            xdg_config_home = os.environ.get('XDG_CONFIG_HOME')
            if not xdg_config_home or not os.path.isabs(xdg_config_home):
                # Per the XDG Base Dir specification, this should be $HOME/.config. Unfortunately, $HOME
                # does not exist on all systems. Therefore, we use ~/.config here. On a POSIX-compliant
                # system, where $HOME must always be set, the XDG spec will be followed precisely.
                xdg_config_home = os.path.join(os.path.expanduser('~'), '.config')
            xdg_config_file = os.path.join(xdg_config_home, 'internetarchive', 'ia.ini')
            candidates.append(xdg_config_file)
            candidates.append(os.path.join(os.path.expanduser('~'), '.config', 'ia.ini'))
            candidates.append(os.path.join(os.path.expanduser('~'), '.ia'))
            for candidate in candidates:
                if os.path.isfile(candidate):
                    config_file = candidate
                    break
            else:
                # None of the candidates exist, default to IA_CONFIG_FILE if set else XDG
                config_file = os.environ.get('IA_CONFIG_FILE', xdg_config_file)
        config.read(config_file)

        if not config.has_section('s3'):
            config.add_section('s3')
            config.set('s3', 'access', None)
            config.set('s3', 'secret', None)
        if not config.has_section('cookies'):
            config.add_section('cookies')
            config.set('cookies', 'logged-in-user', None)
            config.set('cookies', 'logged-in-sig', None)

        if config.has_section('general'):
            for k, v in config.items('general'):
                if k in ['secure']:
                    config.set('general', k, config.getboolean('general', k))
            if not config.get('general', 'screenname'):
                config.set('general', 'screenname', None)
        else:
            config.add_section('general')
            config.set('general', 'screenname', None)

        return config_file, config

    def get_config(self, config=None, config_file=None):
        _config = {} if not config else config
        config_file, config = self.parse_config_file(config_file)

        if not os.path.isfile(config_file):
            return _config

        config_dict = defaultdict(dict)
        for sec in config.sections():
            try:
                for k, v in config.items(sec):
                    if k is None or v is None:
                        continue
                    config_dict[sec][k] = v
            except TypeError:
                pass

        def deep_update(d, u):
            for k, v in u.items():
                if isinstance(v, dict):
                    r = deep_update(d.get(k, {}), v)
                    d[k] = r
                else:
                    d[k] = u[k]
            return d

        # Recursive/deep update.
        deep_update(config_dict, _config)

        return dict((k, v) for k, v in config_dict.items() if v is not None)

    def _get_json(self, url, data=None):
        # Taken from sponsorblock.py with POST support
        max_retries = self.get_param('extractor_retries', 3)
        sleep_interval = self.get_param('sleep_interval_requests') or 0
        for retries in itertools.count():
            try:
                rsp = self._downloader.urlopen(sanitized_Request(url, data=data))
                return json.loads(rsp.read().decode(rsp.info().get_param('charset') or 'utf-8'))
            except network_exceptions as e:
                if isinstance(e, compat_HTTPError) and e.code == 404:
                    return []
                if retries < max_retries:
                    self.report_warning(f'{e}. Retrying...')
                    if sleep_interval > 0:
                        self.to_screen(f'Sleeping {sleep_interval} seconds ...')
                        time.sleep(sleep_interval)
                    continue
                raise PostProcessingError(f'Unable to communicate with Internet Archive: {e}')

    def do_put_request(self, url, headers, data):
        try:
            rsp = self._downloader.urlopen(sanitized_Request(url, headers=headers, data=data))
        except network_exceptions as e:
            e.fp._error = e
            rsp = e.fp
        return rsp, rsp.read().decode(rsp.info().get_param('charset') or 'utf-8')

class QuietByteIO(BinaryIO, IO[bytes]):
    def __init__(self, ydl: YoutubeDL, stream: IO[bytes], filesize: int, throttle: Optional[int] = None) -> None:
        super().__init__()
        self.stream = stream
        self.filesize = filesize
        self.throttle = throttle
        self.counter, self.throttle_count, self.start_time = None, None, None
        self.time_and_size, self.avg_len = [], 10

    def start(self):
        self.counter, self.throttle_count = 0, 0
        self.stream.seek(0, os.SEEK_SET)
        self.start_time = time_millis()

    def report(self):
        pass

    def throttle_check(self):
        self.time_and_size.append((self.counter, time.time()))
        self.time_and_size = tsz = self.time_and_size[-self.avg_len:]
        if not self.throttle:
            return
        if len(tsz) < 4:
            return
        last, early = tsz[0], tsz[-1]
        average_speed = (early[0] - last[0]) / (early[1] - last[1])
        if average_speed < self.throttle:
            self.throttle_count += 1
        if self.throttle_count > 4:
            # throwing ThrottledDownload here could cause re-extraction, which is not desired here
            raise PostProcessingError(ThrottledDownload().msg)

    def end(self):
        pass

    def close(self) -> None:
        self.end()
        self.stream.close()

    def read(self, n: int = -1) -> bytes:
        if n <= 0:
            n = 1048576
        ret = self.stream.read(n)
        if ret:
            self.counter += len(ret)
            self.throttle_check()
            self.report()
        return ret

    def __len__(self):
        return self.filesize

class ProgressByteIO(QuietByteIO, ShowsProgress):
    def __init__(self, ydl: YoutubeDL, stream: IO[bytes], filesize: int, throttle: Optional[int] = None) -> None:
        super().__init__(ydl, stream, filesize, throttle)
        ShowsProgress.__init__(self, ydl)
        self.stream = stream
        self.filesize = filesize
        # call with all=False as this class doesn't have hooks,
        # therefore calling report_progress directly
        self._enable_progress(False)

    def start(self):
        super().start()
        self.report_progress({
            'status': 'processing',
            'elapsed': 0,
            'processed_bytes': 0,
            'total_bytes': self.filesize,
        })

    def report(self):
        tsz = self.time_and_size
        elapsed = time_millis() - self.start_time
        if len(tsz) > 1:
            last, early = tsz[0], tsz[-1]
            average_speed = (early[0] - last[0]) / (early[1] - last[1])
            eta = ((self.filesize - self.counter) / average_speed) if average_speed > 0 else None
        else:
            average_speed = None
            eta = (self.filesize - self.counter) * elapsed / self.counter
            if eta < 0:
                eta = None
        self.report_progress({
            'status': 'processing',
            'elapsed': elapsed,
            'speed': average_speed,
            'processed_bytes': self.counter,
            'eta': eta,
            'total_bytes': self.filesize,
        })

    def end(self):
        # prevent "finished" event to be emitted twice without indication
        if self.counter is None:
            return
        self.report_progress({
            'status': 'finished',
            # do not use self.filesize here since the task could fail at middle of file
            'processed_bytes': self.counter,
            'total_bytes': self.filesize,
        })
        self.counter = None


code:
import functools
import inspect
import re

from ..utils import get_argcount
from .common import PostProcessor
from ..utils import Namespace, filter_dict

class MetadataParserPP(PostProcessor):

    class _functools_partial(functools.partial):
        def __repr__(self) -> str:
            return '<functools.partial object with infodict>'

    def __init__(self, downloader, actions):
        super().__init__(downloader)
        self._actions = []
        for f in actions:
            action, *args = f
            assert action in self.Actions
            self._actions.append(action(self, *args))

    @classmethod
    def validate_action(cls, action, *data):
        """Each action can be:
                (Actions.INTERPRET, from, to) OR
                (Actions.REPLACE, field, search, replace)
        """
        if action not in cls.Actions:
            raise ValueError(f'{action!r} is not a valid action')
        action(cls, *data)  # So this can raise error to validate

    @staticmethod
    def field_to_template(tmpl):
        if re.match(r'[a-zA-Z_]+$', tmpl):
            return f'%({tmpl})s'

        from ..YoutubeDL import YoutubeDL
        err = YoutubeDL.validate_outtmpl(tmpl)
        if err:
            raise err
        return tmpl

    @staticmethod
    def format_to_regex(fmt):
        r"""
        Converts a string like
           '%(title)s - %(artist)s'
        to a regex like
           '(?P<title>.+)\ \-\ (?P<artist>.+)'
        """
        if not re.search(r'%\(\w+\)s', fmt):
            return fmt
        lastpos = 0
        regex = ''
        # replace %(..)s with regex group and escape other string parts
        for match in re.finditer(r'%\((\w+)\)s', fmt):
            regex += re.escape(fmt[lastpos:match.start()])
            regex += rf'(?P<{match.group(1)}>.+)'
            lastpos = match.end()
        if lastpos < len(fmt):
            regex += re.escape(fmt[lastpos:])
        return regex

    def run(self, info):
        for f in self._actions:
            next(filter(lambda x: 0, f(info)), None)
        return [], info

    def interpretter(self, inp, out):
        def f(info):
            data_to_parse = self._downloader.evaluate_outtmpl(template, info)
            self.write_debug(f'Searching for {out_re.pattern!r} in {template!r}')
            match = out_re.search(data_to_parse)
            if match is None:
                self.to_screen(f'Could not interpret {inp!r} as {out!r}')
                return
            for attribute, value in filter_dict(match.groupdict()).items():
                yield (attribute, info.get(attribute, MetadataParserPP.BACKLOG_UNSET))
                info[attribute] = value
                self.to_screen(f'Parsed {attribute} from {template!r}: {value!r}')

        template = self.field_to_template(inp)
        out_re = re.compile(self.format_to_regex(out))
        return f

    def replacer(self, field, search, replace):
        def f(info):
            nonlocal replace
            # let function have info_dict on invocation (for MetadataEditorAugment)
            if inspect.isfunction(replace) and get_argcount(replace) == 2:
                replace = self._functools_partial(replace, info)
            val = info.get(field)
            if val is None:
                self.to_screen(f'Video does not have a {field}')
                return
            elif not isinstance(val, str):
                self.report_warning(f'Cannot replace in field {field} since it is a {type(val).__name__}')
                return
            self.write_debug(f'Replacing all {search!r} in {field} with {replace!r}')
            yield (field, info.get(field, MetadataParserPP.BACKLOG_UNSET))
            info[field], n = search_re.subn(replace, val)
            if n:
                self.to_screen(f'Changed {field} to: {info[field]}')
            else:
                self.to_screen(f'Did not find {search!r} in {field}')

        search_re = re.compile(search)
        return f

    BACKLOG_UNSET = object()
    Actions = Namespace(INTERPRET=interpretter, REPLACE=replacer)

class MetadataFromFieldPP(MetadataParserPP):
    @classmethod
    def to_action(cls, f):
        match = re.match(r'(?s)(?P<in>.*?)(?<!\\):(?P<out>.+)$', f)
        if match is None:
            raise ValueError(f'it should be FROM:TO, not {f!r}')
        return (
            cls.Actions.INTERPRET,
            match.group('in').replace('\\:', ':'),
            match.group('out'),
        )

    def __init__(self, downloader, formats):
        super().__init__(downloader, [self.to_action(f) for f in formats])

# Deprecated
class MetadataFromTitlePP(MetadataParserPP):
    def __init__(self, downloader, titleformat):
        super().__init__(downloader, [(self.Actions.INTERPRET, 'title', titleformat)])
        self.deprecation_warning(
            'yt_dlp.postprocessor.MetadataFromTitlePP is deprecated '
            'and may be removed in a future version. Use yt_dlp.postprocessor.MetadataFromFieldPP instead')

repo_name: Lesmiscore/ytdl


code:
import json
import re
import shlex

from math import inf

from ..utils import parse_duration, render_table
from .common import PostProcessor

class MetapulatorPP(PostProcessor):
    """
    Metadata manuplator via commands
    """

    COMMANDS = {}  # command, object

    def __init__(self, downloader, manual=False, auto=None):
        super().__init__(downloader)
        self.COMMANDS = dict((x.COMMAND_NAME, x()) for x in CLASSES)
        self.manual = manual
        self.auto = auto

    def _yield_commands(self):
        yield from (self.auto or [])
        yield from (self.get_param('metapulator_auto') or [])
        if not (self.get_param('metapulator_manual') or self.manual):
            return
        self.to_screen('Metapulator REPL. Type "help" for help. Type "exit" to finish and continue')
        while True:
            try:
                yield input('>>> ')
            except EOFError:
                self.to_screen('', prefix=False)
                break
            except KeyboardInterrupt:
                self.to_screen('', prefix=False)
                self.to_screen('Ctrl+C has pressed, continuing postprocessing')
                break

    def run(self, info):
        for f in self._yield_commands():
            if not f:
                continue
            try:
                self.execute_line(f, info)
            except Exit:
                break
        return [], info

    def find_command(self, cmd):
        if not cmd:
            return None
        for name, value in self.COMMANDS.items():
            # shortest match
            if name.startswith(cmd):
                return value
        return None

    def execute_line(self, line, info):
        args = shlex.split(line)
        cmd = self.find_command(args[0])
        if cmd is None:
            self._downloader.report_error(f'Command {args[0]!r} does not exist')
            return
        try:
            cmd.run(self, info, args[1:])
        except ShowHelp:
            self.print_help(cmd)

    def print_help(self, cmd):
        if isinstance(cmd, str):
            cmd = self.find_command(cmd)
        if not cmd:
            return
        self.to_screen(f'Help for {cmd.COMMAND_NAME!r}')
        self.to_screen(cmd.HELP, False)

class ShowHelp(Exception):
    pass

class Exit(Exception):
    pass

class MetapulatorCommand:
    COMMAND_NAME = False
    HELP = ''

    def run(self, pp: MetapulatorPP, info: dict, args: list) -> None:
        raise Exception('Override this method in child classes')

class ExitCommand(MetapulatorCommand):
    """
    Stops Metapulator
    """

    COMMAND_NAME = 'exit'
    HELP = "Stops Metapulator"

    def run(self, pp, info, args):
        raise Exit()

class PrintCommand(MetapulatorCommand):
    """
    Equivalent to --print option but in Metapulator
    """

    COMMAND_NAME = 'print'
    HELP = """\
Prints the content of metadata. You can use the same notation as --print
This accepts multiple queries at a time

Examples:
print title
print "%(subtitles.en.-1.ext)s"
print "%(id.3:7:-1)s" "%(formats.:.format_id)s"

Refer to https://github.com/yt-dlp/yt-dlp#output-template for the query details
"""

    def run(self, pp, info, args):
        info = info.copy()

        def format_tmpl(tmpl):
            mobj = re.match(r'\w+(=?)$', tmpl)
            if mobj and mobj.group(1):
                return f'{tmpl[:-1]} = %({tmpl[:-1]})r'
            elif mobj:
                return f'%({tmpl})s'
            return tmpl

        for tmpl in args:
            pp.to_screen(pp._downloader.evaluate_outtmpl(format_tmpl(tmpl), info), prefix=False)

class ChaptersCommand(MetapulatorCommand):
    """
    Command to manuplate and view chapters
    """

    COMMAND_NAME = 'chapters'
    HELP = """\
Manuplates or displays chapters for this video.
Note that this command will not trim the video, but just change it on metadata.

Synopsis: chapters [view]
Show currently available chapters.

Synopsis: chapters add START END [TITLE]
Add a chapter, optionally with a title.
START and END are in seconds.

Synopsis: chapters insert INDEX START END [TITLE]
Insert a new chapter, optionally with a title.
START and END are in seconds.

Synopsis: chapters remove [idx:]INDEX[-INDEX_END]
Remove chapters by indices. Range is inclusive.

Synopsis: chapters remove [t:]TIME-RANGE
Remove chapters by duration. Range is inclusive.

Synopsis: chapters remove [re:]TITLE_REGEX
Remove chapters by title.

Synopsis: chapters sort
Sort chapters. It is highly recommended to run it before finishing manuplation.
"""

    def run(self, pp, info, args):
        ydl = pp._downloader

        def getchap():
            chap = info.get('chapters')
            if not isinstance(chap, list):
                pp.write_debug('setting the new chapter list')
                chap = info['chapters'] = []
            return chap

        if not args or 'view'.startswith(args[0]):
            # view
            chap = info.get('chapters')
            if not chap:
                pp.to_screen('There are no chapters')
                return

            if not isinstance(chap, list):
                pp.report_warning('Extracted chapters are in unexpected structure. Fix extractor code')
                return

            cs = len(chap)
            pp.to_screen(f'There are {cs} chapters:' if cs > 1 else 'There is one chapter:')
            delim = ydl._format_out('\u2502', ydl.Styles.DELIM, '|', test_encoding=True)
            header_line = ydl._list_format_headers('INDEX', delim, 'START', 'END', delim, 'TITLE')

            table = [[
                ch.get('index'),  # INDEX
                delim,
                ch.get('start_time', 0),  # START
                ch.get('end_time', ydl._format_out('???', ydl.Styles.SUPPRESS)),  # END
                delim,
                ch.get('title'),  # TITLE
            ] for ch in chap]
            tbl = render_table(
                header_line, table, hide_empty=True,
                delim=ydl._format_out('\u2500', ydl.Styles.DELIM, '-', test_encoding=True))
            pp.to_screen(tbl, False)
            return
        elif 'add'.startswith(args[0]):
            # add START END [TITLE]
            start, end, *title = args[1:]
            start, end = map(parse_duration, (start, end))
            title = title[0] if title else None
            getchap().append({
                'start_time': start,
                'end_time': end,
                'title': title,
            })
            pp.to_screen(f'Added chapter (start={start} end={end} title={title})')
            return
        elif 'insert'.startswith(args[0]):
            # insert INDEX START END [TITLE]
            index, start, end, *title = args[1:]
            index = int(index)
            start, end = map(parse_duration, (start, end))
            title = title[0] if title else None
            getchap().insert(index, {
                'start_time': start,
                'end_time': end,
                'title': title,
            })
            pp.to_screen(f'Added chapter at index {index} (start={start} end={end} title={title})')
            return
        elif 'remove'.startswith(args[0]):
            # remove [idx:]INDEX[-INDEX_END]|[t:]TIME-RANGE|[re:]TITLE_REGEX

            def find_mode():
                candidates = []
                mobj = re.fullmatch(r'(idx:)?(\d+)(?:-(\d+))?', args[1])
                if mobj:
                    pfx, s1, e1 = mobj.groups()

                    def selector(chap):
                        return s1 < chap[0] and chap[0] < e1

                    if pfx:
                        return selector
                    candidates.append(selector)

                mobj = re.fullmatch(r'(t:)?(\d+)(?:-(\d+))?', args[1])
                if mobj:
                    pfx, s2, e2 = mobj.groups()

                    def selector(chap):
                        chap = chap[1]
                        cs = chap.get('start_time') or 0
                        ce = chap.get('end_time')
                        if ce is None:
                            ce = inf
                        return (s2 < cs and cs < e2) or (s2 < ce and ce < e2)

                    if pfx:
                        return selector
                    candidates.append(selector)

                mobj = re.fullmatch(r'(re:)?(.+)', args[1])
                while mobj:
                    pfx, tre = mobj.groups()
                    try:
                        cre = re.compile(tre)
                    except re.error:
                        break

                    def selector(chap):
                        chap = chap[1]
                        return bool(re.match(cre, chap.get('title') or ''))

                    if pfx:
                        return selector
                    candidates.append(selector)
                    break

                return next(candidates, None)

            selector = find_mode()
            if not selector:
                pp.to_screen(f'Invalid selector: {args[1]}')
                return

            chap = getchap()
            chap[:] = map(lambda x: x[1], filter(selector, enumerate(chap)))

            return
        elif 'sort'.startswith(args[0]):
            # sort
            end = info.get('duration', None) or 0

            def _sort(x):
                return (
                    x.get('start_time') or 0,
                    x.get('end_time') or end,
                    x.get('index') or 0,
                    x.get('title') or '',)

            chap = getchap()
            chap.sort(key=_sort)

            for i, x in enumerate(chap):
                x['index'] = i

            pp.to_screen(f'Sorted {len(chap)} chapters')
            return
        pp.to_screen(f'Unknown command: {args[0]!r}. Use "help {self.COMMAND_NAME}" to see help')

class SetValueCommand(MetapulatorCommand):
    """
    Command to set a value to a key
    """

    COMMAND_NAME = 'setvalue'
    HELP = """\
Synopsis: setvalue KEY [TYPE] VALUE

Command to set a value to a key.

TYPE can be any of: "string", "int", "float" and "json"
Defaults to "string" if skipped.

For "int" and "float", the VALUE must be an interger or a decimal number.

For "json", the VALUE must be a valid JSON.
"""

    def run(self, pp, info, args):
        if len(args) not in (2, 3):
            raise ShowHelp()
        _type = 'string'
        if len(args) == 2:
            key, value_str = args
        elif len(args) == 3:
            key, _type, value_str = args
        _type = _type.lower()
        value = value_str
        if _type == 'int':
            value = int(value_str)
        elif _type == 'float':
            value = float(value_str)
        elif _type == 'json':
            value = json.loads(value_str)
        info[key] = value

class HelpCommand(MetapulatorCommand):
    """
    Command to show help
    """

    COMMAND_NAME = 'help'
    HELP = """\
Synopsis: help [COMMAND]

If COMMAND is not given, it shows all of the available commands.
"""

    def run(self, pp, info, args):
        if not args:
            pp.to_screen('All available commands:')
            pp.to_screen('\n'.join(sorted(pp.COMMANDS.keys())), False)
            return
        cmd = pp.find_command(args[0])
        if not cmd:
            pp.to_screen(f'Command {args[0]!r} not found')
            return
        pp.print_help(cmd)

CLASSES = (ExitCommand, PrintCommand, ChaptersCommand, SetValueCommand, HelpCommand)


code:
from __future__ import unicode_literals
import itertools

import os
import re
import subprocess

from .common import PostProcessor

from ..utils import (
    _get_exe_version_output,
    determine_ext,
    encodeArgument,
    encodeFilename,
    Popen,
    PostProcessingError,
    prepend_extension,
    shell_quote,
    variadic,
)

class MkvToolNixPostProcessorError(PostProcessingError):
    def __init__(self, msg=None, retval=None):
        super().__init__(msg=msg)
        self.retval = retval

class MkvToolNixPostProcessor(PostProcessor):
    _EXECUTABLE = ''

    def __init__(self, downloader=None):
        PostProcessor.__init__(self, downloader)
        self._PROGRESS_LABEL = self.pp_key()
        self._determine_executables()

    def _determine_executables(self):
        self._path = {}
        self._version = None
        self._accepted_formats = ()

        def get_executable_version(path, prog):
            out = _get_exe_version_output(path, ['--version'])
            regexs = [
                r'v((?:\d+\.)+\d)'
            ]
            ver = next((mobj.group(1) for mobj in filter(None, (re.match(regex, out) for regex in regexs))), None)
            self._version = ver
            if prog != 'mkvmerge' or not out:
                return

            # get list of supported formats
            out = _get_exe_version_output(path, ['--list-types'])
            self._accepted_formats = tuple(ext for mobj in re.finditer(r'\[(.+?)\]', out) for ext in mobj.group(1).split())

        ex = self._EXECUTABLE
        location = self.get_param(f'{ex}_location')
        if not location:
            mtn = self.get_param('mkvtoolnix_location')
            if mtn:
                location = os.path.join(self.get_param('mkvtoolnix_location'), self._EXECUTABLE)

        if not location:
            self._path = ex
        else:
            if not os.path.exists(location):
                self.report_warning(
                    '{self._BINARY_NAME}-location %s does not exist! '
                    'Continuing without {self._BINARY_NAME}.' % (location),
                    only_once=True)
                return
            elif os.path.isdir(location):
                dirname, basename = location, None
            else:
                basename = os.path.splitext(os.path.basename(location))[0]
                basename = ex if basename.startswith(ex) else None
                dirname = os.path.dirname(os.path.abspath(location))

            self._path = location if basename else os.path.join(dirname, ex)

        get_executable_version(self._path, ex)

    @property
    def available(self):
        return bool(self._path)

    def run_binary(self, input_path_opts, output_path_opts, *, expected_retcodes=(0,), info_dict=None):
        cmd = [encodeFilename(self._path, True), encodeArgument('-y')]

        oldest_mtime = min(
            self._downloader.stat(path).st_mtime for path, _ in input_path_opts if path)

        if len(output_path_opts) != 1:
            raise MkvToolNixPostProcessorError('Number of output file must be exactly one file.')

        def make_args(file, args, name, number):
            keys = ['_%s%d' % (name, number), '_%s' % name]
            if name == 'i' and self._accepted_formats:
                ext = determine_ext(file, None)
                if ext and ext not in self._accepted_formats:
                    raise MkvToolNixPostProcessorError(f'Format {ext} is not supprted for input. Use ffmpeg to do this post processing.')
            if name == 'o':
                ext = determine_ext(file, None)
                if ext in ('webm', 'webmv', 'webma'):
                    args.append('--webm')
                elif ext not in ('mka', 'mks', 'mkv', 'mk3d'):
                    raise MkvToolNixPostProcessorError(f'Format {ext} is not supprted for output. Use ffmpeg to do this post processing.')
                args.append('-o')
                if number == 1:
                    keys.append('')
            args += self._configuration_args(self._EXECUTABLE, keys)
            return (
                [encodeArgument(arg) for arg in args] + [file])

        for arg_type, path_opts in (('o', output_path_opts), ('i', input_path_opts)):
            cmd += itertools.chain.from_iterable(
                make_args(path, list(opts), arg_type, i + 1)
                for i, (path, opts) in enumerate(path_opts) if path)

        self.write_debug('program command line: %s' % shell_quote(cmd))

        p = Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE)
        stderr = p.communicate_or_kill()[1]
        retval = p.returncode

        if isinstance(stderr, bytes):
            stderr = stderr.decode('utf-8', 'replace')

        if retval not in variadic(expected_retcodes):
            stderr = stderr.strip()
            self.write_debug(stderr)
            raise MkvToolNixPostProcessorError(stderr.split('\n')[-1], retval)

        for out_path, _ in output_path_opts:
            if out_path:
                self.try_utime(out_path, oldest_mtime, oldest_mtime)
        return stderr

class MkvMergePostProcessor(MkvToolNixPostProcessor):

    @PostProcessor._restrict_to(images=False)
    def run(self, info):
        filename = info['filepath']
        temp_filename = prepend_extension(filename, 'temp')
        args = []
        for (i, fmt) in enumerate(info['requested_formats']):
            if fmt.get('acodec') != 'none':
                args.extend(['--audio-tracks', '0'])
            if fmt.get('vcodec') != 'none':
                args.extend(['--video-tracks', '0'])
        self.to_screen('Merging formats into "%s"' % filename)
        self.run_binary(info['__files_to_merge'], temp_filename, args, info_dict=info)
        self._downloader.rename(temp_filename, filename)
        return info['__files_to_merge'], info


code:
import copy
import heapq
import os

from .common import PostProcessor
from .ffmpeg import FFmpegPostProcessor, FFmpegSubtitlesConvertorPP
from .sponsorblock import SponsorBlockPP
from ..utils import PostProcessingError, orderedSet, prepend_extension

_TINY_CHAPTER_DURATION = 1
DEFAULT_SPONSORBLOCK_CHAPTER_TITLE = '[SponsorBlock]: %(category_names)l'

class ModifyChaptersPP(FFmpegPostProcessor):
    def __init__(self, downloader, remove_chapters_patterns=None, remove_sponsor_segments=None, remove_ranges=None,
                 *, sponsorblock_chapter_title=DEFAULT_SPONSORBLOCK_CHAPTER_TITLE, force_keyframes=False):
        FFmpegPostProcessor.__init__(self, downloader)
        self._remove_chapters_patterns = set(remove_chapters_patterns or [])
        self._remove_sponsor_segments = set(remove_sponsor_segments or []) - set(SponsorBlockPP.POI_CATEGORIES.keys())
        self._ranges_to_remove = set(remove_ranges or [])
        self._sponsorblock_chapter_title = sponsorblock_chapter_title
        self._force_keyframes = force_keyframes

    @PostProcessor._restrict_to(images=False)
    def run(self, info):
        # Chapters must be preserved intact when downloading multiple formats of the same video.
        chapters, sponsor_chapters = self._mark_chapters_to_remove(
            copy.deepcopy(info.get('chapters')) or [],
            copy.deepcopy(info.get('sponsorblock_chapters')) or [])
        if not chapters and not sponsor_chapters:
            return [], info

        real_duration = self._get_real_video_duration(info['filepath'])
        if not chapters:
            chapters = [{'start_time': 0, 'end_time': info.get('duration') or real_duration, 'title': info['title']}]

        info['chapters'], cuts = self._remove_marked_arrange_sponsors(chapters + sponsor_chapters)
        if not cuts:
            return [], info

        original_duration, info['duration'] = info.get('duration'), info['chapters'][-1]['end_time']
        if self._duration_mismatch(real_duration, original_duration, 1):
            if not self._duration_mismatch(real_duration, info['duration']):
                self.to_screen(f'Skipping {self.pp_key()} since the video appears to be already cut')
                return [], info
            if not info.get('__real_download'):
                raise PostProcessingError('Cannot cut video since the real and expected durations mismatch. '
                                          'Different chapters may have already been removed')
            else:
                self.write_debug('Expected and actual durations mismatch')

        concat_opts = self._make_concat_opts(cuts, real_duration)
        self.write_debug('Concat spec = %s' % ', '.join(f'{c.get("inpoint", 0.0)}-{c.get("outpoint", "inf")}' for c in concat_opts))

        def remove_chapters(file, is_sub):
            return file, self.remove_chapters(file, cuts, concat_opts, self._force_keyframes and not is_sub)

        in_out_files = [remove_chapters(info['filepath'], False)]
        in_out_files.extend(remove_chapters(in_file, True) for in_file in self._get_supported_subs(info))

        # Renaming should only happen after all files are processed
        files_to_remove = []
        for in_file, out_file in in_out_files:
            mtime = os.stat(in_file).st_mtime
            uncut_file = prepend_extension(in_file, 'uncut')
            os.replace(in_file, uncut_file)
            os.replace(out_file, in_file)
            self.try_utime(in_file, mtime, mtime)
            files_to_remove.append(uncut_file)

        return files_to_remove, info

    def _mark_chapters_to_remove(self, chapters, sponsor_chapters):
        if self._remove_chapters_patterns:
            warn_no_chapter_to_remove = True
            if not chapters:
                self.to_screen('Chapter information is unavailable')
                warn_no_chapter_to_remove = False
            for c in chapters:
                if any(regex.search(c['title']) for regex in self._remove_chapters_patterns):
                    c['remove'] = True
                    warn_no_chapter_to_remove = False
            if warn_no_chapter_to_remove:
                self.to_screen('There are no chapters matching the regex')

        if self._remove_sponsor_segments:
            warn_no_chapter_to_remove = True
            if not sponsor_chapters:
                self.to_screen('SponsorBlock information is unavailable')
                warn_no_chapter_to_remove = False
            for c in sponsor_chapters:
                if c['category'] in self._remove_sponsor_segments:
                    c['remove'] = True
                    warn_no_chapter_to_remove = False
            if warn_no_chapter_to_remove:
                self.to_screen('There are no matching SponsorBlock chapters')

        sponsor_chapters.extend({
            'start_time': start,
            'end_time': end,
            'category': 'manually_removed',
            '_categories': [('manually_removed', start, end)],
            'remove': True,
        } for start, end in self._ranges_to_remove)

        return chapters, sponsor_chapters

    def _get_supported_subs(self, info):
        for sub in (info.get('requested_subtitles') or {}).values():
            sub_file = sub.get('filepath')
            # The file might have been removed by --embed-subs
            if not sub_file or not os.path.exists(sub_file):
                continue
            ext = sub['ext']
            if ext not in FFmpegSubtitlesConvertorPP.SUPPORTED_EXTS:
                self.report_warning(f'Cannot remove chapters from external {ext} subtitles; "{sub_file}" is now out of sync')
                continue
            # TODO: create __real_download for subs?
            yield sub_file

    def _remove_marked_arrange_sponsors(self, chapters):
        # Store cuts separately, since adjacent and overlapping cuts must be merged.
        cuts = []

        def append_cut(c):
            assert 'remove' in c, 'Not a cut is appended to cuts'
            last_to_cut = cuts[-1] if cuts else None
            if last_to_cut and last_to_cut['end_time'] >= c['start_time']:
                last_to_cut['end_time'] = max(last_to_cut['end_time'], c['end_time'])
            else:
                cuts.append(c)
            return len(cuts) - 1

        def excess_duration(c):
            # Cuts that are completely within the chapter reduce chapters' duration.
            # Since cuts can overlap, excess duration may be less that the sum of cuts' durations.
            # To avoid that, chapter stores the index to the fist cut within the chapter,
            # instead of storing excess duration. append_cut ensures that subsequent cuts (if any)
            # will be merged with previous ones (if necessary).
            cut_idx, excess = c.pop('cut_idx', len(cuts)), 0
            while cut_idx < len(cuts):
                cut = cuts[cut_idx]
                if cut['start_time'] >= c['end_time']:
                    break
                if cut['end_time'] > c['start_time']:
                    excess += min(cut['end_time'], c['end_time'])
                    excess -= max(cut['start_time'], c['start_time'])
                cut_idx += 1
            return excess

        new_chapters = []

        def append_chapter(c):
            assert 'remove' not in c, 'Cut is appended to chapters'
            length = c['end_time'] - c['start_time'] - excess_duration(c)
            # Chapter is completely covered by cuts or sponsors.
            if length <= 0:
                return
            start = new_chapters[-1]['end_time'] if new_chapters else 0
            c.update(start_time=start, end_time=start + length)
            new_chapters.append(c)

        # Turn into a priority queue, index is a tie breaker.
        # Plain stack sorted by start_time is not enough: after splitting the chapter,
        # the part returned to the stack is not guaranteed to have start_time
        # less than or equal to the that of the stack's head.
        chapters = [(c['start_time'], i, c) for i, c in enumerate(chapters)]
        heapq.heapify(chapters)

        _, cur_i, cur_chapter = heapq.heappop(chapters)
        while chapters:
            _, i, c = heapq.heappop(chapters)
            # Non-overlapping chapters or cuts can be appended directly. However,
            # adjacent non-overlapping cuts must be merged, which is handled by append_cut.
            if cur_chapter['end_time'] <= c['start_time']:
                (append_chapter if 'remove' not in cur_chapter else append_cut)(cur_chapter)
                cur_i, cur_chapter = i, c
                continue

            # Eight possibilities for overlapping chapters: (cut, cut), (cut, sponsor),
            # (cut, normal), (sponsor, cut), (normal, cut), (sponsor, sponsor),
            # (sponsor, normal), and (normal, sponsor). There is no (normal, normal):
            # normal chapters are assumed not to overlap.
            if 'remove' in cur_chapter:
                # (cut, cut): adjust end_time.
                if 'remove' in c:
                    cur_chapter['end_time'] = max(cur_chapter['end_time'], c['end_time'])
                # (cut, sponsor/normal): chop the beginning of the later chapter
                # (if it's not completely hidden by the cut). Push to the priority queue
                # to restore sorting by start_time: with beginning chopped, c may actually
                # start later than the remaining chapters from the queue.
                elif cur_chapter['end_time'] < c['end_time']:
                    c['start_time'] = cur_chapter['end_time']
                    c['_was_cut'] = True
                    heapq.heappush(chapters, (c['start_time'], i, c))
            # (sponsor/normal, cut).
            elif 'remove' in c:
                cur_chapter['_was_cut'] = True
                # Chop the end of the current chapter if the cut is not contained within it.
                # Chopping the end doesn't break start_time sorting, no PQ push is necessary.
                if cur_chapter['end_time'] <= c['end_time']:
                    cur_chapter['end_time'] = c['start_time']
                    append_chapter(cur_chapter)
                    cur_i, cur_chapter = i, c
                    continue
                # Current chapter contains the cut within it. If the current chapter is
                # a sponsor chapter, check whether the categories before and after the cut differ.
                if '_categories' in cur_chapter:
                    after_c = dict(cur_chapter, start_time=c['end_time'], _categories=[])
                    cur_cats = []
                    for cat_start_end in cur_chapter['_categories']:
                        if cat_start_end[1] < c['start_time']:
                            cur_cats.append(cat_start_end)
                        if cat_start_end[2] > c['end_time']:
                            after_c['_categories'].append(cat_start_end)
                    cur_chapter['_categories'] = cur_cats
                    if cur_chapter['_categories'] != after_c['_categories']:
                        # Categories before and after the cut differ: push the after part to PQ.
                        heapq.heappush(chapters, (after_c['start_time'], cur_i, after_c))
                        cur_chapter['end_time'] = c['start_time']
                        append_chapter(cur_chapter)
                        cur_i, cur_chapter = i, c
                        continue
                # Either sponsor categories before and after the cut are the same or
                # we're dealing with a normal chapter. Just register an outstanding cut:
                # subsequent append_chapter will reduce the duration.
                cur_chapter.setdefault('cut_idx', append_cut(c))
            # (sponsor, normal): if a normal chapter is not completely overlapped,
            # chop the beginning of it and push it to PQ.
            elif '_categories' in cur_chapter and '_categories' not in c:
                if cur_chapter['end_time'] < c['end_time']:
                    c['start_time'] = cur_chapter['end_time']
                    c['_was_cut'] = True
                    heapq.heappush(chapters, (c['start_time'], i, c))
            # (normal, sponsor) and (sponsor, sponsor)
            else:
                assert '_categories' in c, 'Normal chapters overlap'
                cur_chapter['_was_cut'] = True
                c['_was_cut'] = True
                # Push the part after the sponsor to PQ.
                if cur_chapter['end_time'] > c['end_time']:
                    # deepcopy to make categories in after_c and cur_chapter/c refer to different lists.
                    after_c = dict(copy.deepcopy(cur_chapter), start_time=c['end_time'])
                    heapq.heappush(chapters, (after_c['start_time'], cur_i, after_c))
                # Push the part after the overlap to PQ.
                elif c['end_time'] > cur_chapter['end_time']:
                    after_cur = dict(copy.deepcopy(c), start_time=cur_chapter['end_time'])
                    heapq.heappush(chapters, (after_cur['start_time'], cur_i, after_cur))
                    c['end_time'] = cur_chapter['end_time']
                # (sponsor, sponsor): merge categories in the overlap.
                if '_categories' in cur_chapter:
                    c['_categories'] = cur_chapter['_categories'] + c['_categories']
                # Inherit the cuts that the current chapter has accumulated within it.
                if 'cut_idx' in cur_chapter:
                    c['cut_idx'] = cur_chapter['cut_idx']
                cur_chapter['end_time'] = c['start_time']
                append_chapter(cur_chapter)
                cur_i, cur_chapter = i, c
        (append_chapter if 'remove' not in cur_chapter else append_cut)(cur_chapter)
        return self._remove_tiny_rename_sponsors(new_chapters), cuts

    def _remove_tiny_rename_sponsors(self, chapters):
        new_chapters = []
        for i, c in enumerate(chapters):
            # Merge with the previous/next if the chapter is tiny.
            # Only tiny chapters resulting from a cut can be skipped.
            # Chapters that were already tiny in the original list will be preserved.
            if (('_was_cut' in c or '_categories' in c)
                    and c['end_time'] - c['start_time'] < _TINY_CHAPTER_DURATION):
                if not new_chapters:
                    # Prepend tiny chapter to the next one if possible.
                    if i < len(chapters) - 1:
                        chapters[i + 1]['start_time'] = c['start_time']
                        continue
                else:
                    old_c = new_chapters[-1]
                    if i < len(chapters) - 1:
                        next_c = chapters[i + 1]
                        # Not a typo: key names in old_c and next_c are really different.
                        prev_is_sponsor = 'categories' in old_c
                        next_is_sponsor = '_categories' in next_c
                        # Preferentially prepend tiny normals to normals and sponsors to sponsors.
                        if (('_categories' not in c and prev_is_sponsor and not next_is_sponsor)
                                or ('_categories' in c and not prev_is_sponsor and next_is_sponsor)):
                            next_c['start_time'] = c['start_time']
                            continue
                    old_c['end_time'] = c['end_time']
                    continue

            c.pop('_was_cut', None)
            cats = c.pop('_categories', None)
            if cats:
                category = min(cats, key=lambda c: c[2] - c[1])[0]
                cats = orderedSet(x[0] for x in cats)
                c.update({
                    'category': category,
                    'categories': cats,
                    'name': SponsorBlockPP.CATEGORIES[category],
                    'category_names': [SponsorBlockPP.CATEGORIES[c] for c in cats]
                })
                c['title'] = self._downloader.evaluate_outtmpl(self._sponsorblock_chapter_title, c.copy())
                # Merge identically named sponsors.
                if (new_chapters and 'categories' in new_chapters[-1]
                        and new_chapters[-1]['title'] == c['title']):
                    new_chapters[-1]['end_time'] = c['end_time']
                    continue
            new_chapters.append(c)
        return new_chapters

    def remove_chapters(self, filename, ranges_to_cut, concat_opts, force_keyframes=False):
        in_file = filename
        out_file = prepend_extension(in_file, 'temp')
        if force_keyframes:
            in_file = self.force_keyframes(in_file, (t for c in ranges_to_cut for t in (c['start_time'], c['end_time'])))
        self.to_screen(f'Removing chapters from {filename}')
        self.concat_files([in_file] * len(concat_opts), out_file, concat_opts)
        if in_file != filename:
            self._delete_downloaded_files(in_file, msg=None)
        return out_file

    @staticmethod
    def _make_concat_opts(chapters_to_remove, duration):
        opts = [{}]
        for s in chapters_to_remove:
            # Do not create 0 duration chunk at the beginning.
            if s['start_time'] == 0:
                opts[-1]['inpoint'] = f'{s["end_time"]:.6f}'
                continue
            opts[-1]['outpoint'] = f'{s["start_time"]:.6f}'
            # Do not create 0 duration chunk at the end.
            if s['end_time'] < duration:
                opts.append({'inpoint': f'{s["end_time"]:.6f}'})
        return opts


code:
import os
import shutil

from .common import PostProcessor
from ..utils import (
    PostProcessingError,
    decodeFilename,
    encodeFilename,
    make_dir,
)

class MoveFilesAfterDownloadPP(PostProcessor):

    def __init__(self, downloader=None, downloaded=True):
        PostProcessor.__init__(self, downloader)
        self._downloaded = downloaded

    @classmethod
    def pp_key(cls):
        return 'MoveFiles'

    def run(self, info):
        dl_path, dl_name = os.path.split(encodeFilename(info['filepath']))
        finaldir = info.get('__finaldir', dl_path)
        finalpath = os.path.join(finaldir, dl_name)
        if self._downloaded:
            info['__files_to_move'][info['filepath']] = decodeFilename(finalpath)

        make_newfilename = lambda old: decodeFilename(os.path.join(finaldir, os.path.basename(encodeFilename(old))))
        for oldfile, newfile in info['__files_to_move'].items():
            if not newfile:
                newfile = make_newfilename(oldfile)
            if os.path.abspath(encodeFilename(oldfile)) == os.path.abspath(encodeFilename(newfile)):
                continue
            if not os.path.exists(encodeFilename(oldfile)):
                self.report_warning('File "%s" cannot be found' % oldfile)
                continue
            if os.path.exists(encodeFilename(newfile)):
                if self.get_param('overwrites', True):
                    self.report_warning('Replacing existing file "%s"' % newfile)
                    os.remove(encodeFilename(newfile))
                else:
                    self.report_warning(
                        'Cannot move file "%s" out of temporary directory since "%s" already exists. '
                        % (oldfile, newfile))
                    continue
            make_dir(newfile, PostProcessingError)
            self.to_screen(f'Moving file "{oldfile}" to "{newfile}"')
            shutil.move(oldfile, newfile)  # os.rename cannot move between volumes

        info['filepath'] = finalpath
        return [], info


code:
import os
import shlex
import subprocess

from .common import PostProcessor
from ..utils import (
    Popen,
    PostProcessingError,
    check_executable,
    cli_option,
    encodeArgument,
    encodeFilename,
    prepend_extension,
    shell_quote,
    str_or_none,
)

# Deprecated in favor of the native implementation
class SponSkrubPP(PostProcessor):
    _temp_ext = 'spons'
    _exe_name = 'sponskrub'

    def __init__(self, downloader, path='', args=None, ignoreerror=False, cut=False, force=False, _from_cli=False):
        PostProcessor.__init__(self, downloader)
        self.force = force
        self.cutout = cut
        self.args = str_or_none(args) or ''  # For backward compatibility
        self.path = self.get_exe(path)

        if not _from_cli:
            self.deprecation_warning(
                'yt_dlp.postprocessor.SponSkrubPP support is deprecated and may be removed in a future version. '
                'Use yt_dlp.postprocessor.SponsorBlock and yt_dlp.postprocessor.ModifyChaptersPP instead')

        if not ignoreerror and self.path is None:
            if path:
                raise PostProcessingError('sponskrub not found in "%s"' % path)
            else:
                raise PostProcessingError('sponskrub not found. Please install or provide the path using --sponskrub-path')

    def get_exe(self, path=''):
        if not path or not check_executable(path, ['-h']):
            path = os.path.join(path, self._exe_name)
            if not check_executable(path, ['-h']):
                return None
        return path

    @PostProcessor._restrict_to(images=False)
    def run(self, information):
        if self.path is None:
            return [], information

        filename = information['filepath']
        if not os.path.exists(encodeFilename(filename)):  # no download
            return [], information

        if information['extractor_key'].lower() != 'youtube':
            self.to_screen('Skipping sponskrub since it is not a YouTube video')
            return [], information
        if self.cutout and not self.force and not information.get('__real_download', False):
            self.report_warning(
                'Skipping sponskrub since the video was already downloaded. '
                'Use --sponskrub-force to run sponskrub anyway')
            return [], information

        self.to_screen('Trying to %s sponsor sections' % ('remove' if self.cutout else 'mark'))
        if self.cutout:
            self.report_warning('Cutting out sponsor segments will cause the subtitles to go out of sync.')
            if not information.get('__real_download', False):
                self.report_warning('If sponskrub is run multiple times, unintended parts of the video could be cut out.')

        temp_filename = prepend_extension(filename, self._temp_ext)
        if os.path.exists(encodeFilename(temp_filename)):
            os.remove(encodeFilename(temp_filename))

        cmd = [self.path]
        if not self.cutout:
            cmd += ['-chapter']
        cmd += cli_option(self._downloader.params, '-proxy', 'proxy')
        cmd += shlex.split(self.args)  # For backward compatibility
        cmd += self._configuration_args(self._exe_name, use_compat=False)
        cmd += ['--', information['id'], filename, temp_filename]
        cmd = [encodeArgument(i) for i in cmd]

        self.write_debug('sponskrub command line: %s' % shell_quote(cmd))
        stdout, _, returncode = Popen.run(cmd, text=True, stdout=None if self.get_param('verbose') else subprocess.PIPE)

        if not returncode:
            os.replace(temp_filename, filename)
            self.to_screen('Sponsor sections have been %s' % ('removed' if self.cutout else 'marked'))
        elif returncode == 3:
            self.to_screen('No segments in the SponsorBlock database')
        else:
            raise PostProcessingError(
                stdout.strip().splitlines()[0 if stdout.strip().lower().startswith('unrecognised') else -1]
                or f'sponskrub failed with error code {returncode}')
        return [], information


code:
import hashlib
import json
import re
import urllib.parse

from .ffmpeg import FFmpegPostProcessor

class SponsorBlockPP(FFmpegPostProcessor):
    # https://wiki.sponsor.ajay.app/w/Types
    EXTRACTORS = {
        'Youtube': 'YouTube',
    }
    POI_CATEGORIES = {
        'poi_highlight': 'Highlight',
    }
    CATEGORIES = {
        'sponsor': 'Sponsor',
        'intro': 'Intermission/Intro Animation',
        'outro': 'Endcards/Credits',
        'selfpromo': 'Unpaid/Self Promotion',
        'preview': 'Preview/Recap',
        'filler': 'Filler Tangent',
        'interaction': 'Interaction Reminder',
        'music_offtopic': 'Non-Music Section',
        **POI_CATEGORIES,
    }

    def __init__(self, downloader, categories=None, api='https://sponsor.ajay.app'):
        FFmpegPostProcessor.__init__(self, downloader)
        self._categories = tuple(categories or self.CATEGORIES.keys())
        self._API_URL = api if re.match('^https?://', api) else 'https://' + api

    def run(self, info):
        extractor = info['extractor_key']
        if extractor not in self.EXTRACTORS:
            self.to_screen(f'SponsorBlock is not supported for {extractor}')
            return [], info

        self.to_screen('Fetching SponsorBlock segments')
        info['sponsorblock_chapters'] = self._get_sponsor_chapters(info, info.get('duration'))
        return [], info

    def _get_sponsor_chapters(self, info, duration):
        segments = self._get_sponsor_segments(info['id'], self.EXTRACTORS[info['extractor_key']])

        def duration_filter(s):
            start_end = s['segment']
            # Ignore entire video segments (https://wiki.sponsor.ajay.app/w/Types).
            if start_end == (0, 0):
                return False
            # Ignore milliseconds difference at the start.
            if start_end[0] <= 1:
                start_end[0] = 0
            # Make POI chapters 1 sec so that we can properly mark them
            if s['category'] in self.POI_CATEGORIES.keys():
                start_end[1] += 1
            # Ignore milliseconds difference at the end.
            # Never allow the segment to exceed the video.
            if duration and duration - start_end[1] <= 1:
                start_end[1] = duration
            # SponsorBlock duration may be absent or it may deviate from the real one.
            return s['videoDuration'] == 0 or not duration or abs(duration - s['videoDuration']) <= 1

        duration_match = [s for s in segments if duration_filter(s)]
        if len(duration_match) != len(segments):
            self.report_warning('Some SponsorBlock segments are from a video of different duration, maybe from an old version of this video')

        def to_chapter(s):
            (start, end), cat = s['segment'], s['category']
            return {
                'start_time': start,
                'end_time': end,
                'category': cat,
                'title': self.CATEGORIES[cat],
                '_categories': [(cat, start, end)]
            }

        sponsor_chapters = [to_chapter(s) for s in duration_match]
        if not sponsor_chapters:
            self.to_screen('No segments were found in the SponsorBlock database')
        else:
            self.to_screen(f'Found {len(sponsor_chapters)} segments in the SponsorBlock database')
        return sponsor_chapters

    def _get_sponsor_segments(self, video_id, service):
        hash = hashlib.sha256(video_id.encode('ascii')).hexdigest()
        # SponsorBlock API recommends using first 4 hash characters.
        url = f'{self._API_URL}/api/skipSegments/{hash[:4]}?' + urllib.parse.urlencode({
            'service': service,
            'categories': json.dumps(self._categories),
            'actionTypes': json.dumps(['skip', 'poi'])
        })
        for d in self._download_json(url) or []:
            if d['videoID'] == video_id:
                return d['segments']
        return []


code:
import os

from .common import PostProcessor
from ..compat import compat_os_name
from ..utils import (
    PostProcessingError,
    XAttrMetadataError,
    XAttrUnavailableError,
    hyphenate_date,
    write_xattr,
)

class XAttrMetadataPP(PostProcessor):
    """Set extended attributes on downloaded file (if xattr support is found)

    More info about extended attributes for media:
        http://freedesktop.org/wiki/CommonExtendedAttributes/
        http://www.freedesktop.org/wiki/PhreedomDraft/
        http://dublincore.org/documents/usageguide/elements.shtml

    TODO:
        * capture youtube keywords and put them in 'user.dublincore.subject' (comma-separated)
        * figure out which xattrs can be used for 'duration', 'thumbnail', 'resolution'
    """

    XATTR_MAPPING = {
        'user.xdg.referrer.url': 'webpage_url',
        # 'user.xdg.comment': 'description',
        'user.dublincore.title': 'title',
        'user.dublincore.date': 'upload_date',
        'user.dublincore.description': 'description',
        'user.dublincore.contributor': 'uploader',
        'user.dublincore.format': 'format',
    }

    def run(self, info):
        mtime = os.stat(info['filepath']).st_mtime
        self.to_screen('Writing metadata to file\'s xattrs')
        try:
            for xattrname, infoname in self.XATTR_MAPPING.items():
                value = info.get(infoname)
                if value:
                    if infoname == 'upload_date':
                        value = hyphenate_date(value)
                    write_xattr(info['filepath'], xattrname, value.encode())

        except XAttrUnavailableError as e:
            raise PostProcessingError(str(e))
        except XAttrMetadataError as e:
            if e.reason == 'NO_SPACE':
                self.report_warning(
                    'There\'s no disk space left, disk quota exceeded or filesystem xattr limit exceeded. '
                    'Some extended attributes are not written')
            elif e.reason == 'VALUE_TOO_LONG':
                self.report_warning('Unable to write extended attributes due to too long values.')
            else:
                tip = ('You need to use NTFS' if compat_os_name == 'nt'
                       else 'You may have to enable them in your "/etc/fstab"')
                raise PostProcessingError(f'This filesystem doesn\'t support extended attributes. {tip}')

        self.try_utime(info['filepath'], mtime, mtime)
        return [], info


code:
# Public Domain SOCKS proxy protocol implementation
# Adapted from https://gist.github.com/bluec0re/cafd3764412967417fd3
# References:
# SOCKS4 protocol http://www.openssh.com/txt/socks4.protocol
# SOCKS4A protocol http://www.openssh.com/txt/socks4a.protocol
# SOCKS5 protocol https://tools.ietf.org/html/rfc1928
# SOCKS5 username/password authentication https://tools.ietf.org/html/rfc1929

import collections
import socket
import struct

from .compat import compat_ord

__author__ = 'Timo Schmid <coding@timoschmid.de>'

SOCKS4_VERSION = 4
SOCKS4_REPLY_VERSION = 0x00
# Excerpt from SOCKS4A protocol:
# if the client cannot resolve the destination host's domain name to find its
# IP address, it should set the first three bytes of DSTIP to NULL and the last
# byte to a non-zero value.
SOCKS4_DEFAULT_DSTIP = struct.pack('!BBBB', 0, 0, 0, 0xFF)

SOCKS5_VERSION = 5
SOCKS5_USER_AUTH_VERSION = 0x01
SOCKS5_USER_AUTH_SUCCESS = 0x00

class Socks4Command:
    CMD_CONNECT = 0x01
    CMD_BIND = 0x02

class Socks5Command(Socks4Command):
    CMD_UDP_ASSOCIATE = 0x03

class Socks5Auth:
    AUTH_NONE = 0x00
    AUTH_GSSAPI = 0x01
    AUTH_USER_PASS = 0x02
    AUTH_NO_ACCEPTABLE = 0xFF  # For server response

class Socks5AddressType:
    ATYP_IPV4 = 0x01
    ATYP_DOMAINNAME = 0x03
    ATYP_IPV6 = 0x04

class ProxyError(socket.error):
    ERR_SUCCESS = 0x00

    def __init__(self, code=None, msg=None):
        if code is not None and msg is None:
            msg = self.CODES.get(code) or 'unknown error'
        super().__init__(code, msg)

class InvalidVersionError(ProxyError):
    def __init__(self, expected_version, got_version):
        msg = ('Invalid response version from server. Expected {:02x} got '
               '{:02x}'.format(expected_version, got_version))
        super().__init__(0, msg)

class Socks4Error(ProxyError):
    ERR_SUCCESS = 90

    CODES = {
        91: 'request rejected or failed',
        92: 'request rejected because SOCKS server cannot connect to identd on the client',
        93: 'request rejected because the client program and identd report different user-ids'
    }

class Socks5Error(ProxyError):
    ERR_GENERAL_FAILURE = 0x01

    CODES = {
        0x01: 'general SOCKS server failure',
        0x02: 'connection not allowed by ruleset',
        0x03: 'Network unreachable',
        0x04: 'Host unreachable',
        0x05: 'Connection refused',
        0x06: 'TTL expired',
        0x07: 'Command not supported',
        0x08: 'Address type not supported',
        0xFE: 'unknown username or invalid password',
        0xFF: 'all offered authentication methods were rejected'
    }

class ProxyType:
    SOCKS4 = 0
    SOCKS4A = 1
    SOCKS5 = 2

Proxy = collections.namedtuple('Proxy', (
    'type', 'host', 'port', 'username', 'password', 'remote_dns'))

class sockssocket(socket.socket):
    def __init__(self, *args, **kwargs):
        self._proxy = None
        super().__init__(*args, **kwargs)

    def setproxy(self, proxytype, addr, port, rdns=True, username=None, password=None):
        assert proxytype in (ProxyType.SOCKS4, ProxyType.SOCKS4A, ProxyType.SOCKS5)

        self._proxy = Proxy(proxytype, addr, port, username, password, rdns)

    def recvall(self, cnt):
        data = b''
        while len(data) < cnt:
            cur = self.recv(cnt - len(data))
            if not cur:
                raise EOFError(f'{cnt - len(data)} bytes missing')
            data += cur
        return data

    def _recv_bytes(self, cnt):
        data = self.recvall(cnt)
        return struct.unpack(f'!{cnt}B', data)

    @staticmethod
    def _len_and_data(data):
        return struct.pack('!B', len(data)) + data

    def _check_response_version(self, expected_version, got_version):
        if got_version != expected_version:
            self.close()
            raise InvalidVersionError(expected_version, got_version)

    def _resolve_address(self, destaddr, default, use_remote_dns):
        try:
            return socket.inet_aton(destaddr)
        except OSError:
            if use_remote_dns and self._proxy.remote_dns:
                return default
            else:
                return socket.inet_aton(socket.gethostbyname(destaddr))

    def _setup_socks4(self, address, is_4a=False):
        destaddr, port = address

        ipaddr = self._resolve_address(destaddr, SOCKS4_DEFAULT_DSTIP, use_remote_dns=is_4a)

        packet = struct.pack('!BBH', SOCKS4_VERSION, Socks4Command.CMD_CONNECT, port) + ipaddr

        username = (self._proxy.username or '').encode()
        packet += username + b'\x00'

        if is_4a and self._proxy.remote_dns:
            packet += destaddr.encode() + b'\x00'

        self.sendall(packet)

        version, resp_code, dstport, dsthost = struct.unpack('!BBHI', self.recvall(8))

        self._check_response_version(SOCKS4_REPLY_VERSION, version)

        if resp_code != Socks4Error.ERR_SUCCESS:
            self.close()
            raise Socks4Error(resp_code)

        return (dsthost, dstport)

    def _setup_socks4a(self, address):
        self._setup_socks4(address, is_4a=True)

    def _socks5_auth(self):
        packet = struct.pack('!B', SOCKS5_VERSION)

        auth_methods = [Socks5Auth.AUTH_NONE]
        if self._proxy.username and self._proxy.password:
            auth_methods.append(Socks5Auth.AUTH_USER_PASS)

        packet += struct.pack('!B', len(auth_methods))
        packet += struct.pack(f'!{len(auth_methods)}B', *auth_methods)

        self.sendall(packet)

        version, method = self._recv_bytes(2)

        self._check_response_version(SOCKS5_VERSION, version)

        if method == Socks5Auth.AUTH_NO_ACCEPTABLE or (
                method == Socks5Auth.AUTH_USER_PASS and (not self._proxy.username or not self._proxy.password)):
            self.close()
            raise Socks5Error(Socks5Auth.AUTH_NO_ACCEPTABLE)

        if method == Socks5Auth.AUTH_USER_PASS:
            username = self._proxy.username.encode()
            password = self._proxy.password.encode()
            packet = struct.pack('!B', SOCKS5_USER_AUTH_VERSION)
            packet += self._len_and_data(username) + self._len_and_data(password)
            self.sendall(packet)

            version, status = self._recv_bytes(2)

            self._check_response_version(SOCKS5_USER_AUTH_VERSION, version)

            if status != SOCKS5_USER_AUTH_SUCCESS:
                self.close()
                raise Socks5Error(Socks5Error.ERR_GENERAL_FAILURE)

    def _setup_socks5(self, address):
        destaddr, port = address

        ipaddr = self._resolve_address(destaddr, None, use_remote_dns=True)

        self._socks5_auth()

        reserved = 0
        packet = struct.pack('!BBB', SOCKS5_VERSION, Socks5Command.CMD_CONNECT, reserved)
        if ipaddr is None:
            destaddr = destaddr.encode()
            packet += struct.pack('!B', Socks5AddressType.ATYP_DOMAINNAME)
            packet += self._len_and_data(destaddr)
        else:
            packet += struct.pack('!B', Socks5AddressType.ATYP_IPV4) + ipaddr
        packet += struct.pack('!H', port)

        self.sendall(packet)

        version, status, reserved, atype = self._recv_bytes(4)

        self._check_response_version(SOCKS5_VERSION, version)

        if status != Socks5Error.ERR_SUCCESS:
            self.close()
            raise Socks5Error(status)

        if atype == Socks5AddressType.ATYP_IPV4:
            destaddr = self.recvall(4)
        elif atype == Socks5AddressType.ATYP_DOMAINNAME:
            alen = compat_ord(self.recv(1))
            destaddr = self.recvall(alen)
        elif atype == Socks5AddressType.ATYP_IPV6:
            destaddr = self.recvall(16)
        destport = struct.unpack('!H', self.recvall(2))[0]

        return (destaddr, destport)

    def _make_proxy(self, connect_func, address):
        if not self._proxy:
            return connect_func(self, address)

        result = connect_func(self, (self._proxy.host, self._proxy.port))
        if result != 0 and result is not None:
            return result
        setup_funcs = {
            ProxyType.SOCKS4: self._setup_socks4,
            ProxyType.SOCKS4A: self._setup_socks4a,
            ProxyType.SOCKS5: self._setup_socks5,
        }
        setup_funcs[self._proxy.type](address)
        return result

    def connect(self, address):
        self._make_proxy(socket.socket.connect, address)

    def connect_ex(self, address):
        return self._make_proxy(socket.socket.connect_ex, address)


code:
import atexit
import contextlib
import hashlib
import json
import os
import platform
import re
import subprocess
import sys
from zipimport import zipimporter

from .compat import functools  # isort: split
from .compat import compat_realpath, compat_shlex_quote
from .utils import (
    Popen,
    cached_method,
    deprecation_warning,
    shell_quote,
    traverse_obj,
    version_tuple,
)
from .version import UPDATE_HINT, VARIANT, __version__
try:
    from .build_config import variant
except ImportError:
    variant = 'red'

try:
    from .build_config import is_brew
except ImportError:
    is_brew = False

REPOSITORY = 'ytdl-patched/ytdl-patched'
API_URL = f'https://api.github.com/repos/{REPOSITORY}/releases'

@functools.cache
def _get_variant_and_executable_path():
    """@returns (variant, executable_path)"""
    if hasattr(sys, 'frozen'):
        path = sys.executable
        if not hasattr(sys, '_MEIPASS'):
            return 'py2exe', path
        if sys._MEIPASS == os.path.dirname(path):
            return f'{sys.platform}_dir', path
        if sys.platform == 'darwin' and version_tuple(platform.mac_ver()[0]) < (10, 15):
            return 'darwin_legacy_exe', path
        return f'exe_{variant}', path

    path = os.path.dirname(__file__)
    if isinstance(__loader__, zipimporter):
        return 'zip', os.path.join(path, '..')
    elif (os.path.basename(sys.argv[0]) in ('__main__.py', '-m')
          and os.path.exists(os.path.join(path, '../.git/HEAD'))):
        return 'source', path
    elif is_brew:
        return 'homebrew', path
    return 'unknown', path

def detect_variant():
    return VARIANT or _get_variant_and_executable_path()[0]

@functools.cache
def current_git_head():
    if detect_variant() != 'source':
        return
    with contextlib.suppress(Exception):
        stdout, _, _ = Popen.run(
            ['git', 'rev-parse', '--short', 'HEAD'],
            text=True, cwd=os.path.dirname(os.path.abspath(__file__)),
            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        if re.fullmatch('[0-9a-f]+', stdout.strip()):
            return stdout.strip()

_FILE_SUFFIXES = {
    'zip': '',
    'homebrew': '',
    'py2exe': '_min.exe',
    'win32_exe': '.exe',
    'darwin_exe': '_macos',
    'darwin_legacy_exe': '_macos_legacy',
    'linux_exe': '_linux',
    'exe_red': '-red.exe',
    'exe_white': '-white.exe',
}

_NON_UPDATEABLE_REASONS = {
    **{variant: None for variant in _FILE_SUFFIXES},  # Updatable
    **{variant: f'Auto-update is not supported for unpackaged {name} executable; Re-download the latest release'
       for variant, name in {'win32_dir': 'Windows', 'darwin_dir': 'MacOS', 'linux_dir': 'Linux'}.items()},
    'source': 'You cannot update when running from source code; Use git to pull the latest changes',
    'unknown': 'You installed yt-dlp with a package manager or setup.py; Use that to update',
    'other': 'You are using an unofficial build of yt-dlp; Build the executable again',
}

def is_non_updateable():
    if UPDATE_HINT:
        return UPDATE_HINT
    return _NON_UPDATEABLE_REASONS.get(
        detect_variant(), _NON_UPDATEABLE_REASONS['unknown' if VARIANT else 'other'])

def _sha256_file(path):
    h = hashlib.sha256()
    mv = memoryview(bytearray(128 * 1024))
    with open(os.path.realpath(path), 'rb', buffering=0) as f:
        for n in iter(lambda: f.readinto(mv), 0):
            h.update(mv[:n])
    return h.hexdigest()

class Updater:
    def __init__(self, ydl):
        self.ydl = ydl

    @functools.cached_property
    def _tag(self):
        # there's no _update_spec in ytdl-patched
        return 'latest'

    @cached_method
    def _get_version_info(self, tag):
        self.ydl.write_debug(f'Fetching release info: {API_URL}/{tag}')
        return json.loads(self.ydl.urlopen(f'{API_URL}/{tag}').read().decode())

    @property
    def current_version(self):
        """Current version"""
        return __version__

    @property
    def new_version(self):
        """Version of the latest release we can update to"""
        if self._tag.startswith('tags/'):
            return self._tag[5:]
        return self._get_version_info(self._tag)['tag_name']

    @property
    def latest_version(self):
        """Version of the latest release"""
        return self._get_version_info('latest')['tag_name']

    @property
    def has_update(self):
        """Whether there is an update available"""
        return version_tuple(__version__) < version_tuple(self.new_version)

    @functools.cached_property
    def filename(self):
        """Filename of the executable"""
        return compat_realpath(_get_variant_and_executable_path()[1])

    def _download(self, name, tag):
        url = traverse_obj(self._get_version_info(tag), (
            'assets', lambda _, v: v['name'] == name, 'browser_download_url'), get_all=False)
        if not url:
            raise Exception('Unable to find download URL')
        self.ydl.write_debug(f'Downloading {name} from {url}')
        return self.ydl.urlopen(url).read()

    @functools.cached_property
    def release_name(self):
        """The release filename"""
        label = _FILE_SUFFIXES[detect_variant()]
        if label and platform.architecture()[0][:2] == '32':
            label = f'_x86{label}'
        return f'ytdl-patched{label}'

    @functools.cached_property
    def release_hash(self):
        """Hash of the latest release"""
        hash_data = dict(ln.split()[::-1] for ln in self._download('SHA2-256SUMS', self._tag).decode().splitlines())
        return hash_data[self.release_name]

    def _report_error(self, msg, expected=False):
        self.ydl.report_error(msg, tb=False if expected else None)

    def _report_permission_error(self, file):
        self._report_error(f'Unable to write to {file}; Try running as administrator', True)

    def _report_network_error(self, action, delim=';'):
        self._report_error(f'Unable to {action}{delim} Visit  https://github.com/{REPOSITORY}/releases/latest', True)

    def check_update(self):
        """Report whether there is an update available"""
        try:
            self.ydl.to_screen(
                f'Latest version: {self.latest_version}, Current version: {self.current_version}')
            if not self.has_update:
                if self._tag == 'latest':
                    return self.ydl.to_screen(f'yt-dlp is up to date ({__version__})')
                return self.ydl.report_warning(
                    'yt-dlp cannot be updated any further since you are on an older Python version')
        except Exception:
            return self._report_network_error('obtain version info', delim='; Please try again later or')

        if not is_non_updateable():
            self.ydl.to_screen(f'Current Build Hash {_sha256_file(self.filename)}')
        return True

    def update(self):
        """Update yt-dlp executable to the latest version"""
        if not self.check_update():
            return
        err = is_non_updateable()
        if err:
            return self._report_error(err, True)
        self.ydl.to_screen(f'Updating to version {self.new_version} ...')

        variant = detect_variant()
        if variant == 'homebrew':
            stdout = next(filter(None, Popen(['brew', 'tap'], stdout=subprocess.PIPE, encoding='utf-8').communicate()), '')
            if 'nao20010128nao/my' in stdout:
                self.ydl.to_screen('Fixing taps to point to new one')
                ret = Popen(['brew', 'untap', '-f', 'nao20010128nao/my']).wait()
                if ret != 0:
                    return self._report_error('Unable to untap old tap')
                ret = Popen(['brew', 'tap', 'lesmiscore/my']).wait()
                if ret != 0:
                    return self._report_error('Unable to tap the new tap')
            os.execvp('brew', ['brew', 'upgrade', 'lesmiscore/my/ytdl-patched'])

        directory = os.path.dirname(self.filename)
        if not os.access(self.filename, os.W_OK):
            return self._report_permission_error(self.filename)
        elif not os.access(directory, os.W_OK):
            return self._report_permission_error(directory)

        new_filename, old_filename = f'{self.filename}.new', f'{self.filename}.old'
        if variant == 'zip':  # Can be replaced in-place
            new_filename, old_filename = self.filename, None

        try:
            if os.path.exists(old_filename or ''):
                os.remove(old_filename)
        except OSError:
            return self._report_error('Unable to remove the old version')

        try:
            newcontent = self._download(self.release_name, self._tag)
        except OSError:
            return self._report_network_error('download latest version')
        except Exception:
            return self._report_network_error('fetch updates')

        try:
            expected_hash = self.release_hash
        except Exception:
            self.ydl.report_warning('no hash information found for the release')
        else:
            if hashlib.sha256(newcontent).hexdigest() != expected_hash:
                return self._report_network_error('verify the new executable')

        try:
            with open(new_filename, 'wb') as outf:
                outf.write(newcontent)
        except OSError:
            return self._report_permission_error(new_filename)

        if old_filename:
            try:
                os.rename(self.filename, old_filename)
            except OSError:
                return self._report_error('Unable to move current version')

            try:
                os.rename(new_filename, self.filename)
            except OSError:
                self._report_error('Unable to overwrite current version')
                return os.rename(old_filename, self.filename)

        if detect_variant() in ('win32_exe', 'py2exe', 'exe_red', 'exe_white'):
            atexit.register(Popen, f'ping 127.0.0.1 -n 5 -w 1000 & del /F "{old_filename}"',
                            shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        elif old_filename:
            try:
                os.remove(old_filename)
            except OSError:
                self._report_error('Unable to remove the old version')

            try:
                # 0o555 = r-xr-xr-x
                os.chmod(self.filename, (os.stat(self.filename).st_mode | 0o555) & 0o777)
            except OSError:
                return self._report_error(
                    f'Unable to set permissions. Run: sudo chmod a+rx {compat_shlex_quote(self.filename)}')

        self.ydl.to_screen(f'Updated yt-dlp to version {self.new_version}')
        return True

    @functools.cached_property
    def cmd(self):
        """The command-line to run the executable, if known"""
        # There is no sys.orig_argv in py < 3.10. Also, it can be [] when frozen
        if getattr(sys, 'orig_argv', None):
            return sys.orig_argv
        elif hasattr(sys, 'frozen'):
            return sys.argv

    def restart(self):
        """Restart the executable"""
        assert self.cmd, 'Must be frozen or Py >= 3.10'
        self.ydl.write_debug(f'Restarting: {shell_quote(self.cmd)}')
        _, _, returncode = Popen.run(self.cmd)
        return returncode

def run_update(ydl):
    """Update the program file with the latest version from the repository
    @returns    Whether there was a successful update (No update = False)
    """
    return Updater(ydl).update()

# Deprecated
def update_self(to_screen, verbose, opener):
    import traceback

    deprecation_warning(f'"{__name__}.update_self" is deprecated and may be removed '
                        f'in a future version. Use "{__name__}.run_update(ydl)" instead')

    printfn = to_screen

    class FakeYDL():
        to_screen = printfn

        def report_warning(self, msg, *args, **kwargs):
            return printfn(f'WARNING: {msg}', *args, **kwargs)

        def report_error(self, msg, tb=None):
            printfn(f'ERROR: {msg}')
            if not verbose:
                return
            if tb is None:
                # Copied from YoutubeDL.trouble
                if sys.exc_info()[0]:
                    tb = ''
                    if hasattr(sys.exc_info()[1], 'exc_info') and sys.exc_info()[1].exc_info[0]:
                        tb += ''.join(traceback.format_exception(*sys.exc_info()[1].exc_info))
                    tb += traceback.format_exc()
                else:
                    tb_data = traceback.format_list(traceback.extract_stack())
                    tb = ''.join(tb_data)
            if tb:
                printfn(tb)

        def write_debug(self, msg, *args, **kwargs):
            printfn(f'[debug] {msg}', *args, **kwargs)

        def urlopen(self, url):
            return opener.open(url)

    return run_update(FakeYDL())