code:
import base64
import contextlib
import http.cookiejar
import http.cookies
import json
import os
import re
import shutil
import struct
import subprocess
import sys
import tempfile
import time
from datetime import datetime, timedelta, timezone
from enum import Enum, auto
from hashlib import pbkdf2_hmac

from .aes import (
    aes_cbc_decrypt_bytes,
    aes_gcm_decrypt_and_verify_bytes,
    unpad_pkcs7,
)
from .dependencies import (
    _SECRETSTORAGE_UNAVAILABLE_REASON,
    secretstorage,
    sqlite3,
)
from .minicurses import MultilinePrinter, QuietMultilinePrinter
from .utils import (
    Popen,
    YoutubeDLCookieJar,
    error_to_str,
    expand_path,
    is_path_like,
    try_call,
)

CHROMIUM_BASED_BROWSERS = {'brave', 'chrome', 'chromium', 'edge', 'opera', 'vivaldi'}
SUPPORTED_BROWSERS = CHROMIUM_BASED_BROWSERS | {'firefox', 'safari'}

class YDLLogger:
    def __init__(self, ydl=None):
        self._ydl = ydl

    def debug(self, message):
        if self._ydl:
            self._ydl.write_debug(message)

    def info(self, message):
        if self._ydl:
            self._ydl.to_screen(f'[Cookies] {message}')

    def warning(self, message, only_once=False):
        if self._ydl:
            self._ydl.report_warning(message, only_once)

    def error(self, message):
        if self._ydl:
            self._ydl.report_error(message)

    class ProgressBar(MultilinePrinter):
        _DELAY, _timer = 0.1, 0

        def print(self, message):
            if time.time() - self._timer > self._DELAY:
                self.print_at_line(f'[Cookies] {message}', 0)
                self._timer = time.time()

    def progress_bar(self):
        """Return a context manager with a print method. (Optional)"""
        # Do not print to files/pipes, loggers, or when --no-progress is used
        if not self._ydl or self._ydl.params.get('noprogress') or self._ydl.params.get('logger'):
            return
        file = self._ydl._out_files.error
        try:
            if not file.isatty():
                return
        except BaseException:
            return
        return self.ProgressBar(file, preserve_output=False)

def _create_progress_bar(logger):
    if hasattr(logger, 'progress_bar'):
        printer = logger.progress_bar()
        if printer:
            return printer
    printer = QuietMultilinePrinter()
    printer.print = lambda _: None
    return printer

def load_cookies(cookie_file, browser_specification, ydl):
    cookie_jars = []
    if browser_specification is not None:
        browser_name, profile, keyring, container = _parse_browser_specification(*browser_specification)
        cookie_jars.append(
            extract_cookies_from_browser(browser_name, profile, YDLLogger(ydl), keyring=keyring, container=container))

    if cookie_file is not None:
        is_filename = is_path_like(cookie_file)
        if is_filename:
            cookie_file = expand_path(cookie_file)

        jar = YoutubeDLCookieJar(cookie_file)
        if not is_filename or os.access(cookie_file, os.R_OK):
            jar.load(ignore_discard=True, ignore_expires=True)
        cookie_jars.append(jar)

    return _merge_cookie_jars(cookie_jars)

def extract_cookies_from_browser(browser_name, profile=None, logger=YDLLogger(), *, keyring=None, container=None):
    if browser_name == 'firefox':
        return _extract_firefox_cookies(profile, container, logger)
    elif browser_name == 'safari':
        return _extract_safari_cookies(profile, logger)
    elif browser_name in CHROMIUM_BASED_BROWSERS:
        return _extract_chrome_cookies(browser_name, profile, keyring, logger)
    else:
        raise ValueError(f'unknown browser: {browser_name}')

def _extract_firefox_cookies(profile, container, logger):
    logger.info('Extracting cookies from firefox')
    if not sqlite3:
        logger.warning('Cannot extract cookies from firefox without sqlite3 support. '
                       'Please use a python interpreter compiled with sqlite3 support')
        return YoutubeDLCookieJar()

    if profile is None:
        search_root = _firefox_browser_dir()
    elif _is_path(profile):
        search_root = profile
    else:
        search_root = os.path.join(_firefox_browser_dir(), profile)

    cookie_database_path = _find_most_recently_used_file(search_root, 'cookies.sqlite', logger)
    if cookie_database_path is None:
        raise FileNotFoundError(f'could not find firefox cookies database in {search_root}')
    logger.debug(f'Extracting cookies from: "{cookie_database_path}"')

    container_id = None
    if container not in (None, 'none'):
        containers_path = os.path.join(os.path.dirname(cookie_database_path), 'containers.json')
        if not os.path.isfile(containers_path) or not os.access(containers_path, os.R_OK):
            raise FileNotFoundError(f'could not read containers.json in {search_root}')
        with open(containers_path) as containers:
            identities = json.load(containers).get('identities', [])
        container_id = next((context.get('userContextId') for context in identities if container in (
            context.get('name'),
            try_call(lambda: re.fullmatch(r'userContext([^\.]+)\.label', context['l10nID']).group())
        )), None)
        if not isinstance(container_id, int):
            raise ValueError(f'could not find firefox container "{container}" in containers.json')

    with tempfile.TemporaryDirectory(prefix='yt_dlp') as tmpdir:
        cursor = None
        try:
            cursor = _open_database_copy(cookie_database_path, tmpdir)
            if isinstance(container_id, int):
                logger.debug(
                    f'Only loading cookies from firefox container "{container}", ID {container_id}')
                cursor.execute(
                    'SELECT host, name, value, path, expiry, isSecure FROM moz_cookies WHERE originAttributes LIKE ? OR originAttributes LIKE ?',
                    (f'%userContextId={container_id}', f'%userContextId={container_id}&%'))
            elif container == 'none':
                logger.debug('Only loading cookies not belonging to any container')
                cursor.execute(
                    'SELECT host, name, value, path, expiry, isSecure FROM moz_cookies WHERE NOT INSTR(originAttributes,"userContextId=")')
            else:
                cursor.execute('SELECT host, name, value, path, expiry, isSecure FROM moz_cookies')
            jar = YoutubeDLCookieJar()
            with _create_progress_bar(logger) as progress_bar:
                table = cursor.fetchall()
                total_cookie_count = len(table)
                for i, (host, name, value, path, expiry, is_secure) in enumerate(table):
                    progress_bar.print(f'Loading cookie {i: 6d}/{total_cookie_count: 6d}')
                    cookie = http.cookiejar.Cookie(
                        version=0, name=name, value=value, port=None, port_specified=False,
                        domain=host, domain_specified=bool(host), domain_initial_dot=host.startswith('.'),
                        path=path, path_specified=bool(path), secure=is_secure, expires=expiry, discard=False,
                        comment=None, comment_url=None, rest={})
                    jar.set_cookie(cookie)
            logger.info(f'Extracted {len(jar)} cookies from firefox')
            return jar
        finally:
            if cursor is not None:
                cursor.connection.close()

def _firefox_browser_dir():
    if sys.platform in ('cygwin', 'win32'):
        return os.path.expandvars(R'%APPDATA%\Mozilla\Firefox\Profiles')
    elif sys.platform == 'darwin':
        return os.path.expanduser('~/Library/Application Support/Firefox')
    return os.path.expanduser('~/.mozilla/firefox')

def _get_chromium_based_browser_settings(browser_name):
    # https://chromium.googlesource.com/chromium/src/+/HEAD/docs/user_data_dir.md
    if sys.platform in ('cygwin', 'win32'):
        appdata_local = os.path.expandvars('%LOCALAPPDATA%')
        appdata_roaming = os.path.expandvars('%APPDATA%')
        browser_dir = {
            'brave': os.path.join(appdata_local, R'BraveSoftware\Brave-Browser\User Data'),
            'chrome': os.path.join(appdata_local, R'Google\Chrome\User Data'),
            'chromium': os.path.join(appdata_local, R'Chromium\User Data'),
            'edge': os.path.join(appdata_local, R'Microsoft\Edge\User Data'),
            'opera': os.path.join(appdata_roaming, R'Opera Software\Opera Stable'),
            'vivaldi': os.path.join(appdata_local, R'Vivaldi\User Data'),
        }[browser_name]

    elif sys.platform == 'darwin':
        appdata = os.path.expanduser('~/Library/Application Support')
        browser_dir = {
            'brave': os.path.join(appdata, 'BraveSoftware/Brave-Browser'),
            'chrome': os.path.join(appdata, 'Google/Chrome'),
            'chromium': os.path.join(appdata, 'Chromium'),
            'edge': os.path.join(appdata, 'Microsoft Edge'),
            'opera': os.path.join(appdata, 'com.operasoftware.Opera'),
            'vivaldi': os.path.join(appdata, 'Vivaldi'),
        }[browser_name]

    else:
        config = _config_home()
        browser_dir = {
            'brave': os.path.join(config, 'BraveSoftware/Brave-Browser'),
            'chrome': os.path.join(config, 'google-chrome'),
            'chromium': os.path.join(config, 'chromium'),
            'edge': os.path.join(config, 'microsoft-edge'),
            'opera': os.path.join(config, 'opera'),
            'vivaldi': os.path.join(config, 'vivaldi'),
        }[browser_name]

    # Linux keyring names can be determined by snooping on dbus while opening the browser in KDE:
    # dbus-monitor "interface='org.kde.KWallet'" "type=method_return"
    keyring_name = {
        'brave': 'Brave',
        'chrome': 'Chrome',
        'chromium': 'Chromium',
        'edge': 'Microsoft Edge' if sys.platform == 'darwin' else 'Chromium',
        'opera': 'Opera' if sys.platform == 'darwin' else 'Chromium',
        'vivaldi': 'Vivaldi' if sys.platform == 'darwin' else 'Chrome',
    }[browser_name]

    browsers_without_profiles = {'opera'}

    return {
        'browser_dir': browser_dir,
        'keyring_name': keyring_name,
        'supports_profiles': browser_name not in browsers_without_profiles
    }

