code:
import http.client
import random
import socket
import ssl
import time
import urllib.error

from .common import FileDownloader
from ..utils import (
    ContentTooShortError,
    RetryManager,
    ThrottledDownload,
    UnrecoverableHttpError,
    XAttrMetadataError,
    XAttrUnavailableError,
    encodeFilename,
    int_or_none,
    parse_http_range,
    sanitized_Request,
    try_call,
    write_xattr,
)

RESPONSE_READ_EXCEPTIONS = (
    TimeoutError,
    socket.timeout,  # compat: py < 3.10
    ConnectionError,
    ssl.SSLError,
    http.client.HTTPException
)

class HttpFD(FileDownloader):
    def real_download(self, filename, info_dict):
        url = info_dict['url']
        request_data = info_dict.get('request_data', None)
        bad_status_code = info_dict.get('unrecoverable_http_error') or tuple()

        class DownloadContext(dict):
            __getattr__ = dict.get
            __setattr__ = dict.__setitem__
            __delattr__ = dict.__delitem__

        ctx = DownloadContext()
        ctx.filename = filename
        ctx.tmpfilename = self.temp_name(filename)
        ctx.stream = None

        # Do not include the Accept-Encoding header
        headers = {'Youtubedl-no-compression': 'True'}
        add_headers = info_dict.get('http_headers')
        if add_headers:
            headers.update(add_headers)

        is_test = self.params.get('test', False)
        chunk_size = self._TEST_FILE_SIZE if is_test else (
            self.params.get('http_chunk_size')
            or info_dict.get('downloader_options', {}).get('http_chunk_size')
            or 0)

        ctx.open_mode = 'wb'
        ctx.resume_len = 0
        ctx.block_size = self.params.get('buffersize', 1024)
        ctx.start_time = time.time()

        # parse given Range
        req_start, req_end, _ = parse_http_range(headers.get('Range'))

        if self.params.get('continuedl', True):
            # Establish possible resume length
            if self.ydl.isfile(encodeFilename(ctx.tmpfilename)):
                ctx.resume_len = self.ydl.getsize(
                    encodeFilename(ctx.tmpfilename))

        ctx.is_resume = ctx.resume_len > 0
        ctx.show_dest = True

        class SucceedDownload(Exception):
            pass

        class RetryDownload(Exception):
            def __init__(self, source_error):
                self.source_error = source_error

        class NextFragment(Exception):
            pass

        def establish_connection():
            ctx.chunk_size = (random.randint(int(chunk_size * 0.95), chunk_size)
                              if not is_test and chunk_size else chunk_size)
            if ctx.resume_len > 0:
                range_start = ctx.resume_len
                if req_start is not None:
                    # offset the beginning of Range to be within request
                    range_start += req_start
                if ctx.is_resume:
                    self.report_resuming_byte(ctx.resume_len)
                ctx.open_mode = 'ab'
            elif req_start is not None:
                range_start = req_start
            elif ctx.chunk_size > 0:
                range_start = 0
            else:
                range_start = None
            ctx.is_resume = False

            if ctx.chunk_size:
                chunk_aware_end = range_start + ctx.chunk_size - 1
                # we're not allowed to download outside Range
                range_end = chunk_aware_end if req_end is None else min(chunk_aware_end, req_end)
            elif req_end is not None:
                # there's no need for chunked downloads, so download until the end of Range
                range_end = req_end
            else:
                range_end = None

            if try_call(lambda: range_start > range_end):
                ctx.resume_len = 0
                ctx.open_mode = 'wb'
                raise RetryDownload(Exception(f'Conflicting range. (start={range_start} > end={range_end})'))

            if try_call(lambda: range_end >= ctx.content_len):
                range_end = ctx.content_len - 1

            request = sanitized_Request(url, request_data, headers)
            has_range = range_start is not None
            if has_range:
                request.add_header('Range', f'bytes={int(range_start)}-{int_or_none(range_end) or ""}')
            # Establish connection
            try:
                ctx.data = self.ydl.urlopen(request)
                # When trying to resume, Content-Range HTTP header of response has to be checked
                # to match the value of requested Range HTTP header. This is due to a webservers
                # that don't support resuming and serve a whole file with no Content-Range
                # set in response despite of requested Range (see
                # https://github.com/ytdl-org/youtube-dl/issues/6057#issuecomment-126129799)
                if has_range:
                    content_range = ctx.data.headers.get('Content-Range')
                    content_range_start, content_range_end, content_len = parse_http_range(content_range)
                    # Content-Range is present and matches requested Range, resume is possible
                    if range_start == content_range_start and (
                            # Non-chunked download
                            not ctx.chunk_size
                            # Chunked download and requested piece or
                            # its part is promised to be served
                            or content_range_end == range_end
                            or content_len < range_end):
                        ctx.content_len = content_len
                        if content_len or req_end:
                            ctx.data_len = min(content_len or req_end, req_end or content_len) - (req_start or 0)
                        return
                    # Content-Range is either not present or invalid. Assuming remote webserver is
                    # trying to send the whole file, resume is not possible, so wiping the local file
                    # and performing entire redownload
                    self.report_unable_to_resume()
                    ctx.resume_len = 0
                    ctx.open_mode = 'wb'
                ctx.data_len = ctx.content_len = int_or_none(ctx.data.info().get('Content-length', None))
            except urllib.error.HTTPError as err:
                if err.code in bad_status_code:
                    raise UnrecoverableHttpError()
                if err.code == 416:
                    # Unable to resume (requested range not satisfiable)
                    try:
                        # Open the connection again without the range header
                        ctx.data = self.ydl.urlopen(
                            sanitized_Request(url, request_data, headers))
                        content_length = ctx.data.info()['Content-Length']
                    except urllib.error.HTTPError as err:
                        if err.code < 500 or err.code >= 600:
                            raise
                    else:
                        # Examine the reported length
                        if (content_length is not None
                                and (ctx.resume_len - 100 < int(content_length) < ctx.resume_len + 100)):
                            # The file had already been fully downloaded.
                            # Explanation to the above condition: in issue #175 it was revealed that
                            # YouTube sometimes adds or removes a few bytes from the end of the file,
                            # changing the file size slightly and causing problems for some users. So
                            # I decided to implement a suggested change and consider the file
                            # completely downloaded if the file size differs less than 100 bytes from
                            # the one in the hard drive.
                            self.report_file_already_downloaded(ctx.filename)
                            self.try_rename(ctx.tmpfilename, ctx.filename)
                            self._hook_progress({
                                'filename': ctx.filename,
                                'status': 'finished',
                                'downloaded_bytes': ctx.resume_len,
                                'total_bytes': ctx.resume_len,
                            }, info_dict)
                            raise SucceedDownload()
                        else:
                            # The length does not match, we start the download over
                            self.report_unable_to_resume()
                            ctx.resume_len = 0
                            ctx.open_mode = 'wb'
                            return
                elif err.code < 500 or err.code >= 600:
                    # Unexpected HTTP error
                    raise
                raise RetryDownload(err)
            except urllib.error.URLError as err:
                if isinstance(err.reason, ssl.CertificateError):
                    raise
                raise RetryDownload(err)
            # In urllib.request.AbstractHTTPHandler, the response is partially read on request.
            # Any errors that occur during this will not be wrapped by URLError
            except RESPONSE_READ_EXCEPTIONS as err:
                raise RetryDownload(err)

        def close_stream():
            if ctx.stream is not None:
                if not ctx.tmpfilename == '-':
                    ctx.stream.close()
                ctx.stream = None

        def download():
            data_len = ctx.data.info().get('Content-length', None)

            # Range HTTP header may be ignored/unsupported by a webserver
            # (e.g. extractor/scivee.py, extractor/bambuser.py).
            # However, for a test we still would like to download just a piece of a file.
            # To achieve this we limit data_len to _TEST_FILE_SIZE and manually control
            # block size when downloading a file.
            if is_test and (data_len is None or int(data_len) > self._TEST_FILE_SIZE):
                data_len = self._TEST_FILE_SIZE

            if data_len is not None:
                data_len = int(data_len) + ctx.resume_len
                min_data_len = self.params.get('min_filesize')
                max_data_len = self.params.get('max_filesize')
                if min_data_len is not None and data_len < min_data_len:
                    self.to_screen(
                        f'\r[download] File is smaller than min-filesize ({data_len} bytes < {min_data_len} bytes). Aborting.')
                    return False
                if max_data_len is not None and data_len > max_data_len:
                    self.to_screen(
                        f'\r[download] File is larger than max-filesize ({data_len} bytes > {max_data_len} bytes). Aborting.')
                    return False

            byte_counter = 0 + ctx.resume_len
            block_size = ctx.block_size
            start = time.time()

            # measure time over whole while-loop, so slow_down() and best_block_size() work together properly
            now = None  # needed for slow_down() in the first loop run
            before = start  # start measuring

            def retry(e):
                close_stream()
                ctx.resume_len = (byte_counter if ctx.tmpfilename == '-'
                                  else self.ydl.getsize(encodeFilename(ctx.tmpfilename)))
                raise RetryDownload(e)

            while True:
                try:
                    # Download and write
                    data_block = ctx.data.read(block_size if not is_test else min(block_size, data_len - byte_counter))
                except RESPONSE_READ_EXCEPTIONS as err:
                    retry(err)

                byte_counter += len(data_block)

                # exit loop when download is finished
                if len(data_block) == 0:
                    break

                # Open destination file just in time
                if ctx.stream is None:
                    try:
                        ctx.stream, ctx.tmpfilename = self.sanitize_open(
                            ctx.tmpfilename, ctx.open_mode)
                        assert ctx.stream is not None
                        ctx.filename = self.undo_temp_name(ctx.tmpfilename)
                        if ctx.show_dest:
                            self.report_destination(ctx.filename)
                        ctx.show_dest = False
                    except OSError as err:
                        self.report_error('unable to open for writing: %s' % str(err))
                        return False

                    if self.params.get('xattr_set_filesize', False) and data_len is not None:
                        try:
                            write_xattr(ctx.tmpfilename, 'user.ytdl.filesize', str(data_len).encode())
                        except (XAttrUnavailableError, XAttrMetadataError) as err:
                            self.report_error('unable to set filesize xattr: %s' % str(err))

                try:
                    ctx.stream.write(data_block)
                except OSError as err:
                    self.to_stderr('\n')
                    self.report_error('unable to write data: %s' % str(err))
                    return False

                # Apply rate limit
                self.slow_down(start, now, byte_counter - ctx.resume_len)

                # end measuring of one loop run
                now = time.time()
                after = now

                # Adjust block size
                if not self.params.get('noresizebuffer', False):
                    block_size = self.best_block_size(after - before, len(data_block))

                before = after

                # Progress message
                speed = self.calc_speed(start, now, byte_counter - ctx.resume_len)
                if ctx.data_len is None:
                    eta = None
                else:
                    eta = self.calc_eta(start, time.time(), ctx.data_len - ctx.resume_len, byte_counter - ctx.resume_len)

                self._hook_progress({
                    'status': 'downloading',
                    'downloaded_bytes': byte_counter,
                    'total_bytes': ctx.data_len,
                    'tmpfilename': ctx.tmpfilename,
                    'filename': ctx.filename,
                    'eta': eta,
                    'speed': speed,
                    'elapsed': now - ctx.start_time,
                    'ctx_id': info_dict.get('ctx_id'),
                }, info_dict)

                if data_len is not None and byte_counter == data_len:
                    break

                if speed and speed < (self.params.get('throttledratelimit') or 0):
                    # The speed must stay below the limit for 3 seconds
                    # This prevents raising error when the speed temporarily goes down
                    if ctx.throttle_start is None:
                        ctx.throttle_start = now
                    elif now - ctx.throttle_start > 3:
                        if ctx.stream is not None and ctx.tmpfilename != '-':
                            ctx.stream.close()
                        raise ThrottledDownload()
                elif speed:
                    ctx.throttle_start = None

            if not is_test and ctx.chunk_size and ctx.content_len is not None and byte_counter < ctx.content_len:
                ctx.resume_len = byte_counter
                # ctx.block_size = block_size
                raise NextFragment()

            if ctx.stream is None:
                self.to_stderr('\n')
                self.report_error('Did not get any data blocks')
                return False
            if ctx.tmpfilename != '-':
                ctx.stream.close()

            if data_len is not None and byte_counter != data_len:
                err = ContentTooShortError(byte_counter, int(data_len))
                retry(err)

            self.try_rename(ctx.tmpfilename, ctx.filename)

            # Update file modification time
            if self.params.get('updatetime', True):
                info_dict['filetime'] = self.try_utime(ctx.filename, ctx.data.info().get('last-modified', None))

            self._hook_progress({
                'downloaded_bytes': byte_counter,
                'total_bytes': byte_counter,
                'filename': ctx.filename,
                'status': 'finished',
                'elapsed': time.time() - ctx.start_time,
                'ctx_id': info_dict.get('ctx_id'),
            }, info_dict)

            return True

        for retry in RetryManager(self.params.get('retries'), self.report_retry):
            try:
                establish_connection()
                return download()
            except RetryDownload as err:
                retry.error = err.source_error
                continue
            except NextFragment:
                retry.error = None
                retry.attempt -= 1
                continue
            except SucceedDownload:
                return True
            except:  # noqa: E722
                close_stream()
                raise
        return False


