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

Code before:
            self._download_json(
                self._TOKEN_DISTRIBUTION_POINT_URL_STEP_6_7, video_id, note='Restoring lost authorization',
                errnote='Failed to restore authorization', fatal=False,
                data=urlencode_postdata({'grant_type': 'refresh_token', 'refresh_token': self._access_mgmt_tokens.get('refresh_token'), 'client_id': 'web'})) or {})
        self.write_debug(f'Updated tokens: {self._access_mgmt_tokens.keys()}')
        if {'access_token', 'expires_in', 'refresh_expires_in', 'refresh_token', 'token_type', 'id_token', 'not-before-policy', 'session_state', 'scope'} - self._access_mgmt_tokens.keys():
            username, password = self._get_login_info()
            self._logout()
            self._perform_login(username, password)
        authorization_hdr_val = try_get(self._access_mgmt_tokens, lambda tokens: tokens['token_type'] + ' ' + tokens['access_token'])
        if authorization_hdr_val:
            headers['authorization'] = authorization_hdr_val
        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',

Review comment (suggestion):
```suggestion
        # From https://www.rokfin.com/static/js/2.*.chunk.js
```

Code after:


class RokfinStackIE(RokfinPlaylistBaseIE):
    IE_NAME = 'rokfin:stack'
    IE_DESC = 'Rokfin Stacks'
    _VALID_URL = r'https?://(?:www\.)?rokfin\.com/stack/(?P<id>[^/]+)'
    _TESTS = [{
        'url': 'https://www.rokfin.com/stack/271/Tulsi-Gabbard-Portsmouth-Townhall-FULL--Feb-9-2020',
        'playlist_count': 8,
        'info_dict': {
            'id': '271',
        },
    }]

    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)