def _extract_chrome_cookies(browser_name, profile, keyring, logger):
    logger.info(f'Extracting cookies from {browser_name}')

    if not sqlite3:
        logger.warning(f'Cannot extract cookies from {browser_name} without sqlite3 support. '
                       'Please use a python interpreter compiled with sqlite3 support')
        return YoutubeDLCookieJar()

    config = _get_chromium_based_browser_settings(browser_name)

    if profile is None:
        search_root = config['browser_dir']
    elif _is_path(profile):
        search_root = profile
        config['browser_dir'] = os.path.dirname(profile) if config['supports_profiles'] else profile
    else:
        if config['supports_profiles']:
            search_root = os.path.join(config['browser_dir'], profile)
        else:
            logger.error(f'{browser_name} does not support profiles')
            search_root = config['browser_dir']

    cookie_database_path = _find_most_recently_used_file(search_root, 'Cookies', logger)
    if cookie_database_path is None:
        raise FileNotFoundError(f'could not find {browser_name} cookies database in "{search_root}"')
    logger.debug(f'Extracting cookies from: "{cookie_database_path}"')

    decryptor = get_cookie_decryptor(config['browser_dir'], config['keyring_name'], logger, keyring=keyring)

    with tempfile.TemporaryDirectory(prefix='yt_dlp') as tmpdir:
        cursor = None
        try:
            cursor = _open_database_copy(cookie_database_path, tmpdir)
            cursor.connection.text_factory = bytes
            column_names = _get_column_names(cursor, 'cookies')
            secure_column = 'is_secure' if 'is_secure' in column_names else 'secure'
            cursor.execute(f'SELECT host_key, name, value, encrypted_value, path, expires_utc, {secure_column} FROM cookies')
            jar = YoutubeDLCookieJar()
            failed_cookies = 0
            unencrypted_cookies = 0
            with _create_progress_bar(logger) as progress_bar:
                table = cursor.fetchall()
                total_cookie_count = len(table)
                for i, line in enumerate(table):
                    progress_bar.print(f'Loading cookie {i: 6d}/{total_cookie_count: 6d}')
                    is_encrypted, cookie = _process_chrome_cookie(decryptor, *line)
                    if not cookie:
                        failed_cookies += 1
                        continue
                    elif not is_encrypted:
                        unencrypted_cookies += 1
                    jar.set_cookie(cookie)
            if failed_cookies > 0:
                failed_message = f' ({failed_cookies} could not be decrypted)'
            else:
                failed_message = ''
            logger.info(f'Extracted {len(jar)} cookies from {browser_name}{failed_message}')
            counts = decryptor._cookie_counts.copy()
            counts['unencrypted'] = unencrypted_cookies
            logger.debug(f'cookie version breakdown: {counts}')
            return jar
        finally:
            if cursor is not None:
                cursor.connection.close()

def _process_chrome_cookie(decryptor, host_key, name, value, encrypted_value, path, expires_utc, is_secure):
    host_key = host_key.decode()
    name = name.decode()
    value = value.decode()
    path = path.decode()
    is_encrypted = not value and encrypted_value

    if is_encrypted:
        value = decryptor.decrypt(encrypted_value)
        if value is None:
            return is_encrypted, None

    return is_encrypted, http.cookiejar.Cookie(
        version=0, name=name, value=value, port=None, port_specified=False,
        domain=host_key, domain_specified=bool(host_key), domain_initial_dot=host_key.startswith('.'),
        path=path, path_specified=bool(path), secure=is_secure, expires=expires_utc, discard=False,
        comment=None, comment_url=None, rest={})

class ChromeCookieDecryptor:
    """
    Overview:

        Linux:
        - cookies are either v10 or v11
            - v10: AES-CBC encrypted with a fixed key
            - v11: AES-CBC encrypted with an OS protected key (keyring)
            - v11 keys can be stored in various places depending on the activate desktop environment [2]

        Mac:
        - cookies are either v10 or not v10
            - v10: AES-CBC encrypted with an OS protected key (keyring) and more key derivation iterations than linux
            - not v10: 'old data' stored as plaintext

        Windows:
        - cookies are either v10 or not v10
            - v10: AES-GCM encrypted with a key which is encrypted with DPAPI
            - not v10: encrypted with DPAPI

    Sources:
    - [1] https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/
    - [2] https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/key_storage_linux.cc
        - KeyStorageLinux::CreateService
    """

    _cookie_counts = {}

    def decrypt(self, encrypted_value):
        raise NotImplementedError('Must be implemented by sub classes')

def get_cookie_decryptor(browser_root, browser_keyring_name, logger, *, keyring=None):
    if sys.platform == 'darwin':
        return MacChromeCookieDecryptor(browser_keyring_name, logger)
    elif sys.platform in ('win32', 'cygwin'):
        return WindowsChromeCookieDecryptor(browser_root, logger)
    return LinuxChromeCookieDecryptor(browser_keyring_name, logger, keyring=keyring)

class LinuxChromeCookieDecryptor(ChromeCookieDecryptor):
    def __init__(self, browser_keyring_name, logger, *, keyring=None):
        self._logger = logger
        self._v10_key = self.derive_key(b'peanuts')
        password = _get_linux_keyring_password(browser_keyring_name, keyring, logger)
        self._v11_key = None if password is None else self.derive_key(password)
        self._cookie_counts = {'v10': 0, 'v11': 0, 'other': 0}

    @staticmethod
    def derive_key(password):
        # values from
        # https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/os_crypt_linux.cc
        return pbkdf2_sha1(password, salt=b'saltysalt', iterations=1, key_length=16)

    def decrypt(self, encrypted_value):
        version = encrypted_value[:3]
        ciphertext = encrypted_value[3:]

        if version == b'v10':
            self._cookie_counts['v10'] += 1
            return _decrypt_aes_cbc(ciphertext, self._v10_key, self._logger)

        elif version == b'v11':
            self._cookie_counts['v11'] += 1
            if self._v11_key is None:
                self._logger.warning('cannot decrypt v11 cookies: no key found', only_once=True)
                return None
            return _decrypt_aes_cbc(ciphertext, self._v11_key, self._logger)

        else:
            self._cookie_counts['other'] += 1
            return None

class MacChromeCookieDecryptor(ChromeCookieDecryptor):
    def __init__(self, browser_keyring_name, logger):
        self._logger = logger
        password = _get_mac_keyring_password(browser_keyring_name, logger)
        self._v10_key = None if password is None else self.derive_key(password)
        self._cookie_counts = {'v10': 0, 'other': 0}

    @staticmethod
    def derive_key(password):
        # values from
        # https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/os_crypt_mac.mm
        return pbkdf2_sha1(password, salt=b'saltysalt', iterations=1003, key_length=16)

    def decrypt(self, encrypted_value):
        version = encrypted_value[:3]
        ciphertext = encrypted_value[3:]

        if version == b'v10':
            self._cookie_counts['v10'] += 1
            if self._v10_key is None:
                self._logger.warning('cannot decrypt v10 cookies: no key found', only_once=True)
                return None

            return _decrypt_aes_cbc(ciphertext, self._v10_key, self._logger)

        else:
            self._cookie_counts['other'] += 1
            # other prefixes are considered 'old data' which were stored as plaintext
            # https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/os_crypt_mac.mm
            return encrypted_value