code:
from __future__ import division, unicode_literals

import os
import sys
from subprocess import Popen, PIPE

from .external import FFmpegFD
from .dash import DashSegmentsFD
from ..compat import compat_str
from ..longname import split_longname
from ..postprocessor.ffmpeg import EXT_TO_OUT_FORMATS, FFmpegPostProcessor
from ..utils import (
    encodeArgument,
    encodeFilename,
)

class ImageFragmentFD(DashSegmentsFD):
    """ Downloads series of images as a video """

    FD_NAME = 'image_fragment'

class ImageSeriesFD(FFmpegFD):
    """ Downloads series of images as a video """

    FD_NAME = 'image_series'

    def __init__(self, ydl, params):
        super().__init__(ydl, params)
        self._IFFD = ImageFragmentFD(ydl, {**params, 'skip_unavailable_fragments': True})

    def _call_downloader(self, tmpfilename, info_dict):
        filename = self.undo_temp_name(tmpfilename)
        images_filename = filename + '.images'
        segment_result, _ = self._IFFD.download(images_filename, info_dict)
        if not segment_result:
            return False

        # convert to video using ffmpeg
        ffpp = FFmpegPostProcessor(downloader=self)
        if not ffpp.available:
            self.report_error('ffmpeg could not be found. Please install')
            return False
        ffpp.check_version()

        if self.ydl.params.get('escape_long_names', False):
            tmpfilename = split_longname(tmpfilename)

        args = [ffpp.executable, '-y']

        for log_level in ('quiet', 'verbose'):
            if self.params.get(log_level, False):
                args += ['-loglevel', log_level]
                break
        if not self.params.get('verbose'):
            args += ['-hide_banner']

        args += info_dict.get('_ffmpeg_args', [])

        args += ['-f', 'image2pipe']

        if 'frame_count' in info_dict and 'duration' in info_dict:
            framerate_string = '%f/%f' % (info_dict['frame_count'], info_dict['duration'])
        else:
            framerate_string = compat_str(info_dict.get('fps'))
        args += ['-r', framerate_string]

        args += self._configuration_args(('_i1', '_i')) + [
            '-i', ffpp._ffmpeg_filename_argument(images_filename)]

        args += ['-c:v', 'libx265']

        if self.params.get('test', False):
            args += ['-fs', compat_str(self._TEST_FILE_SIZE)]

        ext = info_dict['ext']
        args += [
            '-f', EXT_TO_OUT_FORMATS.get(ext, ext),
            '-crf', framerate_string,
            '-pix_fmt', 'yuv420p']

        args += self._configuration_args(('_o1', '_o', ''))

        args = [encodeArgument(opt) for opt in args]
        args.append(encodeFilename(ffpp._ffmpeg_filename_argument(tmpfilename), True))
        self._debug_cmd(args)

        proc = Popen(args, stdin=PIPE)
        try:
            retval = -1
            retval = proc.wait()
        except BaseException as e:
            # subprocces.run would send the SIGKILL signal to ffmpeg and the
            # mp4 file couldn't be played, but if we ask ffmpeg to quit it
            # produces a file that is playable (this is mostly useful for live
            # streams). Note that Windows is not affected and produces playable
            # files (see https://github.com/ytdl-org/youtube-dl/issues/8300).
            if isinstance(e, KeyboardInterrupt) and sys.platform != 'win32':
                proc.communicate_or_kill(b'q')
            else:
                proc.kill()
                proc.wait()
                raise
        if retval == 0:
            os.remove(images_filename)
        return retval

