[youtube:tab] Extract playlist availability (#504)

Authored by: colethedj

yt_dlp/extractor/youtube.py

@@ -645,6 +645,28 @@ def _report_alerts(self, alerts, expected=True):
     def _extract_and_report_alerts(self, data, *args, **kwargs):
         return self._report_alerts(self._extract_alerts(data), *args, **kwargs)
 
+    def _extract_badges(self, renderer: dict):
+        badges = set()
+        for badge in try_get(renderer, lambda x: x['badges'], list) or []:
+            label = try_get(badge, lambda x: x['metadataBadgeRenderer']['label'], compat_str)
+            if label:
+                badges.add(label.lower())
+        return badges
+
+    @staticmethod
+    def _join_text_entries(runs):
+        text = None
+        for run in runs:
+            if not isinstance(run, dict):
+                continue
+            sub_text = try_get(run, lambda x: x['text'], compat_str)
+            if sub_text:
+                if not text:
+                    text = sub_text
+                    continue
+                text += sub_text
+        return text
+
     def _extract_response(self, item_id, query, note='Downloading API JSON', headers=None,
                           ytcfg=None, check_get_keys=None, ep='browse', fatal=True, api_hostname=None,
                           default_client='WEB'):
@@ -1971,20 +1993,6 @@ def parse_time_text(time_text):
         if len(time_text_split) >= 3:
             return datetime_from_str('now-%s%s' % (time_text_split[0], time_text_split[1]), precision='auto')
 
-    @staticmethod
-    def _join_text_entries(runs):
-        text = None
-        for run in runs:
-            if not isinstance(run, dict):
-                continue
-            sub_text = try_get(run, lambda x: x['text'], compat_str)
-            if sub_text:
-                if not text:
-                    text = sub_text
-                    continue
-                text += sub_text
-        return text
-
     def _extract_comment(self, comment_renderer, parent=None):
         comment_id = comment_renderer.get('commentId')
         if not comment_id:
@@ -2959,21 +2967,20 @@ def chapter_time(mmlir):
         if initial_data and is_private is not None:
             is_membersonly = False
             is_premium = False
-            contents = try_get(initial_data, lambda x: x['contents']['twoColumnWatchNextResults']['results']['results']['contents'], list)
+            contents = try_get(initial_data, lambda x: x['contents']['twoColumnWatchNextResults']['results']['results']['contents'], list) or []
-            for content in contents or []:
+            badge_labels = set()
-                badges = try_get(content, lambda x: x['videoPrimaryInfoRenderer']['badges'], list)
+            for content in contents:
-                for badge in badges or []:
+                if not isinstance(content, dict):
-                    label = try_get(badge, lambda x: x['metadataBadgeRenderer']['label']) or ''
+                    continue
-                    if label.lower() == 'members only':
+                badge_labels.update(self._extract_badges(content.get('videoPrimaryInfoRenderer')))
-                        is_membersonly = True
+            for badge_label in badge_labels:
-                        break
+                if badge_label.lower() == 'members only':
-                    elif label.lower() == 'premium':
+                    is_membersonly = True
-                        is_premium = True
+                elif badge_label.lower() == 'premium':
-                        break
+                    is_premium = True
-                if is_membersonly or is_premium:
+                elif badge_label.lower() == 'unlisted':
-                    break
+                    is_unlisted = True
 
-        # TODO: Add this for playlists
         info['availability'] = self._availability(
             is_private=is_private,
             needs_premium=is_premium,
@@ -3447,6 +3454,17 @@ class YoutubeTabIE(YoutubeBaseInfoExtractor):
             'title': 'Album - Royalty Free Music Library V2 (50 Songs)',
         },
         'playlist_count': 50,
+    }, {
+        'note': 'unlisted single video playlist',
+        'url': 'https://www.youtube.com/playlist?list=PLwL24UFy54GrB3s2KMMfjZscDi1x5Dajf',
+        'info_dict': {
+            'uploader_id': 'UC9zHu_mHU96r19o-wV5Qs1Q',
+            'uploader': 'colethedj',
+            'id': 'PLwL24UFy54GrB3s2KMMfjZscDi1x5Dajf',
+            'title': 'yt-dlp unlisted playlist test',
+            'availability': 'unlisted'
+        },
+        'playlist_count': 1,
     }]
 
     @classmethod
@@ -3768,27 +3786,19 @@ def _extract_selected_tab(tabs):
         else:
             raise ExtractorError('Unable to find selected tab')
 
-    @staticmethod
+    @classmethod
-    def _extract_uploader(data):
+    def _extract_uploader(cls, data):
         uploader = {}
