diff --git a/core/spotify_public_api.py b/core/spotify_public_api.py
index 9a8dff01..5e5d6ee3 100644
--- a/core/spotify_public_api.py
+++ b/core/spotify_public_api.py
@@ -5,10 +5,10 @@ sees the ~100 tracks Spotify bakes into the embed widget — a public playlist
added by link gets truncated. This module gets the *full* track list without
any app credentials by:
- 1. reading the anonymous web-player ``accessToken`` Spotify already embeds in
- its own ``open.spotify.com`` page HTML (server-minted — nothing for us to
- sign or maintain, unlike the rotating TOTP secret the get_access_token
- endpoint now demands), then
+ 1. reading the anonymous web-player ``accessToken`` Spotify ships in its
+ ``open.spotify.com/embed/playlist/{id}`` page (server-minted — nothing for
+ us to sign or maintain, unlike the rotating TOTP secret the now-dead
+ get_access_token endpoint demanded), then
2. paging the public Web API (`/v1/playlists/{id}/tracks`, 100 at a time)
until the whole playlist is pulled.
@@ -80,49 +80,13 @@ def normalize_api_track(item: Any, index: int) -> Optional[Dict[str, Any]]:
}
-def _get_anonymous_token(http_get: Callable, spotify_id: str) -> str:
- """Fetch the public playlist page and pull the anonymous token out of it."""
- resp = http_get(
- f'https://open.spotify.com/playlist/{spotify_id}',
- headers=_BROWSER_HEADERS, timeout=_TIMEOUT,
- )
- resp.raise_for_status()
- token = extract_access_token(resp.text)
- if not token:
- raise RuntimeError('no anonymous access token found in Spotify page')
- return token
-
-
-def fetch_public_playlist_full(
- spotify_id: str,
- *,
- http_get: Callable = requests.get,
-) -> Dict[str, Any]:
- """Pull a public playlist's FULL track list via the anonymous token.
-
- Returns the same shape as ``scrape_spotify_embed`` (id/type/name/subtitle/
- tracks/url/url_hash). Raises on any failure so the caller can fall back.
- """
- token = _get_anonymous_token(http_get, spotify_id)
+def _paginate_api_tracks(http_get: Callable, spotify_id: str, token: str) -> List[Dict[str, Any]]:
+ """Pull the full track list from the Web API, 100 at a time."""
headers = {'Authorization': f'Bearer {token}', **_BROWSER_HEADERS}
-
- meta_resp = http_get(
- f'https://api.spotify.com/v1/playlists/{spotify_id}',
- headers=headers,
- params={'fields': 'name,owner(display_name),tracks(total)'},
- timeout=_TIMEOUT,
- )
- meta_resp.raise_for_status()
- meta = meta_resp.json() or {}
- name = meta.get('name', 'Unknown')
- subtitle = (meta.get('owner') or {}).get('display_name', '')
- total = (meta.get('tracks') or {}).get('total', 0) or 0
-
tracks: List[Dict[str, Any]] = []
offset = 0
- ceiling = min(total, _MAX_TRACKS) if total else _MAX_TRACKS
- while offset < ceiling:
- page_resp = http_get(
+ while offset < _MAX_TRACKS:
+ resp = http_get(
f'https://api.spotify.com/v1/playlists/{spotify_id}/tracks',
headers=headers,
params={
@@ -132,8 +96,8 @@ def fetch_public_playlist_full(
},
timeout=_TIMEOUT,
)
- page_resp.raise_for_status()
- items = (page_resp.json() or {}).get('items') or []
+ resp.raise_for_status()
+ items = (resp.json() or {}).get('items') or []
if not items:
break
for item in items:
@@ -143,9 +107,50 @@ def fetch_public_playlist_full(
offset += _PAGE_LIMIT
if len(items) < _PAGE_LIMIT:
break
+ return tracks
+
+
+def fetch_public_playlist_full(
+ spotify_id: str,
+ *,
+ http_get: Callable = requests.get,
+) -> Dict[str, Any]:
+ """Pull a public playlist's FULL track list with no app credentials.
+
+ Single embed-page fetch yields the anonymous token + name + first-page
+ tracks; the token then paginates the Web API for the whole list. If the API
+ is unavailable (e.g. the anonymous token gets rate-limited / 401s), we fall
+ back to the tracks the embed page already gave us (≤100) — so this is never
+ worse than the embed scraper. Returns ``scrape_spotify_embed``'s shape;
+ raises only when we get neither a token nor any embed tracks (caller then
+ drops to the embed scraper)."""
+ from core.spotify_public_scraper import parse_embed_html
+
+ page = http_get(
+ f'https://open.spotify.com/embed/playlist/{spotify_id}',
+ headers=_BROWSER_HEADERS, timeout=_TIMEOUT,
+ )
+ page.raise_for_status()
+ html = page.text
+
+ token = extract_access_token(html)
+ base = parse_embed_html(html, 'playlist', spotify_id)
+ embed_ok = isinstance(base, dict) and 'error' not in base
+ name = base.get('name', 'Unknown') if embed_ok else 'Unknown'
+ subtitle = base.get('subtitle', '') if embed_ok else ''
+ embed_tracks = base.get('tracks', []) if embed_ok else []
+
+ tracks: List[Dict[str, Any]] = []
+ if token:
+ try:
+ tracks = _paginate_api_tracks(http_get, spotify_id, token)
+ except Exception as e:
+ logger.info("Public API pagination failed (%s); using embed tracks", e)
if not tracks:
- raise RuntimeError('public API returned no usable tracks')
+ tracks = embed_tracks # graceful: at least the embed's ≤100
+ if not tracks:
+ raise RuntimeError('no anonymous token usable and no embed tracks')
source_url = f'https://open.spotify.com/playlist/{spotify_id}'
return {
diff --git a/core/spotify_public_scraper.py b/core/spotify_public_scraper.py
index 8f149466..f822075e 100644
--- a/core/spotify_public_scraper.py
+++ b/core/spotify_public_scraper.py
@@ -82,10 +82,18 @@ def scrape_spotify_embed(spotify_type: str, spotify_id: str) -> dict:
logger.error(f"Failed to fetch Spotify embed: {e}")
return {'error': f'Failed to fetch Spotify page: {str(e)}'}
+ return parse_embed_html(response.text, spotify_type, spotify_id)
+
+
+def parse_embed_html(html: str, spotify_type: str, spotify_id: str) -> dict:
+ """Parse a Spotify embed page's ``__NEXT_DATA__`` into the standard result
+ shape. Shared by the embed scraper and the full public fetch, which pulls
+ the name + first-page tracks from the same embed page it reads the token
+ from (so it needs no extra metadata request)."""
# Extract __NEXT_DATA__ JSON
match = re.search(
r'',
- response.text
+ html
)
if not match:
logger.error("No __NEXT_DATA__ found in Spotify embed response")
diff --git a/tests/test_spotify_public_api.py b/tests/test_spotify_public_api.py
index 3c2a7ed9..57b107ed 100644
--- a/tests/test_spotify_public_api.py
+++ b/tests/test_spotify_public_api.py
@@ -49,6 +49,9 @@ def test_normalize_api_track_skips_unusable():
# Paginated fetch with injected HTTP (no network)
# --------------------------------------------------------------------------
+import json
+
+
class _Resp:
def __init__(self, *, text='', json_data=None, status=200):
self.text, self._json, self.status_code = text, json_data, status
@@ -62,32 +65,50 @@ class _Resp:
return self._json
+def _embed_html(token='BQ_aaaaaaaaaaaaaaaaaaaaaaaa', *, name='My Playlist',
+ subtitle='Owner', n_tracks=3, with_token=True):
+ """Fake embed page: a token blob + a __NEXT_DATA__ the scraper parses."""
+ track_list = [
+ {'uri': f'spotify:track:e{i}', 'title': f'Embed {i}', 'subtitle': 'Artist',
+ 'duration': 1000, 'isExplicit': False}
+ for i in range(n_tracks)
+ ]
+ next_data = {'props': {'pageProps': {'state': {'data': {'entity': {
+ 'type': 'playlist', 'name': name, 'subtitle': subtitle, 'trackList': track_list,
+ }}}}}}
+ tok = f'"accessToken":"{token}",' if with_token else ''
+ return (f''
+ f''
+ f'')
+
+
def _make_items(start, count):
return [{'track': {'id': f't{start + i}', 'name': f'Song {start + i}',
'artists': [{'name': 'Artist'}], 'duration_ms': 1000, 'explicit': False}}
for i in range(count)]
-def _fake_http(*, total, token='BQ_aaaaaaaaaaaaaaaaaaaaaaaa', no_token=False):
- """Build an http_get that serves the page, playlist meta, and track pages."""
+def _fake_http(*, total, embed_tracks=3, with_token=True, api_fail=False):
+ """http_get serving the embed page (token + name + first tracks) and API pages.
+ There is no metadata call anymore — name comes from the embed page."""
def http(url, headers=None, params=None, timeout=None):
- if 'open.spotify.com/playlist/' in url:
- return _Resp(text='' if no_token else f'x={{"accessToken":"{token}"}}')
+ if 'open.spotify.com/embed/' in url:
+ return _Resp(text=_embed_html(n_tracks=embed_tracks, with_token=with_token))
if url.endswith('/tracks'):
+ if api_fail:
+ return _Resp(status=429)
offset = params['offset']
remaining = max(0, total - offset)
return _Resp(json_data={'items': _make_items(offset, min(100, remaining))})
- # playlist meta
- return _Resp(json_data={'name': 'My Playlist', 'owner': {'display_name': 'Owner'},
- 'tracks': {'total': total}})
+ raise AssertionError(f'unexpected URL (no meta call expected): {url}')
return http
def test_full_fetch_paginates_past_100():
result = papi.fetch_public_playlist_full('pl1', http_get=_fake_http(total=250))
- assert result['name'] == 'My Playlist'
+ assert result['name'] == 'My Playlist' # from the embed page — no meta call
assert result['subtitle'] == 'Owner'
- assert len(result['tracks']) == 250 # 3 pages (100+100+50), not capped at 100
+ assert len(result['tracks']) == 250 # API: 100+100+50, not capped at 100
assert result['tracks'][0]['track_number'] == 1
assert result['tracks'][-1]['id'] == 't249'
assert result['type'] == 'playlist' and result['id'] == 'pl1'
@@ -98,14 +119,25 @@ def test_full_fetch_single_page():
assert len(result['tracks']) == 30
-def test_full_fetch_raises_without_token():
- with pytest.raises(Exception):
- papi.fetch_public_playlist_full('pl1', http_get=_fake_http(total=10, no_token=True))
+def test_full_fetch_falls_back_to_embed_tracks_when_api_fails():
+ # Token works but the API 429s -> keep the embed page's tracks (<=100),
+ # never worse than the embed scraper, and DON'T raise.
+ result = papi.fetch_public_playlist_full(
+ 'pl1', http_get=_fake_http(total=250, embed_tracks=100, api_fail=True))
+ assert len(result['tracks']) == 100
+ assert result['name'] == 'My Playlist'
-def test_full_fetch_raises_when_no_tracks():
+def test_full_fetch_raises_without_token_and_no_embed_tracks():
with pytest.raises(Exception):
- papi.fetch_public_playlist_full('pl1', http_get=_fake_http(total=0))
+ papi.fetch_public_playlist_full(
+ 'pl1', http_get=_fake_http(total=10, embed_tracks=0, with_token=False))
+
+
+def test_full_fetch_raises_when_no_tracks_anywhere():
+ with pytest.raises(Exception):
+ papi.fetch_public_playlist_full(
+ 'pl1', http_get=_fake_http(total=0, embed_tracks=0))
# --------------------------------------------------------------------------