repo_name: Lesmiscore/ytdl


code:
import binascii
import io
import struct
import time
import urllib.error

from .fragment import FragmentFD
from ..utils import RetryManager

u8 = struct.Struct('>B')
u88 = struct.Struct('>Bx')
u16 = struct.Struct('>H')
u1616 = struct.Struct('>Hxx')
u32 = struct.Struct('>I')
u64 = struct.Struct('>Q')

s88 = struct.Struct('>bx')
s16 = struct.Struct('>h')
s1616 = struct.Struct('>hxx')
s32 = struct.Struct('>i')

unity_matrix = (s32.pack(0x10000) + s32.pack(0) * 3) * 2 + s32.pack(0x40000000)

TRACK_ENABLED = 0x1
TRACK_IN_MOVIE = 0x2
TRACK_IN_PREVIEW = 0x4

SELF_CONTAINED = 0x1

def box(box_type, payload):
    return u32.pack(8 + len(payload)) + box_type + payload

def full_box(box_type, version, flags, payload):
    return box(box_type, u8.pack(version) + u32.pack(flags)[1:] + payload)

def write_piff_header(stream, params):
    track_id = params['track_id']
    fourcc = params['fourcc']
    duration = params['duration']
    timescale = params.get('timescale', 10000000)
    language = params.get('language', 'und')
    height = params.get('height', 0)
    width = params.get('width', 0)
    stream_type = params['stream_type']
    creation_time = modification_time = int(time.time())

    ftyp_payload = b'isml'  # major brand
    ftyp_payload += u32.pack(1)  # minor version
    ftyp_payload += b'piff' + b'iso2'  # compatible brands
    stream.write(box(b'ftyp', ftyp_payload))  # File Type Box

    mvhd_payload = u64.pack(creation_time)
    mvhd_payload += u64.pack(modification_time)
    mvhd_payload += u32.pack(timescale)
    mvhd_payload += u64.pack(duration)
    mvhd_payload += s1616.pack(1)  # rate
    mvhd_payload += s88.pack(1)  # volume
    mvhd_payload += u16.pack(0)  # reserved
    mvhd_payload += u32.pack(0) * 2  # reserved
    mvhd_payload += unity_matrix
    mvhd_payload += u32.pack(0) * 6  # pre defined
    mvhd_payload += u32.pack(0xffffffff)  # next track id
    moov_payload = full_box(b'mvhd', 1, 0, mvhd_payload)  # Movie Header Box

    tkhd_payload = u64.pack(creation_time)
    tkhd_payload += u64.pack(modification_time)
    tkhd_payload += u32.pack(track_id)  # track id
    tkhd_payload += u32.pack(0)  # reserved
    tkhd_payload += u64.pack(duration)
    tkhd_payload += u32.pack(0) * 2  # reserved
    tkhd_payload += s16.pack(0)  # layer
    tkhd_payload += s16.pack(0)  # alternate group
    tkhd_payload += s88.pack(1 if stream_type == 'audio' else 0)  # volume
    tkhd_payload += u16.pack(0)  # reserved
    tkhd_payload += unity_matrix
    tkhd_payload += u1616.pack(width)
    tkhd_payload += u1616.pack(height)
    trak_payload = full_box(b'tkhd', 1, TRACK_ENABLED | TRACK_IN_MOVIE | TRACK_IN_PREVIEW, tkhd_payload)  # Track Header Box

    mdhd_payload = u64.pack(creation_time)
    mdhd_payload += u64.pack(modification_time)
    mdhd_payload += u32.pack(timescale)
    mdhd_payload += u64.pack(duration)
    mdhd_payload += u16.pack(((ord(language[0]) - 0x60) << 10) | ((ord(language[1]) - 0x60) << 5) | (ord(language[2]) - 0x60))
    mdhd_payload += u16.pack(0)  # pre defined
    mdia_payload = full_box(b'mdhd', 1, 0, mdhd_payload)  # Media Header Box

    hdlr_payload = u32.pack(0)  # pre defined
    if stream_type == 'audio':  # handler type
        hdlr_payload += b'soun'
        hdlr_payload += u32.pack(0) * 3  # reserved
        hdlr_payload += b'SoundHandler\0'  # name
    elif stream_type == 'video':
        hdlr_payload += b'vide'
        hdlr_payload += u32.pack(0) * 3  # reserved
        hdlr_payload += b'VideoHandler\0'  # name
    elif stream_type == 'text':
        hdlr_payload += b'subt'
        hdlr_payload += u32.pack(0) * 3  # reserved
        hdlr_payload += b'SubtitleHandler\0'  # name
    else:
        assert False
    mdia_payload += full_box(b'hdlr', 0, 0, hdlr_payload)  # Handler Reference Box

    if stream_type == 'audio':
        smhd_payload = s88.pack(0)  # balance
        smhd_payload += u16.pack(0)  # reserved
        media_header_box = full_box(b'smhd', 0, 0, smhd_payload)  # Sound Media Header
    elif stream_type == 'video':
        vmhd_payload = u16.pack(0)  # graphics mode
        vmhd_payload += u16.pack(0) * 3  # opcolor
        media_header_box = full_box(b'vmhd', 0, 1, vmhd_payload)  # Video Media Header
    elif stream_type == 'text':
        media_header_box = full_box(b'sthd', 0, 0, b'')  # Subtitle Media Header
    else:
        assert False
    minf_payload = media_header_box

    dref_payload = u32.pack(1)  # entry count
    dref_payload += full_box(b'url ', 0, SELF_CONTAINED, b'')  # Data Entry URL Box
    dinf_payload = full_box(b'dref', 0, 0, dref_payload)  # Data Reference Box
    minf_payload += box(b'dinf', dinf_payload)  # Data Information Box

    stsd_payload = u32.pack(1)  # entry count

    sample_entry_payload = u8.pack(0) * 6  # reserved
    sample_entry_payload += u16.pack(1)  # data reference index
    if stream_type == 'audio':
        sample_entry_payload += u32.pack(0) * 2  # reserved
        sample_entry_payload += u16.pack(params.get('channels', 2))
        sample_entry_payload += u16.pack(params.get('bits_per_sample', 16))
        sample_entry_payload += u16.pack(0)  # pre defined
        sample_entry_payload += u16.pack(0)  # reserved
        sample_entry_payload += u1616.pack(params['sampling_rate'])

        if fourcc == 'AACL':
            sample_entry_box = box(b'mp4a', sample_entry_payload)
    elif stream_type == 'video':
        sample_entry_payload += u16.pack(0)  # pre defined
        sample_entry_payload += u16.pack(0)  # reserved
        sample_entry_payload += u32.pack(0) * 3  # pre defined
        sample_entry_payload += u16.pack(width)
        sample_entry_payload += u16.pack(height)
        sample_entry_payload += u1616.pack(0x48)  # horiz resolution 72 dpi
        sample_entry_payload += u1616.pack(0x48)  # vert resolution 72 dpi
        sample_entry_payload += u32.pack(0)  # reserved
        sample_entry_payload += u16.pack(1)  # frame count
        sample_entry_payload += u8.pack(0) * 32  # compressor name
        sample_entry_payload += u16.pack(0x18)  # depth
        sample_entry_payload += s16.pack(-1)  # pre defined

        codec_private_data = binascii.unhexlify(params['codec_private_data'].encode())
        if fourcc in ('H264', 'AVC1'):
            sps, pps = codec_private_data.split(u32.pack(1))[1:]
            avcc_payload = u8.pack(1)  # configuration version
            avcc_payload += sps[1:4]  # avc profile indication + profile compatibility + avc level indication
            avcc_payload += u8.pack(0xfc | (params.get('nal_unit_length_field', 4) - 1))  # complete representation (1) + reserved (11111) + length size minus one
            avcc_payload += u8.pack(1)  # reserved (0) + number of sps (0000001)
            avcc_payload += u16.pack(len(sps))
            avcc_payload += sps
            avcc_payload += u8.pack(1)  # number of pps
            avcc_payload += u16.pack(len(pps))
            avcc_payload += pps
            sample_entry_payload += box(b'avcC', avcc_payload)  # AVC Decoder Configuration Record
            sample_entry_box = box(b'avc1', sample_entry_payload)  # AVC Simple Entry
        else:
            assert False
    elif stream_type == 'text':
        if fourcc == 'TTML':
            sample_entry_payload += b'http://www.w3.org/ns/ttml\0'  # namespace
            sample_entry_payload += b'\0'  # schema location
            sample_entry_payload += b'\0'  # auxilary mime types(??)
            sample_entry_box = box(b'stpp', sample_entry_payload)
        else:
            assert False
    else:
        assert False
    stsd_payload += sample_entry_box

    stbl_payload = full_box(b'stsd', 0, 0, stsd_payload)  # Sample Description Box

    stts_payload = u32.pack(0)  # entry count
    stbl_payload += full_box(b'stts', 0, 0, stts_payload)  # Decoding Time to Sample Box

    stsc_payload = u32.pack(0)  # entry count
    stbl_payload += full_box(b'stsc', 0, 0, stsc_payload)  # Sample To Chunk Box

    stco_payload = u32.pack(0)  # entry count
    stbl_payload += full_box(b'stco', 0, 0, stco_payload)  # Chunk Offset Box

    minf_payload += box(b'stbl', stbl_payload)  # Sample Table Box

    mdia_payload += box(b'minf', minf_payload)  # Media Information Box

    trak_payload += box(b'mdia', mdia_payload)  # Media Box

    moov_payload += box(b'trak', trak_payload)  # Track Box

    mehd_payload = u64.pack(duration)
    mvex_payload = full_box(b'mehd', 1, 0, mehd_payload)  # Movie Extends Header Box

    trex_payload = u32.pack(track_id)  # track id
    trex_payload += u32.pack(1)  # default sample description index
    trex_payload += u32.pack(0)  # default sample duration
    trex_payload += u32.pack(0)  # default sample size
    trex_payload += u32.pack(0)  # default sample flags
    mvex_payload += full_box(b'trex', 0, 0, trex_payload)  # Track Extends Box

    moov_payload += box(b'mvex', mvex_payload)  # Movie Extends Box
    stream.write(box(b'moov', moov_payload))  # Movie Box