class WindowsChromeCookieDecryptor(ChromeCookieDecryptor):
    def __init__(self, browser_root, logger):
        self._logger = logger
        self._v10_key = _get_windows_v10_key(browser_root, logger)
        self._cookie_counts = {'v10': 0, 'other': 0}

    def decrypt(self, encrypted_value):
        version = encrypted_value[:3]
        ciphertext = encrypted_value[3:]

        if version == b'v10':
            self._cookie_counts['v10'] += 1
            if self._v10_key is None:
                self._logger.warning('cannot decrypt v10 cookies: no key found', only_once=True)
                return None

            # https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/os_crypt_win.cc
            #   kNonceLength
            nonce_length = 96 // 8
            # boringssl
            #   EVP_AEAD_AES_GCM_TAG_LEN
            authentication_tag_length = 16

            raw_ciphertext = ciphertext
            nonce = raw_ciphertext[:nonce_length]
            ciphertext = raw_ciphertext[nonce_length:-authentication_tag_length]
            authentication_tag = raw_ciphertext[-authentication_tag_length:]

            return _decrypt_aes_gcm(ciphertext, self._v10_key, nonce, authentication_tag, self._logger)

        else:
            self._cookie_counts['other'] += 1
            # any other prefix means the data is DPAPI encrypted
            # https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/os_crypt_win.cc
            return _decrypt_windows_dpapi(encrypted_value, self._logger).decode()

def _extract_safari_cookies(profile, logger):
    if profile is not None:
        logger.error('safari does not support profiles')
    if sys.platform != 'darwin':
        raise ValueError(f'unsupported platform: {sys.platform}')

    cookies_path = os.path.expanduser('~/Library/Cookies/Cookies.binarycookies')

    if not os.path.isfile(cookies_path):
        logger.debug('Trying secondary cookie location')
        cookies_path = os.path.expanduser('~/Library/Containers/com.apple.Safari/Data/Library/Cookies/Cookies.binarycookies')
        if not os.path.isfile(cookies_path):
            raise FileNotFoundError('could not find safari cookies database')

    with open(cookies_path, 'rb') as f:
        cookies_data = f.read()

    jar = parse_safari_cookies(cookies_data, logger=logger)
    logger.info(f'Extracted {len(jar)} cookies from safari')
    return jar

class ParserError(Exception):
    pass

class DataParser:
    def __init__(self, data, logger):
        self._data = data
        self.cursor = 0
        self._logger = logger

    def read_bytes(self, num_bytes):
        if num_bytes < 0:
            raise ParserError(f'invalid read of {num_bytes} bytes')
        end = self.cursor + num_bytes
        if end > len(self._data):
            raise ParserError('reached end of input')
        data = self._data[self.cursor:end]
        self.cursor = end
        return data

    def expect_bytes(self, expected_value, message):
        value = self.read_bytes(len(expected_value))
        if value != expected_value:
            raise ParserError(f'unexpected value: {value} != {expected_value} ({message})')

    def read_uint(self, big_endian=False):
        data_format = '>I' if big_endian else '<I'
        return struct.unpack(data_format, self.read_bytes(4))[0]

    def read_double(self, big_endian=False):
        data_format = '>d' if big_endian else '<d'
        return struct.unpack(data_format, self.read_bytes(8))[0]

    def read_cstring(self):
        buffer = []
        while True:
            c = self.read_bytes(1)
            if c == b'\x00':
                return b''.join(buffer).decode()
            else:
                buffer.append(c)

    def skip(self, num_bytes, description='unknown'):
        if num_bytes > 0:
            self._logger.debug(f'skipping {num_bytes} bytes ({description}): {self.read_bytes(num_bytes)!r}')
        elif num_bytes < 0:
            raise ParserError(f'invalid skip of {num_bytes} bytes')

    def skip_to(self, offset, description='unknown'):
        self.skip(offset - self.cursor, description)

    def skip_to_end(self, description='unknown'):
        self.skip_to(len(self._data), description)

def _mac_absolute_time_to_posix(timestamp):
    return int((datetime(2001, 1, 1, 0, 0, tzinfo=timezone.utc) + timedelta(seconds=timestamp)).timestamp())

def _parse_safari_cookies_header(data, logger):
    p = DataParser(data, logger)
    p.expect_bytes(b'cook', 'database signature')
    number_of_pages = p.read_uint(big_endian=True)
    page_sizes = [p.read_uint(big_endian=True) for _ in range(number_of_pages)]
    return page_sizes, p.cursor

def _parse_safari_cookies_page(data, jar, logger):
    p = DataParser(data, logger)
    p.expect_bytes(b'\x00\x00\x01\x00', 'page signature')
    number_of_cookies = p.read_uint()
    record_offsets = [p.read_uint() for _ in range(number_of_cookies)]
    if number_of_cookies == 0:
        logger.debug(f'a cookies page of size {len(data)} has no cookies')
        return

    p.skip_to(record_offsets[0], 'unknown page header field')

    with _create_progress_bar(logger) as progress_bar:
        for i, record_offset in enumerate(record_offsets):
            progress_bar.print(f'Loading cookie {i: 6d}/{number_of_cookies: 6d}')
            p.skip_to(record_offset, 'space between records')
            record_length = _parse_safari_cookies_record(data[record_offset:], jar, logger)
            p.read_bytes(record_length)
    p.skip_to_end('space in between pages')

def _parse_safari_cookies_record(data, jar, logger):
    p = DataParser(data, logger)
    record_size = p.read_uint()
    p.skip(4, 'unknown record field 1')
    flags = p.read_uint()
    is_secure = bool(flags & 0x0001)
    p.skip(4, 'unknown record field 2')
    domain_offset = p.read_uint()
    name_offset = p.read_uint()
    path_offset = p.read_uint()
    value_offset = p.read_uint()
    p.skip(8, 'unknown record field 3')
    expiration_date = _mac_absolute_time_to_posix(p.read_double())
    _creation_date = _mac_absolute_time_to_posix(p.read_double())  # noqa: F841

    try:
        p.skip_to(domain_offset)
        domain = p.read_cstring()

        p.skip_to(name_offset)
        name = p.read_cstring()

        p.skip_to(path_offset)
        path = p.read_cstring()

        p.skip_to(value_offset)
        value = p.read_cstring()
    except UnicodeDecodeError:
        logger.warning('failed to parse Safari cookie because UTF-8 decoding failed', only_once=True)
        return record_size

    p.skip_to(record_size, 'space at the end of the record')

    cookie = http.cookiejar.Cookie(
        version=0, name=name, value=value, port=None, port_specified=False,
        domain=domain, domain_specified=bool(domain), domain_initial_dot=domain.startswith('.'),
        path=path, path_specified=bool(path), secure=is_secure, expires=expiration_date, discard=False,
        comment=None, comment_url=None, rest={})
    jar.set_cookie(cookie)
    return record_size

def parse_safari_cookies(data, jar=None, logger=YDLLogger()):
    """
    References:
        - https://github.com/libyal/dtformats/blob/main/documentation/Safari%20Cookies.asciidoc
            - this data appears to be out of date but the important parts of the database structure is the same
            - there are a few bytes here and there which are skipped during parsing
    """
    if jar is None:
        jar = YoutubeDLCookieJar()
    page_sizes, body_start = _parse_safari_cookies_header(data, logger)
    p = DataParser(data[body_start:], logger)
    for page_size in page_sizes:
        _parse_safari_cookies_page(p.read_bytes(page_size), jar, logger)
    p.skip_to_end('footer')
    return jar

class _LinuxDesktopEnvironment(Enum):
    """
    https://chromium.googlesource.com/chromium/src/+/refs/heads/main/base/nix/xdg_util.h
    DesktopEnvironment
    """
    OTHER = auto()
    CINNAMON = auto()
    GNOME = auto()
    KDE = auto()
    PANTHEON = auto()
    UNITY = auto()
    XFCE = auto()

class _LinuxKeyring(Enum):
    """
    https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/key_storage_util_linux.h
    SelectedLinuxBackend
    """
    KWALLET = auto()
    GNOMEKEYRING = auto()
    BASICTEXT = auto()

SUPPORTED_KEYRINGS = _LinuxKeyring.__members__.keys()

def _get_linux_desktop_environment(env):
    """
    https://chromium.googlesource.com/chromium/src/+/refs/heads/main/base/nix/xdg_util.cc
    GetDesktopEnvironment
    """
    xdg_current_desktop = env.get('XDG_CURRENT_DESKTOP', None)
    desktop_session = env.get('DESKTOP_SESSION', None)
    if xdg_current_desktop is not None:
        xdg_current_desktop = xdg_current_desktop.split(':')[0].strip()

        if xdg_current_desktop == 'Unity':
            if desktop_session is not None and 'gnome-fallback' in desktop_session:
                return _LinuxDesktopEnvironment.GNOME
            else:
                return _LinuxDesktopEnvironment.UNITY
        elif xdg_current_desktop == 'GNOME':
            return _LinuxDesktopEnvironment.GNOME
        elif xdg_current_desktop == 'X-Cinnamon':
            return _LinuxDesktopEnvironment.CINNAMON
        elif xdg_current_desktop == 'KDE':
            return _LinuxDesktopEnvironment.KDE
        elif xdg_current_desktop == 'Pantheon':
            return _LinuxDesktopEnvironment.PANTHEON
        elif xdg_current_desktop == 'XFCE':
            return _LinuxDesktopEnvironment.XFCE
    elif desktop_session is not None:
        if desktop_session in ('mate', 'gnome'):
            return _LinuxDesktopEnvironment.GNOME
        elif 'kde' in desktop_session:
            return _LinuxDesktopEnvironment.KDE
        elif 'xfce' in desktop_session:
            return _LinuxDesktopEnvironment.XFCE
    else:
        if 'GNOME_DESKTOP_SESSION_ID' in env:
            return _LinuxDesktopEnvironment.GNOME
        elif 'KDE_FULL_SESSION' in env:
            return _LinuxDesktopEnvironment.KDE
    return _LinuxDesktopEnvironment.OTHER

