diff --git a/core/spotify_public_api.py b/core/spotify_public_api.py new file mode 100644 index 00000000..9a8dff01 --- /dev/null +++ b/core/spotify_public_api.py @@ -0,0 +1,162 @@ +"""Anonymous full-playlist fetch for the public 'Spotify link' path. + +The embed scraper (``spotify_public_scraper.scrape_spotify_embed``) only ever +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 + 2. paging the public Web API (`/v1/playlists/{id}/tracks`, 100 at a time) + until the whole playlist is pulled. + +Every failure path raises. The only caller +(``spotify_public_scraper.fetch_spotify_public``) catches that and falls back to +the embed scraper, so the worst case is exactly today's behaviour — this never +makes the link path *worse*, only (when Spotify cooperates) better. + +This rides Spotify's undocumented page-embedded token and is expected to break +when they change their page; it degrades to the embed fallback, it does not +crash. Pure helpers (token extraction, normalisation, pagination) take an +injected ``http_get`` so they're unit-testable without the network. +""" + +from __future__ import annotations + +import hashlib +import logging +import re +from typing import Any, Callable, Dict, List, Optional + +import requests + +logger = logging.getLogger(__name__) + +_BROWSER_HEADERS = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ' + '(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', + 'Accept-Language': 'en-US,en;q=0.9', +} + +# Spotify embeds the anonymous token as "accessToken":"BQ..." in the page's +# session/config blob. +_TOKEN_RE = re.compile(r'"accessToken"\s*:\s*"([^"]+)"') + +_PAGE_LIMIT = 100 # Web API max page size +_MAX_TRACKS = 10000 # safety cap so a bad `total` can't loop forever +_TIMEOUT = 20 + + +def extract_access_token(html: str) -> Optional[str]: + """Return the anonymous accessToken embedded in a Spotify page, or None.""" + if not html: + return None + m = _TOKEN_RE.search(html) + token = m.group(1) if m else None + # A truncated/empty token isn't usable. + return token if token and len(token) > 20 else None + + +def normalize_api_track(item: Any, index: int) -> Optional[Dict[str, Any]]: + """Convert a Web API playlist item to the embed scraper's track shape. + + Returns None for items without a usable track id (local files, podcast + episodes, removed tracks) so the caller can skip them. + """ + track = (item or {}).get('track') or {} + track_id = track.get('id') + if not track_id: + return None + artists = [{'name': a.get('name', '')} for a in (track.get('artists') or []) if a.get('name')] + return { + 'id': track_id, + 'name': track.get('name', 'Unknown Track'), + 'artists': artists or [{'name': 'Unknown Artist'}], + 'duration_ms': track.get('duration_ms', 0), + 'is_explicit': bool(track.get('explicit', False)), + 'track_number': index + 1, + } + + +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) + 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( + f'https://api.spotify.com/v1/playlists/{spotify_id}/tracks', + headers=headers, + params={ + 'limit': _PAGE_LIMIT, + 'offset': offset, + 'fields': 'items(track(id,name,artists(name),duration_ms,explicit))', + }, + timeout=_TIMEOUT, + ) + page_resp.raise_for_status() + items = (page_resp.json() or {}).get('items') or [] + if not items: + break + for item in items: + t = normalize_api_track(item, len(tracks)) + if t: + tracks.append(t) + offset += _PAGE_LIMIT + if len(items) < _PAGE_LIMIT: + break + + if not tracks: + raise RuntimeError('public API returned no usable tracks') + + source_url = f'https://open.spotify.com/playlist/{spotify_id}' + return { + 'id': spotify_id, + 'type': 'playlist', + 'name': name, + 'subtitle': subtitle, + 'tracks': tracks, + 'url': source_url, + 'url_hash': hashlib.md5(source_url.encode()).hexdigest()[:12], + } + + +__all__ = ['extract_access_token', 'normalize_api_track', 'fetch_public_playlist_full'] diff --git a/core/spotify_public_scraper.py b/core/spotify_public_scraper.py index fa299817..8f149466 100644 --- a/core/spotify_public_scraper.py +++ b/core/spotify_public_scraper.py @@ -151,3 +151,31 @@ def scrape_spotify_embed(spotify_type: str, spotify_id: str) -> dict: logger.info(f"Scraped Spotify {spotify_type}: {result['name']} ({len(tracks)} tracks)") return result + + +def fetch_spotify_public(spotify_type: str, spotify_id: str) -> dict: + """Fetch a public Spotify link, preferring the FULL track list. + + Playlists are pulled via the anonymous public-API path + (``spotify_public_api.fetch_public_playlist_full``), which paginates past + the embed widget's ~100-track cap. On ANY failure — or for albums, which + the embed already returns whole — this falls back to ``scrape_spotify_embed`` + (today's behaviour). Same return shape either way, so callers don't care + which path produced the data. + """ + if spotify_type == 'playlist': + try: + from core.spotify_public_api import fetch_public_playlist_full + result = fetch_public_playlist_full(spotify_id) + if result and result.get('tracks'): + logger.info( + f"Spotify public API (full): {result.get('name')} " + f"({len(result['tracks'])} tracks)" + ) + return result + except Exception as e: + logger.info( + f"Spotify public full fetch failed ({e}); " + f"falling back to embed scraper (≤100 tracks)" + ) + return scrape_spotify_embed(spotify_type, spotify_id) diff --git a/tests/test_spotify_public_api.py b/tests/test_spotify_public_api.py new file mode 100644 index 00000000..3c2a7ed9 --- /dev/null +++ b/tests/test_spotify_public_api.py @@ -0,0 +1,144 @@ +"""Anonymous full-playlist fetch for the public 'Spotify link' path. + +Covers the testable seams (token extraction, track normalisation, paginated +fetch via an injected http_get) and — most importantly — the embed fallback +orchestration, so a broken anonymous path can never make the link worse than +today. +""" + +from __future__ import annotations + +import pytest + +import core.spotify_public_api as papi +import core.spotify_public_scraper as scraper + + +# -------------------------------------------------------------------------- +# Pure helpers +# -------------------------------------------------------------------------- + +def test_extract_access_token_found(): + html = 'window.foo={"accessToken":"BQ_abcdefghijklmnopqrstuvwxyz","other":1}' + assert papi.extract_access_token(html) == 'BQ_abcdefghijklmnopqrstuvwxyz' + + +def test_extract_access_token_absent_or_short(): + assert papi.extract_access_token('no token here') is None + assert papi.extract_access_token('') is None + assert papi.extract_access_token('{"accessToken":"short"}') is None # too short to be real + + +def test_normalize_api_track_shape(): + item = {'track': {'id': 't1', 'name': 'Song', 'artists': [{'name': 'A'}, {'name': 'B'}], + 'duration_ms': 1000, 'explicit': True}} + t = papi.normalize_api_track(item, 4) + assert t == {'id': 't1', 'name': 'Song', 'artists': [{'name': 'A'}, {'name': 'B'}], + 'duration_ms': 1000, 'is_explicit': True, 'track_number': 5} + + +def test_normalize_api_track_skips_unusable(): + assert papi.normalize_api_track({'track': {'id': None}}, 0) is None # local file / removed + assert papi.normalize_api_track({}, 0) is None + # missing artists -> Unknown Artist fallback + t = papi.normalize_api_track({'track': {'id': 'x', 'name': 'N'}}, 0) + assert t['artists'] == [{'name': 'Unknown Artist'}] + + +# -------------------------------------------------------------------------- +# Paginated fetch with injected HTTP (no network) +# -------------------------------------------------------------------------- + +class _Resp: + def __init__(self, *, text='', json_data=None, status=200): + self.text, self._json, self.status_code = text, json_data, status + + def raise_for_status(self): + if self.status_code >= 400: + import requests + raise requests.HTTPError(str(self.status_code)) + + def json(self): + return self._json + + +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 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 url.endswith('/tracks'): + 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}}) + 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['subtitle'] == 'Owner' + assert len(result['tracks']) == 250 # 3 pages (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' + + +def test_full_fetch_single_page(): + result = papi.fetch_public_playlist_full('pl1', http_get=_fake_http(total=30)) + 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_raises_when_no_tracks(): + with pytest.raises(Exception): + papi.fetch_public_playlist_full('pl1', http_get=_fake_http(total=0)) + + +# -------------------------------------------------------------------------- +# Fallback orchestration (the safety net) +# -------------------------------------------------------------------------- + +def test_fetch_public_uses_full_when_it_succeeds(monkeypatch): + calls = {'embed': 0} + monkeypatch.setattr(papi, 'fetch_public_playlist_full', + lambda pid, **kw: {'name': 'Full', 'tracks': [{'id': 'a'}] * 200}) + monkeypatch.setattr(scraper, 'scrape_spotify_embed', + lambda *a, **k: calls.__setitem__('embed', calls['embed'] + 1) or {'tracks': []}) + out = scraper.fetch_spotify_public('playlist', 'pl1') + assert len(out['tracks']) == 200 + assert calls['embed'] == 0 # full path won — embed never called + + +def test_fetch_public_falls_back_to_embed_on_failure(monkeypatch): + def boom(pid, **kw): + raise RuntimeError('spotify changed their page') + monkeypatch.setattr(papi, 'fetch_public_playlist_full', boom) + monkeypatch.setattr(scraper, 'scrape_spotify_embed', + lambda *a, **k: {'name': 'Embed', 'tracks': [{'id': 'e'}]}) + out = scraper.fetch_spotify_public('playlist', 'pl1') + assert out['name'] == 'Embed' # gracefully fell back + + +def test_fetch_public_album_uses_embed_directly(monkeypatch): + full_called = {'n': 0} + monkeypatch.setattr(papi, 'fetch_public_playlist_full', + lambda pid, **kw: full_called.__setitem__('n', 1) or {}) + monkeypatch.setattr(scraper, 'scrape_spotify_embed', + lambda *a, **k: {'name': 'Album', 'tracks': [{'id': 'x'}]}) + out = scraper.fetch_spotify_public('album', 'al1') + assert out['name'] == 'Album' + assert full_called['n'] == 0 # albums don't attempt the playlist full-fetch diff --git a/web_server.py b/web_server.py index 8d8caa8d..0e8c0d68 100644 --- a/web_server.py +++ b/web_server.py @@ -19151,11 +19151,12 @@ def get_playlist_tracks(playlist_id): time.sleep(0.5) if results is None: # Both attempts failed (often a 403 on followed playlists) — fall back - # to the public embed scraper as a last resort (capped at ~100 tracks). - logger.warning(f"Playlist items unavailable after retry ({items_err}), trying public embed scraper") + # to the no-auth public path (full track list via anonymous token, + # embed scraper if that fails). + logger.warning(f"Playlist items unavailable after retry ({items_err}), trying public fetch") try: - from core.spotify_public_scraper import scrape_spotify_embed - embed_data = scrape_spotify_embed('playlist', playlist_id) + from core.spotify_public_scraper import fetch_spotify_public + embed_data = fetch_spotify_public('playlist', playlist_id) if embed_data and not embed_data.get('error') and embed_data.get('tracks'): for t in embed_data['tracks']: artists = t.get('artists', []) @@ -22270,15 +22271,15 @@ def parse_spotify_public_endpoint(): if not url: return jsonify({"error": "Spotify URL is required"}), 400 - from core.spotify_public_scraper import parse_spotify_url, scrape_spotify_embed + from core.spotify_public_scraper import parse_spotify_url, fetch_spotify_public parsed = parse_spotify_url(url) if not parsed: return jsonify({"error": "Invalid Spotify URL. Please use a playlist or album link from open.spotify.com"}), 400 - logger.info(f"Scraping public Spotify {parsed['type']}: {parsed['id']}") + logger.info(f"Fetching public Spotify {parsed['type']}: {parsed['id']}") - result = scrape_spotify_embed(parsed['type'], parsed['id']) + result = fetch_spotify_public(parsed['type'], parsed['id']) if 'error' in result: return jsonify(result), 400