def extract_box_data(data, box_sequence):
    data_reader = io.BytesIO(data)
    while True:
        box_size = u32.unpack(data_reader.read(4))[0]
        box_type = data_reader.read(4)
        if box_type == box_sequence[0]:
            box_data = data_reader.read(box_size - 8)
            if len(box_sequence) == 1:
                return box_data
            return extract_box_data(box_data, box_sequence[1:])
        data_reader.seek(box_size - 8, 1)

class IsmFD(FragmentFD):
    """
    Download segments in a ISM manifest
    """

    def real_download(self, filename, info_dict):
        segments = info_dict['fragments'][:1] if self.params.get(
            'test', False) else info_dict['fragments']

        ctx = {
            'filename': filename,
            'total_frags': len(segments),
        }

        self._prepare_and_start_frag_download(ctx, info_dict)

        extra_state = ctx.setdefault('extra_state', {
            'ism_track_written': False,
        })

        skip_unavailable_fragments = self.params.get('skip_unavailable_fragments', True)

        frag_index = 0
        for i, segment in enumerate(segments):
            frag_index += 1
            if frag_index <= ctx['fragment_index']:
                continue

            retry_manager = RetryManager(self.params.get('fragment_retries'), self.report_retry,
                                         frag_index=frag_index, fatal=not skip_unavailable_fragments)
            for retry in retry_manager:
                try:
                    success = self._download_fragment(ctx, segment['url'], info_dict)
                    if not success:
                        return False
                    frag_content = self._read_fragment(ctx)

                    if not extra_state['ism_track_written']:
                        tfhd_data = extract_box_data(frag_content, [b'moof', b'traf', b'tfhd'])
                        info_dict['_download_params']['track_id'] = u32.unpack(tfhd_data[4:8])[0]
                        write_piff_header(ctx['dest_stream'], info_dict['_download_params'])
                        extra_state['ism_track_written'] = True
                    self._append_fragment(ctx, frag_content)
                except urllib.error.HTTPError as err:
                    retry.error = err
                    continue

            if retry_manager.error:
                if not skip_unavailable_fragments:
                    return False
                self.report_skip_fragment(frag_index)

        self._finish_frag_download(ctx, info_dict)
        return True


code:
import io
import quopri
import re
import uuid

from .fragment import FragmentFD
from ..compat import imghdr
from ..utils import escapeHTML, formatSeconds, srt_subtitles_timecode, urljoin
from ..version import __version__ as YT_DLP_VERSION