def _choose_linux_keyring(logger):
    """
    https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/key_storage_util_linux.cc
    SelectBackend
    """
    desktop_environment = _get_linux_desktop_environment(os.environ)
    logger.debug(f'detected desktop environment: {desktop_environment.name}')
    if desktop_environment == _LinuxDesktopEnvironment.KDE:
        linux_keyring = _LinuxKeyring.KWALLET
    elif desktop_environment == _LinuxDesktopEnvironment.OTHER:
        linux_keyring = _LinuxKeyring.BASICTEXT
    else:
        linux_keyring = _LinuxKeyring.GNOMEKEYRING
    return linux_keyring

def _get_kwallet_network_wallet(logger):
    """ The name of the wallet used to store network passwords.

    https://chromium.googlesource.com/chromium/src/+/refs/heads/main/components/os_crypt/kwallet_dbus.cc
    KWalletDBus::NetworkWallet
    which does a dbus call to the following function:
    https://api.kde.org/frameworks/kwallet/html/classKWallet_1_1Wallet.html
    Wallet::NetworkWallet
    """
    default_wallet = 'kdewallet'
    try:
        stdout, _, returncode = Popen.run([
            'dbus-send', '--session', '--print-reply=literal',
            '--dest=org.kde.kwalletd5',
            '/modules/kwalletd5',
            'org.kde.KWallet.networkWallet'
        ], text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)

        if returncode:
            logger.warning('failed to read NetworkWallet')
            return default_wallet
        else:
            logger.debug(f'NetworkWallet = "{stdout.strip()}"')
            return stdout.strip()
    except Exception as e:
        logger.warning(f'exception while obtaining NetworkWallet: {e}')
        return default_wallet

def _get_kwallet_password(browser_keyring_name, logger):
    logger.debug('using kwallet-query to obtain password from kwallet')

    if shutil.which('kwallet-query') is None:
        logger.error('kwallet-query command not found. KWallet and kwallet-query '
                     'must be installed to read from KWallet. kwallet-query should be'
                     'included in the kwallet package for your distribution')
        return b''

    network_wallet = _get_kwallet_network_wallet(logger)

    try:
        stdout, _, returncode = Popen.run([
            'kwallet-query',
            '--read-password', f'{browser_keyring_name} Safe Storage',
            '--folder', f'{browser_keyring_name} Keys',
            network_wallet
        ], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)

        if returncode:
            logger.error(f'kwallet-query failed with return code {returncode}. '
                         'Please consult the kwallet-query man page for details')
            return b''
        else:
            if stdout.lower().startswith(b'failed to read'):
                logger.debug('failed to read password from kwallet. Using empty string instead')
                # this sometimes occurs in KDE because chrome does not check hasEntry and instead
                # just tries to read the value (which kwallet returns "") whereas kwallet-query
                # checks hasEntry. To verify this:
                # dbus-monitor "interface='org.kde.KWallet'" "type=method_return"
                # while starting chrome.
                # this may be a bug as the intended behaviour is to generate a random password and store
                # it, but that doesn't matter here.
                return b''
            else:
                logger.debug('password found')
                return stdout.rstrip(b'\n')
    except Exception as e:
        logger.warning(f'exception running kwallet-query: {error_to_str(e)}')
        return b''

def _get_gnome_keyring_password(browser_keyring_name, logger):
    if not secretstorage:
        logger.error(f'secretstorage not available {_SECRETSTORAGE_UNAVAILABLE_REASON}')
        return b''
    # the Gnome keyring does not seem to organise keys in the same way as KWallet,
    # using `dbus-monitor` during startup, it can be observed that chromium lists all keys
    # and presumably searches for its key in the list. It appears that we must do the same.
    # https://github.com/jaraco/keyring/issues/556
    with contextlib.closing(secretstorage.dbus_init()) as con:
        col = secretstorage.get_default_collection(con)
        for item in col.get_all_items():
            if item.get_label() == f'{browser_keyring_name} Safe Storage':
                return item.get_secret()
        else:
            logger.error('failed to read from keyring')
            return b''

def _get_linux_keyring_password(browser_keyring_name, keyring, logger):
    # note: chrome/chromium can be run with the following flags to determine which keyring backend
    # it has chosen to use
    # chromium --enable-logging=stderr --v=1 2>&1 | grep key_storage_
    # Chromium supports a flag: --password-store=<basic|gnome|kwallet> so the automatic detection
    # will not be sufficient in all cases.

    keyring = _LinuxKeyring[keyring] if keyring else _choose_linux_keyring(logger)
    logger.debug(f'Chosen keyring: {keyring.name}')

    if keyring == _LinuxKeyring.KWALLET:
        return _get_kwallet_password(browser_keyring_name, logger)
    elif keyring == _LinuxKeyring.GNOMEKEYRING:
        return _get_gnome_keyring_password(browser_keyring_name, logger)
    elif keyring == _LinuxKeyring.BASICTEXT:
        # when basic text is chosen, all cookies are stored as v10 (so no keyring password is required)
        return None
    assert False, f'Unknown keyring {keyring}'

