diff --git a/core/playlists/sources/__init__.py b/core/playlists/sources/__init__.py new file mode 100644 index 00000000..a087f7d3 --- /dev/null +++ b/core/playlists/sources/__init__.py @@ -0,0 +1,32 @@ +"""Unified playlist-source abstraction. + +Phase 0 of the Discover-to-Sync unification. Each external playlist +provider (Spotify, Tidal, Qobuz, YouTube, Spotify public, iTunes link, +ListenBrainz, Last.fm radio, SoulSync Discovery) gets an adapter that +exposes the same ``PlaylistSource`` Protocol, so callers no longer have +to branch on ``source`` string with an if/elif chain. + +The existing client modules are left untouched — adapters wrap them. +Once every caller speaks the unified interface, the legacy dispatch +sites (``refresh_mirrored.py`` etc.) collapse to a registry lookup. +""" + +from core.playlists.sources.base import ( + PlaylistDetail, + PlaylistMeta, + PlaylistSource, + NormalizedTrack, +) +from core.playlists.sources.registry import ( + PlaylistSourceRegistry, + get_registry, +) + +__all__ = [ + "PlaylistDetail", + "PlaylistMeta", + "PlaylistSource", + "NormalizedTrack", + "PlaylistSourceRegistry", + "get_registry", +] diff --git a/core/playlists/sources/base.py b/core/playlists/sources/base.py new file mode 100644 index 00000000..fb39aa24 --- /dev/null +++ b/core/playlists/sources/base.py @@ -0,0 +1,151 @@ +"""PlaylistSource Protocol + normalized data containers. + +These dataclasses define the *single* shape every adapter must return. +The legacy backing clients each return slightly different dicts / +dataclasses; the adapter's job is to project those into ``PlaylistMeta`` +and ``NormalizedTrack`` so callers don't have to know which source they +got the data from. + +Two distinct shapes: + +- ``PlaylistMeta``: cheap, lightweight — used for "list playlists for a + tab" responses. No tracks. +- ``PlaylistDetail``: meta + full normalized track list. Used after the + user selects a playlist to mirror. + +Discovery flag: + +- ``NormalizedTrack.needs_discovery`` is True for sources that return + raw metadata only (ListenBrainz, Last.fm radio) — the caller must run + the match step before the track is usable in the download pipeline. + Sources that already carry a provider ID (Spotify, Tidal, etc.) set + this to False. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Protocol, runtime_checkable + + +# Canonical source identifiers used as the key in mirrored_playlists.source +# and in the registry. Centralized so a typo in one place doesn't silently +# create a new "source". +SOURCE_SPOTIFY = "spotify" +SOURCE_SPOTIFY_PUBLIC = "spotify_public" +SOURCE_TIDAL = "tidal" +SOURCE_QOBUZ = "qobuz" +SOURCE_YOUTUBE = "youtube" +SOURCE_ITUNES_LINK = "itunes_link" +SOURCE_LISTENBRAINZ = "listenbrainz" +SOURCE_LASTFM = "lastfm" +SOURCE_SOULSYNC_DISCOVERY = "soulsync_discovery" + +ALL_SOURCES = ( + SOURCE_SPOTIFY, + SOURCE_SPOTIFY_PUBLIC, + SOURCE_TIDAL, + SOURCE_QOBUZ, + SOURCE_YOUTUBE, + SOURCE_ITUNES_LINK, + SOURCE_LISTENBRAINZ, + SOURCE_LASTFM, + SOURCE_SOULSYNC_DISCOVERY, +) + + +@dataclass +class PlaylistMeta: + """Lightweight playlist descriptor — no tracks.""" + + source: str + source_playlist_id: str + name: str + track_count: int = 0 + owner: Optional[str] = None + description: Optional[str] = None + image_url: Optional[str] = None + # Original URL for URL-backed sources (youtube, spotify_public, + # itunes_link). Used by the refresh path to re-fetch. + source_url: Optional[str] = None + # Free-form per-source passthrough — adapter can stash whatever the + # native API returned for downstream consumers that need richer data + # (e.g. ListenBrainz creator/MBID, Spotify snapshot_id). + extra: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class NormalizedTrack: + """A single track in normalized shape. + + ``source_track_id`` is the native ID at the source — Spotify track + ID, Tidal ID, YouTube video ID, ListenBrainz recording MBID, etc. + Empty string is allowed for sources that don't have a stable per- + track ID (rare). + """ + + position: int + track_name: str + artist_name: str + album_name: Optional[str] = None + duration_ms: int = 0 + source_track_id: Optional[str] = None + image_url: Optional[str] = None + # True when the track needs a discovery / match step before it can be + # downloaded (e.g. ListenBrainz returns MB recording metadata only — + # no Spotify/iTunes ID, so the matching engine has to run first). + needs_discovery: bool = False + # Passthrough for source-specific extras (explicit flag, popularity, + # external_urls, recording_mbid, etc.). Adapters decide what to stash. + extra: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class PlaylistDetail: + """Full playlist payload — meta + tracks.""" + + meta: PlaylistMeta + tracks: List[NormalizedTrack] = field(default_factory=list) + + +@runtime_checkable +class PlaylistSource(Protocol): + """Contract every playlist source adapter implements. + + Capability flags let callers query the adapter's shape before + invoking it (e.g. ``supports_listing=False`` for URL-only sources + means the Sync page should render a paste-URL input instead of a + playlist picker). + """ + + name: str + supports_listing: bool + supports_refresh: bool + requires_auth: bool + + def is_authenticated(self) -> bool: + """Return True if the adapter can currently call its backend. + + For sources without auth (YouTube, Spotify public, iTunes link), + this is always True. For sources where auth check is expensive, + the adapter may cache (existing clients already do this).""" + + def list_playlists(self) -> List[PlaylistMeta]: + """Return all playlists the user has access to. + + For ``supports_listing=False`` sources, return ``[]`` and let + the caller use ``get_playlist`` with a URL/ID directly.""" + + def get_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + """Fetch full playlist (meta + tracks) by source-native ID. + + For URL-backed sources, ``playlist_id`` is the full URL. For ID- + backed sources it's the native ID string. Returns ``None`` if + the playlist isn't reachable (404, auth failure, etc.).""" + + def refresh_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + """Re-fetch a playlist for the auto-refresh pipeline. + + Default behavior is identical to ``get_playlist``. Sources whose + refresh has side effects (e.g. ListenBrainz cache update, + SoulSync Discovery regeneration) override this.""" diff --git a/core/playlists/sources/itunes_link.py b/core/playlists/sources/itunes_link.py new file mode 100644 index 00000000..0f663165 --- /dev/null +++ b/core/playlists/sources/itunes_link.py @@ -0,0 +1,107 @@ +"""iTunes / Apple Music link playlist source adapter. + +Wraps the iTunes-link parsing logic that currently lives in +``web_server.py`` (commit 718eb0cb). Phase 0 doesn't move it — adapter +takes a parser callable that returns the parsed-playlist dict matching +the ``parse_itunes_link_endpoint`` response shape:: + + { + 'id', 'type', 'name', 'subtitle', 'url', 'url_hash', + 'track_count', 'image_url', + 'tracks': [{ 'id', 'name', 'artists', 'album', 'duration_ms', ... }], + } + +``supports_listing=False`` — Apple Music has no "user library" listing, +the user pastes a URL. +""" + +from __future__ import annotations + +from typing import Any, Callable, List, Optional + +from core.playlists.sources.base import ( + NormalizedTrack, + PlaylistDetail, + PlaylistMeta, + PlaylistSource, + SOURCE_ITUNES_LINK, +) + + +class ITunesLinkPlaylistSource(PlaylistSource): + name = SOURCE_ITUNES_LINK + supports_listing = False + supports_refresh = True + requires_auth = False + + def __init__(self, parser: Callable[[str], Optional[dict]]): + """``parser(url)`` returns the parsed playlist dict, or ``None`` + if the URL is invalid / unreachable. Injected by ``web_server.py`` + at startup, pointing at the existing module-level helpers.""" + self._parser = parser + + def is_authenticated(self) -> bool: + return True + + def list_playlists(self) -> List[PlaylistMeta]: + return [] + + def get_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + """``playlist_id`` is the full Apple Music / iTunes URL.""" + data = self._parser(playlist_id) + if not data: + return None + + source_url = data.get("url") or playlist_id + tracks_raw = data.get("tracks") or [] + + meta = PlaylistMeta( + source=self.name, + source_playlist_id=str(data.get("url_hash") or data.get("id") or ""), + name=data.get("name", "Apple Music Link"), + owner=data.get("subtitle"), + image_url=data.get("image_url") or None, + track_count=int(data.get("track_count", len(tracks_raw))), + source_url=source_url, + extra={ + "itunes_type": data.get("type"), + "itunes_id": data.get("id"), + }, + ) + + tracks = [self._track_from_itunes(t, idx) for idx, t in enumerate(tracks_raw) if t] + return PlaylistDetail(meta=meta, tracks=tracks) + + def refresh_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + return self.get_playlist(playlist_id) + + # ---- projection helpers ------------------------------------------------ + + def _track_from_itunes(self, track: dict, position: int) -> NormalizedTrack: + artists = track.get("artists") or [] + if artists and isinstance(artists[0], dict): + artist_name = ", ".join(a.get("name", "") for a in artists if a.get("name")) + else: + artist_name = ", ".join(str(a) for a in artists if a) + if not artist_name: + artist_name = "Unknown Artist" + album = track.get("album") + album_name: Optional[str] = None + if isinstance(album, dict): + album_name = album.get("name") or None + elif album: + album_name = str(album) + return NormalizedTrack( + position=position, + track_name=track.get("name", "Unknown Track"), + artist_name=artist_name, + album_name=album_name, + duration_ms=int(track.get("duration_ms", 0) or 0), + source_track_id=str(track.get("id", "")), + image_url=track.get("image_url"), + needs_discovery=False, + extra={ + "external_urls": track.get("external_urls"), + "preview_url": track.get("preview_url"), + }, + ) diff --git a/core/playlists/sources/lastfm.py b/core/playlists/sources/lastfm.py new file mode 100644 index 00000000..b5697f6b --- /dev/null +++ b/core/playlists/sources/lastfm.py @@ -0,0 +1,117 @@ +"""Last.fm radio playlist source adapter. + +Last.fm radio playlists are persisted by +``ListenBrainzManager.save_lastfm_radio_playlist`` under +``playlist_type='lastfm_radio'`` in the ``listenbrainz_playlists`` +table — they share the same storage as ListenBrainz playlists but +originate from Last.fm's similar-tracks API. + +Like ListenBrainz, tracks are MB metadata only, so ``needs_discovery`` +is True on every track. +""" + +from __future__ import annotations + +from typing import Any, Callable, Dict, List, Optional + +from core.playlists.sources.base import ( + NormalizedTrack, + PlaylistDetail, + PlaylistMeta, + PlaylistSource, + SOURCE_LASTFM, +) + + +LASTFM_PLAYLIST_TYPE = "lastfm_radio" + + +class LastFMPlaylistSource(PlaylistSource): + name = SOURCE_LASTFM + supports_listing = True + supports_refresh = False # Refresh requires re-running the radio generator + requires_auth = True + + def __init__(self, manager_getter: Callable[[], Any]): + """``manager_getter`` returns the profile's ``ListenBrainzManager`` + (Last.fm radio playlists share that storage layer).""" + self._manager_getter = manager_getter + + def _manager(self): + return self._manager_getter() + + def is_authenticated(self) -> bool: + # Last.fm radio rows exist independently of any auth state — + # they're persisted snapshots. Treat "manager exists" as enough. + return self._manager() is not None + + def list_playlists(self) -> List[PlaylistMeta]: + manager = self._manager() + if manager is None: + return [] + try: + rows = manager.get_cached_playlists(LASTFM_PLAYLIST_TYPE) or [] + except Exception: + rows = [] + return [self._meta_from_cache_row(r) for r in rows] + + def get_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + manager = self._manager() + if manager is None: + return None + try: + rows = manager.get_cached_playlists(LASTFM_PLAYLIST_TYPE) or [] + except Exception: + rows = [] + meta_row = next( + (r for r in rows if str(r.get("playlist_mbid")) == str(playlist_id)), + None, + ) + if meta_row is None: + return None + try: + tracks_raw = manager.get_cached_tracks(playlist_id) or [] + except Exception: + tracks_raw = [] + meta = self._meta_from_cache_row(meta_row) + meta.track_count = len(tracks_raw) + tracks = [self._track_from_cache_row(t, idx) for idx, t in enumerate(tracks_raw)] + return PlaylistDetail(meta=meta, tracks=tracks) + + def refresh_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + # Regenerating a Last.fm radio playlist needs the seed track + + # the Last.fm client — that lives outside this adapter. Phase 1 + # wires up the regeneration; Phase 0 just returns the current + # snapshot. + return self.get_playlist(playlist_id) + + # ---- projection helpers ------------------------------------------------ + + def _meta_from_cache_row(self, row: Dict[str, Any]) -> PlaylistMeta: + return PlaylistMeta( + source=self.name, + source_playlist_id=str(row.get("playlist_mbid", "")), + name=row.get("title") or "Last.fm Radio", + owner=row.get("creator") or "Last.fm", + track_count=int(row.get("track_count", 0) or 0), + extra={ + "annotation": row.get("annotation") or {}, + "last_updated": row.get("last_updated"), + }, + ) + + def _track_from_cache_row(self, row: Dict[str, Any], position: int) -> NormalizedTrack: + return NormalizedTrack( + position=position, + track_name=row.get("track_name", "Unknown Track"), + artist_name=row.get("artist_name", "Unknown Artist"), + album_name=row.get("album_name") or None, + duration_ms=int(row.get("duration_ms", 0) or 0), + source_track_id=row.get("recording_mbid") or None, + image_url=row.get("album_cover_url") or None, + needs_discovery=True, + extra={ + "recording_mbid": row.get("recording_mbid"), + "release_mbid": row.get("release_mbid"), + }, + ) diff --git a/core/playlists/sources/listenbrainz.py b/core/playlists/sources/listenbrainz.py new file mode 100644 index 00000000..861045eb --- /dev/null +++ b/core/playlists/sources/listenbrainz.py @@ -0,0 +1,155 @@ +"""ListenBrainz playlist source adapter. + +Wraps ``core.listenbrainz_manager.ListenBrainzManager``. ListenBrainz +playlists carry only MusicBrainz recording metadata — no Spotify / +iTunes IDs — so every track returned by this adapter has +``needs_discovery=True``. Phase 1+ will route those through the +existing ``run_listenbrainz_discovery_worker`` and persist the matched +provider IDs into ``mirrored_playlist_tracks.extra_data``. + +Construction takes a manager getter callable because the manager is +profile-scoped (one instance per profile, built from credentials stored +in the DB) — there is no process-wide singleton to grab at import time. +""" + +from __future__ import annotations + +from typing import Any, Callable, Dict, List, Optional + +from core.playlists.sources.base import ( + NormalizedTrack, + PlaylistDetail, + PlaylistMeta, + PlaylistSource, + SOURCE_LISTENBRAINZ, +) + + +class ListenBrainzPlaylistSource(PlaylistSource): + name = SOURCE_LISTENBRAINZ + supports_listing = True + supports_refresh = True + requires_auth = True + + # ListenBrainz manager caches three "playlist types" — surface all + # three under this source. The Sync page can group / filter by + # ``meta.extra['playlist_type']`` if it wants per-type sub-tabs. + PLAYLIST_TYPES = ("created_for_user", "user_created", "collaborative") + + def __init__(self, manager_getter: Callable[[], Any]): + """``manager_getter`` returns a live ``ListenBrainzManager`` for + the current profile. ``None`` is allowed and means "no LB + configured" — adapter degrades to empty results.""" + self._manager_getter = manager_getter + + def _manager(self): + return self._manager_getter() + + def is_authenticated(self) -> bool: + manager = self._manager() + if manager is None: + return False + client = getattr(manager, "client", None) + if client is None or not hasattr(client, "is_authenticated"): + return False + return bool(client.is_authenticated()) + + def list_playlists(self) -> List[PlaylistMeta]: + manager = self._manager() + if manager is None: + return [] + out: List[PlaylistMeta] = [] + for ptype in self.PLAYLIST_TYPES: + try: + rows = manager.get_cached_playlists(ptype) or [] + except Exception: + rows = [] + for row in rows: + out.append(self._meta_from_cache_row(row, ptype)) + return out + + def get_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + """``playlist_id`` is the ListenBrainz playlist MBID.""" + manager = self._manager() + if manager is None: + return None + ptype = "" + try: + ptype = manager.get_playlist_type(playlist_id) or "" + except Exception: + ptype = "" + + cached_rows = [] + try: + cached_rows = manager.get_cached_playlists(ptype) if ptype else [] + except Exception: + cached_rows = [] + meta_row = next( + (r for r in cached_rows if str(r.get("playlist_mbid")) == str(playlist_id)), + None, + ) + + try: + tracks_raw = manager.get_cached_tracks(playlist_id) or [] + except Exception: + tracks_raw = [] + + if meta_row is None and not tracks_raw: + return None + + meta = self._meta_from_cache_row( + meta_row or {"playlist_mbid": playlist_id, "track_count": len(tracks_raw)}, + ptype or "listenbrainz", + ) + meta.track_count = len(tracks_raw) + tracks = [self._track_from_cache_row(t, idx) for idx, t in enumerate(tracks_raw)] + return PlaylistDetail(meta=meta, tracks=tracks) + + def refresh_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + """Trigger a manager-side refresh, then return the new snapshot. + + ``update_all_playlists`` is the only refresh entry-point on the + manager — it re-fetches every cached playlist. That's wasteful + for a single-playlist refresh; Phase 1 should add a targeted + ``refresh_playlist(mbid)`` to the manager.""" + manager = self._manager() + if manager is None: + return None + try: + manager.update_all_playlists() + except Exception: + pass + return self.get_playlist(playlist_id) + + # ---- projection helpers ------------------------------------------------ + + def _meta_from_cache_row(self, row: Dict[str, Any], playlist_type: str) -> PlaylistMeta: + return PlaylistMeta( + source=self.name, + source_playlist_id=str(row.get("playlist_mbid", "")), + name=row.get("title") or "ListenBrainz Playlist", + owner=row.get("creator") or None, + track_count=int(row.get("track_count", 0) or 0), + extra={ + "playlist_type": playlist_type, + "annotation": row.get("annotation") or {}, + "last_updated": row.get("last_updated"), + }, + ) + + def _track_from_cache_row(self, row: Dict[str, Any], position: int) -> NormalizedTrack: + return NormalizedTrack( + position=position, + track_name=row.get("track_name", "Unknown Track"), + artist_name=row.get("artist_name", "Unknown Artist"), + album_name=row.get("album_name") or None, + duration_ms=int(row.get("duration_ms", 0) or 0), + source_track_id=row.get("recording_mbid") or None, + image_url=row.get("album_cover_url") or None, + needs_discovery=True, + extra={ + "recording_mbid": row.get("recording_mbid"), + "release_mbid": row.get("release_mbid"), + "additional_metadata": row.get("additional_metadata"), + }, + ) diff --git a/core/playlists/sources/qobuz.py b/core/playlists/sources/qobuz.py new file mode 100644 index 00000000..999d1a3b --- /dev/null +++ b/core/playlists/sources/qobuz.py @@ -0,0 +1,94 @@ +"""Qobuz playlist source adapter. + +Wraps ``core.qobuz_client.QobuzClient``. The client already returns +playlist/track dicts in a normalized Sync-page shape (see +``_normalize_qobuz_playlist`` / ``_normalize_qobuz_track``), so the +adapter is mostly a key remap. +""" + +from __future__ import annotations + +from typing import Any, Callable, Dict, List, Optional + +from core.playlists.sources.base import ( + NormalizedTrack, + PlaylistDetail, + PlaylistMeta, + PlaylistSource, + SOURCE_QOBUZ, +) + + +class QobuzPlaylistSource(PlaylistSource): + name = SOURCE_QOBUZ + supports_listing = True + supports_refresh = True + requires_auth = True + + def __init__(self, client_getter: Callable[[], Any]): + self._client_getter = client_getter + + def _client(self): + return self._client_getter() + + def is_authenticated(self) -> bool: + client = self._client() + if client is None: + return False + return bool(client.is_authenticated()) + + def list_playlists(self) -> List[PlaylistMeta]: + client = self._client() + if client is None or not client.is_authenticated(): + return [] + playlists = client.get_user_playlists() or [] + return [self._meta_from_dict(p) for p in playlists] + + def get_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + client = self._client() + if client is None or not client.is_authenticated(): + return None + playlist = client.get_playlist(playlist_id) + if not playlist: + return None + meta = self._meta_from_dict(playlist) + tracks_raw = playlist.get("tracks") or [] + tracks = [self._track_from_dict(t, idx) for idx, t in enumerate(tracks_raw)] + meta.track_count = len(tracks) + return PlaylistDetail(meta=meta, tracks=tracks) + + def refresh_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + return self.get_playlist(playlist_id) + + # ---- projection helpers ------------------------------------------------ + + def _meta_from_dict(self, p: Dict[str, Any]) -> PlaylistMeta: + return PlaylistMeta( + source=self.name, + source_playlist_id=str(p.get("id", "")), + name=p.get("name", "Qobuz Playlist"), + description=p.get("description") or None, + image_url=p.get("image_url") or None, + track_count=int(p.get("track_count", 0) or 0), + extra={ + "public": bool(p.get("public", False)), + "external_urls": p.get("external_urls", {}), + }, + ) + + def _track_from_dict(self, t: Dict[str, Any], position: int) -> NormalizedTrack: + artists = t.get("artists") or [] + artist_name = ", ".join(artists) if artists else "Unknown Artist" + return NormalizedTrack( + position=position, + track_name=t.get("name", "Unknown Track"), + artist_name=artist_name, + album_name=t.get("album") or None, + duration_ms=int(t.get("duration_ms", 0) or 0), + source_track_id=str(t.get("id", "")), + image_url=t.get("image_url") or None, + needs_discovery=False, + extra={k: v for k, v in t.items() if k not in { + "id", "name", "artists", "album", "duration_ms", "image_url", + }}, + ) diff --git a/core/playlists/sources/registry.py b/core/playlists/sources/registry.py new file mode 100644 index 00000000..e4327f20 --- /dev/null +++ b/core/playlists/sources/registry.py @@ -0,0 +1,77 @@ +"""Registry for playlist source adapters. + +Adapters are registered as zero-arg factories so we can lazy-construct +them. This matters because some adapters need late-binding to globals +that aren't ready at import time (e.g. the YouTube adapter wraps a +parser defined in ``web_server.py`` — importing it eagerly would cause +a circular import). + +Usage:: + + registry = get_registry() + registry.register("spotify", lambda: SpotifyPlaylistSource(...)) + source = registry.get_source("spotify") + +In Phase 0 the registry is set up but not yet consumed by the dispatch +sites. Phase 1+ wires it in. +""" + +from __future__ import annotations + +from threading import Lock +from typing import Callable, Dict, List, Optional + +from core.playlists.sources.base import PlaylistSource + + +class PlaylistSourceRegistry: + """Thread-safe registry mapping source name → cached adapter instance.""" + + def __init__(self) -> None: + self._factories: Dict[str, Callable[[], PlaylistSource]] = {} + self._instances: Dict[str, PlaylistSource] = {} + self._lock = Lock() + + def register(self, name: str, factory: Callable[[], PlaylistSource]) -> None: + """Register an adapter factory under ``name``. + + Re-registering replaces the previous factory and invalidates the + cached instance. Used by tests to swap in stubs.""" + with self._lock: + self._factories[name] = factory + self._instances.pop(name, None) + + def unregister(self, name: str) -> None: + with self._lock: + self._factories.pop(name, None) + self._instances.pop(name, None) + + def get_source(self, name: str) -> Optional[PlaylistSource]: + """Return the adapter for ``name``, building it on first access.""" + with self._lock: + if name in self._instances: + return self._instances[name] + factory = self._factories.get(name) + if factory is None: + return None + instance = factory() + self._instances[name] = instance + return instance + + def known_names(self) -> List[str]: + with self._lock: + return sorted(self._factories.keys()) + + def reset(self) -> None: + """Drop all registrations + cached instances. Test-only.""" + with self._lock: + self._factories.clear() + self._instances.clear() + + +_default_registry = PlaylistSourceRegistry() + + +def get_registry() -> PlaylistSourceRegistry: + """Return the process-wide default registry.""" + return _default_registry diff --git a/core/playlists/sources/soulsync_discovery.py b/core/playlists/sources/soulsync_discovery.py new file mode 100644 index 00000000..c648830c --- /dev/null +++ b/core/playlists/sources/soulsync_discovery.py @@ -0,0 +1,151 @@ +"""SoulSync Discovery (personalized playlists) source adapter. + +Wraps ``core.personalized.manager.PersonalizedPlaylistManager``. Unlike +ListenBrainz / Last.fm, personalized playlists already carry source +IDs (Spotify / iTunes / Deezer track IDs) — they were built from +``discovery_pool`` rows. ``needs_discovery=False`` on every track. + +Playlist IDs here are the integer DB row IDs (``personalized_playlists.id``) +converted to strings so the unified interface stays string-keyed. The +adapter parses them back to ints when calling the manager. +""" + +from __future__ import annotations + +from typing import Any, Callable, List, Optional + +from core.playlists.sources.base import ( + NormalizedTrack, + PlaylistDetail, + PlaylistMeta, + PlaylistSource, + SOURCE_SOULSYNC_DISCOVERY, +) + + +class SoulSyncDiscoveryPlaylistSource(PlaylistSource): + name = SOURCE_SOULSYNC_DISCOVERY + supports_listing = True + supports_refresh = True + requires_auth = False + + def __init__( + self, + manager_getter: Callable[[], Any], + profile_id_getter: Optional[Callable[[], int]] = None, + ): + self._manager_getter = manager_getter + self._profile_id_getter = profile_id_getter or (lambda: 1) + + def _manager(self): + return self._manager_getter() + + def is_authenticated(self) -> bool: + return self._manager() is not None + + def list_playlists(self) -> List[PlaylistMeta]: + manager = self._manager() + if manager is None: + return [] + try: + records = manager.list_playlists(profile_id=self._profile_id_getter()) or [] + except Exception: + return [] + return [self._meta_from_record(r) for r in records] + + def get_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + """``playlist_id`` is the stringified ``personalized_playlists.id``.""" + manager = self._manager() + if manager is None: + return None + try: + row_id = int(playlist_id) + except (TypeError, ValueError): + return None + + records = [] + try: + records = manager.list_playlists(profile_id=self._profile_id_getter()) or [] + except Exception: + records = [] + record = next((r for r in records if int(r.id) == row_id), None) + if record is None: + return None + + try: + tracks_raw = manager.get_playlist_tracks(row_id) or [] + except Exception: + tracks_raw = [] + + meta = self._meta_from_record(record) + meta.track_count = len(tracks_raw) + tracks = [self._track_from_record(t, idx) for idx, t in enumerate(tracks_raw)] + return PlaylistDetail(meta=meta, tracks=tracks) + + def refresh_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + manager = self._manager() + if manager is None: + return None + try: + row_id = int(playlist_id) + except (TypeError, ValueError): + return None + + records = manager.list_playlists(profile_id=self._profile_id_getter()) or [] + record = next((r for r in records if int(r.id) == row_id), None) + if record is None: + return None + + try: + manager.refresh_playlist( + kind=record.kind, + variant=record.variant, + profile_id=record.profile_id, + ) + except Exception: + # Manager already persists ``last_generation_error`` on + # failure; surface the existing snapshot regardless. + pass + return self.get_playlist(playlist_id) + + # ---- projection helpers ------------------------------------------------ + + def _meta_from_record(self, record: Any) -> PlaylistMeta: + return PlaylistMeta( + source=self.name, + source_playlist_id=str(record.id), + name=record.name, + track_count=int(record.track_count or 0), + description=record.kind, + extra={ + "kind": record.kind, + "variant": record.variant, + "profile_id": record.profile_id, + "is_stale": bool(record.is_stale), + "last_generated_at": record.last_generated_at, + "last_synced_at": record.last_synced_at, + "last_generation_source": record.last_generation_source, + "last_generation_error": record.last_generation_error, + }, + ) + + def _track_from_record(self, track: Any, position: int) -> NormalizedTrack: + primary_id = track.primary_id() + return NormalizedTrack( + position=position, + track_name=track.track_name, + artist_name=track.artist_name, + album_name=track.album_name or None, + duration_ms=int(track.duration_ms or 0), + source_track_id=primary_id, + image_url=track.album_cover_url or None, + needs_discovery=False, + extra={ + "spotify_track_id": track.spotify_track_id, + "itunes_track_id": track.itunes_track_id, + "deezer_track_id": track.deezer_track_id, + "popularity": track.popularity, + "track_data_json": track.track_data_json, + "source_hint": track.source, + }, + ) diff --git a/core/playlists/sources/spotify.py b/core/playlists/sources/spotify.py new file mode 100644 index 00000000..ccbf4bfd --- /dev/null +++ b/core/playlists/sources/spotify.py @@ -0,0 +1,99 @@ +"""Spotify playlist source adapter. + +Wraps ``core.spotify_client.SpotifyClient`` — the authenticated Spotify +client used everywhere else. Adapter projects Spotify's ``Playlist`` / +``Track`` dataclasses into ``PlaylistMeta`` / ``NormalizedTrack``. +""" + +from __future__ import annotations + +from typing import Any, Callable, List, Optional + +from core.playlists.sources.base import ( + NormalizedTrack, + PlaylistDetail, + PlaylistMeta, + PlaylistSource, + SOURCE_SPOTIFY, +) + + +class SpotifyPlaylistSource(PlaylistSource): + name = SOURCE_SPOTIFY + supports_listing = True + supports_refresh = True + requires_auth = True + + def __init__(self, client_getter: Callable[[], Any]): + """``client_getter`` returns the live ``SpotifyClient`` singleton. + + We accept a getter (not the client itself) so the adapter can be + constructed at import time, before ``web_server.py`` has wired + up the singleton.""" + self._client_getter = client_getter + + def _client(self): + return self._client_getter() + + def is_authenticated(self) -> bool: + client = self._client() + if client is None: + return False + return bool(client.is_authenticated()) + + def list_playlists(self) -> List[PlaylistMeta]: + client = self._client() + if client is None or not client.is_authenticated(): + return [] + playlists = client.get_user_playlists_metadata_only() + return [self._meta_from_playlist(p) for p in playlists] + + def get_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + client = self._client() + if client is None or not client.is_authenticated(): + return None + playlist = client.get_playlist_by_id(playlist_id) + if playlist is None: + return None + meta = self._meta_from_playlist(playlist) + tracks = [self._track_from_spotify(t, idx) for idx, t in enumerate(playlist.tracks)] + meta.track_count = len(tracks) + return PlaylistDetail(meta=meta, tracks=tracks) + + def refresh_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + return self.get_playlist(playlist_id) + + # ---- projection helpers ------------------------------------------------ + + def _meta_from_playlist(self, playlist: Any) -> PlaylistMeta: + return PlaylistMeta( + source=self.name, + source_playlist_id=str(playlist.id), + name=playlist.name, + owner=playlist.owner, + description=playlist.description, + track_count=int(getattr(playlist, "total_tracks", 0) or 0), + extra={ + "public": bool(getattr(playlist, "public", False)), + "collaborative": bool(getattr(playlist, "collaborative", False)), + }, + ) + + def _track_from_spotify(self, track: Any, position: int) -> NormalizedTrack: + artists = getattr(track, "artists", None) or [] + artist_name = ", ".join(artists) if artists else "Unknown Artist" + return NormalizedTrack( + position=position, + track_name=track.name, + artist_name=artist_name, + album_name=getattr(track, "album", None), + duration_ms=int(getattr(track, "duration_ms", 0) or 0), + source_track_id=str(track.id), + image_url=getattr(track, "image_url", None), + needs_discovery=False, + extra={ + "popularity": getattr(track, "popularity", 0), + "external_urls": getattr(track, "external_urls", None), + "preview_url": getattr(track, "preview_url", None), + }, + ) diff --git a/core/playlists/sources/spotify_public.py b/core/playlists/sources/spotify_public.py new file mode 100644 index 00000000..0b9cec98 --- /dev/null +++ b/core/playlists/sources/spotify_public.py @@ -0,0 +1,95 @@ +"""Spotify public-embed playlist source adapter. + +Wraps ``core.spotify_public_scraper`` — no auth, scrapes the public +embed page. ``supports_listing=False`` because there's no "user +library" to enumerate; the user pastes a URL and the adapter fetches. +""" + +from __future__ import annotations + +import hashlib +from typing import List, Optional + +from core.playlists.sources.base import ( + NormalizedTrack, + PlaylistDetail, + PlaylistMeta, + PlaylistSource, + SOURCE_SPOTIFY_PUBLIC, +) + + +class SpotifyPublicPlaylistSource(PlaylistSource): + name = SOURCE_SPOTIFY_PUBLIC + supports_listing = False + supports_refresh = True + requires_auth = False + + def is_authenticated(self) -> bool: + return True + + def list_playlists(self) -> List[PlaylistMeta]: + return [] + + def get_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + """``playlist_id`` is a Spotify URL or ``open.spotify.com`` URI.""" + from core.spotify_public_scraper import ( + parse_spotify_url, + scrape_spotify_embed, + ) + + parsed = parse_spotify_url(playlist_id) + if not parsed: + return None + + data = scrape_spotify_embed(parsed["type"], parsed["id"]) + if not isinstance(data, dict) or data.get("error"): + return None + + source_url = data.get("url") or playlist_id + url_hash = data.get("url_hash") or hashlib.md5(source_url.encode()).hexdigest()[:12] + tracks_raw = data.get("tracks") or [] + + meta = PlaylistMeta( + source=self.name, + source_playlist_id=url_hash, + name=data.get("name", "Spotify Playlist"), + owner=data.get("subtitle"), + track_count=len(tracks_raw), + source_url=source_url, + extra={ + "spotify_type": data.get("type"), + "spotify_id": data.get("id"), + }, + ) + + tracks = [ + self._track_from_embed(t, idx) + for idx, t in enumerate(tracks_raw) + if t and t.get("id") + ] + return PlaylistDetail(meta=meta, tracks=tracks) + + def refresh_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + return self.get_playlist(playlist_id) + + # ---- projection helpers ------------------------------------------------ + + def _track_from_embed(self, track: dict, position: int) -> NormalizedTrack: + artists = track.get("artists") or [] + artist_name = ", ".join( + a.get("name", "") for a in artists if isinstance(a, dict) + ) or "Unknown Artist" + return NormalizedTrack( + position=position, + track_name=track.get("name", "Unknown Track"), + artist_name=artist_name, + album_name=None, + duration_ms=int(track.get("duration_ms", 0) or 0), + source_track_id=str(track.get("id", "")), + needs_discovery=False, + extra={ + "explicit": bool(track.get("is_explicit", False)), + "track_number": track.get("track_number"), + }, + ) diff --git a/core/playlists/sources/tidal.py b/core/playlists/sources/tidal.py new file mode 100644 index 00000000..7a74d423 --- /dev/null +++ b/core/playlists/sources/tidal.py @@ -0,0 +1,101 @@ +"""Tidal playlist source adapter. + +Wraps ``core.tidal_client.TidalClient``. Tidal recognizes the virtual +``tidal-favorites`` ID inside ``get_playlist`` already, so the adapter +doesn't need to special-case it. +""" + +from __future__ import annotations + +from typing import Any, Callable, List, Optional + +from core.playlists.sources.base import ( + NormalizedTrack, + PlaylistDetail, + PlaylistMeta, + PlaylistSource, + SOURCE_TIDAL, +) + + +class TidalPlaylistSource(PlaylistSource): + name = SOURCE_TIDAL + supports_listing = True + supports_refresh = True + requires_auth = True + + def __init__(self, client_getter: Callable[[], Any]): + self._client_getter = client_getter + + def _client(self): + return self._client_getter() + + def is_authenticated(self) -> bool: + client = self._client() + if client is None: + return False + return bool(client.is_authenticated()) + + def list_playlists(self) -> List[PlaylistMeta]: + client = self._client() + if client is None or not client.is_authenticated(): + return [] + playlists = client.get_user_playlists_metadata_only() or [] + return [self._meta_from_playlist(p) for p in playlists] + + def get_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + client = self._client() + if client is None or not client.is_authenticated(): + return None + playlist = client.get_playlist(playlist_id) + if playlist is None: + return None + meta = self._meta_from_playlist(playlist) + tracks_raw = getattr(playlist, "tracks", None) or [] + tracks = [self._track_from_tidal(t, idx) for idx, t in enumerate(tracks_raw)] + meta.track_count = len(tracks) + return PlaylistDetail(meta=meta, tracks=tracks) + + def refresh_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + return self.get_playlist(playlist_id) + + # ---- projection helpers ------------------------------------------------ + + def _meta_from_playlist(self, playlist: Any) -> PlaylistMeta: + owner_field = getattr(playlist, "owner", None) + owner_name: Optional[str] = None + if isinstance(owner_field, dict): + owner_name = owner_field.get("name") or owner_field.get("id") + elif owner_field: + owner_name = str(owner_field) + tracks_raw = getattr(playlist, "tracks", None) or [] + return PlaylistMeta( + source=self.name, + source_playlist_id=str(playlist.id), + name=playlist.name, + owner=owner_name, + description=getattr(playlist, "description", "") or None, + track_count=len(tracks_raw), + extra={ + "public": bool(getattr(playlist, "public", True)), + "external_urls": getattr(playlist, "external_urls", {}), + }, + ) + + def _track_from_tidal(self, track: Any, position: int) -> NormalizedTrack: + artists = getattr(track, "artists", None) or [] + artist_name = ", ".join(artists) if artists else "Unknown Artist" + return NormalizedTrack( + position=position, + track_name=track.name, + artist_name=artist_name, + album_name=getattr(track, "album", "") or None, + duration_ms=int(getattr(track, "duration_ms", 0) or 0), + source_track_id=str(track.id), + needs_discovery=False, + extra={ + "explicit": bool(getattr(track, "explicit", False)), + "external_urls": getattr(track, "external_urls", {}), + "popularity": getattr(track, "popularity", 0), + }, + ) diff --git a/core/playlists/sources/youtube.py b/core/playlists/sources/youtube.py new file mode 100644 index 00000000..fdf21c5d --- /dev/null +++ b/core/playlists/sources/youtube.py @@ -0,0 +1,87 @@ +"""YouTube playlist source adapter. + +Wraps ``parse_youtube_playlist`` (currently a free function in +``web_server.py`` — Phase 0 doesn't move it, just calls it via an +injected callable to avoid the circular import). ``supports_listing`` +is False — YouTube playlists are URL-input only, no user library. +""" + +from __future__ import annotations + +import hashlib +from typing import Any, Callable, List, Optional + +from core.playlists.sources.base import ( + NormalizedTrack, + PlaylistDetail, + PlaylistMeta, + PlaylistSource, + SOURCE_YOUTUBE, +) + + +class YouTubePlaylistSource(PlaylistSource): + name = SOURCE_YOUTUBE + supports_listing = False + supports_refresh = True + requires_auth = False + + def __init__(self, parser: Callable[[str], Optional[dict]]): + """``parser`` matches the signature of ``parse_youtube_playlist`` + in web_server.py — takes a URL, returns the playlist dict or + ``None``. Injected so adapter can be constructed at import time.""" + self._parser = parser + + def is_authenticated(self) -> bool: + return True + + def list_playlists(self) -> List[PlaylistMeta]: + return [] + + def get_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + """``playlist_id`` is the full YouTube playlist URL.""" + data = self._parser(playlist_id) + if not data: + return None + + source_url = data.get("url") or playlist_id + url_hash = hashlib.md5(source_url.encode()).hexdigest()[:12] + tracks_raw = data.get("tracks") or [] + + meta = PlaylistMeta( + source=self.name, + source_playlist_id=url_hash, + name=data.get("name", "YouTube Playlist"), + track_count=int(data.get("track_count", len(tracks_raw))), + image_url=data.get("image_url") or None, + source_url=source_url, + extra={ + "youtube_playlist_id": data.get("id"), + }, + ) + + tracks = [self._track_from_yt(t, idx) for idx, t in enumerate(tracks_raw) if t] + return PlaylistDetail(meta=meta, tracks=tracks) + + def refresh_playlist(self, playlist_id: str) -> Optional[PlaylistDetail]: + return self.get_playlist(playlist_id) + + # ---- projection helpers ------------------------------------------------ + + def _track_from_yt(self, track: dict, position: int) -> NormalizedTrack: + artists = track.get("artists") or [] + artist_name = ", ".join(artists) if artists else "Unknown Artist" + return NormalizedTrack( + position=position, + track_name=track.get("name", "Unknown Track"), + artist_name=artist_name, + album_name=None, + duration_ms=int(track.get("duration_ms", 0) or 0), + source_track_id=str(track.get("id", "")), + needs_discovery=False, + extra={ + "url": track.get("url"), + "raw_title": track.get("raw_title"), + "raw_artist": track.get("raw_artist"), + }, + ) diff --git a/tests/test_playlist_sources_adapters.py b/tests/test_playlist_sources_adapters.py new file mode 100644 index 00000000..b83afe81 --- /dev/null +++ b/tests/test_playlist_sources_adapters.py @@ -0,0 +1,573 @@ +"""Adapter contract tests for ``core/playlists/sources/``. + +These pin the projection from each backing client's native shape into +the unified ``PlaylistMeta`` / ``NormalizedTrack`` shape. Adapters are +fed minimal fakes (not real clients) so the test is independent of the +live API surface — the goal is to lock in the field mapping so later +phases that consume the unified interface can rely on it. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any, Callable, List, Optional + +import pytest + +from core.playlists.sources import ( + NormalizedTrack, + PlaylistDetail, + PlaylistMeta, + PlaylistSource, + PlaylistSourceRegistry, + get_registry, +) +from core.playlists.sources.itunes_link import ITunesLinkPlaylistSource +from core.playlists.sources.lastfm import LastFMPlaylistSource +from core.playlists.sources.listenbrainz import ListenBrainzPlaylistSource +from core.playlists.sources.qobuz import QobuzPlaylistSource +from core.playlists.sources.soulsync_discovery import SoulSyncDiscoveryPlaylistSource +from core.playlists.sources.spotify import SpotifyPlaylistSource +from core.playlists.sources.spotify_public import SpotifyPublicPlaylistSource +from core.playlists.sources.tidal import TidalPlaylistSource +from core.playlists.sources.youtube import YouTubePlaylistSource + + +# ─── Spotify ──────────────────────────────────────────────────────────── + + +class _FakeSpotifyClient: + """Stand-in for ``core.spotify_client.SpotifyClient``.""" + + def __init__(self, authed: bool = True): + self._authed = authed + + def is_authenticated(self) -> bool: + return self._authed + + def get_user_playlists_metadata_only(self): + return [ + SimpleNamespace( + id="pl1", + name="Drive", + description="long drives", + owner="me", + public=True, + collaborative=False, + tracks=[], + total_tracks=2, + ) + ] + + def get_playlist_by_id(self, playlist_id: str): + track = SimpleNamespace( + id="t1", + name="Run", + artists=["A", "B"], + album="Album", + duration_ms=180_000, + popularity=50, + preview_url=None, + external_urls={"spotify": "url"}, + image_url="img", + ) + return SimpleNamespace( + id=playlist_id, + name="Drive", + description="long drives", + owner="me", + public=True, + collaborative=False, + tracks=[track], + total_tracks=1, + ) + + +def test_spotify_adapter_lists_and_fetches(): + client = _FakeSpotifyClient() + src = SpotifyPlaylistSource(lambda: client) + + assert isinstance(src, PlaylistSource) + assert src.is_authenticated() is True + + metas = src.list_playlists() + assert len(metas) == 1 + m = metas[0] + assert m.source == "spotify" + assert m.source_playlist_id == "pl1" + assert m.name == "Drive" + assert m.track_count == 2 + + detail = src.get_playlist("pl1") + assert detail is not None + assert detail.meta.track_count == 1 + t = detail.tracks[0] + assert t.source_track_id == "t1" + assert t.track_name == "Run" + assert t.artist_name == "A, B" + assert t.album_name == "Album" + assert t.duration_ms == 180_000 + assert t.needs_discovery is False + + +def test_spotify_adapter_handles_unauthed(): + src = SpotifyPlaylistSource(lambda: _FakeSpotifyClient(authed=False)) + assert src.is_authenticated() is False + assert src.list_playlists() == [] + assert src.get_playlist("pl1") is None + + +def test_spotify_adapter_handles_missing_client(): + src = SpotifyPlaylistSource(lambda: None) + assert src.is_authenticated() is False + assert src.list_playlists() == [] + + +# ─── Tidal ────────────────────────────────────────────────────────────── + + +class _FakeTidalClient: + def __init__(self, authed: bool = True): + self._authed = authed + + def is_authenticated(self) -> bool: + return self._authed + + def get_user_playlists_metadata_only(self): + return [ + SimpleNamespace( + id="tpl", + name="Tidal Mix", + description="", + tracks=[], + external_urls={"tidal": "url"}, + owner={"name": "broque"}, + public=True, + ) + ] + + def get_playlist(self, playlist_id: str): + track = SimpleNamespace( + id="ttrk", + name="Wave", + artists=["X"], + album="Ocean", + duration_ms=200_000, + external_urls={}, + popularity=0, + explicit=False, + ) + return SimpleNamespace( + id=playlist_id, + name="Tidal Mix", + description="", + tracks=[track], + external_urls={}, + owner={"name": "broque"}, + public=True, + ) + + +def test_tidal_adapter_projection(): + src = TidalPlaylistSource(lambda: _FakeTidalClient()) + + metas = src.list_playlists() + assert metas[0].owner == "broque" + assert metas[0].source == "tidal" + + detail = src.get_playlist("tpl") + assert detail is not None + assert detail.tracks[0].source_track_id == "ttrk" + assert detail.tracks[0].album_name == "Ocean" + assert detail.tracks[0].needs_discovery is False + + +# ─── Qobuz ────────────────────────────────────────────────────────────── + + +class _FakeQobuzClient: + def is_authenticated(self) -> bool: + return True + + def get_user_playlists(self): + return [{ + "id": "q1", + "name": "Q Mix", + "description": "qobuz", + "public": False, + "track_count": 2, + "image_url": "img", + "external_urls": {"qobuz": "url"}, + }] + + def get_playlist(self, playlist_id: str): + return { + "id": playlist_id, + "name": "Q Mix", + "description": "qobuz", + "public": False, + "track_count": 1, + "image_url": "img", + "external_urls": {"qobuz": "url"}, + "tracks": [{ + "id": "qt1", + "name": "Track", + "artists": ["Q-Artist"], + "album": "Q-Album", + "duration_ms": 300_000, + "image_url": "ti", + }], + } + + +def test_qobuz_adapter_projection(): + src = QobuzPlaylistSource(lambda: _FakeQobuzClient()) + metas = src.list_playlists() + assert metas[0].source_playlist_id == "q1" + + detail = src.get_playlist("q1") + assert detail.tracks[0].source_track_id == "qt1" + assert detail.tracks[0].duration_ms == 300_000 + assert detail.tracks[0].image_url == "ti" + + +# ─── Spotify Public ───────────────────────────────────────────────────── + + +def test_spotify_public_adapter_invalid_url(): + src = SpotifyPublicPlaylistSource() + # invalid URL → parser returns None → adapter returns None + assert src.get_playlist("not-a-spotify-url") is None + assert src.supports_listing is False + assert src.list_playlists() == [] + + +def test_spotify_public_adapter_projects_scrape(monkeypatch): + src = SpotifyPublicPlaylistSource() + + def fake_parse(url: str): + return {"type": "playlist", "id": "xyz"} + + def fake_scrape(spotify_type: str, spotify_id: str): + return { + "id": spotify_id, + "type": "playlist", + "name": "Embed", + "subtitle": "owner", + "tracks": [ + { + "id": "sptrk", + "name": "Song", + "artists": [{"name": "Artist"}], + "duration_ms": 100_000, + "is_explicit": False, + "track_number": 1, + }, + ], + "url": "https://open.spotify.com/playlist/xyz", + "url_hash": "abc123", + } + + monkeypatch.setattr("core.spotify_public_scraper.parse_spotify_url", fake_parse) + monkeypatch.setattr("core.spotify_public_scraper.scrape_spotify_embed", fake_scrape) + + detail = src.get_playlist("https://open.spotify.com/playlist/xyz") + assert detail is not None + assert detail.meta.source_playlist_id == "abc123" + assert detail.meta.source_url == "https://open.spotify.com/playlist/xyz" + assert detail.tracks[0].artist_name == "Artist" + assert detail.tracks[0].source_track_id == "sptrk" + + +# ─── YouTube ──────────────────────────────────────────────────────────── + + +def test_youtube_adapter_projection(): + def parser(url: str): + return { + "id": "yt_pl", + "name": "YT Mix", + "track_count": 1, + "url": url, + "image_url": "thumb", + "tracks": [{ + "id": "vid1", + "name": "Track", + "artists": ["Channel"], + "duration_ms": 240_000, + "url": "https://youtu.be/vid1", + }], + } + + src = YouTubePlaylistSource(parser) + detail = src.get_playlist("https://youtube.com/playlist?list=yt_pl") + assert detail is not None + assert detail.meta.source == "youtube" + assert detail.meta.source_url == "https://youtube.com/playlist?list=yt_pl" + assert len(detail.meta.source_playlist_id) == 12 # md5[:12] + assert detail.tracks[0].source_track_id == "vid1" + + +def test_youtube_adapter_failed_parse(): + src = YouTubePlaylistSource(lambda url: None) + assert src.get_playlist("https://bad") is None + + +# ─── iTunes Link ──────────────────────────────────────────────────────── + + +def test_itunes_link_adapter_projection(): + def parser(url: str): + return { + "id": "1234", + "type": "album", + "name": "Album X", + "subtitle": "Artist", + "url": url, + "url_hash": "abcd1234", + "track_count": 1, + "image_url": "art", + "tracks": [{ + "id": "555", + "name": "Song", + "artists": ["Artist"], + "album": {"name": "Album X"}, + "duration_ms": 220_000, + "image_url": "art", + }], + } + + src = ITunesLinkPlaylistSource(parser) + detail = src.get_playlist("https://music.apple.com/us/album/1234") + assert detail is not None + assert detail.meta.source == "itunes_link" + assert detail.meta.source_playlist_id == "abcd1234" + assert detail.tracks[0].album_name == "Album X" + assert detail.tracks[0].source_track_id == "555" + + +# ─── ListenBrainz ─────────────────────────────────────────────────────── + + +class _FakeLBManager: + def __init__(self, authed: bool = True): + self.client = SimpleNamespace(is_authenticated=lambda: authed) + self._rows = { + "created_for_user": [{ + "playlist_mbid": "lb-1", + "title": "Weekly Discovery", + "creator": "ListenBrainz", + "track_count": 1, + "annotation": {"note": "weekly"}, + "last_updated": "2026-05-26", + }], + "user_created": [], + "collaborative": [], + } + self._tracks = { + "lb-1": [{ + "track_name": "Discovery Track", + "artist_name": "MB Artist", + "album_name": "MB Album", + "duration_ms": 250_000, + "recording_mbid": "rec-1", + "release_mbid": "rel-1", + "album_cover_url": "cover", + "additional_metadata": {}, + }], + } + self.refresh_called = False + + def get_cached_playlists(self, playlist_type: str): + return self._rows.get(playlist_type, []) + + def get_playlist_type(self, mbid: str) -> str: + for ptype, rows in self._rows.items(): + if any(r["playlist_mbid"] == mbid for r in rows): + return ptype + return "" + + def get_cached_tracks(self, mbid: str): + return self._tracks.get(mbid, []) + + def update_all_playlists(self): + self.refresh_called = True + + +def test_listenbrainz_adapter_marks_needs_discovery(): + manager = _FakeLBManager() + src = ListenBrainzPlaylistSource(lambda: manager) + + metas = src.list_playlists() + assert len(metas) == 1 + assert metas[0].source == "listenbrainz" + assert metas[0].extra["playlist_type"] == "created_for_user" + + detail = src.get_playlist("lb-1") + assert detail is not None + assert detail.meta.track_count == 1 + t = detail.tracks[0] + assert t.needs_discovery is True + assert t.source_track_id == "rec-1" + assert t.extra["recording_mbid"] == "rec-1" + + +def test_listenbrainz_adapter_refresh_calls_manager(): + manager = _FakeLBManager() + src = ListenBrainzPlaylistSource(lambda: manager) + src.refresh_playlist("lb-1") + assert manager.refresh_called is True + + +# ─── Last.fm ──────────────────────────────────────────────────────────── + + +class _FakeLastFMManager: + def __init__(self): + self._rows = [{ + "playlist_mbid": "lfm-1", + "title": "Last.fm Radio: Seed", + "creator": "Last.fm", + "track_count": 1, + "annotation": {"seed": "track"}, + "last_updated": "2026-05-26", + }] + self._tracks = [{ + "track_name": "Similar", + "artist_name": "Artist", + "album_name": None, + "duration_ms": 200_000, + "recording_mbid": "lfm-rec-1", + "release_mbid": None, + "album_cover_url": None, + "additional_metadata": {}, + }] + + def get_cached_playlists(self, playlist_type: str): + if playlist_type == "lastfm_radio": + return self._rows + return [] + + def get_cached_tracks(self, mbid: str): + if mbid == "lfm-1": + return self._tracks + return [] + + +def test_lastfm_adapter_projects_radio_rows(): + src = LastFMPlaylistSource(lambda: _FakeLastFMManager()) + + metas = src.list_playlists() + assert len(metas) == 1 + assert metas[0].source == "lastfm" + assert metas[0].owner == "Last.fm" + + detail = src.get_playlist("lfm-1") + assert detail is not None + assert detail.tracks[0].needs_discovery is True + assert detail.tracks[0].source_track_id == "lfm-rec-1" + + +# ─── SoulSync Discovery ───────────────────────────────────────────────── + + +class _FakeDiscoveryManager: + def __init__(self): + self.refresh_calls = [] + self._records = [SimpleNamespace( + id=42, + profile_id=1, + kind="hidden_gems", + variant="", + name="Hidden Gems", + config=None, + track_count=1, + last_generated_at="2026-05-26T00:00:00Z", + last_synced_at=None, + last_generation_source="discovery_pool", + last_generation_error=None, + is_stale=False, + )] + self._tracks = [SimpleNamespace( + track_name="Gem", + artist_name="Indie", + album_name="EP", + spotify_track_id="sp-gem", + itunes_track_id=None, + deezer_track_id=None, + album_cover_url="art", + duration_ms=180_000, + popularity=20, + track_data_json=None, + source="discovery", + primary_id=lambda: "sp-gem", + )] + + def list_playlists(self, profile_id: int = 1): + return self._records + + def get_playlist_tracks(self, playlist_id: int): + return self._tracks if playlist_id == 42 else [] + + def refresh_playlist(self, kind: str, variant: str = "", profile_id: int = 1, config_overrides=None): + self.refresh_calls.append((kind, variant, profile_id)) + return self._records[0] + + +def test_soulsync_discovery_adapter_tracks_dont_need_discovery(): + manager = _FakeDiscoveryManager() + src = SoulSyncDiscoveryPlaylistSource(lambda: manager, profile_id_getter=lambda: 1) + + metas = src.list_playlists() + assert metas[0].source == "soulsync_discovery" + assert metas[0].source_playlist_id == "42" + assert metas[0].extra["kind"] == "hidden_gems" + + detail = src.get_playlist("42") + assert detail is not None + t = detail.tracks[0] + assert t.needs_discovery is False + assert t.source_track_id == "sp-gem" + assert t.album_name == "EP" + assert t.extra["spotify_track_id"] == "sp-gem" + + +def test_soulsync_discovery_adapter_refresh_invokes_manager(): + manager = _FakeDiscoveryManager() + src = SoulSyncDiscoveryPlaylistSource(lambda: manager, profile_id_getter=lambda: 1) + src.refresh_playlist("42") + assert manager.refresh_calls == [("hidden_gems", "", 1)] + + +# ─── Registry ─────────────────────────────────────────────────────────── + + +def test_registry_lazy_construct_and_cache(): + reg = PlaylistSourceRegistry() + constructed = [] + + def factory(): + constructed.append(True) + return SpotifyPlaylistSource(lambda: None) + + reg.register("spotify", factory) + assert constructed == [] # not built yet + + first = reg.get_source("spotify") + second = reg.get_source("spotify") + assert first is second # cached + assert len(constructed) == 1 + + +def test_registry_re_register_invalidates_instance(): + reg = PlaylistSourceRegistry() + reg.register("spotify", lambda: SpotifyPlaylistSource(lambda: None)) + first = reg.get_source("spotify") + reg.register("spotify", lambda: SpotifyPlaylistSource(lambda: None)) + second = reg.get_source("spotify") + assert first is not second + + +def test_registry_unknown_source_returns_none(): + reg = PlaylistSourceRegistry() + assert reg.get_source("nope") is None diff --git a/webui/static/helper.js b/webui/static/helper.js index 612db53c..5eb26aca 100644 --- a/webui/static/helper.js +++ b/webui/static/helper.js @@ -3413,6 +3413,10 @@ function closeHelperSearch() { // projects that span multiple commits before shipping. Strip the flag at // release time and add a real `date:` line at the top of the version block. const WHATS_NEW = { + '2.6.3': [ + { unreleased: true }, + { title: 'Groundwork: unified playlist source layer', desc: 'first slice of a refactor that\'ll let ListenBrainz, Last.fm radio, and SoulSync Discovery playlists live as Sync-page tabs alongside Spotify / Tidal / Qobuz / YouTube — so they can be mirrored + scheduled like the rest. this commit adds the shared adapter layer all those sources will plug into; no UI changes yet. nothing to do on your end.' }, + ], '2.6.2': [ { date: 'May 24, 2026 — 2.6.2 release' }, { title: 'Fix: songs stuck in quarantine loop when picking a different candidate', desc: 'when a track failed AcoustID verification and got quarantined, opening the candidates modal and manually picking a different file would just re-quarantine it — the manual pick path ran full AcoustID verification with no bypass, so if the alternate file disagreed with AcoustID\'s stored metadata too (common for live versions, remasters, regional title differences) the file landed right back in quarantine. user got stuck in the loop. manual picks via the candidates modal now skip AcoustID for that one post-process pass (matching what the Approve button already does for restored quarantine files). integrity and bit-depth gates still run because those check the new file\'s actual condition, not its identity. closes #701.' },