diff --git a/core/spotify_public_api.py b/core/spotify_public_api.py index f2447c88..49a7c504 100644 --- a/core/spotify_public_api.py +++ b/core/spotify_public_api.py @@ -1,70 +1,39 @@ -"""Anonymous full-playlist fetch for the public 'Spotify link' path. +"""Full public-playlist fetch for the 'Spotify link' path, via the OPTIONAL +SpotipyFree library (no Spotify credentials needed). -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: +Why a library: the embed scraper caps at ~100 tracks, and getting the full list +with no login means talking to Spotify's private API the way the web player +does — including client-auth headers Spotify rotates constantly. Rather than +chase those ourselves (we tried; Spotify 429s the bare token), we lean on +SpotipyFree — the maintained no-creds ``spotipy`` drop-in that spotDL uses, +which tracks those rotating bits for us. - 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. +Licensing: SpotipyFree is GPL-3.0, so it is NOT bundled or required by SoulSync +(MIT). It's an OPTIONAL install — if the user has run ``pip install spotipyFree`` +this lights up; otherwise the import fails, this raises, and the caller +(``spotify_public_scraper.fetch_spotify_public``) falls back to the embed +scraper (today's ≤100). So SoulSync ships zero GPL code and stays cleanly MIT. -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. +``client_factory`` is injectable so the orchestration is unit-testable without +the library or the network. """ from __future__ import annotations import hashlib import logging -import re from typing import Any, Callable, Dict, List, Optional -import requests - -# 'soulsync.*' so these lines land in app.log (the bare module name isn't captured). logger = logging.getLogger('soulsync.spotify_public') -_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 +_MAX_TRACKS = 10000 # safety cap 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. + """Convert a spotipy-shape 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. + Returns None for items without a usable track id (local files, removed + tracks, podcast episodes) so the caller can skip them. """ track = (item or {}).get('track') or {} track_id = track.get('id') @@ -81,83 +50,57 @@ def normalize_api_track(item: Any, index: int) -> Optional[Dict[str, Any]]: } -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} - tracks: List[Dict[str, Any]] = [] - offset = 0 - while offset < _MAX_TRACKS: - 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, - ) - if resp.status_code != 200: - logger.warning("public API tracks offset=%s -> HTTP %s", offset, resp.status_code) - resp.raise_for_status() - items = (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 - return tracks +def _default_client(): + """Create a no-credentials SpotipyFree client. + + Raises ImportError when the optional GPL-3.0 library isn't installed — the + caller treats that like any other failure and falls back to the embed + scraper. + """ + from SpotipyFree import Spotify # optional, user-installed (GPL-3.0) + return Spotify() def fetch_public_playlist_full( spotify_id: str, *, - http_get: Callable = requests.get, + client_factory: Optional[Callable[[], Any]] = None, ) -> Dict[str, Any]: - """Pull a public playlist's FULL track list with no app credentials. + """Pull a public playlist's FULL track list with no 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 + Uses a SpotipyFree client (spotipy-compatible: ``playlist`` for metadata, + ``playlist_items`` + ``next`` for paginated tracks). Returns the embed + scraper's shape. Raises on any failure (incl. the library not being + installed) so the caller can fall back to the embed scraper. + """ + client = (client_factory or _default_client)() - 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 [] - logger.info("public fetch %s: token=%s, embed_parsed=%s, embed_tracks=%d", - spotify_id, 'yes' if token else 'NO', embed_ok, len(embed_tracks)) + meta: Dict[str, Any] = {} + try: + meta = client.playlist(spotify_id) or {} + except Exception as e: # metadata is nice-to-have; tracks are the point + logger.debug("playlist metadata fetch failed (%s); continuing", e) + name = meta.get('name', 'Unknown') + subtitle = (meta.get('owner') or {}).get('display_name', '') tracks: List[Dict[str, Any]] = [] - if token: - try: - tracks = _paginate_api_tracks(http_get, spotify_id, token) - logger.info("public API pagination ok: %d tracks", len(tracks)) - except Exception as e: - logger.warning("public API pagination failed (%s); using embed tracks (≤100)", e) + results = client.playlist_items(spotify_id) + while results: + for item in results.get('items', []): + t = normalize_api_track(item, len(tracks)) + if t: + tracks.append(t) + if len(tracks) >= _MAX_TRACKS: + break + if results.get('next'): + results = client.next(results) + else: + break if not tracks: - tracks = embed_tracks # graceful: at least the embed's ≤100 - if not tracks: - raise RuntimeError('no anonymous token usable and no embed tracks') + raise RuntimeError('SpotipyFree returned no usable tracks') + logger.info("SpotipyFree full fetch: %s (%d tracks)", name, len(tracks)) source_url = f'https://open.spotify.com/playlist/{spotify_id}' return { 'id': spotify_id, @@ -170,4 +113,4 @@ def fetch_public_playlist_full( } -__all__ = ['extract_access_token', 'normalize_api_track', 'fetch_public_playlist_full'] +__all__ = ['normalize_api_track', 'fetch_public_playlist_full'] diff --git a/requirements.txt b/requirements.txt index bc1ade54..49a405f6 100644 --- a/requirements.txt +++ b/requirements.txt @@ -65,3 +65,12 @@ tidalapi==0.8.11 flask-socketio==5.6.1 gunicorn==26.0.0 simple-websocket==1.1.0 + +# ── Optional: full public-playlist link imports (no Spotify credentials) ── +# The "Spotify link" tab scrapes Spotify's embed widget, which caps at ~100 +# tracks. SpotipyFree pulls the full list with no login. It is GPL-3.0, so it is +# intentionally NOT installed by default (SoulSync is MIT — bundling GPL would +# force the project to GPL). To enable full link imports, install it yourself: +# pip install SpotipyFree +# Without it, the link tab simply falls back to the ~100-track embed result. +# spotipyFree>=1.1.2,<2 diff --git a/tests/test_spotify_public_api.py b/tests/test_spotify_public_api.py index 57b107ed..6629aabc 100644 --- a/tests/test_spotify_public_api.py +++ b/tests/test_spotify_public_api.py @@ -1,9 +1,9 @@ -"""Anonymous full-playlist fetch for the public 'Spotify link' path. +"""Full public-playlist fetch via the optional SpotipyFree library. -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. +The library is GPL-3.0 and user-installed, so it's never imported in tests — +a fake spotipy-compatible client is injected to exercise normalisation + +pagination, and the embed fallback orchestration is tested separately. So a +missing/broken library can never make the link path worse than the embed ≤100. """ from __future__ import annotations @@ -15,133 +15,84 @@ import core.spotify_public_scraper as scraper # -------------------------------------------------------------------------- -# Pure helpers +# Track normalisation # -------------------------------------------------------------------------- -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} + assert papi.normalize_api_track(item, 4) == { + '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({'track': {'id': None}}, 0) is None # local/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'}] + assert t['artists'] == [{'name': 'Unknown Artist'}] # fallback # -------------------------------------------------------------------------- -# Paginated fetch with injected HTTP (no network) +# Full fetch with an injected fake SpotipyFree client (spotipy-shaped) # -------------------------------------------------------------------------- -import json +class _FakeClient: + """Minimal spotipy-compatible client: playlist() + playlist_items() + next().""" + def __init__(self, total, *, fail_items=False): + self.total, self.fail_items = total, fail_items + def playlist(self, pid): + return {'name': 'My Playlist', 'owner': {'display_name': 'Owner'}} -class _Resp: - def __init__(self, *, text='', json_data=None, status=200): - self.text, self._json, self.status_code = text, json_data, status + def _page(self, offset): + n = min(100, max(0, self.total - offset)) + items = [{'track': {'id': f't{offset + i}', 'name': f'S{offset + i}', + 'artists': [{'name': 'A'}], 'duration_ms': 1000, 'explicit': False}} + for i in range(n)] + nxt = offset + 100 + return {'items': items, 'next': ('u' if nxt < self.total else None), '_next': nxt} - def raise_for_status(self): - if self.status_code >= 400: - import requests - raise requests.HTTPError(str(self.status_code)) + def playlist_items(self, pid): + if self.fail_items: + raise RuntimeError('boom') + return self._page(0) - def json(self): - 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, 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/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))}) - raise AssertionError(f'unexpected URL (no meta call expected): {url}') - return http + def next(self, results): + return self._page(results['_next']) 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' # from the embed page — no meta call + result = papi.fetch_public_playlist_full('pl1', client_factory=lambda: _FakeClient(250)) + assert result['name'] == 'My Playlist' assert result['subtitle'] == 'Owner' - assert len(result['tracks']) == 250 # API: 100+100+50, not capped at 100 + assert len(result['tracks']) == 250 # 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)) + result = papi.fetch_public_playlist_full('pl1', client_factory=lambda: _FakeClient(30)) assert len(result['tracks']) == 30 -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_without_token_and_no_embed_tracks(): +def test_full_fetch_raises_when_library_missing(): + # _default_client would raise ImportError; simulate via the factory. + def missing(): + raise ImportError("No module named 'SpotipyFree'") with pytest.raises(Exception): - papi.fetch_public_playlist_full( - 'pl1', http_get=_fake_http(total=10, embed_tracks=0, with_token=False)) + papi.fetch_public_playlist_full('pl1', client_factory=missing) -def test_full_fetch_raises_when_no_tracks_anywhere(): +def test_full_fetch_raises_when_no_tracks(): with pytest.raises(Exception): - papi.fetch_public_playlist_full( - 'pl1', http_get=_fake_http(total=0, embed_tracks=0)) + papi.fetch_public_playlist_full('pl1', client_factory=lambda: _FakeClient(0)) # -------------------------------------------------------------------------- -# Fallback orchestration (the safety net) +# Fallback orchestration (the safety net) — full path vs embed scraper # -------------------------------------------------------------------------- def test_fetch_public_uses_full_when_it_succeeds(monkeypatch): @@ -151,18 +102,17 @@ def test_fetch_public_uses_full_when_it_succeeds(monkeypatch): 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 + assert len(out['tracks']) == 200 and calls['embed'] == 0 # full won, embed not called def test_fetch_public_falls_back_to_embed_on_failure(monkeypatch): def boom(pid, **kw): - raise RuntimeError('spotify changed their page') + raise RuntimeError('library not installed / spotify changed') 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 + assert out['name'] == 'Embed' # graceful fallback def test_fetch_public_album_uses_embed_directly(monkeypatch): @@ -172,5 +122,4 @@ def test_fetch_public_album_uses_embed_directly(monkeypatch): 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 + assert out['name'] == 'Album' and full_called['n'] == 0 # albums skip full-fetch