def _get_mac_keyring_password(browser_keyring_name, logger):
    logger.debug('using find-generic-password to obtain password from OSX keychain')
    try:
        stdout, _, _ = Popen.run(
            ['security', 'find-generic-password',
             '-w',  # write password to stdout
             '-a', browser_keyring_name,  # match 'account'
             '-s', f'{browser_keyring_name} Safe Storage'],  # match 'service'
            stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
        return stdout.rstrip(b'\n')
    except Exception as e:
        logger.warning(f'exception running find-generic-password: {error_to_str(e)}')
        return None

def _get_windows_v10_key(browser_root, logger):
    path = _find_most_recently_used_file(browser_root, 'Local State', logger)
    if path is None:
        logger.error('could not find local state file')
        return None
    logger.debug(f'Found local state file at "{path}"')
    with open(path, encoding='utf8') as f:
        data = json.load(f)
    try:
        base64_key = data['os_crypt']['encrypted_key']
    except KeyError:
        logger.error('no encrypted key in Local State')
        return None
    encrypted_key = base64.b64decode(base64_key)
    prefix = b'DPAPI'
    if not encrypted_key.startswith(prefix):
        logger.error('invalid key')
        return None
    return _decrypt_windows_dpapi(encrypted_key[len(prefix):], logger)

def pbkdf2_sha1(password, salt, iterations, key_length):
    return pbkdf2_hmac('sha1', password, salt, iterations, key_length)

def _decrypt_aes_cbc(ciphertext, key, logger, initialization_vector=b' ' * 16):
    plaintext = unpad_pkcs7(aes_cbc_decrypt_bytes(ciphertext, key, initialization_vector))
    try:
        return plaintext.decode()
    except UnicodeDecodeError:
        logger.warning('failed to decrypt cookie (AES-CBC) because UTF-8 decoding failed. Possibly the key is wrong?', only_once=True)
        return None

def _decrypt_aes_gcm(ciphertext, key, nonce, authentication_tag, logger):
    try:
        plaintext = aes_gcm_decrypt_and_verify_bytes(ciphertext, key, authentication_tag, nonce)
    except ValueError:
        logger.warning('failed to decrypt cookie (AES-GCM) because the MAC check failed. Possibly the key is wrong?', only_once=True)
        return None

    try:
        return plaintext.decode()
    except UnicodeDecodeError:
        logger.warning('failed to decrypt cookie (AES-GCM) because UTF-8 decoding failed. Possibly the key is wrong?', only_once=True)
        return None

def _decrypt_windows_dpapi(ciphertext, logger):
    """
    References:
        - https://docs.microsoft.com/en-us/windows/win32/api/dpapi/nf-dpapi-cryptunprotectdata
    """

    import ctypes
    import ctypes.wintypes

    class DATA_BLOB(ctypes.Structure):
        _fields_ = [('cbData', ctypes.wintypes.DWORD),
                    ('pbData', ctypes.POINTER(ctypes.c_char))]

    buffer = ctypes.create_string_buffer(ciphertext)
    blob_in = DATA_BLOB(ctypes.sizeof(buffer), buffer)
    blob_out = DATA_BLOB()
    ret = ctypes.windll.crypt32.CryptUnprotectData(
        ctypes.byref(blob_in),  # pDataIn
        None,  # ppszDataDescr: human readable description of pDataIn
        None,  # pOptionalEntropy: salt?
        None,  # pvReserved: must be NULL
        None,  # pPromptStruct: information about prompts to display
        0,  # dwFlags
        ctypes.byref(blob_out)  # pDataOut
    )
    if not ret:
        logger.warning('failed to decrypt with DPAPI', only_once=True)
        return None

    result = ctypes.string_at(blob_out.pbData, blob_out.cbData)
    ctypes.windll.kernel32.LocalFree(blob_out.pbData)
    return result

def _config_home():
    return os.environ.get('XDG_CONFIG_HOME', os.path.expanduser('~/.config'))

def _open_database_copy(database_path, tmpdir):
    # cannot open sqlite databases if they are already in use (e.g. by the browser)
    database_copy_path = os.path.join(tmpdir, 'temporary.sqlite')
    shutil.copy(database_path, database_copy_path)
    conn = sqlite3.connect(database_copy_path)
    return conn.cursor()

def _get_column_names(cursor, table_name):
    table_info = cursor.execute(f'PRAGMA table_info({table_name})').fetchall()
    return [row[1].decode() for row in table_info]

def _find_most_recently_used_file(root, filename, logger):
    # if there are multiple browser profiles, take the most recently used one
    i, paths = 0, []
    with _create_progress_bar(logger) as progress_bar:
        for curr_root, dirs, files in os.walk(root):
            for file in files:
                i += 1
                progress_bar.print(f'Searching for "{filename}": {i: 6d} files searched')
                if file == filename:
                    paths.append(os.path.join(curr_root, file))
    return None if not paths else max(paths, key=lambda path: os.lstat(path).st_mtime)

def _merge_cookie_jars(jars):
    output_jar = YoutubeDLCookieJar()
    for jar in jars:
        for cookie in jar:
            output_jar.set_cookie(cookie)
        if jar.filename is not None:
            output_jar.filename = jar.filename
    return output_jar

def _is_path(value):
    return os.path.sep in value

def _parse_browser_specification(browser_name, profile=None, keyring=None, container=None):
    if browser_name not in SUPPORTED_BROWSERS:
        raise ValueError(f'unsupported browser: "{browser_name}"')
    if keyring not in (None, *SUPPORTED_KEYRINGS):
        raise ValueError(f'unsupported keyring: "{keyring}"')
    if profile is not None and _is_path(expand_path(profile)):
        profile = expand_path(profile)
    return browser_name, profile, keyring, container

class LenientSimpleCookie(http.cookies.SimpleCookie):
    """More lenient version of http.cookies.SimpleCookie"""
    # From https://github.com/python/cpython/blob/v3.10.7/Lib/http/cookies.py
    _LEGAL_KEY_CHARS = r"\w\d!#%&'~_`><@,:/\$\*\+\-\.\^\|\)\(\?\}\{\="
    _LEGAL_VALUE_CHARS = _LEGAL_KEY_CHARS + r"\[\]"

    _RESERVED = {
        "expires",
        "path",
        "comment",
        "domain",
        "max-age",
        "secure",
        "httponly",
        "version",
        "samesite",
    }

    _FLAGS = {"secure", "httponly"}

    # Added 'bad' group to catch the remaining value
    _COOKIE_PATTERN = re.compile(r"""
        \s*                            # Optional whitespace at start of cookie
        (?P<key>                       # Start of group 'key'
        [""" + _LEGAL_KEY_CHARS + r"""]+?# Any word of at least one letter
        )                              # End of group 'key'
        (                              # Optional group: there may not be a value.
        \s*=\s*                          # Equal Sign
        (                                # Start of potential value
        (?P<val>                           # Start of group 'val'
        "(?:[^\\"]|\\.)*"                    # Any doublequoted string
        |                                    # or
        \w{3},\s[\w\d\s-]{9,11}\s[\d:]{8}\sGMT # Special case for "expires" attr
        |                                    # or
        [""" + _LEGAL_VALUE_CHARS + r"""]*     # Any word or empty string
        )                                  # End of group 'val'
        |                                  # or
        (?P<bad>(?:\\;|[^;])*?)            # 'bad' group fallback for invalid values
        )                                # End of potential value
        )?                             # End of optional value group
        \s*                            # Any number of spaces.
        (\s+|;|$)                      # Ending either at space, semicolon, or EOS.
        """, re.ASCII | re.VERBOSE)

    def load(self, data):
        # Workaround for https://github.com/yt-dlp/yt-dlp/issues/4776
        if not isinstance(data, str):
            return super().load(data)

        morsel = None
        index = 0
        length = len(data)

        while 0 <= index < length:
            match = self._COOKIE_PATTERN.search(data, index)
            if not match:
                break

            index = match.end(0)
            if match.group("bad"):
                morsel = None
                continue

            key, value = match.group("key", "val")

            if key[0] == "$":
                if morsel is not None:
                    morsel[key[1:]] = True
                continue

            lower_key = key.lower()
            if lower_key in self._RESERVED:
                if morsel is None:
                    continue

                if value is None:
                    if lower_key not in self._FLAGS:
                        morsel = None
                        continue
                    value = True
                else:
                    value, _ = self.value_decode(value)

                morsel[key] = value

            elif value is not None:
                morsel = self.get(key, http.cookies.Morsel())
                real_value, coded_value = self.value_decode(value)
                morsel.set(key, real_value, coded_value)
                self[key] = morsel

            else:
                morsel = None


code:
# flake8: noqa: F401
"""Imports all optional dependencies for the project.
An attribute "_yt_dlp__identifier" may be inserted into the module if it uses an ambiguous namespace"""

try:
    import brotlicffi as brotli
except ImportError:
    try:
        import brotli
    except ImportError:
        brotli = None

try:
    import certifi
except ImportError:
    certifi = None
else:
    from os.path import exists as _path_exists

    # The certificate may not be bundled in executable
    if not _path_exists(certifi.where()):
        certifi = None

try:
    from Cryptodome.Cipher import AES as Cryptodome_AES
except ImportError:
    try:
        from Crypto.Cipher import AES as Cryptodome_AES
    except (ImportError, SyntaxError):  # Old Crypto gives SyntaxError in newer Python
        Cryptodome_AES = None
    else:
        try:
            # In pycrypto, mode defaults to ECB. See:
            # https://www.pycryptodome.org/en/latest/src/vs_pycrypto.html#:~:text=not%20have%20ECB%20as%20default%20mode
            Cryptodome_AES.new(b'abcdefghijklmnop')
        except TypeError:
            pass
        else:
            Cryptodome_AES._yt_dlp__identifier = 'pycrypto'

try:
    import mutagen
except ImportError:
    mutagen = None

secretstorage = None
try:
    import secretstorage
    _SECRETSTORAGE_UNAVAILABLE_REASON = None
except ImportError:
    _SECRETSTORAGE_UNAVAILABLE_REASON = (
        'as the `secretstorage` module is not installed. '
        'Please install by running `python3 -m pip install secretstorage`')
except Exception as _err:
    _SECRETSTORAGE_UNAVAILABLE_REASON = f'as the `secretstorage` module could not be initialized. {_err}'

try:
    import sqlite3
except ImportError:
    # although sqlite3 is part of the standard library, it is possible to compile python without
    # sqlite support. See: https://github.com/yt-dlp/yt-dlp/issues/544
    sqlite3 = None

try:
    import websockets
except (ImportError, SyntaxError):
    # websockets 3.10 on python 3.6 causes SyntaxError
    # See https://github.com/yt-dlp/yt-dlp/issues/2633
    websockets = None

from .websocket import WebSocket

try:
    import xattr  # xattr or pyxattr
except ImportError:
    xattr = None
else:
    if hasattr(xattr, 'set'):  # pyxattr
        xattr._yt_dlp__identifier = 'pyxattr'

all_dependencies = {k: v for k, v in globals().items() if not k.startswith('_')}

available_dependencies = {k: v for k, v in all_dependencies.items() if v}

__all__ = [
    'all_dependencies',
    'available_dependencies',
    *all_dependencies.keys(),
]

repo_name: Lesmiscore/ytdl


code:
from ..utils import NO_DEFAULT, determine_protocol

def get_suitable_downloader(info_dict, params={}, default=NO_DEFAULT, protocol=None, to_stdout=False):
    info_dict['protocol'] = determine_protocol(info_dict)
    info_copy = info_dict.copy()
    info_copy['to_stdout'] = to_stdout

    protocols = (protocol or info_copy['protocol']).split('+')
    downloaders = [_get_suitable_downloader(info_copy, proto, params, default) for proto in protocols]

    if set(downloaders) == {FFmpegFD} and FFmpegFD.can_merge_formats(info_copy, params):
        return FFmpegFD
    elif (set(downloaders) == {DashSegmentsFD}
          and not (to_stdout and len(protocols) > 1)
          and set(protocols) == {'http_dash_segments_generator'}):
        return DashSegmentsFD
    elif len(downloaders) == 1:
        return downloaders[0]
    return None

# Some of these require get_suitable_downloader
from .common import FileDownloader
from .dash import DashSegmentsFD
from .external import FFmpegFD, get_external_downloader
from .f4m import F4mFD
from .hls import HlsFD
from .http import HttpFD
from .ism import IsmFD
from .mhtml import MhtmlFD
from .niconico import NiconicoLiveFD
from .rtmp import RtmpFD
from .rtsp import RtspFD
from .websocket import WebSocketFragmentFD
from .youtube_live_chat import YoutubeLiveChatFD
from .serial import SerialFD
from .images import ImageSeriesFD

PROTOCOL_MAP = {
    'rtmp': RtmpFD,
    'rtmpe': RtmpFD,
    'rtmp_ffmpeg': FFmpegFD,
    'ffmpeg': FFmpegFD,  # for backward compatibility with old code
    'live_ffmpeg': FFmpegFD,
    'm3u8_native': HlsFD,
    'm3u8': FFmpegFD,
    'mms': RtspFD,
    'rtsp': RtspFD,
    'f4m': F4mFD,
    'http_dash_segments': DashSegmentsFD,
    'http_dash_segments_generator': DashSegmentsFD,
    'ism': IsmFD,
    'mhtml': MhtmlFD,
    'niconico_live': NiconicoLiveFD,
    'websocket_frag': WebSocketFragmentFD,
    'serial': SerialFD,
    'image_series': ImageSeriesFD,
    'youtube_live_chat': YoutubeLiveChatFD,
    'youtube_live_chat_replay': YoutubeLiveChatFD,
}

# exceptions for --live-download-mkv
# adding here will bypass protocol change
LDM_EXCEPTIONS = (
    'http_dash_segments', 'serial',
    'niconico_live', 'fc2_live',
    'youtube_live_chat', 'youtube_live_chat_replay',
    'websocket_frag', 'image_series',
)

def shorten_protocol_name(proto, simplify=False):
    short_protocol_names = {
        'm3u8_native': 'm3u8',
        'm3u8': 'm3u8F',
        'rtmp_ffmpeg': 'rtmpF',
        'http_dash_segments': 'dash',
        'http_dash_segments_generator': 'dashG',
        'niconico_dmc': 'dmc',
        'websocket_frag': 'WSfrag',
    }
    if simplify:
        short_protocol_names.update({
            'https': 'http',
            'ftps': 'ftp',
            'm3u8': 'm3u8',  # Reverse above m3u8 mapping
            'm3u8_native': 'm3u8',
            'http_dash_segments_generator': 'dash',
            'rtmp_ffmpeg': 'rtmp',
            'm3u8_frag_urls': 'm3u8',
            'dash_frag_urls': 'dash',
        })
    return short_protocol_names.get(proto, proto)

def _get_suitable_downloader(info_dict, protocol, params, default):
    """Get the downloader class that can handle the info dict."""
    if default is NO_DEFAULT:
        default = HttpFD

    if (info_dict.get('section_start') or info_dict.get('section_end')) and FFmpegFD.can_download(info_dict):
        return FFmpegFD

    info_dict['protocol'] = protocol
    downloaders = params.get('external_downloader')
    external_downloader = (
        downloaders if isinstance(downloaders, str) or downloaders is None
        else downloaders.get(shorten_protocol_name(protocol, True), downloaders.get('default')))

    if external_downloader is None:
        if info_dict['to_stdout'] and FFmpegFD.can_merge_formats(info_dict, params):
            return FFmpegFD
    elif external_downloader.lower() != 'native':
        ed = get_external_downloader(external_downloader)
        if ed.can_download(info_dict, external_downloader):
            return ed

    if protocol == 'http_dash_segments':
        if info_dict.get('is_live') and (external_downloader or '').lower() != 'native':
            return FFmpegFD

    if protocol in ('m3u8', 'm3u8_native'):
        if info_dict.get('is_live'):
            return FFmpegFD
        elif (external_downloader or '').lower() == 'native':
            return HlsFD
        elif protocol == 'm3u8_native' and get_suitable_downloader(
                info_dict, params, None, protocol='m3u8_frag_urls', to_stdout=info_dict['to_stdout']):
            return HlsFD
        elif params.get('hls_prefer_native') is True:
            return HlsFD
        elif params.get('hls_prefer_native') is False:
            return FFmpegFD

    return PROTOCOL_MAP.get(protocol, default)

__all__ = [
    'FileDownloader',
    'get_suitable_downloader',
    'shorten_protocol_name',
]


code:
from __future__ import division, unicode_literals

import functools
import re
import threading
import typing
import http.server

if typing.TYPE_CHECKING:
    from .common import FileDownloader

from ..postprocessor.metadataparser import MetadataParserPP
from ..utils import (
    sanitized_Request,
)

class Augment():
    _AUGMENT_KEY = None

    def __init__(self, dl: 'FileDownloader', info_dict, params: dict) -> None:
        self.dl = dl
        self.ydl = dl.ydl
        self.to_screen = self.dl.to_screen

        if 'init_callback' in params:
            info_dict, params = params['init_callback'](info_dict, params)

        self.params = params
        self.info_dict = info_dict

        # children classes may:
        # - implement some more initialization tasks
        # - modify info_dict directly to make things pass through Augment
        # at their __init__

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.end()

    def start(self):
        """
            Starts augmented service.
            Calling start() 2 or more times without end()ing is not permitted.
        """
        raise Exception('Implement in inheriting class')

    def end(self):
        """ Stops augmented service, as well as cleanups """
        raise Exception('Implement in inheriting class')

class HeartbeatAugment(Augment):
    """
    Augment for heartbeating.

    Keys:

    interval:  Interval to wait, in seconds.
    callback:  Callable to run periodically. Arguments are: (HeartbeatAugment)
               "url" and "data" are ignored once this key is used.
    url:       (easy mode) URL to reqeust to. Cannot be used with "callback" key
    data:      (optional) POST payload to pass. Use if needed.
    before_dl: Callable to run before download starts. Arguments are: (HeartbeatAugment)
               Can be used even if any of "callback", "url" and "data" are used.
    after_dl: Callable to run after download ends. Arguments are: (HeartbeatAugment)
    """
    _AUGMENT_KEY = 'heartbeat'

    def __init__(self, dl: 'FileDownloader', info_dict, params: dict) -> None:
        super().__init__(dl, info_dict, params)
        params, info_dict = self.params, self.info_dict

        self.interval = params.get('interval', 30)
        self.lock = threading.Lock()
        self.timer = [None]

        if 'callback' in params:
            self.callback = params['callback']
        elif 'url' in params:
            heartbeat_url = params['url']
            heartbeat_data = params.get('data')
            if isinstance(heartbeat_data, str):
                heartbeat_data = heartbeat_data.encode()
            request = sanitized_Request(heartbeat_url, heartbeat_data)

            def callback(a):
                self.ydl.urlopen(request).read()

            self.callback = callback
        else:
            raise Exception('Callback is not provided')

    def start(self):
        self.complete = False

        def heartbeat():
            try:
                self.callback(self)
            except Exception:
                self.to_screen('[download] Heartbeat failed')

            with self.lock:
                if self.complete:
                    self.timer[0] = None
                    self.complete = False
                else:
                    self.timer[0] = threading.Timer(self.interval, heartbeat)
                    self.timer[0]._daemonic = True
                    self.timer[0].start()

        if 'before_dl' in self.params:
            self.params['before_dl'](self)

        heartbeat()

    def end(self):
        with self.lock:
            self.timer[0].cancel()
            self.complete = True
        if 'after_dl' in self.params:
            self.params['after_dl'](self)

class HttpServerAugment(Augment):
    """
    Augment for intermediate HTTP server.

    Keys:

    before_dl: Callable to run before download starts. Arguments are: (HttpServerAugment)
               Can be used even if any of "callback", "url" and "data" are used.
    after_dl:  Callable to run after download ends. Arguments are: (HttpServerAugment)
    tag:       Key for retrieving port number in subsequent Augments.
               Assigned to "_httpserverport_<tag>" key. (optional)
    """
    _AUGMENT_KEY = 'http_server'

    def __init__(self, dl: 'FileDownloader', info_dict, params: dict) -> None:
        super().__init__(dl, info_dict, params)

    def start(self):
        if 'before_dl' in self.params:
            self.params['before_dl'](self)

        self.httpd = http.server.HTTPServer(
            ('127.0.0.1', 0), self.create_handler_class(self.params))
        self.port = self.info_dict[f'_httpserverport_{self.params["tag"]}' if self.params.get('tag') else '_httpserverport'] = \
            self.httpd.socket.getsockname()[1]
        self.server_thread = threading.Thread(
            target=self.httpd.serve_forever, daemon=True)
        self.server_thread.start()

    def end(self):
        if 'after_dl' in self.params:
            self.params['after_dl'](self)
        self.httpd.shutdown()
        self.httpd = None
        self.server_thread = None

    def create_handler_class(self, struct):
        # struct = {
        #     'routes': [{
        #         'method': 'GET',
        #         'route': '/hello/(?P<world>[^/]+)',
        #         'callback': lambda handler: True
        #     }, {
        #         'method': 'GET',
        #         'route': '/static/test',
        #         'data': b'hello world',
        #     }]
        # }

        def respond_constant(value, status_code, headers, handler):
            handler.send_response(status_code)
            for k, v in headers.items():
                handler.send_header(k, v)
            handler.end_headers()
            handler.wfile.write(value)
            return True

        def compile_route(route):
            if route.startswith('re:'):
                route = route[3:]
            else:
                route = re.escape(route)
            return re.compile(route)

        def process_route(regex, callback, method, handler):
            mobj = re.fullmatch(regex, handler.path)
            if not mobj:
                return False
            if method and method != handler.command:
                return False
            try:
                setattr(handler, 'route_params', mobj.groupdict())
                return callback(handler)
            finally:
                delattr(handler, 'route_params')

        def chain(callbacks, handler):
            for cb in callbacks:
                if cb(handler):
                    return True
            return False

        def wrapper(handler):
            assert struct(handler)

        if not callable(struct):
            route_callbacks = []
            for r in struct['routes']:
                if callable(r):
                    route_callbacks.append(r)
                    continue
                if 'data' in r:
                    dd = r['data']
                    if isinstance(dd, str):
                        dd = dd.encode()
                    r['callback'] = functools.partial(
                        respond_constant, dd, r.get('status_code', 200), r.get('headers') or {})

                route_callbacks.append(functools.partial(
                    process_route, compile_route(r['route']), r['callback'], r.get('method')))

            struct = functools.partial(chain, route_callbacks)

        return type('Lol this is allowed here', (ReqHandlerBase, ), {
            'do_GET': wrapper,
            'do_POST': wrapper,
        })

class ReqHandlerBase(http.server.BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        pass

class MetadataEditorAugment(Augment, MetadataParserPP):
    """
    Augment for temporarily rewriting info_dict.
    Values are reverted when end() is called.

    Keys:

    actions: Same as what you pass to MetadataParserPP.
    """
    _AUGMENT_KEY = 'metadata_editor'

    def __init__(self, dl: 'FileDownloader', info_dict, params: dict) -> None:
        super().__init__(dl, info_dict, params)
        MetadataParserPP.__init__(self, dl.ydl, [])
        self.backlog = None

    def start(self):
        # create backlog of modification to revert things back in end()
        self.backlog = [
            x for z in self.params['actions']
            for x in getattr(self, z[0].value)(*z[1:])(self.info_dict)]

    def end(self):
        infodict = self.info_dict
        # rollback
        for k, v in reversed(self.backlog):
            if v is MetadataParserPP.BACKLOG_UNSET:
                infodict.pop(k, None)
            else:
                infodict[k] = v
        # clear backlog
        self.backlog = None

    def to_screen(self, text, prefix=True, *args, **kwargs):
        if not self.get_param('verbose', False):
            return  # don't print anything without -v
        super().to_screen(text, prefix, *args, **kwargs)

    def report_warning(self, text, *args, **kwargs):
        if not self.get_param('verbose', False):
            return  # don't print anything without -v
        super().report_warning(text, *args, **kwargs)

AUGMENT_MAP = {v._AUGMENT_KEY: v for v in (HeartbeatAugment, HttpServerAugment, MetadataEditorAugment)}


code:
import contextlib
import errno
import functools
import random
import re
import time

from ..utils import (
    IDENTITY,
    NO_DEFAULT,
    LockingUnsupportedError,
    RetryManager,
    classproperty,
    decodeArgument,
    encodeFilename,
    get_argcount,
    match_filter_func,
    merge_dicts,
    shell_quote,
    timeconvert,
)

from typing import TYPE_CHECKING
if TYPE_CHECKING:
    from ..YoutubeDL import YoutubeDL
from ..postprocessor._attachments import ShowsProgress
from .augment import AUGMENT_MAP

class FileDownloader(ShowsProgress):
    """File Downloader class.

    File downloader objects are the ones responsible of downloading the
    actual video file and writing it to disk.

    File downloaders accept a lot of parameters. In order not to saturate
    the object constructor with arguments, it receives a dictionary of
    options instead.

    Available options:

    verbose:            Print additional info to stdout.
    quiet:              Do not print messages to stdout.
    ratelimit:          Download speed limit, in bytes/sec.
    continuedl:         Attempt to continue downloads if possible
    throttledratelimit: Assume the download is being throttled below this speed (bytes/sec)
    retries:            Number of times to retry for HTTP error 5xx
    file_access_retries:   Number of times to retry on file access error
    buffersize:         Size of download buffer in bytes.
    noresizebuffer:     Do not automatically resize the download buffer.
    continuedl:         Try to continue downloads if possible.
    noprogress:         Do not print the progress bar.
    nopart:             Do not use temporary .part files.
    updatetime:         Use the Last-modified header to set output file timestamps.
    test:               Download only first bytes to test the downloader.
    min_filesize:       Skip files smaller than this size
    max_filesize:       Skip files larger than this size
    xattr_set_filesize: Set ytdl.filesize user xattribute with expected size.
    external_downloader_args:  A dictionary of downloader keys (in lower case)
                        and a list of additional command-line arguments for the
                        executable. Use 'default' as the name for arguments to be
                        passed to all downloaders. For compatibility with youtube-dl,
                        a single list of args can also be used
    hls_use_mpegts:     Use the mpegts container for HLS videos.
    http_chunk_size:    Size of a chunk for chunk-based HTTP downloading. May be
                        useful for bypassing bandwidth throttling imposed by
                        a webserver (experimental)
    progress_template:  See YoutubeDL.py
    retry_sleep_functions: See YoutubeDL.py

    Subclasses of this one must re-define the real_download method.
    """

    _TEST_FILE_SIZE = 10241
    params = None

    def __init__(self, ydl, params):
        """Create a FileDownloader object with the given options."""
        ShowsProgress.__init__(self, ydl, params)
        self._set_ydl(ydl)
        self._progress_hooks = []
        self.params = params
        self._prepare_multiline_status()
        self.add_progress_hook(self.report_progress)

    def _set_ydl(self, ydl: 'YoutubeDL'):
        self.ydl: 'YoutubeDL' = ydl

        for func in (
            'deprecation_warning',
            'deprecated_feature',
            'report_error',
            'report_file_already_downloaded',
            'report_warning',
            'to_console_title',
            'to_stderr',
            'trouble',
            'write_debug',
        ):
            if not hasattr(self, func):
                setattr(self, func, getattr(ydl, func))

    def to_screen(self, *args, **kargs):
        self.ydl.to_screen(*args, quiet=self.params.get('quiet'), **kargs)

    __to_screen = to_screen

    @classproperty
    def FD_NAME(cls):
        return re.sub(r'(?<=[a-z])(?=[A-Z])', '_', cls.__name__[:-2]).lower()

    def slow_down(self, start_time, now, byte_counter):
        """Sleep if the download speed is over the rate limit."""
        rate_limit = self.params.get('ratelimit')
        if rate_limit is None or byte_counter == 0:
            return
        if now is None:
            now = time.time()
        elapsed = now - start_time
        if elapsed <= 0.0:
            return
        speed = float(byte_counter) / elapsed
        if speed > rate_limit:
            sleep_time = float(byte_counter) / rate_limit - elapsed
            if sleep_time > 0:
                time.sleep(sleep_time)

    def temp_name(self, filename):
        """Returns a temporary filename for the given filename."""
        if self.params.get('nopart', False) or filename == '-' or \
                (self.ydl.exists(encodeFilename(filename)) and not self.ydl.isfile(encodeFilename(filename))):
            return filename
        return filename + '.part'

    def undo_temp_name(self, filename):
        if filename.endswith('.part'):
            return filename[:-len('.part')]
        return filename

    def ytdl_filename(self, filename):
        return filename + '.ytdl'

    def wrap_file_access(action, *, fatal=False):
        def error_callback(err, count, retries, *, fd):
            return RetryManager.report_retry(
                err, count, retries, info=fd.__to_screen,
                warn=lambda e: (time.sleep(0.01), fd.to_screen(f'[download] Unable to {action} file: {e}')),
                error=None if fatal else lambda e: fd.report_error(f'Unable to {action} file: {e}'),
                sleep_func=fd.params.get('retry_sleep_functions', {}).get('file_access'))

        def wrapper(self, func, *args, **kwargs):
            for retry in RetryManager(self.params.get('file_access_retries'), error_callback, fd=self):
                try:
                    return func(self, *args, **kwargs)
                except OSError as err:
                    if err.errno in (errno.EACCES, errno.EINVAL):
                        retry.error = err
                        continue
                    retry.error_callback(err, 1, 0)

        return functools.partial(functools.partialmethod, wrapper)

    @wrap_file_access('open', fatal=True)
    def sanitize_open(self, filename, open_mode):
        f, filename = self.ydl.sanitize_open(filename, open_mode)
        if not getattr(f, 'locked', None):
            self.write_debug(f'{LockingUnsupportedError.msg}. Proceeding without locking', only_once=True)
        return f, filename

    @wrap_file_access('remove')
    def try_remove(self, filename):
        self.ydl.remove(filename)

    @wrap_file_access('rename')
    def try_rename(self, old_filename, new_filename):
        if old_filename == new_filename:
            return
        self.ydl.replace(old_filename, new_filename)

    def try_utime(self, filename, last_modified_hdr):
        """Try to set the last-modified time of the given file."""
        if last_modified_hdr is None:
            return
        if not self.ydl.isfile(encodeFilename(filename)):
            return
        timestr = last_modified_hdr
        if timestr is None:
            return
        filetime = timeconvert(timestr)
        if filetime is None:
            return filetime
        # Ignore obviously invalid dates
        if filetime == 0:
            return
        with contextlib.suppress(Exception):
            self.ydl.utime(filename, (time.time(), filetime))
        return filetime

    def report_destination(self, filename):
        """Report destination filename."""
        self.to_screen('[download] Destination: ' + filename)

    def report_resuming_byte(self, resume_len):
        """Report attempt to resume at given byte."""
        self.to_screen('[download] Resuming download at byte %s' % resume_len)

    def report_retry(self, err, count, retries, frag_index=NO_DEFAULT, fatal=True):
        """Report retry"""
        is_frag = False if frag_index is NO_DEFAULT else 'fragment'
        RetryManager.report_retry(
            err, count, retries, info=self.__to_screen,
            warn=lambda msg: self.__to_screen(f'[download] Got error: {msg}'),
            error=IDENTITY if not fatal else lambda e: self.report_error(f'\r[download] Got error: {e}'),
            sleep_func=self.params.get('retry_sleep_functions', {}).get(is_frag or 'http'),
            suffix=f'fragment{"s" if frag_index is None else f" {frag_index}"}' if is_frag else None)

    def report_unable_to_resume(self):
        """Report it was impossible to resume download."""
        self.to_screen('[download] Unable to resume')

    @staticmethod
    def supports_manifest(manifest):
        """ Whether the downloader can download the fragments from the manifest.
        Redefine in subclasses if needed. """
        pass

    def download(self, filename, info_dict, subtitle=False):
        """Download to a filename using the info from info_dict
        Return True on success and False otherwise
        """

        nooverwrites_and_exists = (
            not self.params.get('overwrites', True)
            and self.ydl.exists(encodeFilename(filename))
        )

        if not hasattr(filename, 'write'):
            continuedl_and_exists = (
                self.params.get('continuedl', True)
                and self.ydl.isfile(encodeFilename(filename))
                and not self.params.get('nopart', False)
            )

            # Check file already present
            if filename != '-' and (nooverwrites_and_exists or continuedl_and_exists):
                self.report_file_already_downloaded(filename)
                self._hook_progress({
                    'filename': filename,
                    'status': 'finished',
                    'total_bytes': self.ydl.getsize(encodeFilename(filename)),
                }, info_dict)
                self._finish_multiline_status()
                return True, False

        if subtitle:
            sleep_interval = self.params.get('sleep_interval_subtitles') or 0
        else:
            min_sleep_interval = self.params.get('sleep_interval') or 0
            sleep_interval = random.uniform(
                min_sleep_interval, self.params.get('max_sleep_interval') or min_sleep_interval)
        if sleep_interval > 0:
            self.to_screen(f'[download] Sleeping {sleep_interval:.2f} seconds ...')
            time.sleep(sleep_interval)

        with self._enter_augmented(info_dict):
            ret = self.real_download(filename, info_dict)
        self._finish_multiline_status()
        return ret, True

    def real_download(self, filename, info_dict):
        """Real download process. Redefine in subclasses."""
        raise NotImplementedError('This method must be implemented by subclasses')

    def _hook_progress(self, status, info_dict):
        # Ideally we want to make a copy of the dict, but that is too slow
        status['info_dict'] = info_dict
        # youtube-dl passes the same status object to all the hooks.
        # Some third party scripts seems to be relying on this.
        # So keep this behavior if possible
        for ph in self._progress_hooks:
            ph(status)

    def add_progress_hook(self, ph):
        # See YoutubeDl.py (search for progress_hooks) for a description of
        # this interface
        self._progress_hooks.append(ph)

    def _debug_cmd(self, args, exe=None):
        if not self.params.get('verbose', False):
            return

        str_args = [decodeArgument(a) for a in args]

        if exe is None:
            exe = self.ydl.basename(str_args[0])

        self.write_debug(f'{exe} command line: {shell_quote(str_args)}')

    def _enter_augmented(self, info_dict):
        augmentation = info_dict.get('augments') or []
        es = contextlib.ExitStack()
        if not augmentation:
            return es
        for a in augmentation:
            if 'condition' in a:
                cond = a['condition']
                if isinstance(cond, (list, str)):
                    # --match-filter compliant value
                    mfilter = match_filter_func(cond)
                    # give the string a way to get downloader key
                    # so that you can do "_downloader == HlsFD"
                    cond = lambda ifd, dl: mfilter(merge_dicts({'_downloader': type(dl).__name__}, ifd))
                # let's not hope some based guy passes a __call__able with (self,) signature
                argcount = get_argcount(cond) or 2
                if not cond(*[info_dict, self][:argcount]):
                    # do not enable Augment as predicate fails
                    continue
            es.enter_context(AUGMENT_MAP[a['key']](self, info_dict, a))
        return es


code:
import time

from . import get_suitable_downloader
from .fragment import FragmentFD
from ..utils import urljoin

class DashSegmentsFD(FragmentFD):
    """
    Download segments in a DASH manifest. External downloaders can take over
    the fragment downloads by supporting the 'dash_frag_urls' protocol
    """

    FD_NAME = 'dashsegments'

    def real_download(self, filename, info_dict):
        if info_dict.get('is_live') and set(info_dict['protocol'].split('+')) != {'http_dash_segments_generator'}:
            self.report_error('Live DASH videos are not supported')

        real_start = time.time()
        real_downloader = get_suitable_downloader(
            info_dict, self.params, None, protocol='dash_frag_urls', to_stdout=(filename == '-'))

        requested_formats = [{**info_dict, **fmt} for fmt in info_dict.get('requested_formats', [])]
        args = []
        for fmt in requested_formats or [info_dict]:
            try:
                fragment_count = 1 if self.params.get('test') else len(fmt['fragments'])
            except TypeError:
                fragment_count = None
            ctx = {
                'filename': fmt.get('filepath') or filename,
                'live': 'is_from_start' if fmt.get('is_from_start') else fmt.get('is_live'),
                'total_frags': fragment_count,
            }

            if real_downloader:
                self._prepare_external_frag_download(ctx)
            else:
                self._prepare_and_start_frag_download(ctx, fmt)
            ctx['start'] = real_start

            fragments_to_download = self._get_fragments(fmt, ctx)

            if real_downloader:
                self.to_screen(
                    f'[{self.FD_NAME}] Fragment downloads will be delegated to {real_downloader.get_basename()}')
                info_dict['fragments'] = list(fragments_to_download)
                fd = real_downloader(self.ydl, self.params)
                return fd.real_download(filename, info_dict)

            args.append([ctx, fragments_to_download, fmt])

        return self.download_and_append_fragments_multiple(*args)

    def _resolve_fragments(self, fragments, ctx):
        fragments = fragments(ctx) if callable(fragments) else fragments
        return [next(iter(fragments))] if self.params.get('test') else fragments

    def _get_fragments(self, fmt, ctx):
        fragment_base_url = fmt.get('fragment_base_url')
        fragments = self._resolve_fragments(fmt['fragments'], ctx)

        frag_index = 0
        for i, fragment in enumerate(fragments):
            frag_index += 1
            if frag_index <= ctx['fragment_index']:
                continue
            fragment_url = fragment.get('url')
            if not fragment_url:
                assert fragment_base_url
                fragment_url = urljoin(fragment_base_url, fragment['path'])

            yield {
                'frag_index': frag_index,
                'fragment_count': fragment.get('fragment_count'),
                'index': i,
                'url': fragment_url,
            }