class MhtmlFD(FragmentFD):
    _STYLESHEET = """\
html, body {
    margin: 0;
    padding: 0;
    height: 100vh;
}

html {
    overflow-y: scroll;
    scroll-snap-type: y mandatory;
}

body {
    scroll-snap-type: y mandatory;
    display: flex;
    flex-flow: column;
}

body > figure {
    max-width: 100vw;
    max-height: 100vh;
    scroll-snap-align: center;
}

body > figure > figcaption {
    text-align: center;
    height: 2.5em;
}

body > figure > img {
    display: block;
    margin: auto;
    max-width: 100%;
    max-height: calc(100vh - 5em);
}
"""
    _STYLESHEET = re.sub(r'\s+', ' ', _STYLESHEET)
    _STYLESHEET = re.sub(r'\B \B|(?<=[\w\-]) (?=[^\w\-])|(?<=[^\w\-]) (?=[\w\-])', '', _STYLESHEET)

    @staticmethod
    def _escape_mime(s):
        return '=?utf-8?Q?' + (b''.join(
            bytes((b,)) if b >= 0x20 else b'=%02X' % b
            for b in quopri.encodestring(s.encode(), header=True)
        )).decode('us-ascii') + '?='

    def _gen_cid(self, i, fragment, frag_boundary):
        return '%u.%s@yt-dlp.github.io.invalid' % (i, frag_boundary)

    def _gen_stub(self, *, fragments, frag_boundary, title):
        output = io.StringIO()

        output.write((
            '<!DOCTYPE html>'
            ''
            ''
            ''  '<meta name="generator" content="yt-dlp {version}">'
            ''  '<title>{title}</title>'
            ''  ''
            ''
        ).format(
            version=escapeHTML(YT_DLP_VERSION),
            styles=self._STYLESHEET,
            title=escapeHTML(title)
        ))

        t0 = 0
        for i, frag in enumerate(fragments):
            output.write('
')
            try:
                t1 = t0 + frag['duration']
                output.write((
                    '
Slide #{num}: {t0} – {t1} (duration: {duration})
'
                ).format(
                    num=i + 1,
                    t0=srt_subtitles_timecode(t0),
                    t1=srt_subtitles_timecode(t1),
                    duration=formatSeconds(frag['duration'], msec=True)
                ))
            except (KeyError, ValueError, TypeError):
                t1 = None
                output.write((
                    '
Slide #{num}
'
                ).format(num=i + 1))
            output.write(''.format(
                cid=self._gen_cid(i, frag, frag_boundary)))
            output.write('
')
            t0 = t1

        return output.getvalue()

    def real_download(self, filename, info_dict):
        fragment_base_url = info_dict.get('fragment_base_url')
        fragments = info_dict['fragments'][:1] if self.params.get(
            'test', False) else info_dict['fragments']
        title = info_dict.get('title', info_dict['format_id'])
        origin = info_dict.get('webpage_url', info_dict['url'])

        ctx = {
            'filename': filename,
            'total_frags': len(fragments),
        }

        self._prepare_and_start_frag_download(ctx, info_dict)

        extra_state = ctx.setdefault('extra_state', {
            'header_written': False,
            'mime_boundary': str(uuid.uuid4()).replace('-', ''),
        })

        frag_boundary = extra_state['mime_boundary']

        if not extra_state['header_written']:
            stub = self._gen_stub(
                fragments=fragments,
                frag_boundary=frag_boundary,
                title=title
            )

            ctx['dest_stream'].write((
                'MIME-Version: 1.0\r\n'
                'From: <nowhere@yt-dlp.github.io.invalid>\r\n'
                'To: <nowhere@yt-dlp.github.io.invalid>\r\n'
                'Subject: {title}\r\n'
                'Content-type: multipart/related; '
                ''  'boundary="{boundary}"; '
                ''  'type="text/html"\r\n'
                'X.yt-dlp.Origin: {origin}\r\n'
                '\r\n'
                '--{boundary}\r\n'
                'Content-Type: text/html; charset=utf-8\r\n'
                'Content-Length: {length}\r\n'
                '\r\n'
                '{stub}\r\n'
            ).format(
                origin=origin,
                boundary=frag_boundary,
                length=len(stub),
                title=self._escape_mime(title),
                stub=stub
            ).encode())
            extra_state['header_written'] = True

        for i, fragment in enumerate(fragments):
            if (i + 1) <= 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'])

            success = self._download_fragment(ctx, fragment_url, info_dict)
            if not success:
                continue
            frag_content = self._read_fragment(ctx)

            frag_header = io.BytesIO()
            frag_header.write(
                b'--%b\r\n' % frag_boundary.encode('us-ascii'))
            frag_header.write(
                b'Content-ID: <%b>\r\n' % self._gen_cid(i, fragment, frag_boundary).encode('us-ascii'))
            frag_header.write(
                b'Content-type: %b\r\n' % f'image/{imghdr.what(h=frag_content) or "jpeg"}'.encode())
            frag_header.write(
                b'Content-length: %u\r\n' % len(frag_content))
            frag_header.write(
                b'Content-location: %b\r\n' % fragment_url.encode('us-ascii'))
            frag_header.write(
                b'X.yt-dlp.Duration: %f\r\n' % fragment['duration'])
            frag_header.write(b'\r\n')
            self._append_fragment(
                ctx, frag_header.getvalue() + frag_content + b'\r\n')

        ctx['dest_stream'].write(
            b'--%b--\r\n\r\n' % frag_boundary.encode('us-ascii'))
        self._finish_frag_download(ctx, info_dict)
        return True


code:
import json
import threading
import time

from .common import FileDownloader
from .external import FFmpegFD
from ..dependencies import WebSocket
from ..utils import (
    str_or_none,
    std_headers,
    DownloadError,
    try_get,
)
from ..compat import compat_str

class NiconicoLiveFD(FileDownloader):
    """ Downloads niconico live without being stopped """

    def real_download(self, filename, info_dict):
        if not WebSocket:  # this is unreachable because this is checked at Extractor
            raise DownloadError('Install websockets or websocket_client package via pip, or install websockat program')

        video_id = info_dict['video_id']
        ws_url = info_dict['url']
        ws_extractor = info_dict['ws']
        cookies = info_dict.get('cookies')
        live_quality = info_dict.get('live_quality', 'high')
        live_latency = info_dict.get('live_latency', 'high')
        dl = FFmpegFD(self.ydl, self.params or {})

        new_info_dict = info_dict.copy()
        new_info_dict.update({
            'protocol': 'live_ffmpeg',
        })

        def communicate_ws(reconnect):
            if reconnect:
                ws = WebSocket(ws_url, {
                    'Cookie': str_or_none(cookies) or '',
                    'Origin': 'https://live2.nicovideo.jp',
                    'Accept': '*/*',
                    'User-Agent': std_headers['User-Agent'],
                })
                if self.ydl.params.get('verbose', False):
                    self.to_screen('[debug] Sending startWatching request')
                ws.send(json.dumps({
                    "type": "startWatching",
                    "data": {
                        "stream": {
                            "quality": live_quality,
                            "protocol": "hls+fmp4",
                            "latency": live_latency,
                            "chasePlay": False
                        },
                        "room": {
                            "protocol": "webSocket",
                            "commentable": True
                        },
                        "reconnect": True,
                    }
                }))
            else:
                ws = ws_extractor
            with ws:
                while True:
                    recv = ws.recv()
                    if not recv:
                        continue
                    data = json.loads(recv)
                    if not data or not isinstance(data, dict):
                        continue
                    if data.get('type') == 'ping':
                        # pong back
                        ws.send(r'{"type":"pong"}')
                        ws.send(r'{"type":"keepSeat"}')
                    elif data.get('type') == 'disconnect':
                        self.write_debug(data)
                        return True
                    elif data.get('type') == 'error':
                        self.write_debug(data)
                        message = try_get(data, lambda x: x["body"]["code"], compat_str) or recv
                        return DownloadError(message)
                    elif self.ydl.params.get('verbose', False):
                        if len(recv) > 100:
                            recv = recv[:100] + '...'
                        self.to_screen('[debug] Server said: %s' % recv)

        def ws_main():
            reconnect = False
            while True:
                try:
                    ret = communicate_ws(reconnect)
                    if ret is True:
                        return
                except BaseException as e:
                    self.to_screen('[%s] %s: Connection error occured, reconnecting after 10 seconds: %s' % ('niconico:live', video_id, str_or_none(e)))
                    time.sleep(10)
                    continue
                finally:
                    reconnect = True

        thread = threading.Thread(target=ws_main, daemon=True)
        thread.start()

        return dl.download(filename, new_info_dict)


code:
import re
import subprocess
import time

from .common import FileDownloader
from ..utils import (
    Popen,
    check_executable,
    encodeArgument,
    encodeFilename,
    get_exe_version,
)

def rtmpdump_version():
    return get_exe_version(
        'rtmpdump', ['--help'], r'(?i)RTMPDump\s*v?([0-9a-zA-Z._-]+)')

class RtmpFD(FileDownloader):
    def real_download(self, filename, info_dict):
        def run_rtmpdump(args):
            start = time.time()
            resume_percent = None
            resume_downloaded_data_len = None
            proc = Popen(args, stderr=subprocess.PIPE)
            cursor_in_new_line = True
            proc_stderr_closed = False
            try:
                while not proc_stderr_closed:
                    # read line from stderr
                    line = ''
                    while True:
                        char = proc.stderr.read(1)
                        if not char:
                            proc_stderr_closed = True
                            break
                        if char in [b'\r', b'\n']:
                            break
                        line += char.decode('ascii', 'replace')
                    if not line:
                        # proc_stderr_closed is True
                        continue
                    mobj = re.search(r'([0-9]+\.[0-9]{3}) kB / [0-9]+\.[0-9]{2} sec \(([0-9]{1,2}\.[0-9])%\)', line)
                    if mobj:
                        downloaded_data_len = int(float(mobj.group(1)) * 1024)
                        percent = float(mobj.group(2))
                        if not resume_percent:
                            resume_percent = percent
                            resume_downloaded_data_len = downloaded_data_len
                        time_now = time.time()
                        eta = self.calc_eta(start, time_now, 100 - resume_percent, percent - resume_percent)
                        speed = self.calc_speed(start, time_now, downloaded_data_len - resume_downloaded_data_len)
                        data_len = None
                        if percent > 0:
                            data_len = int(downloaded_data_len * 100 / percent)
                        self._hook_progress({
                            'status': 'downloading',
                            'downloaded_bytes': downloaded_data_len,
                            'total_bytes_estimate': data_len,
                            'tmpfilename': tmpfilename,
                            'filename': filename,
                            'eta': eta,
                            'elapsed': time_now - start,
                            'speed': speed,
                        }, info_dict)
                        cursor_in_new_line = False
                    else:
                        # no percent for live streams
                        mobj = re.search(r'([0-9]+\.[0-9]{3}) kB / [0-9]+\.[0-9]{2} sec', line)
                        if mobj:
                            downloaded_data_len = int(float(mobj.group(1)) * 1024)
                            time_now = time.time()
                            speed = self.calc_speed(start, time_now, downloaded_data_len)
                            self._hook_progress({
                                'downloaded_bytes': downloaded_data_len,
                                'tmpfilename': tmpfilename,
                                'filename': filename,
                                'status': 'downloading',
                                'elapsed': time_now - start,
                                'speed': speed,
                            }, info_dict)
                            cursor_in_new_line = False
                        elif self.params.get('verbose', False):
                            if not cursor_in_new_line:
                                self.to_screen('')
                            cursor_in_new_line = True
                            self.to_screen('[rtmpdump] ' + line)
                if not cursor_in_new_line:
                    self.to_screen('')
                return proc.wait()
            except BaseException:  # Including KeyboardInterrupt
                proc.kill(timeout=None)
                raise

        url = info_dict['url']
        player_url = info_dict.get('player_url')
        page_url = info_dict.get('page_url')
        app = info_dict.get('app')
        play_path = info_dict.get('play_path')
        tc_url = info_dict.get('tc_url')
        flash_version = info_dict.get('flash_version')
        live = info_dict.get('rtmp_live', False)
        conn = info_dict.get('rtmp_conn')
        protocol = info_dict.get('rtmp_protocol')
        real_time = info_dict.get('rtmp_real_time', False)
        no_resume = info_dict.get('no_resume', False)
        continue_dl = self.params.get('continuedl', True)

        self.report_destination(filename)
        tmpfilename = self.temp_name(filename)
        test = self.params.get('test', False)

        # Check for rtmpdump first
        if not check_executable('rtmpdump', ['-h']):
            self.report_error('RTMP download detected but "rtmpdump" could not be run. Please install')
            return False

        # Download using rtmpdump. rtmpdump returns exit code 2 when
        # the connection was interrupted and resuming appears to be
        # possible. This is part of rtmpdump's normal usage, AFAIK.
        basic_args = [
            'rtmpdump', '--verbose', '-r', url,
            '-o', tmpfilename]
        if player_url is not None:
            basic_args += ['--swfVfy', player_url]
        if page_url is not None:
            basic_args += ['--pageUrl', page_url]
        if app is not None:
            basic_args += ['--app', app]
        if play_path is not None:
            basic_args += ['--playpath', play_path]
        if tc_url is not None:
            basic_args += ['--tcUrl', tc_url]
        if test:
            basic_args += ['--stop', '1']
        if flash_version is not None:
            basic_args += ['--flashVer', flash_version]
        if live:
            basic_args += ['--live']
        if isinstance(conn, list):
            for entry in conn:
                basic_args += ['--conn', entry]
        elif isinstance(conn, str):
            basic_args += ['--conn', conn]
        if protocol is not None:
            basic_args += ['--protocol', protocol]
        if real_time:
            basic_args += ['--realtime']

        args = basic_args
        if not no_resume and continue_dl and not live:
            args += ['--resume']
        if not live and continue_dl:
            args += ['--skip', '1']

        args = [encodeArgument(a) for a in args]

        self._debug_cmd(args, exe='rtmpdump')

        RD_SUCCESS = 0
        RD_FAILED = 1
        RD_INCOMPLETE = 2
        RD_NO_CONNECT = 3

        started = time.time()

        try:
            retval = run_rtmpdump(args)
        except KeyboardInterrupt:
            if not info_dict.get('is_live'):
                raise
            retval = RD_SUCCESS
            self.to_screen('\n[rtmpdump] Interrupted by user')

        if retval == RD_NO_CONNECT:
            self.report_error('[rtmpdump] Could not connect to RTMP server.')
            return False

        while retval in (RD_INCOMPLETE, RD_FAILED) and not test and not live:
            prevsize = self.ydl.getsize(encodeFilename(tmpfilename))
            self.to_screen('[rtmpdump] Downloaded %s bytes' % prevsize)
            time.sleep(5.0)  # This seems to be needed
            args = basic_args + ['--resume']
            if retval == RD_FAILED:
                args += ['--skip', '1']
            args = [encodeArgument(a) for a in args]
            retval = run_rtmpdump(args)
            cursize = self.ydl.getsize(encodeFilename(tmpfilename))
            if prevsize == cursize and retval == RD_FAILED:
                break
            # Some rtmp streams seem abort after ~ 99.8%. Don't complain for those
            if prevsize == cursize and retval == RD_INCOMPLETE and cursize > 1024:
                self.to_screen('[rtmpdump] Could not download the whole video. This can happen for some advertisements.')
                retval = RD_SUCCESS
                break
        if retval == RD_SUCCESS or (test and retval == RD_INCOMPLETE):
            fsize = self.ydl.getsize(encodeFilename(tmpfilename))
            self.to_screen('[rtmpdump] Downloaded %s bytes' % fsize)
            self.try_rename(tmpfilename, filename)
            self._hook_progress({
                'downloaded_bytes': fsize,
                'total_bytes': fsize,
                'filename': filename,
                'status': 'finished',
                'elapsed': time.time() - started,
            }, info_dict)
            return True
        else:
            self.to_stderr('\n')
            self.report_error('rtmpdump exited with code %d' % retval)
            return False


code:
import subprocess

from .common import FileDownloader
from ..utils import check_executable, encodeFilename

class RtspFD(FileDownloader):
    def real_download(self, filename, info_dict):
        url = info_dict['url']
        self.report_destination(filename)
        tmpfilename = self.temp_name(filename)

        if check_executable('mplayer', ['-h']):
            args = [
                'mplayer', '-really-quiet', '-vo', 'null', '-vc', 'dummy',
                '-dumpstream', '-dumpfile', tmpfilename, url]
        elif check_executable('mpv', ['-h']):
            args = [
                'mpv', '-really-quiet', '--vo=null', '--stream-dump=' + tmpfilename, url]
        else:
            self.report_error('MMS or RTSP download detected but neither "mplayer" nor "mpv" could be run. Please install one')
            return False

        self._debug_cmd(args)

        retval = subprocess.call(args)
        if retval == 0:
            fsize = self.ydl.getsize(encodeFilename(tmpfilename))
            self.to_screen(f'\r[{args[0]}] {fsize} bytes')
            self.try_rename(tmpfilename, filename)
            self._hook_progress({
                'downloaded_bytes': fsize,
                'total_bytes': fsize,
                'filename': filename,
                'status': 'finished',
            }, info_dict)
            return True
        else:
            self.to_stderr('\n')
            self.report_error('%s exited with code %d' % (args[0], retval))
            return False


code:
from __future__ import division, unicode_literals

from .common import FileDownloader
from .external import FFmpegFD

class SerialFD(FileDownloader):
    """
    Download videos in the order and join later.
    Ensure all enclosed formats have same codec, resolution and fps.

    Keys for formats:
    - "url" -- Must be "serial:". or download will fail.
    - "items" -- List of formats to download and join. Must have more than one item
    """

    def real_download(self, filename, info_dict):
        from ..downloader import get_suitable_downloader
        if info_dict['url'] != 'serial:':
            self.report_error('"url" key must be "serial:". Please check extractor code.')
            return False

        items = info_dict['items']
        if not items:
            self.report_error('There is no item to download!')
            return False

        if len(items) == 1:
            return get_suitable_downloader(items[0])(self.ydl, self.params).download(filename, items[0])

        names_to_join = []
        for idx, entry in enumerate(items):
            self.to_screen('Downloading step %d' % idx)
            new_name = '%s_%d.%s' % (filename, idx, entry['ext'])
            result = get_suitable_downloader(entry)(self.ydl, self.params).download(new_name, entry)
            if not result:
                self.report_error('Download failed at step %d' % idx)
                return False
            names_to_join.append(new_name)

        txt_file = '%s.txt' % filename
        ffmpeg_request = {
            'protocol': 'live_ffmpeg',
            'url': txt_file,
            'ext': info_dict['ext'],
            'input_params': ['-f', 'concat', '-safe', '0'],
            'output_params': ['-movflags', '+faststart'],
        }
        try:
            with self.ydl.open(txt_file, 'wt') as w:
                w.write(''.join(f"file '{nmj}'\n" for nmj in names_to_join))
            ret = FFmpegFD(self.ydl, self.params).download(filename, ffmpeg_request)[0]

            if ret:
                for nmj in names_to_join:
                    try:
                        self.ydl.remove(nmj)
                    except (IOError, OSError):
                        pass

            return ret
        finally:
            try:
                self.ydl.remove(txt_file)
            except (IOError, OSError):
                pass


code:
import asyncio
import contextlib
import os
import signal
import threading

from .common import FileDownloader
from .external import FFmpegFD
from ..dependencies import websockets

class FFmpegSinkFD(FileDownloader):
    """ A sink to ffmpeg for downloading fragments in any form """

    def real_download(self, filename, info_dict):
        info_copy = info_dict.copy()
        info_copy['url'] = '-'

        async def call_conn(proc, stdin):
            try:
                await self.real_connection(stdin, info_dict)
            except OSError:
                pass
            finally:
                with contextlib.suppress(OSError):
                    stdin.flush()
                    stdin.close()
                os.kill(os.getpid(), signal.SIGINT)

        class FFmpegStdinFD(FFmpegFD):
            @classmethod
            def get_basename(cls):
                return FFmpegFD.get_basename()

            def on_process_started(self, proc, stdin):
                thread = threading.Thread(target=asyncio.run, daemon=True, args=(call_conn(proc, stdin), ))
                thread.start()

        return FFmpegStdinFD(self.ydl, self.params or {}).download(filename, info_copy)

    async def real_connection(self, sink, info_dict):
        """ Override this in subclasses """
        raise NotImplementedError('This method must be implemented by subclasses')

class WebSocketFragmentFD(FFmpegSinkFD):
    async def real_connection(self, sink, info_dict):
        async with websockets.connect(info_dict['url'], extra_headers=info_dict.get('http_headers', {})) as ws:
            while True:
                recv = await ws.recv()
                if isinstance(recv, str):
                    recv = recv.encode('utf8')
                sink.write(recv)


code:
import json
import time
import urllib.error

from .fragment import FragmentFD
from ..utils import (
    RegexNotFoundError,
    RetryManager,
    dict_get,
    int_or_none,
    try_get,
)

class YoutubeLiveChatFD(FragmentFD):
    """ Downloads YouTube live chats fragment by fragment """

    def real_download(self, filename, info_dict):
        video_id = info_dict['video_id']
        self.to_screen('[%s] Downloading live chat' % self.FD_NAME)
        if not self.params.get('skip_download') and info_dict['protocol'] == 'youtube_live_chat':
            self.report_warning('Live chat download runs until the livestream ends. '
                                'If you wish to download the video simultaneously, run a separate yt-dlp instance')

        test = self.params.get('test', False)

        ctx = {
            'filename': filename,
            'live': True,
            'total_frags': None,
        }

        from ..extractor.youtube import YoutubeBaseInfoExtractor

        ie = YoutubeBaseInfoExtractor(self.ydl)

        start_time = int(time.time() * 1000)

        def dl_fragment(url, data=None, headers=None):
            http_headers = info_dict.get('http_headers', {})
            if headers:
                http_headers = http_headers.copy()
                http_headers.update(headers)
            return self._download_fragment(ctx, url, info_dict, http_headers, data)

        def parse_actions_replay(live_chat_continuation):
            offset = continuation_id = click_tracking_params = None
            processed_fragment = bytearray()
            for action in live_chat_continuation.get('actions', []):
                if 'replayChatItemAction' in action:
                    replay_chat_item_action = action['replayChatItemAction']
                    offset = int(replay_chat_item_action['videoOffsetTimeMsec'])
                processed_fragment.extend(
                    json.dumps(action, ensure_ascii=False).encode() + b'\n')
            if offset is not None:
                continuation = try_get(
                    live_chat_continuation,
                    lambda x: x['continuations'][0]['liveChatReplayContinuationData'], dict)
                if continuation:
                    continuation_id = continuation.get('continuation')
                    click_tracking_params = continuation.get('clickTrackingParams')
            self._append_fragment(ctx, processed_fragment)
            return continuation_id, offset, click_tracking_params

        def try_refresh_replay_beginning(live_chat_continuation):
            # choose the second option that contains the unfiltered live chat replay
            refresh_continuation = try_get(
                live_chat_continuation,
                lambda x: x['header']['liveChatHeaderRenderer']['viewSelector']['sortFilterSubMenuRenderer']['subMenuItems'][1]['continuation']['reloadContinuationData'], dict)
            if refresh_continuation:
                # no data yet but required to call _append_fragment
                self._append_fragment(ctx, b'')
                refresh_continuation_id = refresh_continuation.get('continuation')
                offset = 0
                click_tracking_params = refresh_continuation.get('trackingParams')
                return refresh_continuation_id, offset, click_tracking_params
            return parse_actions_replay(live_chat_continuation)

        live_offset = 0

        def parse_actions_live(live_chat_continuation):
            nonlocal live_offset
            continuation_id = click_tracking_params = None
            processed_fragment = bytearray()
            for action in live_chat_continuation.get('actions', []):
                timestamp = self.parse_live_timestamp(action)
                if timestamp is not None:
                    live_offset = timestamp - start_time
                # compatibility with replay format
                pseudo_action = {
                    'replayChatItemAction': {'actions': [action]},
                    'videoOffsetTimeMsec': str(live_offset),
                    'isLive': True,
                }
                processed_fragment.extend(
                    json.dumps(pseudo_action, ensure_ascii=False).encode() + b'\n')
            continuation_data_getters = [
                lambda x: x['continuations'][0]['invalidationContinuationData'],
                lambda x: x['continuations'][0]['timedContinuationData'],
            ]
            continuation_data = try_get(live_chat_continuation, continuation_data_getters, dict)
            if continuation_data:
                continuation_id = continuation_data.get('continuation')
                click_tracking_params = continuation_data.get('clickTrackingParams')
                timeout_ms = int_or_none(continuation_data.get('timeoutMs'))
                if timeout_ms is not None:
                    time.sleep(timeout_ms / 1000)
            self._append_fragment(ctx, processed_fragment)
            return continuation_id, live_offset, click_tracking_params

        def download_and_parse_fragment(url, frag_index, request_data=None, headers=None):
            for retry in RetryManager(self.params.get('fragment_retries'), self.report_retry, frag_index=frag_index):
                try:
                    success = dl_fragment(url, request_data, headers)
                    if not success:
                        return False, None, None, None
                    raw_fragment = self._read_fragment(ctx)
                    try:
                        data = ie.extract_yt_initial_data(video_id, raw_fragment.decode('utf-8', 'replace'))
                    except RegexNotFoundError:
                        data = None
                    if not data:
                        data = json.loads(raw_fragment)
                    live_chat_continuation = try_get(
                        data,
                        lambda x: x['continuationContents']['liveChatContinuation'], dict) or {}

                    func = (info_dict['protocol'] == 'youtube_live_chat' and parse_actions_live
                            or frag_index == 1 and try_refresh_replay_beginning
                            or parse_actions_replay)
                    return (True, *func(live_chat_continuation))
                except urllib.error.HTTPError as err:
                    retry.error = err
                    continue
            return False, None, None, None

        self._prepare_and_start_frag_download(ctx, info_dict)

        success = dl_fragment(info_dict['url'])
        if not success:
            return False
        raw_fragment = self._read_fragment(ctx)
        try:
            data = ie.extract_yt_initial_data(video_id, raw_fragment.decode('utf-8', 'replace'))
        except RegexNotFoundError:
            return False
        continuation_id = try_get(
            data,
            lambda x: x['contents']['twoColumnWatchNextResults']['conversationBar']['liveChatRenderer']['continuations'][0]['reloadContinuationData']['continuation'])
        # no data yet but required to call _append_fragment
        self._append_fragment(ctx, b'')

        ytcfg = ie.extract_ytcfg(video_id, raw_fragment.decode('utf-8', 'replace'))

        if not ytcfg:
            return False
        api_key = try_get(ytcfg, lambda x: x['INNERTUBE_API_KEY'])
        innertube_context = try_get(ytcfg, lambda x: x['INNERTUBE_CONTEXT'])
        if not api_key or not innertube_context:
            return False
        visitor_data = try_get(innertube_context, lambda x: x['client']['visitorData'], str)
        if info_dict['protocol'] == 'youtube_live_chat_replay':
            url = 'https://www.youtube.com/youtubei/v1/live_chat/get_live_chat_replay?key=' + api_key
            chat_page_url = 'https://www.youtube.com/live_chat_replay?continuation=' + continuation_id
        elif info_dict['protocol'] == 'youtube_live_chat':
            url = 'https://www.youtube.com/youtubei/v1/live_chat/get_live_chat?key=' + api_key
            chat_page_url = 'https://www.youtube.com/live_chat?continuation=' + continuation_id

        frag_index = offset = 0
        click_tracking_params = None
        while continuation_id is not None:
            frag_index += 1
            request_data = {
                'context': innertube_context,
                'continuation': continuation_id,
            }
            if frag_index > 1:
                request_data['currentPlayerState'] = {'playerOffsetMs': str(max(offset - 5000, 0))}
                if click_tracking_params:
                    request_data['context']['clickTracking'] = {'clickTrackingParams': click_tracking_params}
                headers = ie.generate_api_headers(ytcfg=ytcfg, visitor_data=visitor_data)
                headers.update({'content-type': 'application/json'})
                fragment_request_data = json.dumps(request_data, ensure_ascii=False).encode() + b'\n'
                success, continuation_id, offset, click_tracking_params = download_and_parse_fragment(
                    url, frag_index, fragment_request_data, headers)
            else:
                success, continuation_id, offset, click_tracking_params = download_and_parse_fragment(
                    chat_page_url, frag_index)
            if not success:
                return False
            if test:
                break

        self._finish_frag_download(ctx, info_dict)
        return True

    @staticmethod
    def parse_live_timestamp(action):
        action_content = dict_get(
            action,
            ['addChatItemAction', 'addLiveChatTickerItemAction', 'addBannerToLiveChatCommand'])
        if not isinstance(action_content, dict):
            return None
        item = dict_get(action_content, ['item', 'bannerRenderer'])
        if not isinstance(item, dict):
            return None
        renderer = dict_get(item, [
            # text
            'liveChatTextMessageRenderer', 'liveChatPaidMessageRenderer',
            'liveChatMembershipItemRenderer', 'liveChatPaidStickerRenderer',
            # ticker
            'liveChatTickerPaidMessageItemRenderer',
            'liveChatTickerSponsorItemRenderer',
            # banner
            'liveChatBannerRenderer',
        ])
        if not isinstance(renderer, dict):
            return None
        parent_item_getters = [
            lambda x: x['showItemEndpoint']['showLiveChatItemEndpoint']['renderer'],
            lambda x: x['contents'],
        ]
        parent_item = try_get(renderer, parent_item_getters, dict)
        if parent_item:
            renderer = dict_get(parent_item, [
                'liveChatTextMessageRenderer', 'liveChatPaidMessageRenderer',
                'liveChatMembershipItemRenderer', 'liveChatPaidStickerRenderer',
            ])
            if not isinstance(renderer, dict):
                return None
        return int_or_none(renderer.get('timestampUsec'), 1000)


code:
from ..compat.compat_utils import passthrough_module

from typing import List, Type, TYPE_CHECKING
if TYPE_CHECKING:
    from .common import SelfHostedInfoExtractor

passthrough_module(__name__, '.extractors')
del passthrough_module

def gen_extractor_classes():
    """ Return a list of supported extractors.
    The order does matter; the first extractor matched is the one handling the URL.
    """
    from .extractors import _ALL_CLASSES

    return _ALL_CLASSES

def gen_selfhosted_extractor_classes() -> 'List[Type[SelfHostedInfoExtractor]]':
    """
    Return a list of extractors for self-hosted services.
    """
    from .extractors import _SELFHOSTED_CLASSES

    return _SELFHOSTED_CLASSES

def gen_extractors():
    """ Return a list of an instance of every supported extractor.
    The order does matter; the first extractor matched is the one handling the URL.
    """
    return [klass() for klass in gen_extractor_classes()]

def list_extractor_classes(age_limit=None):
    """Return a list of extractors that are suitable for the given age, sorted by extractor name"""
    from .generic import GenericIE

    yield from sorted(filter(
        lambda ie: ie.is_suitable(age_limit) and ie != GenericIE,
        gen_extractor_classes()), key=lambda ie: ie.IE_NAME.lower())
    yield GenericIE

def list_extractors(age_limit=None):
    """Return a list of extractor instances that are suitable for the given age, sorted by extractor name"""
    return [ie() for ie in list_extractor_classes(age_limit)]

def get_info_extractor(ie_name):
    """Returns the info extractor class with the given ie_name"""
    from . import extractors

    return getattr(extractors, f'{ie_name}IE')