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