soulsync/core/playlists/sources/deezer.py
Broque Thomas 8c41b05fe8 Refactor refresh_mirrored to use unified PlaylistSource registry
Phase 1a of the Discover-to-Sync unification. The mirrored-playlist
refresh handler used to branch per-source through a ~190-line
if/elif chain (Spotify, Spotify public, Deezer, Tidal, YouTube).
Each branch hand-built its own ``extra_data`` JSON for the matched-
data block. With every new source we considered for Sync-page mirror
support (ListenBrainz, Last.fm radio, SoulSync Discovery, iTunes
link), that chain would have grown a new elif.

This commit lifts the per-source logic into the existing adapter
layer and collapses the dispatch to a registry lookup:

- ``core/playlists/sources/deezer.py`` — new adapter so the registry
  covers every source the refresh handler previously branched on.
- ``core/playlists/sources/bootstrap.py`` — single helper that builds
  a populated registry from injected getter callables. Both
  ``web_server.py`` boot and the automation test fixtures call it,
  so the two construction paths can't drift.
- ``core/playlists/sources/base.py`` — ``to_mirror_track_dict``
  projection helper centralises the NormalizedTrack → DB-row
  conversion (including the discovered/matched_data and
  spotify_hint extra_data shapes the downstream sync + wishlist
  consumers already expect).
- Spotify adapter now populates ``extra['discovered']`` + an
  ``extra['matched_data']`` block when fetching via the authed API,
  so Spotify mirrors keep landing pre-discovered (matches the
  pre-refactor contract pinned by
  ``test_spotify_refresh_writes_to_db``).
- Spotify-public adapter populates ``extra['spotify_hint']`` so the
  discovery worker can skip its search step and jump straight to
  enrichment for the known track ID.
- All artist-name fields now project to first-artist-only across
  every adapter — matches the pre-refactor mirror_playlist DB shape
  (``t.artists[0]``).

``refresh_mirrored.py`` shrinks ~190 → ~80 lines and keeps:

- the file/beatport unrefreshable-source filter,
- URL extraction from ``description`` via ``require_refresh_url``
  for spotify_public + youtube,
- the Spotify-public → authed-Spotify fallback when the user is
  signed in (handler-level branch, not in any adapter),
- the Tidal-not-authenticated soft-skip log (skip, not error),
- existing-extra_data preservation across refreshes,
- the ``playlist_changed`` automation event emit on track-set delta.

Test scaffolding:

- ``_build_deps`` in ``tests/automation/test_handlers_playlist.py``
  now builds a default registry from the passed clients via
  ``build_playlist_source_registry``, so existing refresh tests
  exercise the same path without per-test changes. New tests cover
  Tidal-not-authed soft-skip, Deezer refresh writes plain tracks,
  YouTube refresh reads URL from description, and Spotify-public
  uses authed Spotify when signed in.
- 4 new adapter tests for Deezer projection +
  ``to_mirror_track_dict`` (minimal track, Spotify matched_data,
  Spotify-public spotify_hint).
- ``playlist_source_registry`` field on ``AutomationDeps`` defaults
  to ``None`` so the other 5 automation test files (which don't
  exercise refresh_mirrored) keep working unchanged.

220 tests across automation + adapter suites green.
2026-05-26 12:52:39 -07:00

108 lines
4 KiB
Python

"""Deezer playlist source adapter.
Wraps ``core.deezer_client.DeezerClient.get_playlist``. Deezer's public
API needs no auth, so ``is_authenticated`` always returns True. Listing
the *user's* playlists requires OAuth — surfaced via the underlying
``is_user_authenticated`` flag — but the get-by-id flow works on any
public playlist regardless.
"""
from __future__ import annotations
from typing import Any, Callable, Dict, List, Optional
from core.playlists.sources.base import (
NormalizedTrack,
PlaylistDetail,
PlaylistMeta,
PlaylistSource,
SOURCE_DEEZER,
)
class DeezerPlaylistSource(PlaylistSource):
name = SOURCE_DEEZER
supports_listing = True # user playlists need OAuth; falls back to []
supports_refresh = True
requires_auth = False
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
# Deezer's `is_authenticated` is True even with no OAuth token —
# the public API works without one. Use that as our liveness signal.
return bool(client.is_authenticated())
def list_playlists(self) -> List[PlaylistMeta]:
client = self._client()
if client is None:
return []
# User playlists need OAuth; `get_user_playlists` returns [] when
# the stub-interface variant is in use. Honor whatever the client
# actually returns.
try:
playlists = client.get_user_playlists() or []
except Exception:
return []
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:
return None
data = client.get_playlist(playlist_id)
if not data:
return None
meta = self._meta_from_dict(data)
tracks_raw = data.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_playlist(self, playlist: Any) -> PlaylistMeta:
"""Project a ``DeezerClient.Playlist`` dataclass into PlaylistMeta."""
return PlaylistMeta(
source=self.name,
source_playlist_id=str(getattr(playlist, "id", "")),
name=getattr(playlist, "name", "Deezer Playlist"),
description=getattr(playlist, "description", None),
track_count=int(getattr(playlist, "total_tracks", 0) or 0),
owner=getattr(playlist, "owner", None),
)
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", "Deezer Playlist"),
description=p.get("description") or None,
owner=p.get("owner") or None,
image_url=p.get("image_url") or None,
track_count=int(p.get("track_count", 0) or 0),
)
def _track_from_dict(self, t: Dict[str, Any], position: int) -> NormalizedTrack:
artists = t.get("artists") or []
artist_name = artists[0] 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", "")),
needs_discovery=False,
extra={"track_number": t.get("track_number")},
)