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

Code before:

        videos = self._download_json(
            'https://amp-api.music.apple.com/v1/catalog/us/uploaded-videos',
            video_id, headers={
                'Authorization': f'Bearer {jwt}',
                'Origin': self._BASE_URL,
            }, query={'ids': video_id, 'l': 'en-US'})
        attributes = traverse_obj(videos, (
            'data', ..., 'attributes', any, {require('video information')}))

        formats = []
        quality = qualities(list(self._QUALITIES.keys()))
        for format_id, src_url in traverse_obj(attributes, (
            'assetTokens', {dict.items}, lambda _, v: url_or_none(v[1]),
        )):
            formats.append({
                'ext': 'm4v',
                'format_id': format_id,
                'height': self._QUALITIES.get(format_id),
                'quality': quality(format_id),
                'url': src_url,
                **parse_resolution(update_url(src_url, query=None), lenient=True),
            })

        return {
            'id': video_id,
            'formats': formats,
            'thumbnail': self._html_search_meta(
                ['og:image', 'og:image:secure_url', 'twitter:image'], webpage),
            **traverse_obj(attributes, {
                'title': ('name', {str}),
                'duration': ('durationInMilliseconds', {float_or_none(scale=1000)}),
                'upload_date': ('uploadDate', {unified_strdate}),
                'uploader': (('artistName', 'uploadingArtistName'), {str}, any),
                'webpage_url': ('postUrl', {url_or_none}),
            }),
        }


Review comment (security):
Since the token is long-lasting IIRC (at least a few weeks, right?), I think we should cache it (to disk, even).

What do you think about:
```py
    _CACHE_SECTION = 'appleconnect'
    _JWT_CACHE_KEY = 'jwt'
    _jwt = None

    @staticmethod
    def _jwt_is_expired(token):
        return jwt_decode_hs256(token)['exp'] - time.time() < 120

    def _get_token(self, webpage, video_id):
        if not self._jwt:
            self._jwt = self.cache.load(self._CACHE_SECTION, self._JWT_CACHE_KEY)
        if self._jwt and not self._jwt_is_expired(self._jwt):
            return self._jwt

        js_url = traverse_obj(webpage, (
            {find_element(tag='script', attr='crossorigin', value='', html=True)},
            {extract_attributes}, 'src', {urljoin(self._BASE_URL)}, {require('JS URL')}))
        js = self._download_webpage(
            js_url, video_id, 'Downloading token JS', 'Unable to download token JS')

        header = jwt_encode({}, '', headers={'alg': 'ES256', 'kid': 'WebPlayKid'}).split('.')[0]
        self._jwt = self._search_regex(
            fr'(["\'])(?P<jwt>{header}(?:\.[\w-]+){{2}})\1', js, 'JSON Web Token', group='jwt')
        if self._jwt_is_expired(self._jwt):
            raise ExtractorError('The newly fetched token is already expired')

        self.cache.store(self._CACHE_SECTION, self._JWT_CACHE_KEY, self._jwt)
        return self._jwt
```

and then:

```suggestion
        videos = self._download_json(
            'https://amp-api.music.apple.com/v1/catalog/us/uploaded-videos',
            video_id, headers={
                'Authorization': f'Bearer {self._get_token(webpage, video_id)}',
                'Origin': self._BASE_URL,
            }, query={'ids': video_id, 'l': 'en-US'})
```

Code after:
            return self._jwt

        js_url = traverse_obj(webpage, (
            {find_element(tag='script', attr='crossorigin', value='', html=True)},
            {extract_attributes}, 'src', {urljoin(self._BASE_URL)}, {require('JS URL')}))
        js = self._download_webpage(
            js_url, video_id, 'Downloading token JS', 'Unable to download token JS')

        header = jwt_encode({}, '', headers={'alg': 'ES256', 'kid': 'WebPlayKid'}).split('.')[0]
        self._jwt = self._search_regex(
            fr'(["\'])(?P<jwt>{header}(?:\.[\w-]+){{2}})\1', js, 'JSON Web Token', group='jwt')
        if self._jwt_is_expired(self._jwt):
            raise ExtractorError('The fetched token is already expired')

        return self._jwt

    def _real_extract(self, url):
        video_id = self._match_id(url)
        webpage = self._download_webpage(url, video_id)

        videos = self._download_json(
            'https://amp-api.music.apple.com/v1/catalog/us/uploaded-videos',
            video_id, headers={
                'Authorization': f'Bearer {self._get_token(webpage, video_id)}',
                'Origin': self._BASE_URL,
            }, query={'ids': video_id, 'l': 'en-US'})
        attributes = traverse_obj(videos, (
            'data', ..., 'attributes', any, {require('video information')}))

        formats = []
        quality = qualities(list(self._QUALITIES.keys()))
        for format_id, src_url in traverse_obj(attributes, (
            'assetTokens', {dict.items}, lambda _, v: url_or_none(v[1]),
        )):
            formats.append({
                'ext': 'm4v',
                'format_id': format_id,
                'height': self._QUALITIES.get(format_id),
                'quality': quality(format_id),
                'url': src_url,
                **parse_resolution(update_url(src_url, query=None), lenient=True),
            })

        return {
            'id': video_id,
            'formats': formats,
            'thumbnail': self._html_search_meta(
                ['og:image', 'og:image:secure_url', 'twitter:image'], webpage),
            **traverse_obj(attributes, {
                'title': ('name', {str}),
                'duration': ('durationInMilliseconds', {float_or_none(scale=1000)}),