Merge pull request #779 from Nezreka/feature/spotify-public-full-playlist

Feature/spotify public full playlist

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']

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'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>',
-        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)

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

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

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).
+            # to the no-auth public path (full track list via anonymous token,
-            logger.warning(f"Playlist items unavailable after retry ({items_err}), trying public embed scraper")
+            # 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
+                from core.spotify_public_scraper import fetch_spotify_public
-                embed_data = scrape_spotify_embed('playlist', playlist_id)
+                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