-        sidebar_renderer = try_get(
+        renderer = cls._extract_sidebar_info_renderer(data, 'playlistSidebarSecondaryInfoRenderer') or {}
-            data, lambda x: x['sidebar']['playlistSidebarRenderer']['items'], list)
+        owner = try_get(
-        if sidebar_renderer:
+            renderer, lambda x: x['videoOwner']['videoOwnerRenderer']['title']['runs'][0], dict)
-            for item in sidebar_renderer:
+        if owner:
-                if not isinstance(item, dict):
+            uploader['uploader'] = owner.get('text')
-                    continue
+            uploader['uploader_id'] = try_get(
-                renderer = item.get('playlistSidebarSecondaryInfoRenderer')
+                owner, lambda x: x['navigationEndpoint']['browseEndpoint']['browseId'], compat_str)
-                if not isinstance(renderer, dict):
+            uploader['uploader_url'] = urljoin(
-                    continue
+                'https://www.youtube.com/',
-                owner = try_get(
+                try_get(owner, lambda x: x['navigationEndpoint']['browseEndpoint']['canonicalBaseUrl'], compat_str))
-                    renderer, lambda x: x['videoOwner']['videoOwnerRenderer']['title']['runs'][0], dict)
-                if owner:
-                    uploader['uploader'] = owner.get('text')
-                    uploader['uploader_id'] = try_get(
-                        owner, lambda x: x['navigationEndpoint']['browseEndpoint']['browseId'], compat_str)
-                    uploader['uploader_url'] = urljoin(
-                        'https://www.youtube.com/',
-                        try_get(owner, lambda x: x['navigationEndpoint']['browseEndpoint']['canonicalBaseUrl'], compat_str))
         return {k: v for k, v in uploader.items() if v is not None}
 
     def _extract_from_tabs(self, item_id, webpage, data, tabs):
@@ -3814,8 +3824,8 @@ def _extract_from_tabs(self, item_id, webpage, data, tabs):
             thumbnails_list = (
                 try_get(renderer, lambda x: x['avatar']['thumbnails'], list)
                 or try_get(
-                    data,
+                    self._extract_sidebar_info_renderer(data, 'playlistSidebarPrimaryInfoRenderer'),
-                    lambda x: x['sidebar']['playlistSidebarRenderer']['items'][0]['playlistSidebarPrimaryInfoRenderer']['thumbnailRenderer']['playlistVideoThumbnailRenderer']['thumbnail']['thumbnails'],
+                    lambda x: x['thumbnailRenderer']['playlistVideoThumbnailRenderer']['thumbnail']['thumbnails'],
                     list)
                 or [])
 
@@ -3839,7 +3849,6 @@ def _extract_from_tabs(self, item_id, webpage, data, tabs):
                 or playlist_id)
         title += format_field(selected_tab, 'title', ' - %s')
         title += format_field(selected_tab, 'expandedText', ' - %s')
-
         metadata = {
             'playlist_id': playlist_id,
             'playlist_title': title,
@@ -3850,6 +3859,9 @@ def _extract_from_tabs(self, item_id, webpage, data, tabs):
             'thumbnails': thumbnails,
             'tags': tags,
         }
