#798: Spotify Free metadata mode — backend (opt-in, shared spotify tables)

Adds an opt-in no-creds Spotify metadata path: SpotifyClient serves SpotipyFree
data (real Spotify IDs) when metadata.spotify_free is enabled AND SpotipyFree is
installed AND there's no real Spotify auth. The data lands in the SAME spotify_*
columns, so the enrichment worker, search, and #775 lookups work UNCHANGED —
they just receive free-sourced data. The worker's only change is its availability
gate.

- core/spotify_free_metadata.py: SpotifyFreeMetadataClient + normalize_artist +
  pure gate fns (should_use_free_fallback / should_offer_spotify_metadata /
  spotify_free_installed).
- SpotifyClient: _free_enabled (opt-in, default OFF) / _free_available /
  is_spotify_metadata_available / _free_active + in-client routing for
  search_artists/tracks + get_album/artist/track_details/album_tracks/artist_albums.
- 3 scoped availability gates use is_spotify_metadata_available(): enrichment
  worker loop, search resolve_client, watchlist. The ~40 discovery/user-library
  sites stay auth-only (no sprawl, no user-data risk).

Authed users are untouched (gate closed when auth healthy). UI surfacing comes
next. Pure gates + normalizer tested; orchestrator resolve test added.
This commit is contained in:
BoulderBadgeDad 2026-06-05 13:32:37 -07:00
parent ad1bd97aac
commit 0eff9e3708
7 changed files with 499 additions and 8 deletions

View file

@ -65,7 +65,9 @@ class SearchDeps:
def resolve_client(source_name: str, deps: SearchDeps) -> tuple[Any, bool]:
"""Return (client, is_available) for an explicit metadata source request."""
if source_name == 'spotify':
if deps.spotify_client and deps.spotify_client.is_spotify_authenticated():
# Available when real auth OR the no-creds SpotipyFree fallback can serve
# (the client routes to free internally when auth is missing/limited).
if deps.spotify_client and deps.spotify_client.is_spotify_metadata_available():
return deps.spotify_client, True
return None, False
if source_name == 'itunes':

View file

@ -569,6 +569,61 @@ class SpotifyClient:
return self._itunes # Fall back to iTunes if no Discogs token
return self._itunes
def _free_enabled(self) -> bool:
"""Opt-in switch for the no-creds SpotipyFree ('Spotify Free') source.
Default OFF the user explicitly enables it (Settings), so there's no
surprise web-scraping for people who simply haven't connected Spotify."""
try:
return bool(config_manager.get('metadata.spotify_free', False))
except Exception:
return False
def _free_available(self) -> bool:
"""Whether the no-creds fallback can be offered at all: enabled in
config AND the SpotipyFree package is installed."""
from core.spotify_free_metadata import spotify_free_installed
return self._free_enabled() and spotify_free_installed()
def is_spotify_metadata_available(self) -> bool:
"""Whether SoulSync can serve Spotify metadata — real auth OR the
no-creds fallback. Availability gates (search resolve, enrichment
worker, watchlist) use THIS instead of ``is_spotify_authenticated()``
so the fallback is reachable. Does NOT change auth semantics."""
from core.spotify_free_metadata import should_offer_spotify_metadata
try:
authed = self.is_spotify_authenticated()
except Exception:
authed = False
return should_offer_spotify_metadata(authed, self._free_available())
def _free_active(self) -> bool:
"""Whether the no-creds SpotipyFree fallback may serve THIS request.
Only when official Spotify can't (no auth, or rate-limited — note
``is_spotify_authenticated()`` already returns False during a rate-limit
ban) AND the fallback is available. When authed + healthy the official
path returns before any fallback, so this never opens. Delegates the
pure decision to ``core.spotify_free_metadata.should_use_free_fallback``."""
from core.spotify_free_metadata import should_use_free_fallback
if not self._free_available():
return False
try:
authed = self.is_spotify_authenticated()
except Exception:
authed = False
return should_use_free_fallback(authed, _is_globally_rate_limited())
@property
def _free_meta(self):
"""Lazy SpotipyFree-backed metadata client (soft import — absence is
not fatal; methods degrade to the iTunes/Deezer fallback)."""
client = getattr(self, '_free_meta_client', None)
if client is None:
from core.spotify_free_metadata import SpotifyFreeMetadataClient
client = SpotifyFreeMetadataClient()
self._free_meta_client = client
return client
def reload_config(self):
"""Reload configuration and re-initialize client"""
self._invalidate_auth_cache()
@ -1276,6 +1331,17 @@ class SpotifyClient:
logger.error(f"Error searching tracks via Spotify: {e}")
# Fall through to fallback
# No-creds Spotify (SpotipyFree) before the iTunes/Deezer fallback —
# only when official Spotify is unavailable (no auth / rate-limited).
if allow_fallback and self._free_active():
try:
objs = [Track.from_spotify_track(t)
for t in self._free_meta.search_tracks(query, effective_limit)]
if objs:
return objs
except Exception as e:
logger.debug("SpotipyFree track search failed: %s", e)
# Fallback (iTunes or Deezer — configured in settings)
if allow_fallback:
logger.debug(f"Using {self._fallback_source} fallback for track search: {query}")
@ -1336,6 +1402,21 @@ class SpotifyClient:
logger.error(f"Error searching artists via Spotify: {e}")
# Fall through to iTunes fallback
# No-creds Spotify (SpotipyFree): keep Spotify catalog/matching when
# official Spotify can't serve us (no auth / rate-limited), before the
# iTunes/Deezer fallback. Gated by _free_active() so it never runs while
# auth is healthy.
if allow_fallback and self._free_active():
try:
objs = [Artist.from_spotify_artist(a)
for a in self._free_meta.search_artists(query, limit)]
if objs:
query_lower = query.lower().strip()
objs.sort(key=lambda a: (0 if a.name.lower().strip() == query_lower else 1))
return objs
except Exception as e:
logger.debug("SpotipyFree artist search failed: %s", e)
# Fallback (iTunes or Deezer)
if allow_fallback:
logger.debug(f"Using {self._fallback_source} fallback for artist search: {query}")
@ -1437,6 +1518,13 @@ class SpotifyClient:
logger.error(f"Error fetching track details via Spotify: {e}")
# Fall through to iTunes fallback
# No-creds Spotify (SpotipyFree) for a real Spotify track id when
# official Spotify is unavailable (no auth / rate-limited).
if allow_fallback and self._free_active() and not self._is_itunes_id(track_id):
free = self._free_meta.get_track_details(track_id)
if free:
return free
# Fallback - only if ID is numeric (non-Spotify format)
if allow_fallback and self._is_itunes_id(track_id):
logger.debug(f"Using {fallback_src} fallback for track details: {track_id}")
@ -1528,6 +1616,13 @@ class SpotifyClient:
logger.error(f"Error fetching album via Spotify: {e}")
# Fall through to fallback
# No-creds Spotify (SpotipyFree) for a real Spotify album id when
# official Spotify is unavailable (no auth / rate-limited).
if allow_fallback and self._free_active() and not self._is_itunes_id(album_id):
free = self._free_meta.get_album(album_id)
if free:
return free
# Fallback - only if ID is numeric (non-Spotify format)
if allow_fallback and self._is_itunes_id(album_id):
logger.debug(f"Using {fallback_src} fallback for album: {album_id}")
@ -1605,6 +1700,13 @@ class SpotifyClient:
logger.error(f"Error fetching album tracks via Spotify: {e}")
# Fall through to iTunes fallback
# No-creds Spotify (SpotipyFree) for a real Spotify album id when
# official Spotify is unavailable (no auth / rate-limited).
if allow_fallback and self._free_active() and not self._is_itunes_id(album_id):
free = self._free_meta.get_album_tracks(album_id)
if free:
return free
# Fallback - only if ID is numeric (non-Spotify format)
if allow_fallback and self._is_itunes_id(album_id):
logger.debug(f"Using {fallback_src} fallback for album tracks: {album_id}")
@ -1691,6 +1793,18 @@ class SpotifyClient:
logger.error(f"Error fetching artist albums via Spotify: {e}")
# Fall through to iTunes fallback
# No-creds Spotify (SpotipyFree) for a real Spotify artist id when
# official Spotify is unavailable (no auth / rate-limited). The free
# discography is already album-shaped — project to Album dataclasses.
if allow_fallback and self._free_active() and not self._is_itunes_id(artist_id):
try:
free = [Album.from_spotify_album(a)
for a in self._free_meta.get_artist_albums_list(artist_id, limit)]
if free:
return free
except Exception as e:
logger.debug("SpotipyFree artist albums failed: %s", e)
# Fallback - only if ID is numeric (non-Spotify format)
if allow_fallback and self._is_itunes_id(artist_id):
logger.debug(f"Using {fallback_src} fallback for artist albums: {artist_id}")
@ -1745,6 +1859,13 @@ class SpotifyClient:
logger.error(f"Error fetching artist via Spotify: {e}")
# Fall through to iTunes fallback
# No-creds Spotify (SpotipyFree) for a real Spotify artist id when
# official Spotify is unavailable (no auth / rate-limited).
if allow_fallback and self._free_active() and not self._is_itunes_id(artist_id):
free = self._free_meta.get_artist(artist_id)
if free:
return free
# Fallback - only if ID is numeric (non-Spotify format)
if allow_fallback and self._is_itunes_id(artist_id):
logger.debug(f"Using {fallback_src} fallback for artist: {artist_id}")

