danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Refactor artist discography lookup

Browse source 
Move artist discography resolution into core metadata_service, introduce MetadataLookupOptions, and keep web_server focused on request handling. Add focused tests for the new service boundary and preserve current fallback behavior for now.
This commit is contained in:
Antti Kettunen 2026-04-18 18:41:31 +03:00
parent 1905678c7b
commit 2b575a59ae
3 changed files with 494 additions and 146 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

235
core/metadata_service.py
View file

@ -3,11 +3,12 @@ Metadata Service - Centralized metadata source selection
ALL metadata source decisions flow through this module. Other files import
get_primary_source() and get_primary_client() instead of reimplementing
the logic. This prevents bugs where different files have different defaults
or auth checks.
the logic. This prevents bugs where different files have different defaults,
auth checks, or source-fallback behavior.
"""
import threading
from dataclasses import dataclass
from typing import List, Optional, Dict, Any, Literal
from core.spotify_client import SpotifyClient
from core.itunes_client import iTunesClient
@ -24,6 +25,17 @@ _client_cache_lock = threading.RLock()
_client_cache: Dict[str, Any] = {}
@dataclass(frozen=True)
class MetadataLookupOptions:
"""Generic metadata lookup policy shared by metadata services."""
source_override: Optional[str] = None
allow_fallback: bool = True
skip_cache: bool = False
max_pages: int = 0
limit: int = 50
# =============================================================================
# CANONICAL SOURCE SELECTION — all code should use these two functions
# =============================================================================
@ -138,6 +150,7 @@ def get_album_tracks_for_source(source: str, album_id: str):
def get_artist_albums_for_source(
source: str,
artist_id: str,
artist_name: str = '',
album_type: str = 'album,single',
limit: int = 50,
skip_cache: bool = False,
@ -146,7 +159,8 @@ def get_artist_albums_for_source(
"""Get artist albums for an exact source.
Returns a provider-native album list or None if the source is unavailable.
No fallback swaps.
Tries the requested artist ID first, then falls back to artist-name
search using the same flow for every provider when artist_name is provided.
Set skip_cache=True only for freshness-sensitive flows that need newly
released albums to show up immediately.
@ -155,7 +169,7 @@ def get_artist_albums_for_source(
if not client or not artist_id or not hasattr(client, 'get_artist_albums'):
return None
try:
def _fetch_for_artist(target_artist_id: str):
kwargs = {
'album_type': album_type,
'limit': limit,
@ -164,11 +178,222 @@ def get_artist_albums_for_source(
kwargs['allow_fallback'] = False
kwargs['skip_cache'] = skip_cache
kwargs['max_pages'] = max_pages
return client.get_artist_albums(artist_id, **kwargs)
return client.get_artist_albums(target_artist_id, **kwargs)
try:
albums = _fetch_for_artist(artist_id) or []
if albums:
return albums
if not artist_name:
return albums
search_results = _search_artists_for_source(source, client, artist_name, limit=5)
if not search_results:
return albums
best = _pick_best_artist_match(search_results, artist_name)
if not best:
return albums
found_artist_id = _extract_lookup_value(best, 'id', 'artist_id')
if not found_artist_id:
return albums
resolved = _fetch_for_artist(found_artist_id) or []
if resolved:
logger.debug("Found %s artist '%s' (id=%s)", source, _extract_lookup_value(best, 'name', 'artist_name', 'title'), found_artist_id)
return resolved
except Exception:
return None
def _get_source_chain_for_lookup(options: MetadataLookupOptions) -> List[str]:
primary_source = get_primary_source()
source_chain = list(get_source_priority(primary_source))
override = (options.source_override or '').strip().lower()
if override:
source_chain = [override] + [source for source in source_chain if source != override]
if not options.allow_fallback:
source_chain = source_chain[:1]
return source_chain
def _extract_lookup_value(value: Any, *names: str, default: Any = None) -> Any:
if value is None:
return default
for name in names:
if isinstance(value, dict):
if name in value and value[name] is not None:
return value[name]
else:
candidate = getattr(value, name, None)
if candidate is not None:
return candidate
return default
def _normalize_artist_name(value: Any) -> str:
return (value or '').strip().casefold()
def _search_artists_for_source(source: str, client: Any, artist_name: str, limit: int = 5) -> List[Any]:
if not client or not hasattr(client, 'search_artists'):
return []
try:
kwargs = {'limit': limit}
if source == 'spotify':
kwargs['allow_fallback'] = False
return client.search_artists(artist_name, **kwargs) or []
except Exception as exc:
logger.debug("Could not search %s for %s: %s", source, artist_name, exc)
return []
def _pick_best_artist_match(search_results: List[Any], artist_name: str) -> Optional[Any]:
"""Prefer an exact artist-name match, otherwise use the first result."""
if not search_results:
return None
target_name = _normalize_artist_name(artist_name)
for artist in search_results:
candidate_name = _normalize_artist_name(
_extract_lookup_value(artist, 'name', 'artist_name', 'title')
)
if candidate_name == target_name:
return artist
return search_results[0]
def _build_discography_release_dict(release: Any, artist_id: str) -> Optional[Dict[str, Any]]:
release_id = _extract_lookup_value(release, 'id', 'album_id', 'release_id')
if not release_id:
return None
artist_ids = _extract_lookup_value(release, 'artist_ids') or []
if isinstance(artist_ids, (str, bytes)):
artist_ids = [artist_ids]
else:
try:
artist_ids = list(artist_ids)
except TypeError:
artist_ids = [artist_ids]
if artist_ids and str(artist_ids[0]) != str(artist_id):
return None
album_type = _extract_lookup_value(release, 'album_type', default='album') or 'album'
release_date = _extract_lookup_value(release, 'release_date')
return {
'id': release_id,
'name': _extract_lookup_value(release, 'name', 'title', default=release_id),
'release_date': release_date,
'album_type': album_type,
'image_url': _extract_lookup_value(release, 'image_url', 'thumb_url', 'cover_image'),
'total_tracks': _extract_lookup_value(release, 'total_tracks', default=0) or 0,
'external_urls': _extract_lookup_value(release, 'external_urls', default={}) or {},
}
def _sort_discography_releases(releases: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
def get_release_year(item):
if item.get('release_date'):
try:
return int(str(item['release_date'])[:4])
except (ValueError, IndexError, TypeError):
return 0
return 0
return sorted(releases, key=get_release_year, reverse=True)
def get_artist_discography(
artist_id: str,
artist_name: str = '',
options: Optional[MetadataLookupOptions] = None,
) -> Dict[str, Any]:
"""Get a normalized artist discography with source resolution and fallback.
Each provider uses the same lookup flow:
1. try the requested artist ID
2. if that misses, search by artist name
3. retry with the provider-specific artist ID from the search result
"""
options = options or MetadataLookupOptions()
source_priority = _get_source_chain_for_lookup(options)
albums: List[Any] = []
active_source: Optional[str] = None
if not albums:
for source in source_priority:
client = get_client_for_source(source)
if not client:
continue
try:
albums = get_artist_albums_for_source(
source,
artist_id,
artist_name=artist_name,
limit=options.limit,
skip_cache=options.skip_cache,
max_pages=options.max_pages,
) or []
except Exception as exc:
logger.debug("%s direct lookup failed for artist %s: %s", source, artist_id, exc)
albums = []
if albums:
active_source = source
logger.info("Got %s albums from %s for artist %s", len(albums), source, artist_id)
break
album_list: List[Dict[str, Any]] = []
singles_list: List[Dict[str, Any]] = []
seen_albums = set()
for release in albums or []:
release_data = _build_discography_release_dict(release, artist_id)
if not release_data:
continue
release_id = release_data['id']
if release_id in seen_albums:
continue
seen_albums.add(release_id)
album_type = release_data.get('album_type') or 'album'
if album_type in ['single', 'ep']:
singles_list.append(release_data)
else:
album_list.append(release_data)
album_list = _sort_discography_releases(album_list)
singles_list = _sort_discography_releases(singles_list)
logger.debug(
"Total albums returned for artist %s: %s (source=%s)",
artist_id,
len(album_list) + len(singles_list),
active_source,
)
return {
'albums': album_list,
'singles': singles_list,
'source': active_source or (source_priority[0] if source_priority else 'unknown'),
'source_priority': source_priority,
}
def get_deezer_client():
"""Get cached Deezer client.

250
tests/test_metadata_service_discography.py Normal file
View file

@ -0,0 +1,250 @@
import sys
import types
import pytest
if "spotipy" not in sys.modules:
spotipy = types.ModuleType("spotipy")
class _DummySpotify:
def __init__(self, *args, **kwargs):
pass
oauth2 = types.ModuleType("spotipy.oauth2")
class _DummyOAuth:
def __init__(self, *args, **kwargs):
pass
spotipy.Spotify = _DummySpotify
oauth2.SpotifyOAuth = _DummyOAuth
oauth2.SpotifyClientCredentials = _DummyOAuth
spotipy.oauth2 = oauth2
sys.modules["spotipy"] = spotipy
sys.modules["spotipy.oauth2"] = oauth2
if "config.settings" not in sys.modules:
config_pkg = types.ModuleType("config")
settings_mod = types.ModuleType("config.settings")
class _DummyConfigManager:
def get(self, key, default=None):
return default
settings_mod.config_manager = _DummyConfigManager()
config_pkg.settings = settings_mod
sys.modules["config"] = config_pkg
sys.modules["config.settings"] = settings_mod
from core import metadata_service
from core.metadata_service import MetadataLookupOptions
@pytest.fixture(autouse=True)
def _clear_metadata_client_cache():
metadata_service.clear_cached_metadata_clients()
yield
metadata_service.clear_cached_metadata_clients()
class _FakeSourceClient:
def __init__(self, album_results=None, artist_search_results=None, discography_results=None):
self.album_results = list(album_results or [])
self.artist_search_results = list(artist_search_results or [])
self.discography_results = list(discography_results or [])
self.album_calls = []
self.artist_search_calls = []
self.discography_calls = []
def get_artist_albums(self, artist_id, **kwargs):
self.album_calls.append((artist_id, dict(kwargs)))
return list(self.album_results)
def search_artists(self, query, **kwargs):
self.artist_search_calls.append((query, dict(kwargs)))
return list(self.artist_search_results)
def search_discography(self, query, **kwargs):
self.discography_calls.append((query, dict(kwargs)))
return list(self.discography_results)
def _album(album_id, name, release_date, album_type="album"):
return types.SimpleNamespace(
id=album_id,
name=name,
release_date=release_date,
album_type=album_type,
image_url=f"https://img.example/{album_id}.jpg",
total_tracks=12,
external_urls={"spotify": f"https://example/{album_id}"},
artist_ids=["artist-1"],
)
def _artist(artist_id, name):
return types.SimpleNamespace(id=artist_id, name=name)
def test_get_artist_discography_uses_primary_then_fallback(monkeypatch):
deezer = _FakeSourceClient()
spotify = _FakeSourceClient(
album_results=[
_album("album-old", "Older Album", "2022-01-01"),
_album("single-one", "Single One", "2024-06-01", album_type="single"),
_album("album-new", "New Album", "2024-08-01"),
]
)
itunes = _FakeSourceClient()
clients = {
"deezer": deezer,
"spotify": spotify,
"itunes": itunes,
}
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: clients.get(source))
result = metadata_service.get_artist_discography("artist-1", "Artist One", MetadataLookupOptions())
assert result["source"] == "spotify"
assert result["source_priority"] == ["deezer", "spotify", "itunes"]
assert [album["id"] for album in result["albums"]] == ["album-new", "album-old"]
assert [single["id"] for single in result["singles"]] == ["single-one"]
assert spotify.album_calls == [(
"artist-1",
{
"album_type": "album,single",
"limit": 50,
"allow_fallback": False,
"skip_cache": False,
"max_pages": 0,
},
)]
def test_get_artist_discography_uses_name_search_when_direct_lookup_missing(monkeypatch):
class _SearchThenAlbumClient(_FakeSourceClient):
def get_artist_albums(self, artist_id, **kwargs):
self.album_calls.append((artist_id, dict(kwargs)))
if artist_id == "deezer-artist-1":
return [_album("deezer-album-1", "Deezer Album", "2023-05-01")]
return []
deezer = _SearchThenAlbumClient(
artist_search_results=[_artist("deezer-artist-1", "Artist One")],
)
clients = {"deezer": deezer}
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: clients.get(source))
result = metadata_service.get_artist_discography("artist-1", "Artist One", MetadataLookupOptions())
assert result["source"] == "deezer"
assert [album["id"] for album in result["albums"]] == ["deezer-album-1"]
assert deezer.artist_search_calls == [("Artist One", {"limit": 5})]
assert deezer.album_calls == [
(
"artist-1",
{
"album_type": "album,single",
"limit": 50,
},
),
(
"deezer-artist-1",
{
"album_type": "album,single",
"limit": 50,
},
),
]
def test_get_artist_discography_respects_source_override_without_fallback(monkeypatch):
deezer = _FakeSourceClient()
itunes = _FakeSourceClient(album_results=[_album("itunes-album-1", "iTunes Album", "2024-02-01")])
spotify = _FakeSourceClient()
clients = {
"deezer": deezer,
"itunes": itunes,
"spotify": spotify,
}
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: clients.get(source))
result = metadata_service.get_artist_discography(
"artist-1",
"Artist One",
MetadataLookupOptions(source_override="itunes", allow_fallback=False),
)
assert result["source"] == "itunes"
assert result["source_priority"] == ["itunes"]
assert [album["id"] for album in result["albums"]] == ["itunes-album-1"]
assert deezer.album_calls == []
assert spotify.album_calls == []
assert itunes.album_calls == [
(
"artist-1",
{
"album_type": "album,single",
"limit": 50,
},
)
]
def test_get_artist_discography_uses_hydrabase_fast_path_when_active(monkeypatch):
class _HydrabaseLikeClient(_FakeSourceClient):
def get_artist_albums(self, artist_id, **kwargs):
self.album_calls.append((artist_id, dict(kwargs)))
if artist_id == "hydrabase-artist-1":
return [
_album("hydrabase-album-1", "Hydra Album", "2024-03-01"),
_album("hydrabase-single-1", "Hydra Single", "2024-04-01", album_type="single"),
]
return []
hydrabase = _HydrabaseLikeClient(
artist_search_results=[_artist("hydrabase-artist-1", "Artist One")],
)
clients = {"deezer": None, "spotify": None, "itunes": None}
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes", "hydrabase"])
def fake_get_client_for_source(source):
if source == "hydrabase":
return hydrabase
return clients.get(source)
monkeypatch.setattr(metadata_service, "get_client_for_source", fake_get_client_for_source)
result = metadata_service.get_artist_discography("artist-1", "Artist One", MetadataLookupOptions())
assert result["source"] == "hydrabase"
assert [album["id"] for album in result["albums"]] == ["hydrabase-album-1"]
assert [single["id"] for single in result["singles"]] == ["hydrabase-single-1"]
assert hydrabase.album_calls == [
(
"artist-1",
{
"album_type": "album,single",
"limit": 50,
},
),
(
"hydrabase-artist-1",
{
"album_type": "album,single",
"limit": 50,
},
)
]
assert hydrabase.artist_search_calls == [("Artist One", {"limit": 5})]

155
web_server.py
View file

@ -10949,23 +10949,6 @@ def get_artist_discography(artist_id):
if hydrabase_worker and dev_mode_enabled and artist_name:
hydrabase_worker.enqueue(artist_name, 'artist.albums')
from core.metadata_service import (
get_artist_albums_for_source,
get_client_for_source,
get_primary_source,
get_source_priority,
)
def _search_artists_for_source(client, source: str, name: str, limit: int = 5):
if not client or not hasattr(client, 'search_artists'):
return []
kwargs = {'limit': limit}
if source == 'spotify':
kwargs['allow_fallback'] = False
return client.search_artists(name, **kwargs) or []
primary_source = get_primary_source()
source_priority = list(get_source_priority(primary_source))
effective_override_source = source_override
if source_override == 'hydrabase':
plugin = request.args.get('plugin', '').strip().lower()
@ -10976,135 +10959,25 @@ def get_artist_discography(artist_id):
else:
effective_override_source = 'spotify'
if effective_override_source and effective_override_source in source_priority:
source_priority = [effective_override_source] + [
source for source in source_priority if source != effective_override_source
]
from core.metadata_service import MetadataLookupOptions, get_artist_discography as _get_artist_discography
logger.debug(
"Fetching discography for artist %s (name=%s, source_override=%s, priority=%s)",
discography = _get_artist_discography(
artist_id,
artist_name,
source_override or 'auto',
source_priority,
artist_name=artist_name,
options=MetadataLookupOptions(
source_override=effective_override_source,
allow_fallback=True,
skip_cache=False,
max_pages=0,
limit=50,
),
)
albums = []
active_source = None
album_list = discography['albums']
singles_list = discography['singles']
active_source = discography['source']
source_priority = discography['source_priority']
# Hydrabase can return a ready-made discography when active.
if not source_override and _is_hydrabase_active() and artist_name:
try:
albums = hydrabase_client.search_discography(artist_name, limit=50)
if albums:
active_source = 'hydrabase'
logger.info("Got %s albums from Hydrabase for artist %s", len(albums), artist_id)
except Exception as e:
logger.debug("Hydrabase discography failed for artist %s: %s", artist_id, e)
# Walk sources in configured priority order, keeping strict per-source lookups.
if not albums:
for source in source_priority:
if source == 'hydrabase':
continue
client = get_client_for_source(source)
try:
albums = get_artist_albums_for_source(source, artist_id)
except Exception as e:
logger.debug("%s direct lookup failed for artist %s: %s", source, artist_id, e)
albums = []
if not albums and artist_name:
try:
search_artists = _search_artists_for_source(client, source, artist_name, limit=5)
if search_artists:
best = next(
(a for a in search_artists if a.name.lower() == artist_name.lower()),
search_artists[0]
)
logger.debug("Found %s artist '%s' (id=%s)", source, best.name, best.id)
albums = get_artist_albums_for_source(source, best.id)
except Exception as e:
logger.debug("%s name search failed for artist %s: %s", source, artist_name, e)
if albums:
active_source = source
logger.info("Got %s albums from %s for artist %s", len(albums), source, artist_id)
break
logger.debug("Total albums returned for artist %s: %s (source=%s)", artist_id, len(albums), active_source)
if not albums:
return jsonify({
"albums": [],
"singles": [],
"source": active_source or (source_priority[0] if source_priority else "unknown")
})
# Separate albums from singles/EPs
album_list = []
singles_list = []
# Track seen albums to avoid duplicates (especially for "appears_on")
seen_albums = set()
for album in albums:
# Skip duplicates
if album.id in seen_albums:
continue
seen_albums.add(album.id)
# Skip albums where this artist isn't the primary (first-listed) artist
if hasattr(album, 'artist_ids') and album.artist_ids:
if album.artist_ids[0] != artist_id:
continue
album_data = {
"id": album.id,
"name": album.name,
"release_date": album.release_date if hasattr(album, 'release_date') else None,
"album_type": album.album_type if hasattr(album, 'album_type') else 'album',
"image_url": album.image_url if hasattr(album, 'image_url') else None,
"total_tracks": album.total_tracks if hasattr(album, 'total_tracks') else 0,
"external_urls": album.external_urls if hasattr(album, 'external_urls') else {}
}
# Skip obvious compilation issues but be more lenient for now
if hasattr(album, 'album_type') and album.album_type == 'compilation':
logger.debug("Found compilation '%s' for artist %s - including for now", album.name, artist_id)
# Categorize by album type
if hasattr(album, 'album_type'):
if album.album_type in ['single', 'ep']:
singles_list.append(album_data)
else: # 'album' or approved 'compilation'
album_list.append(album_data)
else:
# Default to album if no type specified
album_list.append(album_data)
# Sort by release date (newest first)
def get_release_year(item):
if item['release_date']:
try:
# Handle different date formats (YYYY, YYYY-MM, YYYY-MM-DD)
return int(item['release_date'][:4])
except (ValueError, IndexError):
return 0
return 0
album_list.sort(key=get_release_year, reverse=True)
singles_list.sort(key=get_release_year, reverse=True)
logger.info("Found %s albums and %s singles for artist %s", len(album_list), len(singles_list), artist_id)
# Debug: Log the final album list
for album in album_list:
logger.debug("Album: %s (%s) - %s", album['name'], album['album_type'], album['release_date'])
for single in singles_list:
logger.debug("Single/EP: %s (%s) - %s", single['name'], single['album_type'], single['release_date'])
# Gather artist enrichment info from cache + library
artist_info = {}
try: