Repository: yt-dlp/yt-dlp
File: yt_dlp/extractor/youtube/_video.py (Python)

Code before:
        streaming_data = traverse_obj(player_responses, (..., 'streamingData'))
        *formats, subtitles = self._extract_formats_and_subtitles(streaming_data, video_id, player_url, live_status, duration)
        if all(f.get('has_drm') for f in formats):
            # If there are no formats that definitely don't have DRM, all have DRM
            for f in formats:
                f['has_drm'] = True

        return live_broadcast_details, live_status, streaming_data, formats, subtitles

    def _download_initial_data(self, video_id, webpage, webpage_client, webpage_ytcfg):
        initial_data = None
        if webpage and 'initial_data' not in self._configuration_arg('webpage_skip'):
            initial_data = self.extract_yt_initial_data(video_id, webpage, fatal=False)
            if not traverse_obj(initial_data, 'contents'):
                self.report_warning('Incomplete data received in embedded initial data; re-fetching using API.')
                initial_data = None
        if not initial_data and 'initial_data' not in self._configuration_arg('player_skip'):
            query = {'videoId': video_id}
            query.update(self._get_checkok_params())
            initial_data = self._extract_response(
                item_id=video_id, ep='next', fatal=False,
                ytcfg=webpage_ytcfg, query=query, check_get_keys='contents',
                note='Downloading initial data API JSON', default_client=webpage_client)
        return initial_data

    def _is_premium_subscriber(self, initial_data=None):
        if not self.is_authenticated or not initial_data:
            return False

        tlr = traverse_obj(
            initial_data, ('topbar', 'desktopTopbarRenderer', 'logo', 'topbarLogoRenderer'))
        return (
            traverse_obj(tlr, ('iconImage', 'iconType')) == 'YOUTUBE_PREMIUM_LOGO'
            or 'premium' in (self._get_text(tlr, 'tooltipText') or '').lower()
        )

    def _initial_extract(self, url, smuggled_data, webpage_url, webpage_client, video_id):
        # This function is also used by live-from-start refresh
        webpage = self._download_initial_webpage(webpage_url, webpage_client, video_id)
        webpage_ytcfg = self.extract_ytcfg(video_id, webpage) or self._get_default_ytcfg(webpage_client)

        initial_data = self._download_initial_data(video_id, webpage, webpage_client, webpage_ytcfg)

        is_premium_subscriber = self._is_premium_subscriber(webpage_ytcfg, initial_data)
        if is_premium_subscriber:
            self.write_debug('Detected YouTube Premium subscription')

        player_responses, player_url = self._extract_player_responses(
            self._get_requested_clients(url, smuggled_data, is_premium_subscriber),
            video_id, webpage, webpage_client, webpage_ytcfg, is_premium_subscriber)


Review comment (suggestion):
```suggestion
    def _is_premium_subscriber(self, initial_data):
```

Code after:
        streaming_data = traverse_obj(player_responses, (..., 'streamingData'))
        *formats, subtitles = self._extract_formats_and_subtitles(streaming_data, video_id, player_url, live_status, duration)
        if all(f.get('has_drm') for f in formats):
            # If there are no formats that definitely don't have DRM, all have DRM
            for f in formats:
                f['has_drm'] = True

        return live_broadcast_details, live_status, streaming_data, formats, subtitles

    def _download_initial_data(self, video_id, webpage, webpage_client, webpage_ytcfg):
        initial_data = None
        if webpage and 'initial_data' not in self._configuration_arg('webpage_skip'):
            initial_data = self.extract_yt_initial_data(video_id, webpage, fatal=False)
            if not traverse_obj(initial_data, 'contents'):
                self.report_warning('Incomplete data received in embedded initial data; re-fetching using API.')
                initial_data = None
        if not initial_data and 'initial_data' not in self._configuration_arg('player_skip'):
            query = {'videoId': video_id}
            query.update(self._get_checkok_params())
            initial_data = self._extract_response(
                item_id=video_id, ep='next', fatal=False,
                ytcfg=webpage_ytcfg, query=query, check_get_keys='contents',
                note='Downloading initial data API JSON', default_client=webpage_client)
        return initial_data

    def _is_premium_subscriber(self, initial_data):
        if not self.is_authenticated or not initial_data:
            return False

        tlr = traverse_obj(
            initial_data, ('topbar', 'desktopTopbarRenderer', 'logo', 'topbarLogoRenderer'))
        return (
            traverse_obj(tlr, ('iconImage', 'iconType')) == 'YOUTUBE_PREMIUM_LOGO'
            or 'premium' in (self._get_text(tlr, 'tooltipText') or '').lower()
        )

    def _initial_extract(self, url, smuggled_data, webpage_url, webpage_client, video_id):
        # This function is also used by live-from-start refresh
        webpage = self._download_initial_webpage(webpage_url, webpage_client, video_id)
        webpage_ytcfg = self.extract_ytcfg(video_id, webpage) or self._get_default_ytcfg(webpage_client)

        initial_data = self._download_initial_data(video_id, webpage, webpage_client, webpage_ytcfg)

        is_premium_subscriber = self._is_premium_subscriber(initial_data)
        if is_premium_subscriber:
            self.write_debug('Detected YouTube Premium subscription')

        player_responses, player_url = self._extract_player_responses(
            self._get_requested_clients(url, smuggled_data, is_premium_subscriber),
            video_id, webpage, webpage_client, webpage_ytcfg, is_premium_subscriber)