minutia

yt_dlp/extractor/kukululive.py

@@ -4,9 +4,12 @@ from .common import InfoExtractor
 from ..utils import (
     ExtractorError,
     clean_html,
+    filter_dict,
     get_element_by_id,
     int_or_none,
+    join_nonempty,
     js_to_json,
+    qualities,
     url_or_none,
     urljoin,
 )
@@ -44,27 +47,22 @@ class KukuluLiveIE(InfoExtractor):
         'only_matching': True,
     }]
 
-    def _get_quality_meta(self, video_id, desc, code, force_h264=''):
-        description = desc if force_h264 == '' else f'{desc} (force_h264)'
+    def _get_quality_meta(self, video_id, desc, code, force_h264=None):
+        desc += ' (force_h264)' if force_h264 else ''
         qs = self._download_webpage(
             'https://live.erinn.biz/live.player.fplayer.php', video_id,
-            query={
+            f'Downloading {desc} quality metadata', f'Unable to download {desc} quality metadata',
+            query=filter_dict({
                 'hash': video_id,
                 'action': f'get{code}liveByAjax',
                 'force_h264': force_h264,
-            },
-            note=f'Downloading {description} quality metadata',
-            errnote=f'Unable to download {description} quality metadata')
+            }))
         return urllib.parse.parse_qs(qs)
 
     def _add_quality_formats(self, formats, quality_meta):
-        vcodec = traverse_obj(quality_meta, ('vcodec', 0))
-        quality = traverse_obj(quality_meta, ('now_quality', 0))
-        quality_priority = {
-            'high': 3,
-            'h264': 2,
-            'low': 1,
-        }.get(quality, 0)
+        vcodec = traverse_obj(quality_meta, ('vcodec', 0, {str}))
+        quality = traverse_obj(quality_meta, ('now_quality', 0, {str}))
+        quality_priority = qualities(('low', 'h264', 'high'))(quality)
         if traverse_obj(quality_meta, ('hlsaddr', 0, {url_or_none})):
             formats.append({
                 'format_id': quality,
@@ -75,7 +73,7 @@ class KukuluLiveIE(InfoExtractor):
             })
         if traverse_obj(quality_meta, ('hlsaddr_audioonly', 0, {url_or_none})):
             formats.append({
-                'format_id': f'{quality}-audioonly',
+                'format_id': join_nonempty(quality, 'audioonly'),
                 'url': quality_meta['hlsaddr_audioonly'][0],
                 'ext': 'm4a',
                 'vcodec': 'none',
@@ -86,25 +84,22 @@ class KukuluLiveIE(InfoExtractor):
         video_id = self._match_id(url)
         html = self._download_webpage(url, video_id)
 
-        title = clean_html(get_element_by_id('livetitle', html.replace('<SPAN', '<span').replace('SPAN>', 'span>')))
+        if '>タイムシフトが見つかりませんでした。<' in html:
+            raise ExtractorError('This stream has expired', expected=True)
+
+        title = clean_html(
+            get_element_by_id('livetitle', html.replace('<SPAN', '<span').replace('SPAN>', 'span>')))
         description = self._html_search_meta('Description', html)
         thumbnail = self._html_search_meta(['og:image', 'twitter:image'], html)
 
-        is_live_stream = 'var timeshift = false;' in html
-        is_vod = 'var timeshift = true;' in html
-
-        if is_live_stream:
-            qualities = [
-                ('high', 'Z'),
-                ('low', 'ForceLow'),
-            ]
+        if self._search_regex(r'(var\s+timeshift\s*=\s*false)', html, 'is livestream', default=False):
             formats = []
-            for (desc, code) in qualities:
+            for (desc, code) in [('high', 'Z'), ('low', 'ForceLow')]:
                 quality_meta = self._get_quality_meta(video_id, desc, code)
                 self._add_quality_formats(formats, quality_meta)
-                if desc == 'high' and quality_meta.get('vcodec')[0] == 'HEVC':
-                    h264_meta = self._get_quality_meta(video_id, desc, code, force_h264='1')
-                    self._add_quality_formats(formats, h264_meta)
+                if desc == 'high' and traverse_obj(quality_meta, ('vcodec', 0)) == 'HEVC':
+                    self._add_quality_formats(
+                        formats, self._get_quality_meta(video_id, desc, code, force_h264='1'))
 
             return {
                 'id': video_id,
@@ -115,42 +110,31 @@ class KukuluLiveIE(InfoExtractor):
                 'formats': formats,
             }
 
-        if is_vod:
+        # VOD extraction
         player_html = self._download_webpage(
             'https://live.erinn.biz/live.timeshift.fplayer.php', video_id,
-                query={'hash': video_id},
-                note='Downloading player html',
-                errnote='Unable to download player html')
+            'Downloading player html', 'Unable to download player html', query={'hash': video_id})
 
-            sources_json = self._search_json(
+        sources = traverse_obj(self._search_json(
             r'var\s+fplayer_source\s*=', player_html, 'stream data', video_id,
-                contains_pattern=r'\[(?s:.+)\]', transform_source=js_to_json)
+            contains_pattern=r'\[(?s:.+)\]', transform_source=js_to_json), lambda _, v: v['file'])
 
-            def parse_segment(segment, segment_id, segment_title):
-                path = segment.get('file')
-                if not path:
-                    return None
-                formats = [{
-                    'url': urljoin('https://live.erinn.biz', path),
-                    'ext': 'mp4',
-                    'protocol': 'm3u8_native',
-                }]
-                return {
-                    'id': segment_id,
-                    'title': segment_title,
+        def entries(segments, playlist=True):
+            for i, segment in enumerate(segments, 1):
+                yield {
+                    'id': f'{video_id}_{i}' if playlist else video_id,
+                    'title': f'{title} (Part {i})' if playlist else title,
                     'description': description,
                     'timestamp': traverse_obj(segment, ('time_start', {int_or_none})),
                     'thumbnail': thumbnail,
-                    'formats': formats,
+                    'formats': [{
+                        'url': urljoin('https://live.erinn.biz', segment['file']),
+                        'ext': 'mp4',
+                        'protocol': 'm3u8_native',
+                    }],
                 }
 
-            if len(sources_json) == 1:
-                return parse_segment(sources_json[0], video_id, title)
+        if len(sources) == 1:
+            return next(entries(sources, playlist=False))
 
-            entries = []
-            for i, segment in enumerate(sources_json):
-                if entry := parse_segment(segment, f'{video_id}_{i}', f'{title} (Part {i + 1})'):
-                    entries.append(entry)
-            return self.playlist_result(entries, video_id, title, description, multi_video=True)
-
-        raise ExtractorError('Could not detect media type')
+        return self.playlist_result(entries(sources), video_id, title, description, multi_video=True)