diff --git a/core/spotify_public_api.py b/core/spotify_public_api.py new file mode 100644 index 00000000..d0dc49c1 --- /dev/null +++ b/core/spotify_public_api.py @@ -0,0 +1,118 @@ +"""Full public-playlist fetch for the 'Spotify link' path, via the OPTIONAL +SpotipyFree library (no Spotify credentials needed). + +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. + +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. + +``client_factory`` is injectable so the orchestration is unit-testable without +the library or the network. +""" + +from __future__ import annotations + +import hashlib +import logging +from typing import Any, Callable, Dict, List, Optional + +logger = logging.getLogger('soulsync.spotify_public') + +_MAX_TRACKS = 10000 # safety cap + + +def normalize_api_track(item: Any, index: int) -> Optional[Dict[str, Any]]: + """Convert a spotipy-shape playlist item to the embed scraper's track shape. + + 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') + 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 _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, + *, + client_factory: Optional[Callable[[], Any]] = None, +) -> Dict[str, Any]: + """Pull a public playlist's FULL track list with no credentials. + + 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)() + + meta: Dict[str, Any] = {} + try: + # limit=1: we only want name/owner here — tracks come from the paginated + # playlist_items call below, so don't pull the whole list twice. + meta = client.playlist(spotify_id, limit=1) 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]] = [] + 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: + 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, + 'type': 'playlist', + 'name': name, + 'subtitle': subtitle, + 'tracks': tracks, + 'url': source_url, + 'url_hash': hashlib.md5(source_url.encode()).hexdigest()[:12], + } + + +__all__ = ['normalize_api_track', 'fetch_public_playlist_full'] diff --git a/core/spotify_public_scraper.py b/core/spotify_public_scraper.py index fa299817..1d5d653b 100644 --- a/core/spotify_public_scraper.py +++ b/core/spotify_public_scraper.py @@ -9,7 +9,8 @@ import logging import hashlib import requests -logger = logging.getLogger(__name__) +# 'soulsync.*' so these lines land in app.log (the bare module name isn't captured). +logger = logging.getLogger('soulsync.spotify_public') def parse_spotify_url(url: str) -> dict: @@ -82,10 +83,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") @@ -151,3 +160,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/requirements.txt b/requirements.txt index bc1ade54..148d3c0c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -65,3 +65,13 @@ tidalapi==0.8.11 flask-socketio==5.6.1 gunicorn==26.0.0 simple-websocket==1.1.0 + +# Full public-playlist link imports (no Spotify credentials). The "Spotify link" +# tab scrapes Spotify's embed widget, which caps at ~100 tracks; SpotipyFree +# (the no-creds spotipy drop-in spotDL uses) pulls the full list. It is GPL-3.0, +# used here as a normal pip dependency — aggregation, exactly like spotDL (also +# MIT). It is NOT vendored into SoulSync's own code, so the project stays MIT. +# The code soft-imports it and falls back to the embed scraper (~100 tracks) if +# it's missing or fails, so this is never a hard runtime requirement. +spotipyFree>=1.1.2,<2 +websockets>=12 # required by SpotipyFree/spotapi, not pulled in automatically diff --git a/tests/test_spotify_public_api.py b/tests/test_spotify_public_api.py new file mode 100644 index 00000000..e4c37487 --- /dev/null +++ b/tests/test_spotify_public_api.py @@ -0,0 +1,125 @@ +"""Full public-playlist fetch via the optional SpotipyFree library. + +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 + +import pytest + +import core.spotify_public_api as papi +import core.spotify_public_scraper as scraper + + +# -------------------------------------------------------------------------- +# Track normalisation +# -------------------------------------------------------------------------- + +def test_normalize_api_track_shape(): + item = {'track': {'id': 't1', 'name': 'Song', 'artists': [{'name': 'A'}, {'name': 'B'}], + 'duration_ms': 1000, 'explicit': True}} + 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/removed + assert papi.normalize_api_track({}, 0) is None + t = papi.normalize_api_track({'track': {'id': 'x', 'name': 'N'}}, 0) + assert t['artists'] == [{'name': 'Unknown Artist'}] # fallback + + +# -------------------------------------------------------------------------- +# Full fetch with an injected fake SpotipyFree client (spotipy-shaped) +# -------------------------------------------------------------------------- + +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, limit=-1, offset=0, *args, **kwargs): + return {'name': 'My Playlist', 'owner': {'display_name': 'Owner'}} + + 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 playlist_items(self, pid): + if self.fail_items: + raise RuntimeError('boom') + return self._page(0) + + def next(self, results): + return self._page(results['_next']) + + +def test_full_fetch_paginates_past_100(): + 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 # 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', client_factory=lambda: _FakeClient(30)) + assert len(result['tracks']) == 30 + + +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', client_factory=missing) + + +def test_full_fetch_raises_when_no_tracks(): + with pytest.raises(Exception): + papi.fetch_public_playlist_full('pl1', client_factory=lambda: _FakeClient(0)) + + +# -------------------------------------------------------------------------- +# Fallback orchestration (the safety net) — full path vs embed scraper +# -------------------------------------------------------------------------- + +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 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('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' # graceful fallback + + +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' and full_called['n'] == 0 # albums skip 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