From 2b575a59aeb1d74ccc301f4c5116df9a694ee736 Mon Sep 17 00:00:00 2001 From: Antti Kettunen Date: Sat, 18 Apr 2026 18:41:31 +0300 Subject: [PATCH] Refactor artist discography lookup 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. --- core/metadata_service.py | 235 ++++++++++++++++++- tests/test_metadata_service_discography.py | 250 +++++++++++++++++++++ web_server.py | 155 ++----------- 3 files changed, 494 insertions(+), 146 deletions(-) create mode 100644 tests/test_metadata_service_discography.py diff --git a/core/metadata_service.py b/core/metadata_service.py index b49d1444..3b3070a7 100644 --- a/core/metadata_service.py +++ b/core/metadata_service.py @@ -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. diff --git a/tests/test_metadata_service_discography.py b/tests/test_metadata_service_discography.py new file mode 100644 index 00000000..7267b5bd --- /dev/null +++ b/tests/test_metadata_service_discography.py @@ -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})] diff --git a/web_server.py b/web_server.py index 7262b2fd..fc498b67 100644 --- a/web_server.py +++ b/web_server.py @@ -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: