]> jfr.im git - yt-dlp.git/blobdiff - yt_dlp/downloader/hls.py
[ie/mlbtv] Fix extraction (#10296)
[yt-dlp.git] / yt_dlp / downloader / hls.py
index 8b7d51de360f40e4956deabcf704a334faebe20a..0a00d5dabbcd705fd72bbd3b5eb658e53f92804f 100644 (file)
@@ -1,45 +1,43 @@
-from __future__ import unicode_literals
-
-import re
 import binascii
-try:
-    from Crypto.Cipher import AES
-    can_decrypt_frag = True
-except ImportError:
-    can_decrypt_frag = False
-try:
-    import concurrent.futures
-    can_threaded_download = True
-except ImportError:
-    can_threaded_download = False
-
-from ..downloader import _get_real_downloader
-from .fragment import FragmentFD
-from .external import FFmpegFD
+import io
+import re
+import urllib.parse
 
-from ..compat import (
-    compat_urllib_error,
-    compat_urlparse,
-    compat_struct_pack,
-)
+from . import get_suitable_downloader
+from .external import FFmpegFD
+from .fragment import FragmentFD
+from .. import webvtt
+from ..dependencies import Cryptodome
 from ..utils import (
+    bug_reports_message,
     parse_m3u8_attributes,
-    sanitize_open,
+    remove_start,
+    traverse_obj,
     update_url_query,
+    urljoin,
 )
 
 
 class HlsFD(FragmentFD):
     """
     Download segments in a m3u8 manifest. External downloaders can take over
-    the fragment downloads by supporting the 'frag_urls' protocol and
+    the fragment downloads by supporting the 'm3u8_frag_urls' protocol and
     re-defining 'supports_manifest' function
     """
 
     FD_NAME = 'hlsnative'
 
     @staticmethod
-    def can_download(manifest, info_dict, allow_unplayable_formats=False, with_crypto=can_decrypt_frag):
+    def _has_drm(manifest):  # TODO: https://github.com/yt-dlp/yt-dlp/pull/5039
+        return bool(re.search('|'.join((
+            r'#EXT-X-(?:SESSION-)?KEY:.*?URI="skd://',  # Apple FairPlay
+            r'#EXT-X-(?:SESSION-)?KEY:.*?KEYFORMAT="com\.apple\.streamingkeydelivery"',  # Apple FairPlay
+            r'#EXT-X-(?:SESSION-)?KEY:.*?KEYFORMAT="com\.microsoft\.playready"',  # Microsoft PlayReady
+            r'#EXT-X-FAXS-CM:',  # Adobe Flash Access
+        )), manifest))
+
+    @classmethod
+    def can_download(cls, manifest, info_dict, allow_unplayable_formats=False):
         UNSUPPORTED_FEATURES = [
             # r'#EXT-X-BYTERANGE',  # playlists composed of byte ranges of media files [2]
 
@@ -61,44 +59,64 @@ def can_download(manifest, info_dict, allow_unplayable_formats=False, with_crypt
         ]
         if not allow_unplayable_formats:
             UNSUPPORTED_FEATURES += [
-                r'#EXT-X-KEY:METHOD=(?!NONE|AES-128)',  # encrypted streams [1]
+                r'#EXT-X-KEY:METHOD=(?!NONE|AES-128)',  # encrypted streams [1], but not necessarily DRM
             ]
 
         def check_results():
             yield not info_dict.get('is_live')
-            is_aes128_enc = '#EXT-X-KEY:METHOD=AES-128' in manifest
-            yield with_crypto or not is_aes128_enc
-            yield not (is_aes128_enc and r'#EXT-X-BYTERANGE' in manifest)
             for feature in UNSUPPORTED_FEATURES:
                 yield not re.search(feature, manifest)
+            if not allow_unplayable_formats:
+                yield not cls._has_drm(manifest)
         return all(check_results())
 
     def real_download(self, filename, info_dict):
         man_url = info_dict['url']
-        self.to_screen('[%s] Downloading m3u8 manifest' % self.FD_NAME)
+        self.to_screen(f'[{self.FD_NAME}] Downloading m3u8 manifest')
 
         urlh = self.ydl.urlopen(self._prepare_url(info_dict, man_url))
-        man_url = urlh.geturl()
+        man_url = urlh.url
         s = urlh.read().decode('utf-8', 'ignore')
 
-        if not self.can_download(s, info_dict, self.params.get('allow_unplayable_formats')):
-            if info_dict.get('extra_param_to_segment_url') or info_dict.get('_decryption_key_url'):
-                self.report_error('pycryptodome not found. Please install it.')
+        can_download, message = self.can_download(s, info_dict, self.params.get('allow_unplayable_formats')), None
+        if can_download:
+            has_ffmpeg = FFmpegFD.available()
+            no_crypto = not Cryptodome.AES and '#EXT-X-KEY:METHOD=AES-128' in s
+            if no_crypto and has_ffmpeg:
+                can_download, message = False, 'The stream has AES-128 encryption and pycryptodomex is not available'
+            elif no_crypto:
+                message = ('The stream has AES-128 encryption and neither ffmpeg nor pycryptodomex are available; '
+                           'Decryption will be performed natively, but will be extremely slow')
+            elif info_dict.get('extractor_key') == 'Generic' and re.search(r'(?m)#EXT-X-MEDIA-SEQUENCE:(?!0$)', s):
+                install_ffmpeg = '' if has_ffmpeg else 'install ffmpeg and '
+                message = ('Live HLS streams are not supported by the native downloader. If this is a livestream, '
+                           f'please {install_ffmpeg}add "--downloader ffmpeg --hls-use-mpegts" to your command')
+        if not can_download:
+            if self._has_drm(s) and not self.params.get('allow_unplayable_formats'):
+                if info_dict.get('has_drm') and self.params.get('test'):
+                    self.to_screen(f'[{self.FD_NAME}] This format is DRM protected', skip_eol=True)
+                else:
+                    self.report_error(
+                        'This format is DRM protected; Try selecting another format with --format or '
+                        'add --check-formats to automatically fallback to the next best format', tb=False)
                 return False
-            if self.can_download(s, info_dict, with_crypto=True):
-                self.report_warning('pycryptodome is needed to download this file with hlsnative')
-            self.report_warning(
-                'hlsnative has detected features it does not support, '
-                'extraction will be delegated to ffmpeg')
+            message = message or 'Unsupported features have been detected'
             fd = FFmpegFD(self.ydl, self.params)
-            # TODO: Make progress updates work without hooking twice
-            # for ph in self._progress_hooks:
-            #     fd.add_progress_hook(ph)
+            self.report_warning(f'{message}; extraction will be delegated to {fd.get_basename()}')
             return fd.real_download(filename, info_dict)
+        elif message:
+            self.report_warning(message)
 
-        real_downloader = _get_real_downloader(info_dict, 'frag_urls', self.params, None)
+        is_webvtt = info_dict['ext'] == 'vtt'
+        if is_webvtt:
+            real_downloader = None  # Packing the fragments is not currently supported for external downloader
+        else:
+            real_downloader = get_suitable_downloader(
+                info_dict, self.params, None, protocol='m3u8_frag_urls', to_stdout=(filename == '-'))
         if real_downloader and not real_downloader.supports_manifest(s):
             real_downloader = None
+        if real_downloader:
+            self.to_screen(f'[{self.FD_NAME}] Fragment downloads will be delegated to {real_downloader.get_basename()}')
 
         def is_ad_fragment_start(s):
             return (s.startswith('#ANVATO-SEGMENT-INFO') and 'type=ad' in s
@@ -137,20 +155,27 @@ def is_ad_fragment_end(s):
         if real_downloader:
             self._prepare_external_frag_download(ctx)
         else:
-            self._prepare_and_start_frag_download(ctx)
+            self._prepare_and_start_frag_download(ctx, info_dict)
 
-        fragment_retries = self.params.get('fragment_retries', 0)
-        skip_unavailable_fragments = self.params.get('skip_unavailable_fragments', True)
-        test = self.params.get('test', False)
+        extra_state = ctx.setdefault('extra_state', {})
 
         format_index = info_dict.get('format_index')
-        extra_query = None
-        extra_param_to_segment_url = info_dict.get('extra_param_to_segment_url')
-        if extra_param_to_segment_url:
-            extra_query = compat_urlparse.parse_qs(extra_param_to_segment_url)
+        extra_segment_query = None
+        if extra_param_to_segment_url := info_dict.get('extra_param_to_segment_url'):
+            extra_segment_query = urllib.parse.parse_qs(extra_param_to_segment_url)
+        extra_key_query = None
+        if extra_param_to_key_url := info_dict.get('extra_param_to_key_url'):
+            extra_key_query = urllib.parse.parse_qs(extra_param_to_key_url)
         i = 0
         media_sequence = 0
         decrypt_info = {'METHOD': 'NONE'}
+        external_aes_key = traverse_obj(info_dict, ('hls_aes', 'key'))
+        if external_aes_key:
+            external_aes_key = binascii.unhexlify(remove_start(external_aes_key, '0x'))
+            assert len(external_aes_key) in (16, 24, 32), 'Invalid length for HLS AES-128 key'
+        external_aes_iv = traverse_obj(info_dict, ('hls_aes', 'iv'))
+        if external_aes_iv:
+            external_aes_iv = binascii.unhexlify(remove_start(external_aes_iv, '0x').zfill(32))
         byte_range = {}
         discontinuity_count = 0
         frag_index = 0
@@ -166,12 +191,9 @@ def is_ad_fragment_end(s):
                     frag_index += 1
                     if frag_index <= ctx['fragment_index']:
                         continue
-                    frag_url = (
-                        line
-                        if re.match(r'^https?://', line)
-                        else compat_urlparse.urljoin(man_url, line))
-                    if extra_query:
-                        frag_url = update_url_query(frag_url, extra_query)
+                    frag_url = urljoin(man_url, line)
+                    if extra_segment_query:
+                        frag_url = update_url_query(frag_url, extra_segment_query)
 
                     fragments.append({
                         'frag_index': frag_index,
@@ -180,30 +202,20 @@ def is_ad_fragment_end(s):
                         'byte_range': byte_range,
                         'media_sequence': media_sequence,
                     })
+                    media_sequence += 1
 
                 elif line.startswith('#EXT-X-MAP'):
                     if format_index and discontinuity_count != format_index:
                         continue
                     if frag_index > 0:
                         self.report_error(
-                            'initialization fragment found after media fragments, unable to download')
+                            'Initialization fragment found after media fragments, unable to download')
                         return False
                     frag_index += 1
                     map_info = parse_m3u8_attributes(line[11:])
-                    frag_url = (
-                        map_info.get('URI')
-                        if re.match(r'^https?://', map_info.get('URI'))
-                        else compat_urlparse.urljoin(man_url, map_info.get('URI')))
-                    if extra_query:
-                        frag_url = update_url_query(frag_url, extra_query)
-
-                    fragments.append({
-                        'frag_index': frag_index,
-                        'url': frag_url,
-                        'decrypt_info': decrypt_info,
-                        'byte_range': byte_range,
-                        'media_sequence': media_sequence
-                    })
+                    frag_url = urljoin(man_url, map_info.get('URI'))
+                    if extra_segment_query:
+                        frag_url = update_url_query(frag_url, extra_segment_query)
 
                     if map_info.get('BYTERANGE'):
                         splitted_byte_range = map_info.get('BYTERANGE').split('@')
@@ -213,19 +225,33 @@ def is_ad_fragment_end(s):
                             'end': sub_range_start + int(splitted_byte_range[0]),
                         }
 
+                    fragments.append({
+                        'frag_index': frag_index,
+                        'url': frag_url,
+                        'decrypt_info': decrypt_info,
+                        'byte_range': byte_range,
+                        'media_sequence': media_sequence,
+                    })
+                    media_sequence += 1
+
                 elif line.startswith('#EXT-X-KEY'):
                     decrypt_url = decrypt_info.get('URI')
                     decrypt_info = parse_m3u8_attributes(line[11:])
                     if decrypt_info['METHOD'] == 'AES-128':
-                        if 'IV' in decrypt_info:
+                        if external_aes_iv:
+                            decrypt_info['IV'] = external_aes_iv
+                        elif 'IV' in decrypt_info:
                             decrypt_info['IV'] = binascii.unhexlify(decrypt_info['IV'][2:].zfill(32))
-                        if not re.match(r'^https?://', decrypt_info['URI']):
-                            decrypt_info['URI'] = compat_urlparse.urljoin(
-                                man_url, decrypt_info['URI'])
-                        if extra_query:
-                            decrypt_info['URI'] = update_url_query(decrypt_info['URI'], extra_query)
-                        if decrypt_url != decrypt_info['URI']:
-                            decrypt_info['KEY'] = None
+                        if external_aes_key:
+                            decrypt_info['KEY'] = external_aes_key
+                        else:
+                            decrypt_info['URI'] = urljoin(man_url, decrypt_info['URI'])
+                            if extra_key_query or extra_segment_query:
+                                # Fall back to extra_segment_query to key for backwards compat
+                                decrypt_info['URI'] = update_url_query(
+                                    decrypt_info['URI'], extra_key_query or extra_segment_query)
+                            if decrypt_url != decrypt_info['URI']:
+                                decrypt_info['KEY'] = None
 
                 elif line.startswith('#EXT-X-MEDIA-SEQUENCE'):
                     media_sequence = int(line[22:])
@@ -243,123 +269,113 @@ def is_ad_fragment_end(s):
                 elif line.startswith('#EXT-X-DISCONTINUITY'):
                     discontinuity_count += 1
                 i += 1
-                media_sequence += 1
 
         # We only download the first fragment during the test
-        if test:
+        if self.params.get('test', False):
             fragments = [fragments[0] if fragments else None]
 
         if real_downloader:
-            info_copy = info_dict.copy()
-            info_copy['fragments'] = fragments
+            info_dict['fragments'] = fragments
             fd = real_downloader(self.ydl, self.params)
             # TODO: Make progress updates work without hooking twice
             # for ph in self._progress_hooks:
             #     fd.add_progress_hook(ph)
-            success = fd.real_download(filename, info_copy)
-            if not success:
-                return False
-        else:
-            def download_fragment(fragment):
-                frag_index = fragment['frag_index']
-                frag_url = fragment['url']
-                decrypt_info = fragment['decrypt_info']
-                byte_range = fragment['byte_range']
-                media_sequence = fragment['media_sequence']
-
-                ctx['fragment_index'] = frag_index
-
-                count = 0
-                headers = info_dict.get('http_headers', {})
-                if byte_range:
-                    headers['Range'] = 'bytes=%d-%d' % (byte_range['start'], byte_range['end'] - 1)
-                while count <= fragment_retries:
-                    try:
-                        success, frag_content = self._download_fragment(
-                            ctx, frag_url, info_dict, headers)
-                        if not success:
-                            return False, frag_index
-                        break
-                    except compat_urllib_error.HTTPError as err:
-                        # Unavailable (possibly temporary) fragments may be served.
-                        # First we try to retry then either skip or abort.
-                        # See https://github.com/ytdl-org/youtube-dl/issues/10165,
-                        # https://github.com/ytdl-org/youtube-dl/issues/10448).
-                        count += 1
-                        if count <= fragment_retries:
-                            self.report_retry_fragment(err, frag_index, count, fragment_retries)
-                if count > fragment_retries:
-                    return False, frag_index
-
-                if decrypt_info['METHOD'] == 'AES-128':
-                    iv = decrypt_info.get('IV') or compat_struct_pack('>8xq', media_sequence)
-                    decrypt_info['KEY'] = decrypt_info.get('KEY') or self.ydl.urlopen(
-                        self._prepare_url(info_dict, info_dict.get('_decryption_key_url') or decrypt_info['URI'])).read()
-                    # Don't decrypt the content in tests since the data is explicitly truncated and it's not to a valid block
-                    # size (see https://github.com/ytdl-org/youtube-dl/pull/27660). Tests only care that the correct data downloaded,
-                    # not what it decrypts to.
-                    if not test:
-                        frag_content = AES.new(
-                            decrypt_info['KEY'], AES.MODE_CBC, iv).decrypt(frag_content)
-
-                return frag_content, frag_index
-
-            def append_fragment(frag_content, frag_index):
-                if frag_content:
-                    fragment_filename = '%s-Frag%d' % (ctx['tmpfilename'], frag_index)
-                    try:
-                        file, frag_sanitized = sanitize_open(fragment_filename, 'rb')
-                        ctx['fragment_filename_sanitized'] = frag_sanitized
-                        file.close()
-                        self._append_fragment(ctx, frag_content)
-                        return True
-                    except FileNotFoundError:
-                        if skip_unavailable_fragments:
-                            self.report_skip_fragment(frag_index)
-                            return True
-                        else:
-                            self.report_error(
-                                'fragment %s not found, unable to continue' % frag_index)
-                            return False
-                else:
-                    if skip_unavailable_fragments:
-                        self.report_skip_fragment(frag_index)
-                        return True
-                    else:
-                        self.report_error(
-                            'fragment %s not found, unable to continue' % frag_index)
-                        return False
+            return fd.real_download(filename, info_dict)
 
-            max_workers = self.params.get('concurrent_fragment_downloads', 1)
-            if can_threaded_download and max_workers > 1:
-                self.report_warning('The download speed shown is only of one thread. This is a known issue')
-                with concurrent.futures.ThreadPoolExecutor(max_workers) as pool:
-                    futures = [pool.submit(download_fragment, fragment) for fragment in fragments]
-                    # timeout must be 0 to return instantly
-                    done, not_done = concurrent.futures.wait(futures, timeout=0)
-                    try:
-                        while not_done:
-                            # Check every 1 second for KeyboardInterrupt
-                            freshly_done, not_done = concurrent.futures.wait(not_done, timeout=1)
-                            done |= freshly_done
-                    except KeyboardInterrupt:
-                        for future in not_done:
-                            future.cancel()
-                        # timeout must be none to cancel
-                        concurrent.futures.wait(not_done, timeout=None)
-                        raise KeyboardInterrupt
-                results = [future.result() for future in futures]
-
-                for frag_content, frag_index in results:
-                    result = append_fragment(frag_content, frag_index)
-                    if not result:
-                        return False
+        if is_webvtt:
+            def pack_fragment(frag_content, frag_index):
+                output = io.StringIO()
+                adjust = 0
+                overflow = False
+                mpegts_last = None
+                for block in webvtt.parse_fragment(frag_content):
+                    if isinstance(block, webvtt.CueBlock):
+                        extra_state['webvtt_mpegts_last'] = mpegts_last
+                        if overflow:
+                            extra_state['webvtt_mpegts_adjust'] += 1
+                            overflow = False
+                        block.start += adjust
+                        block.end += adjust
+
+                        dedup_window = extra_state.setdefault('webvtt_dedup_window', [])
+
+                        ready = []
+
+                        i = 0
+                        is_new = True
+                        while i < len(dedup_window):
+                            wcue = dedup_window[i]
+                            wblock = webvtt.CueBlock.from_json(wcue)
+                            i += 1
+                            if wblock.hinges(block):
+                                wcue['end'] = block.end
+                                is_new = False
+                                continue
+                            if wblock == block:
+                                is_new = False
+                                continue
+                            if wblock.end > block.start:
+                                continue
+                            ready.append(wblock)
+                            i -= 1
+                            del dedup_window[i]
+
+                        if is_new:
+                            dedup_window.append(block.as_json)
+                        for block in ready:
+                            block.write_into(output)
+
+                        # we only emit cues once they fall out of the duplicate window
+                        continue
+                    elif isinstance(block, webvtt.Magic):
+                        # take care of MPEG PES timestamp overflow
+                        if block.mpegts is None:
+                            block.mpegts = 0
+                        extra_state.setdefault('webvtt_mpegts_adjust', 0)
+                        block.mpegts += extra_state['webvtt_mpegts_adjust'] << 33
+                        if block.mpegts < extra_state.get('webvtt_mpegts_last', 0):
+                            overflow = True
+                            block.mpegts += 1 << 33
+                        mpegts_last = block.mpegts
+
+                        if frag_index == 1:
+                            extra_state['webvtt_mpegts'] = block.mpegts or 0
+                            extra_state['webvtt_local'] = block.local or 0
+                            # XXX: block.local = block.mpegts = None ?
+                        else:
+                            if block.mpegts is not None and block.local is not None:
+                                adjust = (
+                                    (block.mpegts - extra_state.get('webvtt_mpegts', 0))
+                                    - (block.local - extra_state.get('webvtt_local', 0))
+                                )
+                            continue
+                    elif isinstance(block, webvtt.HeaderBlock):
+                        if frag_index != 1:
+                            # XXX: this should probably be silent as well
+                            # or verify that all segments contain the same data
+                            self.report_warning(bug_reports_message(
+                                f'Discarding a {type(block).__name__} block found in the middle of the stream; '
+                                'if the subtitles display incorrectly,'))
+                            continue
+                    block.write_into(output)
+
+                return output.getvalue().encode()
+
+            def fin_fragments():
+                dedup_window = extra_state.get('webvtt_dedup_window')
+                if not dedup_window:
+                    return b''
+
+                output = io.StringIO()
+                for cue in dedup_window:
+                    webvtt.CueBlock.from_json(cue).write_into(output)
+
+                return output.getvalue().encode()
+
+            if len(fragments) == 1:
+                self.download_and_append_fragments(ctx, fragments, info_dict)
             else:
-                for fragment in fragments:
-                    frag_content, frag_index = download_fragment(fragment)
-                    result = append_fragment(frag_content, frag_index)
-                    if not result:
-                        return False
-
-            self._finish_frag_download(ctx)
-        return True
+                self.download_and_append_fragments(
+                    ctx, fragments, info_dict, pack_func=pack_fragment, finish_func=fin_fragments)
+        else:
+            return self.download_and_append_fragments(ctx, fragments, info_dict)