From dd7f0483864ee189637a63dcdff822e8fc49be0b Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Tue, 2 Jun 2026 21:27:06 -0700 Subject: [PATCH 1/5] Full public playlist fetch for the 'Spotify link' path (no creds), embed fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The no-auth 'add by link' path scrapes Spotify's embed widget, which only ever contains ~100 tracks and can't paginate — so big public playlists got truncated. This adds an in-house anonymous fetch that pulls the FULL list: - core/spotify_public_api.py: reads the anonymous web-player accessToken Spotify already embeds in its own open.spotify.com page HTML (no app credentials, and no rotating TOTP secret for us to maintain), then paginates /v1/playlists/{id}/tracks 100 at a time until the whole playlist is pulled. Returns the embed scraper's exact shape. Pure helpers + injected http_get so it's unit-testable without the network. - core/spotify_public_scraper.fetch_spotify_public(): tries the full fetch for playlists; on ANY failure (or for albums) falls back to scrape_spotify_embed. Worst case == today's behaviour, so the link path can't regress. - web_server: the link-tab endpoint and the authed flow's last-resort scrape now both go through fetch_spotify_public. Scoped entirely to the spotify_public_* (no-auth) path — the authenticated playlist sync is untouched. 11 tests (token extraction, normalisation, pagination past 100, and the embed-fallback orchestration). Caveat: rides Spotify's undocumented page-embedded token — expected to break when they change their page; it degrades to the embed fallback, never crashes. Needs a live click-through to confirm the token path works end to end (can't hit Spotify from the test env). --- core/spotify_public_api.py | 162 +++++++++++++++++++++++++++++++ core/spotify_public_scraper.py | 28 ++++++ tests/test_spotify_public_api.py | 144 +++++++++++++++++++++++++++ web_server.py | 15 +-- 4 files changed, 342 insertions(+), 7 deletions(-) create mode 100644 core/spotify_public_api.py create mode 100644 tests/test_spotify_public_api.py 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 From 8b060ee79a3b5731e63d6b46baa9c53f736f40da Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Tue, 2 Jun 2026 21:59:56 -0700 Subject: [PATCH 2/5] Fix: pull anonymous token from the EMBED page; drop meta call; graceful partial MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live debugging the 'shows 100' report: - The full playlist page no longer embeds an accessToken, and get_access_token / server-time now 403/404. The EMBED page (open.spotify.com/embed/playlist/{id}) still ships a usable anonymous token. Was fetching the wrong page -> no token -> raised -> embed fallback (100). Now reads the embed page for the token. - Confirmed live: token extraction + embed parse work; the token is accepted by the Web API (429 rate-limit, not 401). Could not show >100 from here because the test IP got rate-limited from probing; needs a clean-IP click-through. While in there, made it more robust against the rate-limiting that's clearly in play: - Refactored scrape_spotify_embed -> reusable parse_embed_html. - fetch_public_playlist_full now does ONE embed fetch for token + name + first page (no separate metadata call = fewer requests = less 429 surface), then paginates the API. If the API is unavailable/rate-limited, it keeps the embed page's tracks (<=100) instead of raising — so the result is always >= today's behaviour, never worse. - 12 tests incl. the new API-fails-but-embed-tracks-survive path. Caveat unchanged: rides Spotify's undocumented embed-page token; degrades to the embed fallback, never crashes. --- core/spotify_public_api.py | 99 +++++++++++++++++--------------- core/spotify_public_scraper.py | 10 +++- tests/test_spotify_public_api.py | 60 ++++++++++++++----- 3 files changed, 107 insertions(+), 62 deletions(-) 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)) # -------------------------------------------------------------------------- From 951293c56a3b208e58b751cd001561010e6fc206 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Tue, 2 Jun 2026 22:09:51 -0700 Subject: [PATCH 3/5] Diagnostics: route public-fetch logs to soulsync namespace + log HTTP status The full-fetch's logs used a bare module logger that app.log doesn't capture, so we couldn't see whether the API path succeeded or why it fell back. Route them to 'soulsync.spotify_public' and log: token found?, embed parsed?, the API HTTP status on a non-200, and pagination result. Lets us see the exact failure (e.g. 401 vs 429) on the next link-tab test. --- core/spotify_public_api.py | 10 ++++++++-- core/spotify_public_scraper.py | 3 ++- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/core/spotify_public_api.py b/core/spotify_public_api.py index 5e5d6ee3..f2447c88 100644 --- a/core/spotify_public_api.py +++ b/core/spotify_public_api.py @@ -32,7 +32,8 @@ from typing import Any, Callable, Dict, List, Optional 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') _BROWSER_HEADERS = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ' @@ -96,6 +97,8 @@ def _paginate_api_tracks(http_get: Callable, spotify_id: str, token: str) -> Lis }, 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: @@ -139,13 +142,16 @@ def fetch_public_playlist_full( 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]] = [] 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.info("Public API pagination failed (%s); using embed tracks", e) + logger.warning("public API pagination failed (%s); using embed tracks (≤100)", e) if not tracks: tracks = embed_tracks # graceful: at least the embed's ≤100 diff --git a/core/spotify_public_scraper.py b/core/spotify_public_scraper.py index f822075e..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: From 06f11dc95a1e14e4b3f0756349e2516911a9b2c0 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Tue, 2 Jun 2026 22:43:34 -0700 Subject: [PATCH 4/5] Full public playlists via optional SpotipyFree (no creds), MIT-clean MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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). --- core/spotify_public_api.py | 173 +++++++++++-------------------- requirements.txt | 9 ++ tests/test_spotify_public_api.py | 147 +++++++++----------------- 3 files changed, 115 insertions(+), 214 deletions(-) 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 From 77b8d7dd1f653349689109d2908d939735589a6e Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Tue, 2 Jun 2026 22:50:04 -0700 Subject: [PATCH 5/5] SpotipyFree integration confirmed working (236 tracks live); deps + meta tweak MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Verified end-to-end: fetch_public_playlist_full pulled all 236 tracks of the test playlist via SpotipyFree (the library handles the client-auth that 429'd the raw approach). Name + tracks correct. - requirements.txt: declare spotipyFree>=1.1.2 as a normal pip dependency (like spotDL, also MIT — aggregation, not vendored) + websockets (a transitive dep SpotipyFree/spotapi needs that pip doesn't pull automatically). Code still soft-imports + falls back to embed, so it's never a hard runtime requirement. - meta fetch uses limit=1 (name/owner only) so we don't pull the whole list twice. 9 tests green. --- core/spotify_public_api.py | 4 +++- requirements.txt | 17 +++++++++-------- tests/test_spotify_public_api.py | 2 +- 3 files changed, 13 insertions(+), 10 deletions(-) diff --git a/core/spotify_public_api.py b/core/spotify_public_api.py index 49a7c504..d0dc49c1 100644 --- a/core/spotify_public_api.py +++ b/core/spotify_public_api.py @@ -77,7 +77,9 @@ def fetch_public_playlist_full( meta: Dict[str, Any] = {} try: - meta = client.playlist(spotify_id) or {} + # 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') diff --git a/requirements.txt b/requirements.txt index 49a405f6..148d3c0c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -66,11 +66,12 @@ 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 +# 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 index 6629aabc..e4c37487 100644 --- a/tests/test_spotify_public_api.py +++ b/tests/test_spotify_public_api.py @@ -43,7 +43,7 @@ class _FakeClient: def __init__(self, total, *, fail_items=False): self.total, self.fail_items = total, fail_items - def playlist(self, pid): + def playlist(self, pid, limit=-1, offset=0, *args, **kwargs): return {'name': 'My Playlist', 'owner': {'display_name': 'Owner'}} def _page(self, offset):