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

Code before:
                    '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']
        video = options['video']
        startDate = video.get('startDate')
        currentDate = video.get('currentDate')
        if startDate and currentDate and startDate > currentDate:
            raise ExtractorError(f'This video is not available yet. Release date: {startDate}', expected=True)

        user = options['user']
        if not user.get('hasAccess'):
            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
        links_data = None
        for _ in range(3):
            padded_message = intlist_to_bytes(pkcs1pad(message, 128))
            n, e = self._RSA_KEY
            encrypted_message = long_to_bytes(pow(bytes_to_long(padded_message), e, n))

Review comment (refactor):
Are these values unix timestamps? And if so, should we rely on `currentDate` instead of `time.time()`? IMO we should do this a bit more safely (and use snake_case):
```suggestion
        start_date = traverse_obj(options, ('video', 'startDate', {int}))
        current_date = traverse_obj(options, ('video', 'currentDate', {int}))
        if start_date and current_date and start_date > current_date:
            raise ExtractorError(f'This video is not available yet. Release date: {start_date}', expected=True)
```
and if these are unix timestamps, `Release date: {start_date}` is not useful IMO

Perhaps we should even check this after the `if not formats:` statement below?



Code after:
                    '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
        links_data = None
        for _ in range(3):
            padded_message = intlist_to_bytes(pkcs1pad(message, 128))
            n, e = self._RSA_KEY
            encrypted_message = long_to_bytes(pow(bytes_to_long(padded_message), e, n))