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

Code before:
                self._API_BASE_URL + 'authentication/login', None,
                'Logging in', self._LOGIN_ERR_MESSAGE, fatal=False,
                data=urlencode_postdata({
                    'password': password,
                    'rememberMe': False,
                    'source': 'Web',
                    'username': username,
                })) or {}).get('accessToken')
            if access_token:
                self._HEADERS = {'authorization': 'Bearer ' + access_token}
        except ExtractorError as e:
            message = None
            if isinstance(e.cause, HTTPError) and e.cause.status == 401:
                resp = self._parse_json(
                    e.cause.response.read().decode(), None, fatal=False) or {}
                message = resp.get('message') or resp.get('code')
            self.report_warning(message or self._LOGIN_ERR_MESSAGE)

    def _real_extract(self, url):
        lang, video_id = self._match_valid_url(url).group('lang', 'id')
        username, password = self._get_login_info()
        # ADN requires login for German site (at least for now)
        if lang == 'de' and (not username or not password):
            self.raise_login_required(method='password')

        video_base_url = self._PLAYER_BASE_URL + 'video/%s/' % video_id
        player = self._download_json(
            video_base_url + 'configuration', video_id,
            'Downloading player config JSON metadata',
            headers=self._HEADERS)['player']
        options = player['options']

        subscription_msg = 'This video requires a subscription'
        # Ad supported videos are not yet available in German
        if lang == 'de' and 'ads' in options:
            raise ExtractorError(subscription_msg, expected=True)

        user = options['user']
        if not user.get('hasAccess'):
            if username and password:
                raise ExtractorError(subscription_msg, expected=True)

            self.raise_login_required(method='password')

        token = self._download_json(
            user.get('refreshTokenUrl') or (self._PLAYER_BASE_URL + 'refresh/token'),
            video_id, 'Downloading access token', headers={
                'X-Player-Refresh-Token': user['refreshToken'],
            }, data=b'')['token']

        links_url = try_get(options, lambda x: x['video']['url']) or (video_base_url + 'link')

Review comment (refactor):
since you said this in the linked issue:

> Currently there is no free German content, so we could just hardcode the login requirement for .de Links. But this may change once the site launches officially. There was free content a few weeks ago, so this may change at any time.

I don't think we should hardcode this behavior

We should check for the truthiness of `formats` and `self._HEADERS` after format extraction, and `raise_login_required` if both are falsy
```suggestion
        video_base_url = self._PLAYER_BASE_URL + 'video/%s/' % video_id
```

Code after:
        try:
            access_token = (self._download_json(
                self._API_BASE_URL + 'authentication/login', None,
                'Logging in', self._LOGIN_ERR_MESSAGE, fatal=False,
                data=urlencode_postdata({
                    'password': password,
                    'rememberMe': False,
                    'source': 'Web',
                    'username': username,
                })) or {}).get('accessToken')
            if access_token:
                self._HEADERS = {'authorization': 'Bearer ' + access_token}
        except ExtractorError as e:
            message = None
            if isinstance(e.cause, HTTPError) and e.cause.status == 401:
                resp = self._parse_json(
                    e.cause.response.read().decode(), None, fatal=False) or {}
                message = resp.get('message') or resp.get('code')
            self.report_warning(message or self._LOGIN_ERR_MESSAGE)

    def _real_extract(self, url):
        lang, video_id = self._match_valid_url(url).group('lang', 'id')
        video_base_url = self._PLAYER_BASE_URL + 'video/%s/' % video_id
        player = self._download_json(
            video_base_url + 'configuration', video_id,
            'Downloading player config JSON metadata',
            headers=self._HEADERS)['player']
        options = player['options']

        user = options['user']
        if not user.get('hasAccess'):
            start_date = traverse_obj(options, ('video', 'startDate', {str}))
            if (parse_iso8601(start_date) or 0) > time.time():
                raise ExtractorError(f'This video is not available yet. Release date: {start_date}', expected=True)
            self.raise_login_required('This video requires a subscription', method='password')

        token = self._download_json(
            user.get('refreshTokenUrl') or (self._PLAYER_BASE_URL + 'refresh/token'),
            video_id, 'Downloading access token', headers={
                'X-Player-Refresh-Token': user['refreshToken'],
            }, data=b'')['token']

        links_url = try_get(options, lambda x: x['video']['url']) or (video_base_url + 'link')
        self._K = ''.join(random.choices('0123456789abcdef', k=16))
        message = bytes_to_intlist(json.dumps({
            'k': self._K,
            't': token,
        }))

        # Sometimes authentication fails for no good reason, retry with
        # a different random padding