View file

@ -0,0 +1,210 @@
"""PROTOTYPE — no-credentials Spotify metadata via SpotipyFree / spotapi.
Goal: stand in as a READ-ONLY Spotify metadata source when the user has no
Spotify auth (or is rate-limited), mapping SpotipyFree's outputs onto the same
Spotify-compatible shapes the rest of SoulSync already consumes
(see core/spotify_client.py + core/search/sources.py).
Unofficial / web-player scraping best-effort, fragile, and NOT a substitute
for the user-account features (those need a real login).
Capabilities (verified live, 2026-06):
search_tracks SpotipyFree.search (already official-shaped)
search_artists spotapi.Public().artist_search (normalized here)
search_albums no album-name search exists upstream returns []
get_album SpotipyFree.album (already official-shaped)
get_artist_albums_list SpotipyFree.artist_albums (already official-shaped)
get_track_details SpotipyFree.track (already official-shaped)
get_artist SpotipyFree.artist (RAW GraphQL normalized here)
get_artist_top_tracks unavailable
audio_features unavailable (Spotify deprecated it anyway)
This module is import-safe even when SpotipyFree isn't installed — it
soft-imports inside the client factory.
"""
from __future__ import annotations
import importlib.util
import logging
from typing import Any, Optional
logger = logging.getLogger(__name__)
_installed_cache: Optional[bool] = None
def spotify_free_installed() -> bool:
"""Cheap, cached check: is the optional SpotipyFree package importable?
Used by the availability gates never constructs a client or hits the
network. Absence just means we degrade to the iTunes/Deezer fallback.
"""
global _installed_cache
if _installed_cache is None:
_installed_cache = importlib.util.find_spec('SpotipyFree') is not None
return _installed_cache
def should_use_free_fallback(authenticated: bool, rate_limited: bool) -> bool:
"""The per-request gate: the no-creds SpotipyFree source may serve a request
ONLY when official Spotify can't — i.e. the user has no Spotify auth, or
we're currently rate-limited. When authed AND healthy the official path
returns before any fallback, so this never opens.
"""
return (not authenticated) or rate_limited
def should_offer_spotify_metadata(authenticated: bool, free_available: bool) -> bool:
"""The availability gate: SoulSync can serve *some* Spotify metadata when
either real auth is present, or the no-creds fallback is available. The
upstream gates (search resolve, enrichment worker, watchlist) use this so
the fallback is actually reachable without changing what
``is_spotify_authenticated()`` means anywhere.
"""
return authenticated or free_available
# --------------------------------------------------------------------------
# Normalizers (pure — unit-testable against captured fixtures)
# --------------------------------------------------------------------------
def normalize_artist(raw: dict) -> dict:
"""Map a raw SpotipyFree/spotapi artist object (from ``artist()`` or an
``artist_search`` item's ``data``) onto the Spotify-compatible artist dict
SoulSync expects: ``{id, name, images, genres, followers, external_urls}``.
Artist-search items carry no usable image (only color swatches), so
``images`` may be empty there SoulSync lazy-loads artist art separately.
Genres aren't provided by the web player at all.
"""
raw = raw or {}
profile = raw.get('profile') or {}
uri = raw.get('uri') or ''
artist_id = raw.get('id') or (uri.split(':')[-1] if uri else '')
name = profile.get('name') or raw.get('name') or ''
images = []
avatar = (raw.get('visuals') or {}).get('avatarImage') or {}
for src in (avatar.get('sources') or []):
if src.get('url'):
images.append({
'url': src['url'],
'height': src.get('height'),
'width': src.get('width'),
})
followers = (raw.get('stats') or {}).get('followers')
return {
'id': str(artist_id),
'name': name,
'images': images,
'genres': [], # web player doesn't expose genres
'followers': {'total': followers or 0},
'external_urls': (
{'spotify': f'https://open.spotify.com/artist/{artist_id}'}
if artist_id else {}
),
}
# --------------------------------------------------------------------------
# Client
# --------------------------------------------------------------------------
class SpotifyFreeMetadataClient:
"""Read-only Spotify metadata via SpotipyFree, normalized to SoulSync's
Spotify-compatible shapes. Methods mirror the metadata-source interface."""
def __init__(self):
self._sf = None # SpotipyFree.Spotify instance
self._public = None # spotapi.Public() for artist_search
# -- lazy clients (soft import; absence is not fatal) ------------------
def _sf_client(self):
if self._sf is None:
from SpotipyFree import Spotify # optional, user-installed
self._sf = Spotify()
return self._sf
def _public_client(self):
if self._public is None:
import spotapi
self._public = spotapi.Public()
return self._public
def is_available(self) -> bool:
try:
self._sf_client()
return True
except Exception as e:
logger.debug(f"SpotipyFree unavailable: {e}")
return False
# -- search -----------------------------------------------------------
def search_tracks(self, query: str, limit: int = 10) -> list[dict]:
try:
res = self._sf_client().search(query, limit=limit) or {}
items = ((res.get('tracks') or {}).get('items')) or []
return items[:limit]
except Exception as e:
logger.debug(f"SpotipyFree search_tracks failed: {e}")
return []
def search_artists(self, query: str, limit: int = 10) -> list[dict]:
try:
pages = self._public_client().artist_search(query)
first = next(iter(pages), [])
out = []
for item in first[:limit]:
data = item.get('data') if isinstance(item, dict) else None
if data:
out.append(normalize_artist(data))
return out
except Exception as e:
logger.debug(f"SpotipyFree search_artists failed: {e}")
return []
def search_albums(self, query: str, limit: int = 10) -> list[dict]:
# No album-name search exists in SpotipyFree/spotapi. Albums are only
# reachable by id or via an artist's discography.
return []
# -- entity lookups ---------------------------------------------------
def get_album(self, album_id: str, include_tracks: bool = True) -> Optional[dict]:
try:
return self._sf_client().album(album_id)
except Exception as e:
logger.debug(f"SpotipyFree get_album({album_id}) failed: {e}")
return None
def get_track_details(self, track_id: str) -> Optional[dict]:
try:
return self._sf_client().track(track_id)
except Exception as e:
logger.debug(f"SpotipyFree get_track_details({track_id}) failed: {e}")
return None
def get_album_tracks(self, album_id: str) -> Optional[dict]:
try:
return self._sf_client().album_tracks(album_id)
except Exception as e:
logger.debug(f"SpotipyFree get_album_tracks({album_id}) failed: {e}")
return None
def get_artist(self, artist_id: str) -> Optional[dict]:
try:
raw = self._sf_client().artist(artist_id)
return normalize_artist(raw) if raw else None
except Exception as e:
logger.debug(f"SpotipyFree get_artist({artist_id}) failed: {e}")
return None
def get_artist_albums_list(self, artist_id: str, limit: int = 50) -> list[dict]:
try:
res = self._sf_client().artist_albums(artist_id) or {}
return (res.get('items') or [])[:limit]
except Exception as e:
logger.debug(f"SpotipyFree get_artist_albums_list({artist_id}) failed: {e}")
return []

View file

@ -215,10 +215,12 @@ class SpotifyWorker:
# We intentionally avoid calling is_spotify_authenticated() here
# because it makes an API probe that can re-trigger rate limits
# and lock users in an infinite rate-limit loop.
if not self.client.is_spotify_authenticated():
# Available = real auth OR the no-creds SpotipyFree fallback
# (enrichment is metadata-only, so the free source can serve it).
if not self.client.is_spotify_metadata_available():
self.client.reload_config()
if not self.client.is_spotify_authenticated():
logger.debug("Spotify not authenticated, sleeping 30s...")
if not self.client.is_spotify_metadata_available():
logger.debug("Spotify metadata unavailable, sleeping 30s...")
interruptible_sleep(self._stop_event, 30)
continue

View file

@ -489,12 +489,16 @@ class WatchlistScanner:
self._spotify_disabled_reason = reason
def _spotify_available_for_run(self) -> bool:
"""Check if Spotify should be used for this run."""
"""Check if Spotify should be used for this run.
Available = real auth OR the no-creds SpotipyFree fallback (new-release
detection is metadata-only get_artist_albums so the free source can
serve it; the client routes internally when auth is missing/limited)."""
if self._spotify_disabled_for_run:
return False
if not self.spotify_client:
return False
return self.spotify_client.is_spotify_authenticated()
return self.spotify_client.is_spotify_metadata_available()
def _spotify_is_primary_source(self) -> bool:
"""Check if Spotify is both authenticated and the configured primary metadata source.

View file

@ -50,7 +50,7 @@ class _Track:
class _Client:
def __init__(self, *, name='fake', artists=None, albums=None, tracks=None,
fail_search=False, authed=True, connected=True):
fail_search=False, authed=True, connected=True, meta_available=None):
self.name = name
self._artists = artists or []
self._albums = albums or []
@ -58,6 +58,10 @@ class _Client:
self._fail = fail_search
self._authed = authed
self._connected = connected
# When unset, metadata availability tracks auth (the common case). Set
# explicitly to model the no-creds SpotipyFree fallback: not authed but
# metadata still available.
self._meta_available = meta_available
def search_artists(self, q, limit=10):
if self._fail:
@ -77,6 +81,9 @@ class _Client:
def is_spotify_authenticated(self):
return self._authed
def is_spotify_metadata_available(self):
return self._authed if self._meta_available is None else self._meta_available
def is_connected(self):
return self._connected
@ -157,12 +164,22 @@ def test_resolve_spotify_authed_returns_client():
def test_resolve_spotify_unauthed_returns_none():
deps = _build_deps(spotify_client=_Client(authed=False))
# No auth AND no free fallback available → Spotify source unavailable.
deps = _build_deps(spotify_client=_Client(authed=False, meta_available=False))
client, ok = orchestrator.resolve_client('spotify', deps)
assert client is None
assert ok is False
def test_resolve_spotify_unauthed_but_free_available_returns_client():
# #798: no Spotify auth but the no-creds SpotipyFree fallback is available →
# the Spotify source stays usable (the client routes to free internally).
deps = _build_deps(spotify_client=_Client(authed=False, meta_available=True))
client, ok = orchestrator.resolve_client('spotify', deps)
assert client is deps.spotify_client
assert ok is True
def test_resolve_spotify_missing_returns_none():
deps = _build_deps(spotify_client=None)
client, ok = orchestrator.resolve_client('spotify', deps)

View file

@ -0,0 +1,135 @@
"""Tests for core/spotify_free_metadata.py — the no-creds Spotify fallback.
Pure-unit only: the live SpotipyFree calls hit the network and can't run in CI,
but the two things that actually need pinning are pure the activation gate and
the artist normalizer (the one shape SpotipyFree returns raw). Fixtures below
are trimmed from real captured responses (2026-06).
"""
from __future__ import annotations
from core.spotify_free_metadata import (
normalize_artist,
should_offer_spotify_metadata,
should_use_free_fallback,
spotify_free_installed,
)
# ---------------------------------------------------------------------------
# should_use_free_fallback — the activation gate
# ---------------------------------------------------------------------------
def test_gate_open_when_no_auth():
assert should_use_free_fallback(authenticated=False, rate_limited=False) is True
def test_gate_open_when_rate_limited_even_if_authed():
assert should_use_free_fallback(authenticated=True, rate_limited=True) is True
def test_gate_closed_when_authed_and_healthy():
# The critical guarantee: with working Spotify auth and no rate limit, the
# free source must NEVER activate.
assert should_use_free_fallback(authenticated=True, rate_limited=False) is False
def test_gate_open_when_no_auth_and_rate_limited():
assert should_use_free_fallback(authenticated=False, rate_limited=True) is True
# ---------------------------------------------------------------------------
# should_offer_spotify_metadata — the availability gate the callers use
# ---------------------------------------------------------------------------
def test_offer_when_authed_even_if_no_free():
assert should_offer_spotify_metadata(authenticated=True, free_available=False) is True
def test_offer_when_free_available_even_if_no_auth():
# The whole point: no auth but free available → Spotify source stays usable.
assert should_offer_spotify_metadata(authenticated=False, free_available=True) is True
def test_not_offered_when_neither():
assert should_offer_spotify_metadata(authenticated=False, free_available=False) is False
def test_full_composition_authed_healthy_never_uses_free():
# The critical no-regression guarantee end to end: authed + healthy →
# offered (official), and the per-request gate stays CLOSED.
assert should_offer_spotify_metadata(authenticated=True, free_available=True) is True
assert should_use_free_fallback(authenticated=True, rate_limited=False) is False
def test_spotify_free_installed_returns_bool():
# Whatever the env, it's a cached bool and doesn't raise / hit network.
assert isinstance(spotify_free_installed(), bool)
# ---------------------------------------------------------------------------
# normalize_artist — raw web-player GraphQL → Spotify-compatible artist dict
# ---------------------------------------------------------------------------
# Trimmed from a real `SpotipyFree.artist()` response.
_RAW_ARTIST = {
'__typename': 'Artist',
'id': '4Z8W4fKeB5YxbusRsdQVPb',
'uri': 'spotify:artist:4Z8W4fKeB5YxbusRsdQVPb',
'profile': {'name': 'Radiohead', 'biography': {'text': '...'}},
'stats': {'followers': 16239336, 'monthlyListeners': 45139421},
'visuals': {
'avatarImage': {
'sources': [
{'height': 640, 'width': 640, 'url': 'https://i.scdn.co/image/big'},
{'height': 160, 'width': 160, 'url': 'https://i.scdn.co/image/small'},
]
}
},
}
# Trimmed from a real `artist_search` item's `data` (no usable image — only
# color swatches under visualIdentity; SoulSync lazy-loads art separately).
_RAW_SEARCH_ARTIST = {
'__typename': 'Artist',
'uri': 'spotify:artist:1dfeR4HaWDbWqFHLkxsg1d',
'profile': {'name': 'Queen'},
'visualIdentity': {'squareCoverImage': {'extractedColorSet': {}}},
}
def test_normalize_artist_full_shape():
out = normalize_artist(_RAW_ARTIST)
assert out['id'] == '4Z8W4fKeB5YxbusRsdQVPb'
assert out['name'] == 'Radiohead'
assert out['followers'] == {'total': 16239336}
assert out['images'][0]['url'] == 'https://i.scdn.co/image/big'
assert len(out['images']) == 2
assert out['external_urls'] == {
'spotify': 'https://open.spotify.com/artist/4Z8W4fKeB5YxbusRsdQVPb'
}
assert out['genres'] == [] # web player doesn't provide genres
def test_normalize_artist_id_from_uri_when_no_id_field():
out = normalize_artist(_RAW_SEARCH_ARTIST)
assert out['id'] == '1dfeR4HaWDbWqFHLkxsg1d'
assert out['name'] == 'Queen'
assert out['images'] == [] # search items carry no usable image
assert out['external_urls']['spotify'].endswith('1dfeR4HaWDbWqFHLkxsg1d')
def test_normalize_artist_handles_empty_and_none():
for bad in (None, {}, {'profile': None, 'stats': None, 'visuals': None}):
out = normalize_artist(bad)
assert out['name'] == ''
assert out['images'] == []
assert out['followers'] == {'total': 0}
assert out['external_urls'] == {}
def test_normalize_artist_skips_imageless_sources():
raw = {'uri': 'spotify:artist:x', 'profile': {'name': 'A'},
'visuals': {'avatarImage': {'sources': [{'height': 1}, {'url': 'u'}]}}}
out = normalize_artist(raw)
assert out['images'] == [{'url': 'u', 'height': None, 'width': None}]