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

Code before:
        return self._download_json(
            url_or_request, video_id, note='Downloading JSON metadata' + (' [logged in]' if 'authorization' in headers else ''),
            errnote='Unable to download JSON metadata' + (' [logged in]' if 'authorization' in headers else ''),
            headers=headers, query=query)

    def _get_OAuth_tokens(self):
        PARTIAL_USER_CONSENT_URL_STEP_4_5 = urllib.parse.urlparse('https://secure.rokfin.com/auth/realms/rokfin-web/protocol/openid-connect/auth?client_id=web&redirect_uri=https%3A%2F%2Frokfin.com%2Fsilent-check-sso.html&response_mode=fragment&response_type=code&scope=openid&prompt=none')

        # ---------------------------- CODE FLOW AUTHORIZATION ----------------------------
        # https://openid.net/specs/openid-connect-core-1_0.html#CodeFlowAuth (Sec. 3.1.1)

        # Steps 4 & 5 authorize yt-dlp to act on user's behalf.

        # random_str() came from https://www.rokfin.com/static/js/2.*.chunk.js
        def random_str():
            rnd_lst = [random.choice('0123456789abcdef') for _ in range(36)]
            rnd_lst[14] = '4'
            rnd_lst[19] = '0123456789abcdef'[3 & ord(rnd_lst[19]) | 8]
            rnd_lst[8] = rnd_lst[13] = rnd_lst[18] = rnd_lst[23] = '-'
            return ''.join(rnd_lst)

        user_consent_url_step_4_5 = PARTIAL_USER_CONSENT_URL_STEP_4_5._replace(query=urllib.parse.urlencode((lambda d: d.update(state=random_str(), nonce=random_str()) or d)(dict(urllib.parse.parse_qsl(PARTIAL_USER_CONSENT_URL_STEP_4_5.query))))).geturl()

        # By making this HTTP request, the user authorizes yt-dlp to act on behalf of the user:
        urlh = (self._download_webpage_handle(
            user_consent_url_step_4_5, None, note='granting user authorization', errnote='user authorization rejected by Rokfin', fatal=False, encoding='utf-8') or (None, None))[1]

        def parse_authorization_code(urlh):
            codes = urllib.parse.parse_qs(urllib.parse.urldefrag(urlh.geturl()).fragment).get('code')
            return codes[0] if codes else None

        authorization_code = parse_authorization_code(urlh)
        if not authorization_code:
            self.write_debug('authorization failed: missing authorization code')
            return None

        # Steps 6 & 7 request and acquire ID Token & Access Token:
        access_token = self._download_json(
            self._TOKEN_DISTRIBUTION_POINT_URL_STEP_6_7, None, note='getting access credentials', fatal=False, encoding='utf-8',
            data=urlencode_postdata({'code': authorization_code, 'grant_type': 'authorization_code', 'client_id': 'web', 'redirect_uri': 'https://rokfin.com/silent-check-sso.html'})) or {}

        # Usage: --extractor-args "rokfin:print-user-info"  # For testing by those who don't have premium access.
        if self._configuration_arg('print_user_info'):
            self.to_screen(self._download_json('https://prod-api-v2.production.rokfin.com/api/v2/user/me', None, note='obtaining user info', errnote='could not obtain user info', fatal=False, headers={'authorization': access_token['token_type'] + ' ' + access_token['access_token']}))

        return access_token

    def _clear_cookies(self):
        self._downloader.cookiejar.clear, lambda f: f(domain='secure.rokfin.com')



Review comment (suggestion):
1. why is this non-fatal?
2. If you don't need the webpage content, use _request_webpage
3. _request/download_webpage accepts a `query` param. No need of the messy URL manipulations above

Code after:
    }]

    def _real_extract(self, url):
        list_id = self._match_id(url)
        return self.playlist_result(self._get_video_data(
            self._download_json(f'{_API_BASE_URL}stack/{list_id}', list_id)), list_id)


class RokfinChannelIE(RokfinPlaylistBaseIE):
    IE_NAME = 'rokfin:channel'
    IE_DESC = 'Rokfin Channels'
    _VALID_URL = r'https?://(?:www\.)?rokfin\.com/(?!((feed/?)|(discover/?)|(channels/?))$)(?P<id>[^/]+)/?$'
    _TESTS = [{
        'url': 'https://rokfin.com/TheConvoCouch',
        'playlist_mincount': 100,
        'info_dict': {
            'id': '12071-new',
            'title': 'TheConvoCouch - New',
            'description': 'md5:bb622b1bca100209b91cd685f7847f06',
        },
    }]

    _TABS = {
        'new': 'posts',
        'top': 'top',
        'videos': 'video',
        'podcasts': 'audio',
        'streams': 'stream',
        'stacks': 'stack',
    }

    def _real_initialize(self):
        self._validate_extractor_args()

    def _validate_extractor_args(self):
        requested_tabs = self._configuration_arg('tab', None)
        if requested_tabs is not None and (len(requested_tabs) > 1 or requested_tabs[0] not in self._TABS):
            raise ExtractorError(f'Invalid extractor-arg "tab". Must be one of {", ".join(self._TABS)}', expected=True)

    def _entries(self, channel_id, channel_name, tab):
        pages_total = None
        for page_n in itertools.count(0):
            if tab in ('posts', 'top'):
                data_url = f'{_API_BASE_URL}user/{channel_name}/{tab}?page={page_n}&size=50'
            else:
                data_url = f'{_API_BASE_URL}post/search/{tab}?page={page_n}&size=50&creator={channel_id}'
            metadata = self._download_json(
                data_url, channel_name,
                note=f'Downloading video metadata page {page_n + 1}{format_field(pages_total, template=" of %s")}')

            yield from self._get_video_data(metadata)