+        availability = self._extract_availability(data)
+        if availability:
+            metadata['availability'] = availability
         if not channel_id:
             metadata.update(self._extract_uploader(data))
         metadata.update({
@@ -3921,49 +3933,86 @@ def _extract_from_playlist(self, item_id, url, data, playlist, webpage):
             self._extract_mix_playlist(playlist, playlist_id, data, webpage),
             playlist_id=playlist_id, playlist_title=title)
 
+    def _extract_availability(self, data):
+        """
+        Gets the availability of a given playlist/tab.
+        Note: Unless YouTube tells us explicitly, we do not assume it is public
+        @param data: response
+        """
+        is_private = is_unlisted = None
+        renderer = self._extract_sidebar_info_renderer(data, 'playlistSidebarPrimaryInfoRenderer') or {}
+        badge_labels = self._extract_badges(renderer)
+
+        # Personal playlists, when authenticated, have a dropdown visibility selector instead of a badge
+        privacy_dropdown_entries = try_get(
+            renderer, lambda x: x['privacyForm']['dropdownFormFieldRenderer']['dropdown']['dropdownRenderer']['entries'], list) or []
+        for renderer_dict in privacy_dropdown_entries:
+            is_selected = try_get(
+                renderer_dict, lambda x: x['privacyDropdownItemRenderer']['isSelected'], bool) or False
+            if not is_selected:
+                continue
+            label = self._join_text_entries(
+                try_get(renderer_dict, lambda x: x['privacyDropdownItemRenderer']['label']['runs'], list) or [])
+            if label:
+                badge_labels.add(label.lower())
+                break
+
+        for badge_label in badge_labels:
+            if badge_label == 'unlisted':
+                is_unlisted = True
+            elif badge_label == 'private':
+                is_private = True
+            elif badge_label == 'public':
+                is_unlisted = is_private = False
+        return self._availability(is_private, False, False, False, is_unlisted)
+
+    @staticmethod
+    def _extract_sidebar_info_renderer(data, info_renderer, expected_type=dict):
+        sidebar_renderer = try_get(
+            data, lambda x: x['sidebar']['playlistSidebarRenderer']['items'], list) or []
+        for item in sidebar_renderer:
+            renderer = try_get(item, lambda x: x[info_renderer], expected_type)
+            if renderer:
+                return renderer
+
     def _reload_with_unavailable_videos(self, item_id, data, webpage):
         """
         Get playlist with unavailable videos if the 'show unavailable videos' button exists.
         """
-        sidebar_renderer = try_get(
-            data, lambda x: x['sidebar']['playlistSidebarRenderer']['items'], list)
-        if not sidebar_renderer:
-            return
         browse_id = params = None
-        for item in sidebar_renderer:
+        renderer = self._extract_sidebar_info_renderer(data, 'playlistSidebarPrimaryInfoRenderer')
-            if not isinstance(item, dict):
+        if not renderer:
+            return
+        menu_renderer = try_get(
+            renderer, lambda x: x['menu']['menuRenderer']['items'], list) or []
+        for menu_item in menu_renderer:
+            if not isinstance(menu_item, dict):
                 continue
-            renderer = item.get('playlistSidebarPrimaryInfoRenderer')
+            nav_item_renderer = menu_item.get('menuNavigationItemRenderer')
-            menu_renderer = try_get(
+            text = try_get(
-                renderer, lambda x: x['menu']['menuRenderer']['items'], list) or []
+                nav_item_renderer, lambda x: x['text']['simpleText'], compat_str)
-            for menu_item in menu_renderer:
+            if not text or text.lower() != 'show unavailable videos':
-                if not isinstance(menu_item, dict):
+                continue
-                    continue
+            browse_endpoint = try_get(
-                nav_item_renderer = menu_item.get('menuNavigationItemRenderer')
+                nav_item_renderer, lambda x: x['navigationEndpoint']['browseEndpoint'], dict) or {}
-                text = try_get(
+            browse_id = browse_endpoint.get('browseId')
-                    nav_item_renderer, lambda x: x['text']['simpleText'], compat_str)
+            params = browse_endpoint.get('params')
-                if not text or text.lower() != 'show unavailable videos':
+            break
-                    continue
-                browse_endpoint = try_get(
-                    nav_item_renderer, lambda x: x['navigationEndpoint']['browseEndpoint'], dict) or {}
-                browse_id = browse_endpoint.get('browseId')
-                params = browse_endpoint.get('params')
-                break
 
-            ytcfg = self._extract_ytcfg(item_id, webpage)
+        ytcfg = self._extract_ytcfg(item_id, webpage)
-            headers = self._generate_api_headers(
+        headers = self._generate_api_headers(
-                ytcfg, account_syncid=self._extract_account_syncid(ytcfg),
+            ytcfg, account_syncid=self._extract_account_syncid(ytcfg),
-                identity_token=self._extract_identity_token(webpage, item_id=item_id),
+            identity_token=self._extract_identity_token(webpage, item_id=item_id),
-                visitor_data=try_get(
+            visitor_data=try_get(
-                    self._extract_context(ytcfg), lambda x: x['client']['visitorData'], compat_str))
+                self._extract_context(ytcfg), lambda x: x['client']['visitorData'], compat_str))
-            query = {
+        query = {
-                'params': params or 'wgYCCAA=',
+            'params': params or 'wgYCCAA=',
-                'browseId': browse_id or 'VL%s' % item_id
+            'browseId': browse_id or 'VL%s' % item_id
-            }
+        }
-            return self._extract_response(
+        return self._extract_response(
-                item_id=item_id, headers=headers, query=query,
+            item_id=item_id, headers=headers, query=query,
-                check_get_keys='contents', fatal=False,
+            check_get_keys='contents', fatal=False,
-                note='Downloading API JSON with unavailable videos')
+            note='Downloading API JSON with unavailable videos')
 
     def _extract_webpage(self, url, item_id):
         retries = self.get_param('extractor_retries', 3)
@@ -4100,7 +4149,6 @@ def get_mobj(url):
         if 'no-youtube-unavailable-videos' not in compat_opts:
             data = self._reload_with_unavailable_videos(item_id, data, webpage) or data
         self._extract_and_report_alerts(data)
-
         tabs = try_get(
             data, lambda x: x['contents']['twoColumnBrowseResultsRenderer']['tabs'], list)
         if tabs: