Spotify Free: close the album-search gap via the artist's discography
SpotipyFree has no album-name search (search_albums returned []), and SpotifyClient.search_albums had no Free branch at all — so when Spotify Free was the active source (no-auth / rate-limit ban / spent budget), album matching couldn't use Spotify and dropped straight to the iTunes/Deezer fallback. The enrichment worker's per-album matching was effectively blind on Free. Fix — work the gap using ONLY the free source: - spotify_free_metadata: new search_albums_via_artist(artist, album) resolves the best-matching artist, scans their discography (get_artist_albums_list — which Free DOES support), and ranks by album-name similarity. Pure rank_albums_by_name helper is unit-testable; the network method composes it. - SpotifyClient.search_albums: add a Free branch mirroring search_artists/tracks (official → Free-if-active → iTunes/Deezer), plus optional artist/album params so the Free path has the names it needs. Bare-query callers skip it unchanged. - spotify_worker: _process_album_individual passes artist+album so the bridge can match albums on Free. Tests: rank ordering + limit, via-artist picks the right artist and ranks, empty on missing artist/album/no-match, and SpotifyClient.search_albums returns Free albums (not the iTunes fallback) when free is active. 210 Spotify tests pass.
This commit is contained in:
parent
2bc86e3022
commit
4a6d6fc17b
4 changed files with 156 additions and 4 deletions
|
|
@ -1548,11 +1548,17 @@ class SpotifyClient:
|
|||
return []
|
||||
|
||||
@rate_limited
|
||||
def search_albums(self, query: str, limit: int = 10, allow_fallback: bool = True) -> List[Album]:
|
||||
def search_albums(self, query: str, limit: int = 10, allow_fallback: bool = True,
|
||||
artist: str = None, album: str = None) -> List[Album]:
|
||||
"""Search for albums.
|
||||
|
||||
When allow_fallback is True, falls back to the configured metadata source
|
||||
if Spotify is unavailable or returns an error.
|
||||
|
||||
``artist`` + ``album`` (the names, passed separately) enable the no-creds
|
||||
Spotify Free path: SpotipyFree has no album-name search, so when Free is
|
||||
active it resolves the album via the artist's discography. Callers without
|
||||
that context (a bare query) skip the Free album path.
|
||||
"""
|
||||
cache = get_metadata_cache()
|
||||
# Check Spotify cache first so cached data remains usable even when
|
||||
|
|
@ -1592,7 +1598,21 @@ class SpotifyClient:
|
|||
except Exception as e:
|
||||
_detect_and_set_rate_limit(e, 'search_albums')
|
||||
logger.error(f"Error searching albums via Spotify: {e}")
|
||||
# Fall through to iTunes fallback
|
||||
# Fall through to free / iTunes fallback
|
||||
|
||||
# No-creds Spotify (SpotipyFree): keep Spotify catalog/matching when
|
||||
# official Spotify can't serve us (no auth / rate-limited / budget spent),
|
||||
# before the iTunes/Deezer fallback. Albums have no name-search upstream,
|
||||
# so resolve via the artist's discography — needs artist + album names.
|
||||
# Gated by _free_active() so it never runs while auth is healthy.
|
||||
if allow_fallback and self._free_active() and artist and album:
|
||||
try:
|
||||
objs = [Album.from_spotify_album(a)
|
||||
for a in self._free_meta.search_albums_via_artist(artist, album, min(limit, 10))]
|
||||
if objs:
|
||||
return objs
|
||||
except Exception as e:
|
||||
logger.debug("SpotipyFree album search failed: %s", e)
|
||||
|
||||
# Fallback (iTunes or Deezer)
|
||||
if allow_fallback:
|
||||
|
|
|
|||
|
|
@ -27,10 +27,31 @@ from __future__ import annotations
|
|||
|
||||
import importlib.util
|
||||
import logging
|
||||
from difflib import SequenceMatcher
|
||||
from typing import Any, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _norm_name(s: str) -> str:
|
||||
"""Lowercase + collapse whitespace for loose name comparison."""
|
||||
return ' '.join((s or '').lower().split())
|
||||
|
||||
|
||||
def rank_albums_by_name(albums: list[dict], album_name: str, limit: int = 10) -> list[dict]:
|
||||
"""Rank Spotify-shaped album dicts by name similarity to ``album_name`` (best
|
||||
first) and return up to ``limit``. Pure (no network) so it's unit-testable.
|
||||
|
||||
Used to turn an artist's whole discography into the most-relevant album
|
||||
candidates, since SpotipyFree has no album-name search of its own."""
|
||||
target = _norm_name(album_name)
|
||||
scored = [
|
||||
(SequenceMatcher(None, _norm_name((alb or {}).get('name', '')), target).ratio(), alb)
|
||||
for alb in (albums or [])
|
||||
]
|
||||
scored.sort(key=lambda t: t[0], reverse=True)
|
||||
return [alb for _score, alb in scored[:limit]]
|
||||
|
||||
_installed_cache: Optional[bool] = None
|
||||
|
||||
|
||||
|
|
@ -186,9 +207,46 @@ class SpotifyFreeMetadataClient:
|
|||
|
||||
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.
|
||||
# reachable by id or via an artist's discography — see
|
||||
# search_albums_via_artist() for the artist-scoped workaround.
|
||||
return []
|
||||
|
||||
def search_albums_via_artist(self, artist_name: str, album_name: str,
|
||||
limit: int = 10) -> list[dict]:
|
||||
"""Find an album by name using ONLY the free source: SpotipyFree has no
|
||||
album-name search, so resolve the artist (best name match) and scan their
|
||||
discography for the album. Returns Spotify-shaped album dicts ranked by
|
||||
similarity to ``album_name`` (best first); the caller applies its own
|
||||
match threshold. Empty when artist/album is missing or no artist matches.
|
||||
|
||||
This is what lets the budget/rate-limit bridge match albums on Spotify
|
||||
Free instead of being unable to (the gap that previously dropped album
|
||||
enrichment straight through to the iTunes/Deezer fallback)."""
|
||||
if not (artist_name and album_name):
|
||||
return []
|
||||
try:
|
||||
artists = self.search_artists(artist_name, limit=5)
|
||||
except Exception as e:
|
||||
logger.debug(f"SpotipyFree search_albums_via_artist artist lookup failed: {e}")
|
||||
return []
|
||||
if not artists:
|
||||
return []
|
||||
target_artist = _norm_name(artist_name)
|
||||
best = max(
|
||||
artists,
|
||||
key=lambda a: SequenceMatcher(
|
||||
None, _norm_name(a.get('name', '')), target_artist).ratio(),
|
||||
)
|
||||
artist_id = best.get('id')
|
||||
if not artist_id:
|
||||
return []
|
||||
try:
|
||||
albums = self.get_artist_albums_list(artist_id, limit=50)
|
||||
except Exception as e:
|
||||
logger.debug(f"SpotipyFree search_albums_via_artist discography failed: {e}")
|
||||
return []
|
||||
return rank_albums_by_name(albums, album_name, limit)
|
||||
|
||||
# -- entity lookups ---------------------------------------------------
|
||||
def get_album(self, album_id: str, include_tracks: bool = True) -> Optional[dict]:
|
||||
try:
|
||||
|
|
|
|||
|
|
@ -768,7 +768,10 @@ class SpotifyWorker:
|
|||
return
|
||||
|
||||
query = f"{artist_name} {album_name}" if artist_name else album_name
|
||||
results = self.client.search_albums(query, limit=5)
|
||||
# Pass artist + album names separately too, so the no-creds Spotify Free
|
||||
# path can resolve the album via the artist's discography (SpotipyFree has
|
||||
# no album-name search) when bridging a budget/rate-limit ban.
|
||||
results = self.client.search_albums(query, limit=5, artist=artist_name, album=album_name)
|
||||
|
||||
if not results:
|
||||
self._mark_status('album', album_id, 'not_found')
|
||||
|
|
|
|||
|
|
@ -234,3 +234,74 @@ def test_selected_but_package_missing_is_graceful():
|
|||
def test_connected_ratelimited_but_no_package_no_bridge():
|
||||
avail, free = _gate(authed=False, has_creds=True, selected=False, installed=False, rate_limited=True)
|
||||
assert avail is False and free is False
|
||||
|
||||
|
||||
# ── #(free album search): rank + artist-discography workaround ──────────────
|
||||
|
||||
from core.spotify_free_metadata import ( # noqa: E402
|
||||
rank_albums_by_name, SpotifyFreeMetadataClient,
|
||||
)
|
||||
|
||||
|
||||
def test_rank_albums_by_name_orders_best_first_and_limits():
|
||||
albums = [{'name': 'Random Access Memories'}, {'name': 'GNX'},
|
||||
{'name': 'GNX (Deluxe)'}, {'name': 'untitled unmastered.'}]
|
||||
out = rank_albums_by_name(albums, 'GNX', limit=2)
|
||||
assert [a['name'] for a in out] == ['GNX', 'GNX (Deluxe)']
|
||||
|
||||
|
||||
def test_rank_albums_by_name_handles_empty():
|
||||
assert rank_albums_by_name([], 'GNX') == []
|
||||
assert rank_albums_by_name(None, 'GNX') == []
|
||||
|
||||
|
||||
def test_search_albums_via_artist_resolves_through_discography():
|
||||
c = SpotifyFreeMetadataClient()
|
||||
c.search_artists = lambda q, limit=5: [
|
||||
{'id': 'art_other', 'name': 'Some Other Band'},
|
||||
{'id': 'art_k', 'name': 'Kendrick Lamar'},
|
||||
]
|
||||
c.get_artist_albums_list = lambda aid, limit=50: (
|
||||
[{'id': 'al1', 'name': 'DAMN.'}, {'id': 'al2', 'name': 'GNX'}]
|
||||
if aid == 'art_k' else [])
|
||||
out = c.search_albums_via_artist('Kendrick Lamar', 'GNX', limit=3)
|
||||
assert out and out[0]['name'] == 'GNX' # picked the right artist + ranked
|
||||
|
||||
|
||||
def test_search_albums_via_artist_empty_without_artist_or_album():
|
||||
c = SpotifyFreeMetadataClient()
|
||||
c.search_artists = lambda q, limit=5: [{'id': 'a', 'name': 'X'}]
|
||||
assert c.search_albums_via_artist('', 'GNX') == []
|
||||
assert c.search_albums_via_artist('X', '') == []
|
||||
|
||||
|
||||
def test_search_albums_via_artist_empty_when_no_artist_match():
|
||||
c = SpotifyFreeMetadataClient()
|
||||
c.search_artists = lambda q, limit=5: [] # nothing found
|
||||
assert c.search_albums_via_artist('Nobody', 'GNX') == []
|
||||
|
||||
|
||||
def test_client_search_albums_uses_free_via_artist_when_active():
|
||||
"""SpotifyClient.search_albums bridges album matching through Spotify Free
|
||||
(artist discography) when free is active, instead of dropping to iTunes/Deezer."""
|
||||
c = SpotifyClient.__new__(SpotifyClient)
|
||||
fake_free = SpotifyFreeMetadataClient()
|
||||
fake_free.search_albums_via_artist = lambda artist, album, limit: [
|
||||
{'id': 'al2', 'name': 'GNX',
|
||||
'artists': [{'name': 'Kendrick Lamar', 'id': 'art_k'}]}
|
||||
]
|
||||
c._free_meta_client = fake_free
|
||||
fake_cache = type('C', (), {'get_search_results': lambda *a, **k: None})()
|
||||
|
||||
with patch.object(SpotifyClient, 'is_spotify_authenticated', return_value=False), \
|
||||
patch('core.spotify_client.config_manager') as cm, \
|
||||
patch('core.spotify_client._is_globally_rate_limited', return_value=False), \
|
||||
patch('core.spotify_client.get_metadata_cache', return_value=fake_cache), \
|
||||
patch.object(_sfm, 'spotify_free_installed', return_value=True):
|
||||
cm.get_spotify_config.return_value = {}
|
||||
cm.get.side_effect = lambda k, d=None: True if k == 'metadata.spotify_free' else d
|
||||
results = c.search_albums('Kendrick Lamar GNX', limit=5,
|
||||
artist='Kendrick Lamar', album='GNX')
|
||||
|
||||
assert len(results) == 1 and results[0].name == 'GNX'
|
||||
assert results[0].id == 'al2'
|
||||
|
|
|
|||
Loading…
Reference in a new issue