yt_dlp/extractor/spotify.py

   1 import functools
   2 import json
   3 import re
   4
   5 from .common import InfoExtractor
   6 from ..utils import (
   7     OnDemandPagedList,
   8     clean_podcast_url,
   9     float_or_none,
  10     int_or_none,
  11     strip_or_none,
  12     traverse_obj,
  13     try_get,
  14     unified_strdate,
  15 )
  16
  17
  18 class SpotifyBaseIE(InfoExtractor):
  19     _ACCESS_TOKEN = None
  20     _OPERATION_HASHES = {
  21         'Episode': '8276d4423d709ae9b68ec1b74cc047ba0f7479059a37820be730f125189ac2bf',
  22         'MinimalShow': '13ee079672fad3f858ea45a55eb109553b4fb0969ed793185b2e34cbb6ee7cc0',
  23         'ShowEpisodes': 'e0e5ce27bd7748d2c59b4d44ba245a8992a05be75d6fabc3b20753fc8857444d',
  24     }
  25     _VALID_URL_TEMPL = r'https?://open\.spotify\.com/(?:embed-podcast/|embed/|)%s/(?P<id>[^/?&#]+)'
  26     _EMBED_REGEX = [r'<iframe[^>]+src="(?P<url>https?://open\.spotify.com/embed/[^"]+)"']
  27
  28     def _real_initialize(self):
  29         self._ACCESS_TOKEN = self._download_json(
  30             'https://open.spotify.com/get_access_token', None)['accessToken']
  31
  32     def _call_api(self, operation, video_id, variables, **kwargs):
  33         return self._download_json(
  34             'https://api-partner.spotify.com/pathfinder/v1/query', video_id, query={
  35                 'operationName': 'query' + operation,
  36                 'variables': json.dumps(variables),
  37                 'extensions': json.dumps({
  38                     'persistedQuery': {
  39                         'sha256Hash': self._OPERATION_HASHES[operation],
  40                     },
  41                 })
  42             }, headers={'authorization': 'Bearer ' + self._ACCESS_TOKEN},
  43             **kwargs)['data']
  44
  45     def _extract_episode(self, episode, series):
  46         episode_id = episode['id']
  47         title = episode['name'].strip()
  48
  49         formats = []
  50         audio_preview = episode.get('audioPreview') or {}
  51         audio_preview_url = audio_preview.get('url')
  52         if audio_preview_url:
  53             f = {
  54                 'url': audio_preview_url.replace('://p.scdn.co/mp3-preview/', '://anon-podcast.scdn.co/'),
  55                 'vcodec': 'none',
  56             }
  57             audio_preview_format = audio_preview.get('format')
  58             if audio_preview_format:
  59                 f['format_id'] = audio_preview_format
  60                 mobj = re.match(r'([0-9A-Z]{3})_(?:[A-Z]+_)?(\d+)', audio_preview_format)
  61                 if mobj:
  62                     f.update({
  63                         'abr': int(mobj.group(2)),
  64                         'ext': mobj.group(1).lower(),
  65                     })
  66             formats.append(f)
  67
  68         for item in (try_get(episode, lambda x: x['audio']['items']) or []):
  69             item_url = item.get('url')
  70             if not (item_url and item.get('externallyHosted')):
  71                 continue
  72             formats.append({
  73                 'url': clean_podcast_url(item_url),
  74                 'vcodec': 'none',
  75             })
  76
  77         thumbnails = []
  78         for source in (try_get(episode, lambda x: x['coverArt']['sources']) or []):
  79             source_url = source.get('url')
  80             if not source_url:
  81                 continue
  82             thumbnails.append({
  83                 'url': source_url,
  84                 'width': int_or_none(source.get('width')),
  85                 'height': int_or_none(source.get('height')),
  86             })
  87
  88         return {
  89             'id': episode_id,
  90             'title': title,
  91             'formats': formats,
  92             'thumbnails': thumbnails,
  93             'description': strip_or_none(episode.get('description')),
  94             'duration': float_or_none(try_get(
  95                 episode, lambda x: x['duration']['totalMilliseconds']), 1000),
  96             'release_date': unified_strdate(try_get(
  97                 episode, lambda x: x['releaseDate']['isoString'])),
  98             'series': series,
  99         }
 100
 101
 102 class SpotifyIE(SpotifyBaseIE):
 103     IE_NAME = 'spotify'
 104     IE_DESC = 'Spotify episodes'
 105     _VALID_URL = SpotifyBaseIE._VALID_URL_TEMPL % 'episode'
 106     _TESTS = [{
 107         'url': 'https://open.spotify.com/episode/4Z7GAJ50bgctf6uclHlWKo',
 108         'md5': '74010a1e3fa4d9e1ab3aa7ad14e42d3b',
 109         'info_dict': {
 110             'id': '4Z7GAJ50bgctf6uclHlWKo',
 111             'ext': 'mp3',
 112             'title': 'From the archive: Why time management is ruining our lives',
 113             'description': 'md5:b120d9c4ff4135b42aa9b6d9cde86935',
 114             'duration': 2083.605,
 115             'release_date': '20201217',
 116             'series': "The Guardian's Audio Long Reads",
 117         }
 118     }, {
 119         'url': 'https://open.spotify.com/embed/episode/4TvCsKKs2thXmarHigWvXE?si=7eatS8AbQb6RxqO2raIuWA',
 120         'only_matching': True,
 121     }]
 122
 123     def _real_extract(self, url):
 124         episode_id = self._match_id(url)
 125         episode = self._call_api('Episode', episode_id, {
 126             'uri': 'spotify:episode:' + episode_id
 127         })['episode']
 128         return self._extract_episode(
 129             episode, try_get(episode, lambda x: x['podcast']['name']))
 130
 131
 132 class SpotifyShowIE(SpotifyBaseIE):
 133     IE_NAME = 'spotify:show'
 134     IE_DESC = 'Spotify shows'
 135     _VALID_URL = SpotifyBaseIE._VALID_URL_TEMPL % 'show'
 136     _TEST = {
 137         'url': 'https://open.spotify.com/show/4PM9Ke6l66IRNpottHKV9M',
 138         'info_dict': {
 139             'id': '4PM9Ke6l66IRNpottHKV9M',
 140             'title': 'The Story from the Guardian',
 141             'description': 'The Story podcast is dedicated to our finest audio documentaries, investigations and long form stories',
 142         },
 143         'playlist_mincount': 36,
 144     }
 145     _PER_PAGE = 100
 146
 147     def _fetch_page(self, show_id, page=0):
 148         return self._call_api('ShowEpisodes', show_id, {
 149             'limit': 100,
 150             'offset': page * self._PER_PAGE,
 151             'uri': f'spotify:show:{show_id}',
 152         }, note=f'Downloading page {page + 1} JSON metadata')['podcast']
 153
 154     def _real_extract(self, url):
 155         show_id = self._match_id(url)
 156         first_page = self._fetch_page(show_id)
 157
 158         def _entries(page):
 159             podcast = self._fetch_page(show_id, page) if page else first_page
 160             yield from map(
 161                 functools.partial(self._extract_episode, series=podcast.get('name')),
 162                 traverse_obj(podcast, ('episodes', 'items', ..., 'episode')))
 163
 164         return self.playlist_result(
 165             OnDemandPagedList(_entries, self._PER_PAGE),
 166             show_id, first_page.get('name'), first_page.get('description'))