danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Full public playlists via optional SpotipyFree (no creds), MIT-clean

Browse source 
The in-house anonymous-token path is blocked by Spotify (429 without the web
player's rotating client-auth). Switch the full-fetch to SpotipyFree — the
maintained no-creds spotipy drop-in spotDL uses, which tracks that machinery.

- core/spotify_public_api.fetch_public_playlist_full now uses a SpotipyFree
  client (playlist + playlist_items + next), normalising the spotipy-shaped
  items to the embed scraper's shape. Injectable client_factory keeps it
  unit-testable without the library or network. Dropped the dead in-house
  token/pagination code.
- Licensing: SpotipyFree is GPL-3.0, so it is NOT bundled/required (SoulSync is
  MIT). Optional, user-installed: the import is soft, and on ImportError (or any
  failure) fetch_spotify_public falls back to the embed scraper (~100). So the
  shipped project stays cleanly MIT and the link path never regresses.
- requirements.txt: documents it as a commented optional extra
  (pip install SpotipyFree) with the GPL/MIT rationale.
- 9 tests: normalisation, pagination past 100, library-missing -> raises (->
  fallback), and the embed-fallback orchestration.

Needs a live click-through with SpotipyFree installed to confirm the exact
class/method names match (SpotipyFree.Spotify / playlist / playlist_items).
This commit is contained in:
BoulderBadgeDad 2026-06-02 22:43:34 -07:00
parent 951293c56a
commit 06f11dc95a
3 changed files with 115 additions and 214 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

173
core/spotify_public_api.py
View file

@ -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 Why a library: the embed scraper caps at ~100 tracks, and getting the full list
sees the ~100 tracks Spotify bakes into the embed widget — a public playlist with no login means talking to Spotify's private API the way the web player
added by link gets truncated. This module gets the *full* track list without does — including client-auth headers Spotify rotates constantly. Rather than
any app credentials by: 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 Licensing: SpotipyFree is GPL-3.0, so it is NOT bundled or required by SoulSync
``open.spotify.com/embed/playlist/{id}`` page (server-minted — nothing for (MIT). It's an OPTIONAL install — if the user has run ``pip install spotipyFree``
us to sign or maintain, unlike the rotating TOTP secret the now-dead this lights up; otherwise the import fails, this raises, and the caller
get_access_token endpoint demanded), then (``spotify_public_scraper.fetch_spotify_public``) falls back to the embed
2. paging the public Web API (`/v1/playlists/{id}/tracks`, 100 at a time) scraper (today's ≤100). So SoulSync ships zero GPL code and stays cleanly MIT.
until the whole playlist is pulled.
Every failure path raises. The only caller ``client_factory`` is injectable so the orchestration is unit-testable without
(``spotify_public_scraper.fetch_spotify_public``) catches that and falls back to the library or the network.
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 from __future__ import annotations
import hashlib import hashlib
import logging import logging
import re
from typing import Any, Callable, Dict, List, Optional 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') logger = logging.getLogger('soulsync.spotify_public')
_BROWSER_HEADERS = { _MAX_TRACKS = 10000 # safety cap
'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]]: 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 Returns None for items without a usable track id (local files, removed
episodes, removed tracks) so the caller can skip them. tracks, podcast episodes) so the caller can skip them.
""" """
track = (item or {}).get('track') or {} track = (item or {}).get('track') or {}
track_id = track.get('id') 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]]: def _default_client():
"""Pull the full track list from the Web API, 100 at a time.""" """Create a no-credentials SpotipyFree client.
headers = {'Authorization': f'Bearer {token}', **_BROWSER_HEADERS}
tracks: List[Dict[str, Any]] = [] Raises ImportError when the optional GPL-3.0 library isn't installed — the
offset = 0 caller treats that like any other failure and falls back to the embed
while offset < _MAX_TRACKS: scraper.
resp = http_get( """
f'https://api.spotify.com/v1/playlists/{spotify_id}/tracks', from SpotipyFree import Spotify # optional, user-installed (GPL-3.0)
headers=headers, return Spotify()
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 fetch_public_playlist_full( def fetch_public_playlist_full(
spotify_id: str, spotify_id: str,
*, *,
http_get: Callable = requests.get, client_factory: Optional[Callable[[], Any]] = None,
) -> Dict[str, Any]: ) -> 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 Uses a SpotipyFree client (spotipy-compatible: ``playlist`` for metadata,
tracks; the token then paginates the Web API for the whole list. If the API ``playlist_items`` + ``next`` for paginated tracks). Returns the embed
is unavailable (e.g. the anonymous token gets rate-limited / 401s), we fall scraper's shape. Raises on any failure (incl. the library not being
back to the tracks the embed page already gave us (≤100) — so this is never installed) so the caller can fall back to the embed scraper.
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 client = (client_factory or _default_client)()
drops to the embed scraper)."""
from core.spotify_public_scraper import parse_embed_html
page = http_get( meta: Dict[str, Any] = {}
f'https://open.spotify.com/embed/playlist/{spotify_id}', try:
headers=_BROWSER_HEADERS, timeout=_TIMEOUT, meta = client.playlist(spotify_id) or {}
) except Exception as e: # metadata is nice-to-have; tracks are the point
page.raise_for_status() logger.debug("playlist metadata fetch failed (%s); continuing", e)
html = page.text name = meta.get('name', 'Unknown')
subtitle = (meta.get('owner') or {}).get('display_name', '')
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))
tracks: List[Dict[str, Any]] = [] tracks: List[Dict[str, Any]] = []
if token: results = client.playlist_items(spotify_id)
try: while results:
tracks = _paginate_api_tracks(http_get, spotify_id, token) for item in results.get('items', []):
logger.info("public API pagination ok: %d tracks", len(tracks)) t = normalize_api_track(item, len(tracks))
except Exception as e: if t:
logger.warning("public API pagination failed (%s); using embed tracks (≤100)", e) tracks.append(t)
if len(tracks) >= _MAX_TRACKS:
break
if results.get('next'):
results = client.next(results)
else:
break
if not tracks: if not tracks:
tracks = embed_tracks # graceful: at least the embed's ≤100 raise RuntimeError('SpotipyFree returned no usable tracks')
if not tracks:
raise RuntimeError('no anonymous token usable and no embed tracks')
logger.info("SpotipyFree full fetch: %s (%d tracks)", name, len(tracks))
source_url = f'https://open.spotify.com/playlist/{spotify_id}' source_url = f'https://open.spotify.com/playlist/{spotify_id}'
return { return {
'id': spotify_id, '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']

9
requirements.txt
View file

@ -65,3 +65,12 @@ tidalapi==0.8.11
flask-socketio==5.6.1 flask-socketio==5.6.1
gunicorn==26.0.0 gunicorn==26.0.0
simple-websocket==1.1.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

147
tests/test_spotify_public_api.py
View file

@ -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 The library is GPL-3.0 and user-installed, so it's never imported in tests —
fetch via an injected http_get) and — most importantly — the embed fallback a fake spotipy-compatible client is injected to exercise normalisation +
orchestration, so a broken anonymous path can never make the link worse than pagination, and the embed fallback orchestration is tested separately. So a
today. missing/broken library can never make the link path worse than the embed ≤100.
""" """
from __future__ import annotations 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('<html>no token here</html>') 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(): def test_normalize_api_track_shape():
item = {'track': {'id': 't1', 'name': 'Song', 'artists': [{'name': 'A'}, {'name': 'B'}], item = {'track': {'id': 't1', 'name': 'Song', 'artists': [{'name': 'A'}, {'name': 'B'}],
'duration_ms': 1000, 'explicit': True}} 'duration_ms': 1000, 'explicit': True}}
t = papi.normalize_api_track(item, 4) assert papi.normalize_api_track(item, 4) == {
assert t == {'id': 't1', 'name': 'Song', 'artists': [{'name': 'A'}, {'name': 'B'}], 'id': 't1', 'name': 'Song', 'artists': [{'name': 'A'}, {'name': 'B'}],
'duration_ms': 1000, 'is_explicit': True, 'track_number': 5} 'duration_ms': 1000, 'is_explicit': True, 'track_number': 5,
}
def test_normalize_api_track_skips_unusable(): 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 assert papi.normalize_api_track({}, 0) is None
# missing artists -> Unknown Artist fallback
t = papi.normalize_api_track({'track': {'id': 'x', 'name': 'N'}}, 0) 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 _page(self, offset):
def __init__(self, *, text='', json_data=None, status=200): n = min(100, max(0, self.total - offset))
self.text, self._json, self.status_code = text, json_data, status 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): def playlist_items(self, pid):
if self.status_code >= 400: if self.fail_items:
import requests raise RuntimeError('boom')
raise requests.HTTPError(str(self.status_code)) return self._page(0)
def json(self): def next(self, results):
return self._json return self._page(results['_next'])
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'<html><script>var s={{{tok}"x":1}}</script>'
f'<script id="__NEXT_DATA__" type="application/json">{json.dumps(next_data)}</script>'
f'</html>')
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 test_full_fetch_paginates_past_100(): def test_full_fetch_paginates_past_100():
result = papi.fetch_public_playlist_full('pl1', http_get=_fake_http(total=250)) result = papi.fetch_public_playlist_full('pl1', client_factory=lambda: _FakeClient(250))
assert result['name'] == 'My Playlist' # from the embed page — no meta call assert result['name'] == 'My Playlist'
assert result['subtitle'] == 'Owner' 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'][0]['track_number'] == 1
assert result['tracks'][-1]['id'] == 't249' assert result['tracks'][-1]['id'] == 't249'
assert result['type'] == 'playlist' and result['id'] == 'pl1' assert result['type'] == 'playlist' and result['id'] == 'pl1'
def test_full_fetch_single_page(): 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 assert len(result['tracks']) == 30
def test_full_fetch_falls_back_to_embed_tracks_when_api_fails(): def test_full_fetch_raises_when_library_missing():
# Token works but the API 429s -> keep the embed page's tracks (<=100), # _default_client would raise ImportError; simulate via the factory.
# never worse than the embed scraper, and DON'T raise. def missing():
result = papi.fetch_public_playlist_full( raise ImportError("No module named 'SpotipyFree'")
'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():
with pytest.raises(Exception): with pytest.raises(Exception):
papi.fetch_public_playlist_full( papi.fetch_public_playlist_full('pl1', client_factory=missing)
'pl1', http_get=_fake_http(total=10, embed_tracks=0, with_token=False))
def test_full_fetch_raises_when_no_tracks_anywhere(): def test_full_fetch_raises_when_no_tracks():
with pytest.raises(Exception): with pytest.raises(Exception):
papi.fetch_public_playlist_full( papi.fetch_public_playlist_full('pl1', client_factory=lambda: _FakeClient(0))
'pl1', http_get=_fake_http(total=0, embed_tracks=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): 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', monkeypatch.setattr(scraper, 'scrape_spotify_embed',
lambda *a, **k: calls.__setitem__('embed', calls['embed'] + 1) or {'tracks': []}) lambda *a, **k: calls.__setitem__('embed', calls['embed'] + 1) or {'tracks': []})
out = scraper.fetch_spotify_public('playlist', 'pl1') out = scraper.fetch_spotify_public('playlist', 'pl1')
assert len(out['tracks']) == 200 assert len(out['tracks']) == 200 and calls['embed'] == 0 # full won, embed not called
assert calls['embed'] == 0 # full path won — embed never called
def test_fetch_public_falls_back_to_embed_on_failure(monkeypatch): def test_fetch_public_falls_back_to_embed_on_failure(monkeypatch):
def boom(pid, **kw): 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(papi, 'fetch_public_playlist_full', boom)
monkeypatch.setattr(scraper, 'scrape_spotify_embed', monkeypatch.setattr(scraper, 'scrape_spotify_embed',
lambda *a, **k: {'name': 'Embed', 'tracks': [{'id': 'e'}]}) lambda *a, **k: {'name': 'Embed', 'tracks': [{'id': 'e'}]})
out = scraper.fetch_spotify_public('playlist', 'pl1') 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): 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', monkeypatch.setattr(scraper, 'scrape_spotify_embed',
lambda *a, **k: {'name': 'Album', 'tracks': [{'id': 'x'}]}) lambda *a, **k: {'name': 'Album', 'tracks': [{'id': 'x'}]})
out = scraper.fetch_spotify_public('album', 'al1') out = scraper.fetch_spotify_public('album', 'al1')
assert out['name'] == 'Album' assert out['name'] == 'Album' and full_called['n'] == 0 # albums skip full-fetch
assert full_called['n'] == 0 # albums don't attempt the playlist full-fetch