Merge pull request #419 from kettui/refactor/metadata-service-split-and-metadata-client-management-optimizations

Split metadata service logic into separate modules, move client management out of web_server
This commit is contained in:
BoulderBadgeDad 2026-04-29 12:36:03 -07:00 committed by GitHub
commit 58a4c1905b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
32 changed files with 3163 additions and 2674 deletions

View file

@ -9,7 +9,7 @@ Used by ``/api/artist-detail/<id>`` when the URL is called with a ``source``
query parameter and the library DB lookup misses. Enriches the response with query parameter and the library DB lookup misses. Enriches the response with
whatever metadata we can pull on demand: whatever metadata we can pull on demand:
* Image URL (via ``metadata_service.get_artist_image_url``) * Image URL (via ``core.metadata.artist_image.get_artist_image_url``)
* Source-specific artist info genres + follower count from the named * Source-specific artist info genres + follower count from the named
source's ``get_artist`` / ``get_artist_info`` helper source's ``get_artist`` / ``get_artist_info`` helper
* Last.fm bio + listeners + playcount + URL (by artist name) * Last.fm bio + listeners + playcount + URL (by artist name)
@ -27,6 +27,9 @@ import logging
from typing import Any, Dict, Optional, Tuple from typing import Any, Dict, Optional, Tuple
from core.artist_source_lookup import SOURCE_ID_FIELD from core.artist_source_lookup import SOURCE_ID_FIELD
from core.metadata import artist_image as metadata_artist_image
from core.metadata import discography as metadata_discography
from core.metadata.lookup import MetadataLookupOptions
logger = logging.getLogger("artist_source_detail") logger = logging.getLogger("artist_source_detail")
@ -48,21 +51,12 @@ def build_source_only_artist_detail(
``jsonify`` or equivalent. Status is 200 on success, 404 when the ``jsonify`` or equivalent. Status is 200 on success, 404 when the
source's discography lookup returned no releases. source's discography lookup returned no releases.
""" """
# Deferred import — keeps the top-level module importable in test rigs
# that stub out only what they need (same pattern `_find_library_artist`
# uses).
from core.metadata_service import (
MetadataLookupOptions,
get_artist_detail_discography,
get_artist_image_url,
)
resolved_name = (artist_name or artist_id or "").strip() resolved_name = (artist_name or artist_id or "").strip()
# 1. Image URL via the same helper /api/artist/<id>/image uses. # 1. Image URL via the same helper /api/artist/<id>/image uses.
image_url: Optional[str] = None image_url: Optional[str] = None
try: try:
image_url = get_artist_image_url(artist_id, source_override=source) image_url = metadata_artist_image.get_artist_image_url(artist_id, source_override=source)
except Exception as e: except Exception as e:
logger.debug(f"Artist image lookup failed for {source}:{artist_id}: {e}") logger.debug(f"Artist image lookup failed for {source}:{artist_id}: {e}")
@ -124,7 +118,7 @@ def build_source_only_artist_detail(
# 4. Discography from the specified source. Skip variant dedup so the # 4. Discography from the specified source. Skip variant dedup so the
# page shows every release the source returns — matches the inline # page shows every release the source returns — matches the inline
# Artists-page behaviour that this view was modelled after. # Artists-page behaviour that this view was modelled after.
discography_result = get_artist_detail_discography( discography_result = metadata_discography.get_artist_detail_discography(
artist_id, artist_id,
artist_name=resolved_name or artist_id, artist_name=resolved_name or artist_id,
options=MetadataLookupOptions( options=MetadataLookupOptions(

View file

@ -6,7 +6,7 @@ from typing import Dict, List, Optional, Any
from functools import wraps from functools import wraps
from dataclasses import dataclass from dataclasses import dataclass
from utils.logging_config import get_logger from utils.logging_config import get_logger
from core.metadata_cache import get_metadata_cache from core.metadata.cache import get_metadata_cache
logger = get_logger("deezer_client") logger = get_logger("deezer_client")

View file

@ -411,7 +411,7 @@ class DeezerDownloadClient:
album_ids.add(str(aid)) album_ids.add(str(aid))
album_release_dates = {} album_release_dates = {}
try: try:
from core.metadata_cache import get_metadata_cache from core.metadata.cache import get_metadata_cache
cache = get_metadata_cache() cache = get_metadata_cache()
except Exception: except Exception:
cache = None cache = None

View file

@ -12,7 +12,7 @@ import re
import time import time
import threading import threading
import requests import requests
from core.metadata_cache import get_metadata_cache from core.metadata.cache import get_metadata_cache
from typing import List, Dict, Any, Optional from typing import List, Dict, Any, Optional
from dataclasses import dataclass from dataclasses import dataclass
from functools import wraps from functools import wraps

View file

@ -4,18 +4,13 @@ from __future__ import annotations
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
from core.metadata import registry as metadata_registry
from utils.logging_config import get_logger from utils.logging_config import get_logger
logger = get_logger("imports.resolution") logger = get_logger("imports.resolution")
def _get_metadata_service():
from core import metadata_service
return metadata_service
def _extract_lookup_value(value: Any, *names: str, default: Any = None) -> Any: def _extract_lookup_value(value: Any, *names: str, default: Any = None) -> Any:
if value is None: if value is None:
return default return default
@ -73,9 +68,8 @@ def _get_source_chain_for_lookup(
source_override: Optional[str] = None, source_override: Optional[str] = None,
allow_fallback: bool = True, allow_fallback: bool = True,
) -> List[str]: ) -> List[str]:
metadata_service = _get_metadata_service() primary_source = metadata_registry.get_primary_source()
primary_source = metadata_service.get_primary_source() source_chain = list(metadata_registry.get_source_priority(primary_source))
source_chain = list(metadata_service.get_source_priority(primary_source))
override = (source_override or '').strip().lower() override = (source_override or '').strip().lower()
if override: if override:
@ -323,14 +317,13 @@ def get_single_track_import_context(
source_override: Optional[str] = None, source_override: Optional[str] = None,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""Build an import context for singles using source-priority metadata lookup.""" """Build an import context for singles using source-priority metadata lookup."""
metadata_service = _get_metadata_service()
source_priority = _get_source_chain_for_lookup(source_override=source_override, allow_fallback=True) source_priority = _get_source_chain_for_lookup(source_override=source_override, allow_fallback=True)
title = (title or '').strip() title = (title or '').strip()
artist = (artist or '').strip() artist = (artist or '').strip()
if override_id: if override_id:
chosen_source = (override_source or 'spotify').strip().lower() or 'spotify' chosen_source = (override_source or 'spotify').strip().lower() or 'spotify'
client = metadata_service.get_client_for_source(chosen_source) client = metadata_registry.get_client_for_source(chosen_source)
if client and hasattr(client, 'get_track_details'): if client and hasattr(client, 'get_track_details'):
try: try:
track_data = client.get_track_details(str(override_id)) track_data = client.get_track_details(str(override_id))
@ -358,7 +351,7 @@ def get_single_track_import_context(
logger.debug("Override track lookup failed on %s for %s: %s", chosen_source, override_id, exc) logger.debug("Override track lookup failed on %s for %s: %s", chosen_source, override_id, exc)
for source in source_priority: for source in source_priority:
client = metadata_service.get_client_for_source(source) client = metadata_registry.get_client_for_source(source)
if not client: if not client:
continue continue

View file

@ -5,7 +5,7 @@ import threading
from functools import wraps from functools import wraps
from dataclasses import dataclass from dataclasses import dataclass
from utils.logging_config import get_logger from utils.logging_config import get_logger
from core.metadata_cache import get_metadata_cache from core.metadata.cache import get_metadata_cache
logger = get_logger("itunes_client") logger = get_logger("itunes_client")

View file

@ -1,2 +1,88 @@
"""Metadata helper package.""" """Metadata package public surface."""
from core.metadata.album_tracks import (
get_album_for_source,
get_album_tracks_for_source,
get_artist_album_tracks,
get_artist_albums_for_source,
resolve_album_reference,
)
from core.metadata.artist_image import get_artist_image_url
from core.metadata.cache import MetadataCache, get_metadata_cache
from core.metadata.completion import (
check_album_completion,
check_artist_discography_completion,
check_single_completion,
iter_artist_discography_completion_events,
)
from core.metadata.discography import (
get_artist_detail_discography,
get_artist_discography,
)
from core.metadata.lookup import MetadataLookupOptions
from core.metadata.registry import (
METADATA_SOURCE_PRIORITY,
clear_cached_metadata_client,
clear_cached_metadata_clients,
clear_cached_profile_spotify_client,
get_client_for_source,
get_deezer_client,
get_discogs_client,
get_hydrabase_client,
get_itunes_client,
get_primary_client,
get_primary_source,
get_spotify_client_for_profile,
get_registered_runtime_client,
get_source_priority,
get_spotify_client,
is_hydrabase_enabled,
register_profile_spotify_credentials_provider,
register_runtime_clients,
)
from core.metadata.service import MetadataProvider, MetadataService, get_metadata_service
from core.metadata.similar_artists import (
get_musicmap_similar_artists,
iter_musicmap_similar_artist_events,
)
__all__ = [
"METADATA_SOURCE_PRIORITY",
"MetadataCache",
"MetadataLookupOptions",
"MetadataProvider",
"MetadataService",
"check_album_completion",
"check_artist_discography_completion",
"check_single_completion",
"clear_cached_metadata_client",
"clear_cached_metadata_clients",
"clear_cached_profile_spotify_client",
"get_album_for_source",
"get_album_tracks_for_source",
"get_artist_album_tracks",
"get_artist_albums_for_source",
"get_artist_detail_discography",
"get_artist_discography",
"get_artist_image_url",
"get_client_for_source",
"get_deezer_client",
"get_discogs_client",
"get_hydrabase_client",
"get_itunes_client",
"get_metadata_cache",
"get_metadata_service",
"get_musicmap_similar_artists",
"get_primary_client",
"get_primary_source",
"get_spotify_client_for_profile",
"get_registered_runtime_client",
"get_spotify_client",
"get_source_priority",
"iter_artist_discography_completion_events",
"iter_musicmap_similar_artist_events",
"is_hydrabase_enabled",
"register_profile_spotify_credentials_provider",
"register_runtime_clients",
"resolve_album_reference",
]

View file

@ -0,0 +1,566 @@
"""Album-track lookup helpers for metadata API."""
from __future__ import annotations
from typing import Any, Dict, List, Optional
from core.metadata import registry as metadata_registry
from core.metadata.lookup import MetadataLookupOptions
from utils.logging_config import get_logger
logger = get_logger("metadata.album_tracks")
__all__ = [
"get_album_for_source",
"get_album_tracks_for_source",
"get_artist_album_tracks",
"get_artist_albums_for_source",
"resolve_album_reference",
]
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 _get_source_chain_for_lookup(options: MetadataLookupOptions) -> List[str]:
primary_source = metadata_registry.get_primary_source()
source_chain = list(metadata_registry.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 _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 _search_albums_for_source(source: str, client: Any, query: str, limit: int = 5) -> List[Any]:
if not client or not hasattr(client, 'search_albums'):
return []
try:
kwargs = {'limit': limit}
if source == 'spotify':
kwargs['allow_fallback'] = False
return client.search_albums(query, **kwargs) or []
except Exception as exc:
logger.debug("Could not search %s for %s: %s", source, query, exc)
return []
def _pick_best_artist_match(search_results: List[Any], artist_name: str) -> Optional[Any]:
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 _extract_track_items(api_tracks: Any) -> List[Dict[str, Any]]:
if not api_tracks:
return []
if isinstance(api_tracks, dict):
return api_tracks.get('items') or []
if isinstance(api_tracks, list):
return api_tracks
return []
def _normalize_track_artists(track_item: Any) -> List[str]:
artists = _extract_lookup_value(track_item, 'artists', default=[]) or []
if isinstance(artists, (str, bytes)):
artists = [artists]
elif isinstance(artists, dict):
artists = [artists]
else:
try:
artists = list(artists)
except TypeError:
artists = [artists]
normalized = []
for artist in artists:
artist_name = _extract_lookup_value(artist, 'name', 'artist_name', 'title')
if not artist_name and isinstance(artist, str):
artist_name = artist
if artist_name:
normalized.append(str(artist_name))
return normalized
def _extract_album_track_items(album_data: Any, tracks_data: Any = None) -> List[Dict[str, Any]]:
embedded_tracks = _extract_lookup_value(album_data, 'tracks', default=None)
if isinstance(embedded_tracks, dict):
items = embedded_tracks.get('items') or []
if items:
return items
elif isinstance(embedded_tracks, list):
if embedded_tracks:
return embedded_tracks
return _extract_track_items(tracks_data)
def _normalize_context_artists(artists: Any) -> List[Dict[str, Any]]:
if not artists:
return []
if isinstance(artists, (str, bytes)):
artists = [artists]
elif isinstance(artists, dict):
artists = [artists]
else:
try:
artists = list(artists)
except TypeError:
artists = [artists]
normalized: List[Dict[str, Any]] = []
for artist in artists:
if isinstance(artist, dict):
name = _extract_lookup_value(artist, 'name', 'artist_name', 'title', default='') or ''
artist_id = _extract_lookup_value(artist, 'id', 'artist_id', default='') or ''
entry: Dict[str, Any] = {}
if name:
entry['name'] = str(name)
if artist_id:
entry['id'] = str(artist_id)
genres = _extract_lookup_value(artist, 'genres', default=None)
if genres is not None:
entry['genres'] = genres
if entry:
normalized.append(entry)
continue
name = str(artist).strip()
if name:
normalized.append({'name': name})
return normalized
def _build_album_info(album_data: Any, album_id: str, album_name: str = '', artist_name: str = '') -> Dict[str, Any]:
images = _extract_lookup_value(album_data, 'images', default=[]) or []
if not isinstance(images, list):
images = list(images) if images else []
artists = _normalize_context_artists(_extract_lookup_value(album_data, 'artists', default=[]))
if not artists and artist_name:
artists = [{'name': artist_name}]
primary_artist = artists[0] if artists else {}
resolved_artist_name = (
_extract_lookup_value(primary_artist, 'name', default='')
or artist_name
or _extract_lookup_value(album_data, 'artist_name', 'artist', default='')
or ''
)
resolved_artist_id = str(
_extract_lookup_value(primary_artist, 'id', default='')
or _extract_lookup_value(album_data, 'artist_id', default='')
or ''
).strip()
image_url = None
if images:
image_url = _extract_lookup_value(images[0], 'url')
if not image_url:
image_url = _extract_lookup_value(album_data, 'image_url', 'thumb_url')
return {
'id': _extract_lookup_value(album_data, 'id', 'album_id', 'collectionId', 'release_id', default=album_id) or album_id,
'name': _extract_lookup_value(album_data, 'name', 'title', default=album_name or album_id) or album_name or album_id,
'artist': resolved_artist_name or '',
'artist_name': resolved_artist_name or '',
'artist_id': resolved_artist_id,
'artists': artists,
'image_url': image_url,
'images': images,
'release_date': _extract_lookup_value(album_data, 'release_date', default='') or '',
'album_type': _extract_lookup_value(album_data, 'album_type', default='album') or 'album',
'total_tracks': _extract_lookup_value(album_data, 'total_tracks', 'track_count', default=0) or 0,
}
def _build_album_track_entry(track_item: Any, album_info: Dict[str, Any], source: str) -> Dict[str, Any]:
explicit_value = _extract_lookup_value(track_item, 'explicit', 'trackExplicitness', default=False)
if isinstance(explicit_value, str):
explicit_value = explicit_value.lower() == 'explicit'
return {
'id': _extract_lookup_value(track_item, 'id', 'track_id', 'trackId', default='') or '',
'name': _extract_lookup_value(track_item, 'name', 'track_name', 'trackName', default='Unknown Track') or 'Unknown Track',
'artists': _normalize_track_artists(track_item),
'duration_ms': _extract_lookup_value(track_item, 'duration_ms', 'trackTimeMillis', default=0) or 0,
'track_number': _extract_lookup_value(track_item, 'track_number', 'trackNumber', default=0) or 0,
'disc_number': _extract_lookup_value(track_item, 'disc_number', 'discNumber', default=1) or 1,
'explicit': bool(explicit_value),
'preview_url': _extract_lookup_value(track_item, 'preview_url', 'previewUrl'),
'external_urls': _extract_lookup_value(track_item, 'external_urls', default={}) or {},
'uri': _extract_lookup_value(track_item, 'uri', default='') or '',
'album': album_info,
'source': source,
'provider': source,
'_source': source,
}
def _build_album_tracks_payload(
album_data: Any,
tracks_data: Any,
source: str,
album_id: str,
album_name: str = '',
artist_name: str = '',
) -> Dict[str, Any]:
album_info = _build_album_info(album_data, album_id, album_name=album_name, artist_name=artist_name)
album_info['source'] = source
album_info['_source'] = source
album_info['provider'] = source
track_items = _extract_album_track_items(album_data, tracks_data)
tracks = [_build_album_track_entry(track, album_info, source) for track in track_items]
return {
'success': bool(tracks),
'album': album_info,
'tracks': tracks,
'source': source,
}
def get_album_tracks_for_source(source: str, album_id: str):
"""Get album tracks for an exact source."""
client = metadata_registry.get_client_for_source(source)
if not client:
return None
try:
fetch = getattr(client, 'get_album_tracks_dict', None) if source == 'hydrabase' else getattr(client, 'get_album_tracks', None)
if not fetch:
return None
if source == 'spotify':
return fetch(album_id, allow_fallback=False)
return fetch(album_id)
except Exception:
return None
def get_album_for_source(source: str, album_id: str):
"""Get album metadata for an exact source."""
client = metadata_registry.get_client_for_source(source)
if not client or not hasattr(client, 'get_album'):
return None
try:
if source == 'spotify':
return client.get_album(album_id, allow_fallback=False)
return client.get_album(album_id)
except Exception:
return None
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,
max_pages: int = 0,
):
"""Get artist albums for an exact source."""
client = metadata_registry.get_client_for_source(source)
if not client or not hasattr(client, 'get_artist_albums'):
return None
def _fetch_for_artist(target_artist_id: str):
kwargs = {
'album_type': album_type,
'limit': limit,
}
if source == 'spotify':
kwargs['allow_fallback'] = False
kwargs['skip_cache'] = skip_cache
kwargs['max_pages'] = max_pages
return client.get_artist_albums(target_artist_id, **kwargs)
try:
if artist_id:
albums = _fetch_for_artist(artist_id) or []
if albums:
return albums
else:
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 resolve_album_reference(
album_id: str,
preferred_source: Optional[str] = None,
album_name: str = '',
artist_name: str = '',
) -> tuple[Optional[str], Optional[str]]:
"""Resolve a local database album ID or name-based reference to a provider ID."""
try:
from database.music_database import get_database
database = get_database()
with database._get_connection() as conn:
cursor = conn.cursor()
cursor.execute("PRAGMA table_info(albums)")
album_columns = {row[1] for row in cursor.fetchall()}
source_chain = list(metadata_registry.get_source_priority(preferred_source or metadata_registry.get_primary_source()))
override = (preferred_source or '').strip().lower()
if override:
source_chain = [override] + [source for source in source_chain if source != override]
source_columns = {
'spotify': ('spotify_album_id',),
'deezer': ('deezer_id', 'deezer_album_id'),
'itunes': ('itunes_album_id',),
'discogs': ('discogs_id',),
'hydrabase': ('soul_id', 'hydrabase_album_id'),
}
select_columns = ["a.title", "ar.name as artist_name"]
for columns in source_columns.values():
for column in columns:
if column in album_columns:
select_columns.append(f"a.{column}")
cursor.execute(
"""
SELECT {select_columns}
FROM albums a
JOIN artists ar ON a.artist_id = ar.id
WHERE a.id = ?
""".format(select_columns=", ".join(select_columns)),
(album_id,),
)
row = cursor.fetchone()
if row:
for source in source_chain:
for column in source_columns.get(source, ()):
if column not in row.keys():
continue
value = row[column]
if value:
return value, source
search_title = album_name or row['title']
search_artist = artist_name or row['artist_name']
query = f"{search_artist} {search_title}".strip()
for source in source_chain:
client = metadata_registry.get_client_for_source(source)
if not client:
continue
results = _search_albums_for_source(source, client, query, limit=5)
if results:
for album in results:
candidate_name = str(_extract_lookup_value(album, 'name', 'title', default='') or '').strip().lower()
if candidate_name and candidate_name == str(search_title).strip().lower():
return _extract_lookup_value(album, 'id', 'album_id', 'release_id'), source
best = results[0]
return _extract_lookup_value(best, 'id', 'album_id', 'release_id'), source
if not album_name and not artist_name:
return None, None
query = " ".join(part for part in (artist_name, album_name) if part).strip() or album_id
for source in source_chain:
client = metadata_registry.get_client_for_source(source)
if not client:
continue
results = _search_albums_for_source(source, client, query, limit=5)
if results:
for album in results:
candidate_name = str(_extract_lookup_value(album, 'name', 'title', default='') or '').strip().lower()
if album_name and candidate_name == album_name.strip().lower():
return _extract_lookup_value(album, 'id', 'album_id', 'release_id'), source
best = results[0]
return _extract_lookup_value(best, 'id', 'album_id', 'release_id'), source
except Exception as e:
logger.debug("Error resolving album reference %s: %s", album_id, e)
return None, None
def get_artist_album_tracks(
album_id: str,
artist_name: str = '',
album_name: str = '',
source_override: Optional[str] = None,
) -> Dict[str, Any]:
"""Get a normalized album-track payload using source-priority lookup."""
source_chain = _get_source_chain_for_lookup(
MetadataLookupOptions(source_override=source_override, allow_fallback=True)
)
preferred_source = source_chain[0] if source_chain else None
for source in source_chain:
client = metadata_registry.get_client_for_source(source)
if not client:
continue
album_data = get_album_for_source(source, album_id)
if not album_data:
continue
tracks_data = None
if not _extract_album_track_items(album_data):
tracks_data = get_album_tracks_for_source(source, album_id)
payload = _build_album_tracks_payload(
album_data,
tracks_data,
source,
album_id,
album_name=album_name,
artist_name=artist_name,
)
if payload['tracks']:
payload['success'] = True
payload['source_priority'] = source_chain
payload['resolved_album_id'] = album_id
return payload
resolved_album_id, resolved_source = resolve_album_reference(
album_id,
preferred_source=preferred_source,
album_name=album_name,
artist_name=artist_name,
)
if resolved_album_id:
retry_sources = []
if resolved_source:
retry_sources.append(resolved_source)
retry_sources.extend(source for source in source_chain if source not in retry_sources)
for source in retry_sources:
client = metadata_registry.get_client_for_source(source)
if not client:
continue
album_data = get_album_for_source(source, resolved_album_id)
if not album_data:
continue
tracks_data = None
if not _extract_album_track_items(album_data):
tracks_data = get_album_tracks_for_source(source, resolved_album_id)
payload = _build_album_tracks_payload(
album_data,
tracks_data,
source,
resolved_album_id,
album_name=album_name,
artist_name=artist_name,
)
if payload['tracks']:
payload['success'] = True
payload['source_priority'] = source_chain
payload['resolved_album_id'] = resolved_album_id
return payload
# Keep trying the remaining sources in case another provider has the track listing.
continue
if resolved_album_id:
return {
'success': False,
'error': 'No tracks found for album — it may be region-restricted or unavailable on this metadata source',
'status_code': 404,
'source_priority': source_chain,
'resolved_album_id': resolved_album_id,
'tracks': [],
'album': {
'id': resolved_album_id,
'name': album_name or resolved_album_id,
'image_url': None,
'images': [],
'release_date': '',
'album_type': 'album',
'total_tracks': 0,
},
}
return {
'success': False,
'error': 'Album not found',
'status_code': 404,
'source_priority': source_chain,
'resolved_album_id': None,
'tracks': [],
'album': {
'id': album_id,
'name': album_name or album_id,
'image_url': None,
'images': [],
'release_date': '',
'album_type': 'album',
'total_tracks': 0,
},
}

View file

@ -0,0 +1,138 @@
"""Artist image lookup helpers for metadata API."""
from __future__ import annotations
from typing import Any, Optional
from core.metadata import registry as metadata_registry
from core.metadata.discography import _extract_lookup_value
from utils.logging_config import get_logger
logger = get_logger("metadata.artist_image")
__all__ = [
"get_artist_image_url",
]
def _extract_artist_image_url(artist_data: Any) -> Optional[str]:
if not artist_data:
return None
images = _extract_lookup_value(artist_data, 'images', default=[]) or []
if not isinstance(images, list):
try:
images = list(images)
except TypeError:
images = []
if images:
first_image = images[0]
image_url = _extract_lookup_value(first_image, 'url')
if image_url:
return image_url
return _extract_lookup_value(
artist_data,
'image_url',
'thumb_url',
'cover_image',
'picture_xl',
'picture_big',
'picture_medium',
)
def _get_artist_image_from_source(source: str, artist_id: str) -> Optional[str]:
client = metadata_registry.get_client_for_source(source)
if not client:
return None
try:
if source == 'spotify':
artist_data = client.get_artist(artist_id, allow_fallback=False)
else:
artist_data = client.get_artist(artist_id)
except Exception as exc:
logger.debug("Could not fetch artist image for %s on %s: %s", artist_id, source, exc)
artist_data = None
image_url = _extract_artist_image_url(artist_data)
if image_url:
return image_url
if hasattr(client, '_get_artist_image_from_albums'):
try:
return client._get_artist_image_from_albums(artist_id)
except Exception as exc:
logger.debug("Could not fetch artist album art for %s on %s: %s", artist_id, source, exc)
return None
def _lookup_artist_image_by_name(name: str) -> Optional[str]:
"""Look up an artist image by name across fallback sources."""
name = (name or '').strip()
if not name:
return None
skip_sources = {'musicbrainz', 'soulseek', 'youtube_videos', 'hydrabase'}
for source in metadata_registry.get_source_priority(metadata_registry.get_primary_source()):
if source in skip_sources:
continue
client = metadata_registry.get_client_for_source(source)
if not client or not hasattr(client, 'search_artists'):
continue
try:
results = client.search_artists(name, limit=1) or []
if results:
top = results[0]
image_url = getattr(top, 'image_url', None) or (
top.get('image_url') if isinstance(top, dict) else None
)
if image_url:
return image_url
except Exception as exc:
logger.debug("Artist image lookup by name failed on %s for %r: %s", source, name, exc)
continue
return None
def get_artist_image_url(
artist_id: str,
source_override: Optional[str] = None,
plugin: Optional[str] = None,
artist_name: Optional[str] = None,
) -> Optional[str]:
"""Resolve an artist image URL using the configured source priority."""
if not artist_id:
return None
if artist_id.startswith('soul_'):
return None
source_override = (source_override or '').strip().lower()
plugin = (plugin or '').strip().lower()
if source_override == 'hydrabase':
if plugin in ('deezer', 'itunes'):
return _get_artist_image_from_source(plugin, artist_id)
if artist_id.isdigit():
return _get_artist_image_from_source('itunes', artist_id)
return None
if source_override == 'musicbrainz':
if not artist_name:
return None
return _lookup_artist_image_by_name(artist_name)
if source_override:
return _get_artist_image_from_source(source_override, artist_id)
for source in metadata_registry.get_source_priority(metadata_registry.get_primary_source()):
image_url = _get_artist_image_from_source(source, artist_id)
if image_url:
return image_url
return None

478
core/metadata/completion.py Normal file
View file

@ -0,0 +1,478 @@
"""Completion helpers for metadata lookups."""
from __future__ import annotations
from typing import Any, Dict, List, Optional
from core.metadata import registry as metadata_registry
from core.metadata.album_tracks import get_album_tracks_for_source
from core.metadata.discography import _extract_release_artist_name
from core.metadata.lookup import MetadataLookupOptions
from utils.logging_config import get_logger
logger = get_logger("metadata.completion")
__all__ = [
"check_album_completion",
"check_artist_discography_completion",
"check_single_completion",
"iter_artist_discography_completion_events",
]
def _extract_track_items(api_tracks: Any) -> List[Dict[str, Any]]:
if not api_tracks:
return []
if isinstance(api_tracks, dict):
return api_tracks.get('items') or []
if isinstance(api_tracks, list):
return api_tracks
return []
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 _get_completion_source_chain(source_override: Optional[str] = None) -> List[str]:
primary_source = metadata_registry.get_primary_source()
source_chain = list(metadata_registry.get_source_priority(primary_source))
override = (source_override or '').strip().lower()
if override:
source_chain = [override] + [source for source in source_chain if source != override]
return source_chain
def _resolve_completion_artist_name(
discography: Dict[str, Any],
artist_name: str,
) -> str:
resolved_name = (artist_name or '').strip()
if resolved_name and resolved_name.lower() != 'unknown artist':
return resolved_name
release_items = list((discography or {}).get('albums', []) or []) + list((discography or {}).get('singles', []) or [])
if not release_items:
return resolved_name or 'Unknown Artist'
release_artist_name = _extract_release_artist_name(release_items[0])
if release_artist_name:
logger.debug("Using release artist metadata '%s' for completion", release_artist_name)
return release_artist_name
return resolved_name or 'Unknown Artist'
def _resolve_completion_track_total(release: Dict[str, Any], source_chain: List[str]) -> int:
total_tracks = _extract_lookup_value(release, 'total_tracks', default=0) or 0
if total_tracks:
return int(total_tracks)
release_id = _extract_lookup_value(release, 'id', 'album_id', 'release_id')
if not release_id:
return 0
for source in source_chain:
try:
api_tracks = get_album_tracks_for_source(source, str(release_id))
items = _extract_track_items(api_tracks)
if items:
logger.debug("Resolved track count for release %s from %s", release_id, source)
return len(items)
except Exception as exc:
logger.debug("Could not resolve track count for release %s from %s: %s", release_id, source, exc)
return 0
def check_album_completion(
db,
album_data: Dict[str, Any],
artist_name: str,
source_override: Optional[str] = None,
source_chain: Optional[List[str]] = None,
candidate_albums: Optional[List[Any]] = None,
) -> Dict[str, Any]:
"""Check completion status for a single album."""
try:
source_chain = source_chain or _get_completion_source_chain(source_override)
album_name = album_data.get('name', '')
total_tracks = _resolve_completion_track_total(album_data, source_chain)
album_id = album_data.get('id', '')
# If total_tracks is 0 (Discogs masters don't include track counts),
# try to fetch the real count from the prioritized metadata sources.
if total_tracks == 0 and album_id:
logger.debug("No track count found for '%s' (%s)", album_name, album_id)
logger.debug(f"Checking album: '{album_name}' ({total_tracks} tracks)")
formats = []
try:
from config.settings import config_manager
active_server = config_manager.get_active_media_server()
db_album, confidence, owned_tracks, expected_tracks, is_complete, formats = db.check_album_exists_with_completeness(
title=album_name,
artist=artist_name,
expected_track_count=total_tracks if total_tracks > 0 else None,
confidence_threshold=0.7,
server_source=active_server,
candidate_albums=candidate_albums,
)
except Exception as db_error:
logger.error(f"Database error for album '{album_name}': {db_error}")
return {
"id": album_id,
"name": album_name,
"status": "error",
"owned_tracks": 0,
"expected_tracks": total_tracks,
"completion_percentage": 0,
"confidence": 0.0,
"found_in_db": False,
"error_message": str(db_error),
"formats": [],
}
if expected_tracks > 0:
completion_percentage = (owned_tracks / expected_tracks) * 100
elif total_tracks > 0:
completion_percentage = (owned_tracks / total_tracks) * 100
else:
completion_percentage = 100 if owned_tracks > 0 else 0
if owned_tracks > 0 and owned_tracks >= (expected_tracks or total_tracks):
status = "completed"
elif owned_tracks > 0:
status = "partial"
else:
status = "missing"
logger.debug(
"Album completion result: owned=%s expected=%s total=%s completion=%.1f status=%s",
owned_tracks,
expected_tracks or total_tracks,
total_tracks,
completion_percentage,
status,
)
return {
"id": album_id,
"name": album_name,
"status": status,
"owned_tracks": owned_tracks,
"expected_tracks": expected_tracks or total_tracks,
"completion_percentage": round(completion_percentage, 1),
"confidence": round(confidence, 2) if confidence else 0.0,
"found_in_db": db_album is not None,
"formats": formats,
}
except Exception as e:
logger.error(f"Error checking album completion for '{album_data.get('name', 'Unknown')}': {e}")
return {
"id": album_data.get('id', ''),
"name": album_data.get('name', 'Unknown'),
"status": "error",
"owned_tracks": 0,
"expected_tracks": album_data.get('total_tracks', 0),
"completion_percentage": 0,
"confidence": 0.0,
"found_in_db": False,
"formats": [],
}
def check_single_completion(
db,
single_data: Dict[str, Any],
artist_name: str,
source_override: Optional[str] = None,
source_chain: Optional[List[str]] = None,
candidate_albums: Optional[List[Any]] = None,
candidate_tracks: Optional[List[Any]] = None,
) -> Dict[str, Any]:
"""Check completion status for a single/EP."""
try:
source_chain = source_chain or _get_completion_source_chain(source_override)
single_name = single_data.get('name', '')
raw_total_tracks = single_data.get('total_tracks', 1)
total_tracks = raw_total_tracks if raw_total_tracks is not None else 1
single_id = single_data.get('id', '')
album_type = single_data.get('album_type', 'single')
formats = []
if total_tracks == 0:
total_tracks = _resolve_completion_track_total(single_data, source_chain) or 1
logger.debug(
"Checking %s: name=%r tracks=%s",
album_type,
single_name,
total_tracks,
)
if album_type == 'ep' or total_tracks > 1:
try:
from config.settings import config_manager
active_server = config_manager.get_active_media_server()
db_album, confidence, owned_tracks, expected_tracks, is_complete, formats = db.check_album_exists_with_completeness(
title=single_name,
artist=artist_name,
expected_track_count=total_tracks,
confidence_threshold=0.7,
server_source=active_server,
candidate_albums=candidate_albums,
)
except Exception as db_error:
logger.error(f"Database error for EP '{single_name}': {db_error}")
owned_tracks, expected_tracks, confidence = 0, total_tracks, 0.0
db_album = None
if expected_tracks > 0:
completion_percentage = (owned_tracks / expected_tracks) * 100
else:
completion_percentage = (owned_tracks / total_tracks) * 100
if owned_tracks > 0 and owned_tracks >= (expected_tracks or total_tracks):
status = "completed"
elif owned_tracks > 0:
status = "partial"
else:
status = "missing"
logger.debug(
"EP completion result: owned=%s expected=%s total=%s completion=%.1f status=%s",
owned_tracks,
expected_tracks or total_tracks,
total_tracks,
completion_percentage,
status,
)
return {
"id": single_id,
"name": single_name,
"status": status,
"owned_tracks": owned_tracks,
"expected_tracks": expected_tracks or total_tracks,
"completion_percentage": round(completion_percentage, 1),
"confidence": round(confidence, 2) if confidence else 0.0,
"found_in_db": db_album is not None,
"type": album_type,
"formats": formats,
}
else:
try:
from config.settings import config_manager
active_server = config_manager.get_active_media_server()
db_track, confidence = db.check_track_exists(
title=single_name,
artist=artist_name,
confidence_threshold=0.7,
server_source=active_server,
candidate_tracks=candidate_tracks,
)
except Exception as db_error:
logger.error(f"Database error for single '{single_name}': {db_error}")
db_track, confidence = None, 0.0
owned_tracks = 1 if db_track else 0
expected_tracks = 1
completion_percentage = 100 if db_track else 0
status = "completed" if db_track else "missing"
if db_track and db_track.file_path:
import os
ext = os.path.splitext(db_track.file_path)[1].lstrip('.').upper()
if ext == 'MP3' and db_track.bitrate:
formats = [f"MP3-{db_track.bitrate}"]
elif ext:
formats = [ext]
logger.debug(
"Single completion result: owned=%s expected=1 completion=%.1f status=%s",
owned_tracks,
completion_percentage,
status,
)
return {
"id": single_id,
"name": single_name,
"status": status,
"owned_tracks": owned_tracks,
"expected_tracks": expected_tracks,
"completion_percentage": round(completion_percentage, 1),
"confidence": round(confidence, 2) if confidence else 0.0,
"found_in_db": db_track is not None,
"type": album_type,
"formats": formats,
}
except Exception as e:
logger.error(f"Error checking single/EP completion for '{single_data.get('name', 'Unknown')}': {e}")
return {
"id": single_data.get('id', ''),
"name": single_data.get('name', 'Unknown'),
"status": "error",
"owned_tracks": 0,
"expected_tracks": single_data.get('total_tracks', 1),
"completion_percentage": 0,
"confidence": 0.0,
"found_in_db": False,
"type": single_data.get('album_type', 'single'),
"formats": [],
}
def iter_artist_discography_completion_events(
discography: Dict[str, Any],
artist_name: str = 'Unknown Artist',
source_override: Optional[str] = None,
db=None,
):
"""Yield completion-stream events for artist discography ownership checks."""
if db is None:
from database.music_database import get_database
db = get_database()
source_chain = _get_completion_source_chain(source_override)
resolved_artist_name = _resolve_completion_artist_name(discography or {}, artist_name)
albums = list((discography or {}).get('albums', []) or [])
singles = list((discography or {}).get('singles', []) or [])
total_items = len(albums) + len(singles)
processed_count = 0
import time as _time_metadata
candidate_albums = None
candidate_tracks = None
try:
from config.settings import config_manager as _cm_metadata
_active_server = _cm_metadata.get_active_media_server()
_t0 = _time_metadata.perf_counter()
candidate_albums = db.get_candidate_albums_for_artist(resolved_artist_name, server_source=_active_server)
_t1 = _time_metadata.perf_counter()
print(f"[artist-completion-stream] Pre-fetched {len(candidate_albums) if candidate_albums is not None else 0} library albums for '{resolved_artist_name}' in {(_t1 - _t0) * 1000:.0f}ms")
if candidate_albums:
_t2 = _time_metadata.perf_counter()
candidate_tracks = db.get_candidate_tracks_for_albums([a.id for a in candidate_albums])
_t3 = _time_metadata.perf_counter()
print(f"[artist-completion-stream] Pre-fetched {len(candidate_tracks) if candidate_tracks is not None else 0} library tracks in {(_t3 - _t2) * 1000:.0f}ms")
except Exception as _pre_err:
print(f"[artist-completion-stream] Failed to pre-fetch candidates for '{resolved_artist_name}': {_pre_err}")
candidate_albums = None
candidate_tracks = None
yield {
'type': 'start',
'total_items': total_items,
'artist_name': resolved_artist_name,
}
_loop_start = _time_metadata.perf_counter()
for album in albums:
try:
completion_data = check_album_completion(
db,
album,
resolved_artist_name,
source_override=source_override,
source_chain=source_chain,
candidate_albums=candidate_albums,
)
completion_data['type'] = 'album_completion'
completion_data['container_type'] = 'albums'
processed_count += 1
completion_data['progress'] = round((processed_count / total_items) * 100, 1) if total_items else 100
yield completion_data
except Exception as e:
yield {
'type': 'error',
'container_type': 'albums',
'id': album.get('id', ''),
'name': album.get('name', 'Unknown'),
'error': str(e),
}
for single in singles:
try:
completion_data = check_single_completion(
db,
single,
resolved_artist_name,
source_override=source_override,
source_chain=source_chain,
candidate_albums=candidate_albums,
candidate_tracks=candidate_tracks,
)
completion_data['type'] = 'single_completion'
completion_data['container_type'] = 'singles'
processed_count += 1
completion_data['progress'] = round((processed_count / total_items) * 100, 1) if total_items else 100
yield completion_data
except Exception as e:
yield {
'type': 'error',
'container_type': 'singles',
'id': single.get('id', ''),
'name': single.get('name', 'Unknown'),
'error': str(e),
}
_loop_elapsed = _time_metadata.perf_counter() - _loop_start
print(f"[artist-completion-stream] Processed {total_items} items for '{resolved_artist_name}' in {_loop_elapsed * 1000:.0f}ms")
yield {
'type': 'complete',
'processed_count': processed_count,
'artist_name': resolved_artist_name,
}
def check_artist_discography_completion(
discography: Dict[str, Any],
artist_name: str = 'Unknown Artist',
source_override: Optional[str] = None,
db=None,
) -> Dict[str, Any]:
"""Return completion results for an artist discography without streaming."""
albums_completion = []
singles_completion = []
for event in iter_artist_discography_completion_events(
discography,
artist_name=artist_name,
source_override=source_override,
db=db,
):
if event.get('type') == 'album_completion':
albums_completion.append(event)
elif event.get('type') == 'single_completion':
singles_completion.append(event)
return {
'albums': albums_completion,
'singles': singles_completion,
}

View file

@ -0,0 +1,435 @@
"""Discography lookup helpers for metadata API."""
from __future__ import annotations
from typing import Any, Dict, List, Optional
from core.metadata import registry as metadata_registry
from core.metadata.album_tracks import get_artist_albums_for_source
from core.metadata.lookup import MetadataLookupOptions
from utils.logging_config import get_logger
logger = get_logger("metadata.discography")
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 _get_source_chain_for_lookup(options: MetadataLookupOptions) -> List[str]:
primary_source = metadata_registry.get_primary_source()
source_chain = list(metadata_registry.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 _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 _search_albums_for_source(source: str, client: Any, query: str, limit: int = 5) -> List[Any]:
if not client or not hasattr(client, 'search_albums'):
return []
try:
kwargs = {'limit': limit}
if source == 'spotify':
kwargs['allow_fallback'] = False
return client.search_albums(query, **kwargs) or []
except Exception as exc:
logger.debug("Could not search %s for %s: %s", source, query, 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
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),
'artist_name': _extract_release_artist_name(release),
'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 _extract_release_artist_name(release: Any) -> str:
artist_name = _extract_lookup_value(release, 'artist_name', 'artist', default='') or ''
artist_name = str(artist_name).strip()
if artist_name:
return artist_name
artists = _extract_lookup_value(release, 'artists', default=[]) or []
if isinstance(artists, (str, bytes)):
return str(artists).strip()
if isinstance(artists, dict):
return str(_extract_lookup_value(artists, 'name', 'artist_name', 'title', default='') or '').strip()
try:
artists = list(artists)
except TypeError:
artists = [artists]
if not artists:
return ''
first_artist = artists[0]
inferred_name = _extract_lookup_value(first_artist, 'name', 'artist_name', 'title')
if not inferred_name and isinstance(first_artist, str):
inferred_name = first_artist
return str(inferred_name).strip() if inferred_name else ''
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 _dedup_variant_releases(releases: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Collapse obvious edition variants into a single canonical release card.
This keeps a clean UI while still preserving distinct releases when the
cleaned titles diverge enough that they are likely not variants.
"""
if not releases:
return []
import re
from difflib import SequenceMatcher
variant_suffix_pattern = re.compile(
r'\s*[\(\[][^()\[\]]*\b(?:edition|editions|deluxe|remaster|remastered|'
r'explicit|clean|version|anniversary|collector|expanded|redux)\b[^()\[\]]*[\)\]]\s*$',
re.IGNORECASE,
)
legacy_suffix_pattern = re.compile(
r'\s*-\s*(explicit|clean|deluxe edition|single)\s*$',
re.IGNORECASE,
)
variant_keyword_pattern = re.compile(
r'\b(?:edition|editions|deluxe|remaster|remastered|explicit|clean|version|'
r'anniversary|collector|expanded|redux)\b',
re.IGNORECASE,
)
def _clean_title(title: Any) -> str:
cleaned = str(title or '').strip().lower()
while True:
new_cleaned = variant_suffix_pattern.sub('', cleaned).strip()
new_cleaned = legacy_suffix_pattern.sub('', new_cleaned).strip()
if new_cleaned == cleaned:
break
cleaned = new_cleaned
cleaned = re.sub(r'\s+', ' ', cleaned).strip()
return cleaned
def _has_variant_suffix(title: Any) -> bool:
raw = str(title or '').strip()
return bool(re.search(r'[\(\[][^\)\]]*' + variant_keyword_pattern.pattern + r'[^\)\]]*[\)\]]\s*$', raw, flags=re.IGNORECASE))
def _is_compilation(release: Dict[str, Any]) -> bool:
title = str(_extract_lookup_value(release, 'name', 'title', default='') or '').lower()
album_type = str(_extract_lookup_value(release, 'album_type', default='') or '').lower()
return (
album_type == 'compilation'
or 'best of' in title
or 'greatest hits' in title
or 'collection' in title
or 'anthology' in title
or 'essential' in title
)
def _variant_score(release: Dict[str, Any]) -> tuple:
title = str(_extract_lookup_value(release, 'name', 'title', default='') or '').lower()
has_explicit = 'explicit' in title
has_clean = 'clean' in title and not has_explicit
track_count = int(_extract_lookup_value(release, 'track_count', 'total_tracks', default=0) or 0)
release_date = str(_extract_lookup_value(release, 'release_date', default='') or '')
has_variant_suffix = _has_variant_suffix(title)
# Higher is better.
return (
1 if not _is_compilation(release) else 0,
1 if not has_variant_suffix else 0,
2 if has_explicit else (1 if not has_clean else 0),
track_count,
release_date,
)
grouped: Dict[tuple, Dict[str, Any]] = {}
ordered_keys: List[tuple] = []
for release in releases:
title = _extract_lookup_value(release, 'name', 'title', default='') or ''
release_date = _extract_lookup_value(release, 'release_date')
year = _extract_lookup_value(release, 'year')
if not year and release_date:
year = str(release_date)[:4]
year = str(year) if year is not None else ''
cleaned_title = _clean_title(title) or str(title).strip().lower()
key = (cleaned_title, year)
existing = grouped.get(key)
if existing is None:
grouped[key] = release
ordered_keys.append(key)
continue
# If the cleaned titles are still materially different, keep both.
existing_clean = _clean_title(_extract_lookup_value(existing, 'name', 'title', default='') or '')
if SequenceMatcher(None, cleaned_title, existing_clean).ratio() < 0.85:
alt_key = (str(title).strip().lower(), year)
if alt_key not in grouped:
grouped[alt_key] = release
ordered_keys.append(alt_key)
continue
if _variant_score(release) > _variant_score(existing):
grouped[key] = release
return [grouped[key] for key in ordered_keys]
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)
source_artist_ids = options.artist_source_ids or {}
albums: List[Any] = []
active_source: Optional[str] = None
if not albums:
for source in source_priority:
client = metadata_registry.get_client_for_source(source)
if not client:
continue
source_artist_id = (source_artist_ids.get(source) or '').strip()
lookup_artist_id = source_artist_id if source_artist_id else (artist_id if not source_artist_ids else '')
if source_artist_id:
logger.debug("Using %s artist id %s for discography lookup", source, source_artist_id)
try:
albums = get_artist_albums_for_source(
source,
lookup_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 _build_artist_detail_release_card(release: Dict[str, Any]) -> Optional[Dict[str, Any]]:
release_id = _extract_lookup_value(release, 'id', 'album_id', 'release_id')
if not release_id:
return None
album_type = (_extract_lookup_value(release, 'album_type', default='album') or 'album').lower()
release_date = _extract_lookup_value(release, 'release_date')
release_year = None
if release_date:
try:
release_year = str(release_date)[:4]
except Exception:
release_year = None
if not release_year:
release_year = _extract_lookup_value(release, 'year')
if release_year is not None:
release_year = str(release_year)
card = {
'id': release_id,
'name': _extract_lookup_value(release, 'name', 'title', default=release_id),
'title': _extract_lookup_value(release, 'name', 'title', default=release_id),
'album_type': album_type,
'image_url': _extract_lookup_value(release, 'image_url', 'thumb_url', 'cover_image'),
'year': release_year,
'track_count': _extract_lookup_value(release, 'track_count', 'total_tracks', default=0) or 0,
'owned': None,
'track_completion': 'checking',
}
if release_date:
card['release_date'] = release_date
elif release_year:
card['release_date'] = f"{release_year}-01-01"
return card
def get_artist_detail_discography(
artist_id: str,
artist_name: str = '',
options: Optional[MetadataLookupOptions] = None,
) -> Dict[str, Any]:
"""Get artist-detail-ready discography cards from the source-priority lookup flow."""
source_discography = get_artist_discography(
artist_id,
artist_name=artist_name,
options=options,
)
albums: List[Dict[str, Any]] = []
eps: List[Dict[str, Any]] = []
singles: List[Dict[str, Any]] = []
seen_ids = set()
for release in list(source_discography.get('albums', []) or []) + list(source_discography.get('singles', []) or []):
card = _build_artist_detail_release_card(release)
if not card:
continue
release_id = card['id']
if release_id in seen_ids:
continue
seen_ids.add(release_id)
album_type = (card.get('album_type') or 'album').lower()
if album_type == 'ep':
eps.append(card)
elif album_type == 'single':
singles.append(card)
else:
albums.append(card)
if options is None or options.dedup_variants:
albums = _dedup_variant_releases(albums)
eps = _dedup_variant_releases(eps)
singles = _dedup_variant_releases(singles)
albums = _sort_discography_releases(albums)
eps = _sort_discography_releases(eps)
singles = _sort_discography_releases(singles)
has_releases = bool(albums or eps or singles)
return {
'success': has_releases,
'albums': albums,
'eps': eps,
'singles': singles,
'source': source_discography.get('source', 'unknown'),
'source_priority': source_discography.get('source_priority', []),
'error': None if has_releases else f'No releases found for artist "{artist_name or artist_id}"',
}

22
core/metadata/lookup.py Normal file
View file

@ -0,0 +1,22 @@
"""Shared metadata lookup policy objects."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict, Optional
__all__ = ["MetadataLookupOptions"]
@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
artist_source_ids: Optional[Dict[str, str]] = None
dedup_variants: bool = True

355
core/metadata/registry.py Normal file
View file

@ -0,0 +1,355 @@
"""Metadata client registry and source selection.
Owns shared metadata client singletons, runtime client registration, and
canonical source selection. Package-internal code should use this module
instead of importing `web_server`.
"""
from __future__ import annotations
import threading
import hashlib
from typing import Any, Callable, Dict, Optional
from utils.logging_config import get_logger
logger = get_logger("metadata.registry")
MetadataClientFactory = Callable[[], Any]
METADATA_SOURCE_PRIORITY = ("deezer", "itunes", "spotify", "discogs", "hydrabase")
_UNSET = object()
_client_cache_lock = threading.RLock()
_client_cache: Dict[str, Any] = {}
_runtime_clients_lock = threading.RLock()
_runtime_clients: Dict[str, Any] = {
"spotify": None,
"hydrabase": None,
}
_dev_mode_enabled_provider: Callable[[], bool] = lambda: False
_profile_spotify_credentials_provider: Callable[[int], Any] = lambda profile_id: None
def register_runtime_clients(
*,
spotify_client: Any = _UNSET,
hydrabase_client: Any = _UNSET,
dev_mode_enabled_provider: Optional[Callable[[], bool]] = _UNSET,
) -> None:
"""Register app-owned runtime clients.
`None` is a valid value and clears the registered client. Omitted
arguments leave the current registration unchanged.
"""
global _dev_mode_enabled_provider
with _runtime_clients_lock:
if spotify_client is not _UNSET:
_runtime_clients["spotify"] = spotify_client
if hydrabase_client is not _UNSET:
_runtime_clients["hydrabase"] = hydrabase_client
if dev_mode_enabled_provider is not _UNSET:
_dev_mode_enabled_provider = dev_mode_enabled_provider or (lambda: False)
def register_profile_spotify_credentials_provider(
provider: Optional[Callable[[int], Any]] = _UNSET,
) -> None:
"""Register a callable that returns per-profile Spotify credentials."""
global _profile_spotify_credentials_provider
with _runtime_clients_lock:
if provider is not _UNSET:
_profile_spotify_credentials_provider = provider or (lambda profile_id: None)
def get_registered_runtime_client(name: str) -> Any:
with _runtime_clients_lock:
return _runtime_clients.get(name)
def clear_cached_metadata_clients() -> None:
"""Clear lazily-created client singletons.
Runtime clients registered by the host app stay in place.
"""
with _client_cache_lock:
_client_cache.clear()
def clear_cached_metadata_client(cache_key: str) -> None:
"""Clear one lazily-created client singleton by cache key."""
with _client_cache_lock:
_client_cache.pop(cache_key, None)
def clear_cached_profile_spotify_client(profile_id: int) -> None:
"""Clear any cached Spotify client for a specific profile."""
prefix = f"spotify_profile::{profile_id}::"
with _client_cache_lock:
for key in [key for key in _client_cache if key.startswith(prefix)]:
_client_cache.pop(key, None)
def _get_config_value(key: str, default: Any = None) -> Any:
try:
from config.settings import config_manager
return config_manager.get(key, default)
except Exception:
return default
def _get_spotify_factory(client_factory: Optional[MetadataClientFactory]) -> MetadataClientFactory:
if client_factory is not None:
return client_factory
from core.spotify_client import SpotifyClient
return SpotifyClient
def _get_itunes_factory(client_factory: Optional[MetadataClientFactory]) -> MetadataClientFactory:
if client_factory is not None:
return client_factory
from core.itunes_client import iTunesClient
return iTunesClient
def _get_deezer_factory(client_factory: Optional[MetadataClientFactory]) -> MetadataClientFactory:
if client_factory is not None:
return client_factory
from core.deezer_client import DeezerClient
return DeezerClient
def _get_discogs_factory(client_factory: Optional[MetadataClientFactory]) -> MetadataClientFactory:
if client_factory is not None:
return client_factory
from core.discogs_client import DiscogsClient
return DiscogsClient
def get_spotify_client(client_factory: Optional[MetadataClientFactory] = None):
"""Get shared Spotify client.
Prefers the app-registered runtime client. Falls back to a lazily
cached singleton if no runtime client was registered.
"""
runtime_client = get_registered_runtime_client("spotify")
if runtime_client is not None:
return runtime_client
cache_key = "spotify"
factory = _get_spotify_factory(client_factory)
with _client_cache_lock:
client = _client_cache.get(cache_key)
if client is None:
client = factory()
_client_cache[cache_key] = client
return client
def _build_profile_spotify_cache_key(profile_id: int, creds: Dict[str, Any]) -> str:
fingerprint = hashlib.sha256(
f"{profile_id}:{creds.get('client_id', '')}:{creds.get('client_secret', '')}:{creds.get('redirect_uri', '')}".encode(
"utf-8"
)
).hexdigest()
return f"spotify_profile::{profile_id}::{fingerprint}"
def get_spotify_client_for_profile(profile_id: Optional[int] = None):
"""Get a profile-specific Spotify client or fall back to the global one."""
if profile_id is None or profile_id == 1:
return get_spotify_client()
try:
creds = _profile_spotify_credentials_provider(profile_id)
if not creds or not creds.get("client_id"):
return get_spotify_client()
except Exception:
return get_spotify_client()
cache_key = _build_profile_spotify_cache_key(profile_id, creds)
with _client_cache_lock:
client = _client_cache.get(cache_key)
if client is not None and getattr(client, "sp", None) is not None:
return client
try:
from core.spotify_client import SpotifyClient
from spotipy.oauth2 import SpotifyOAuth
import spotipy
auth_manager = SpotifyOAuth(
client_id=creds["client_id"],
client_secret=creds["client_secret"],
redirect_uri=creds.get("redirect_uri", "http://127.0.0.1:8888/callback"),
scope="user-library-read user-read-private playlist-read-private playlist-read-collaborative user-read-email user-follow-read",
cache_path=f"config/.spotify_cache_profile_{profile_id}",
state=f"profile_{profile_id}",
)
profile_client = SpotifyClient()
profile_client.sp = spotipy.Spotify(auth_manager=auth_manager, retries=0, requests_timeout=15)
profile_client.user_id = None
with _client_cache_lock:
_client_cache[cache_key] = profile_client
logger.info("Created per-profile Spotify client for profile %s", profile_id)
return profile_client
except Exception as e:
logger.error("Failed to create per-profile Spotify client for profile %s: %s", profile_id, e)
return get_spotify_client()
def get_deezer_client(client_factory: Optional[MetadataClientFactory] = None):
"""Get cached Deezer client keyed by current access token."""
current_token = _get_config_value("deezer.access_token", None)
cache_key = f"deezer::{current_token or ''}"
factory = _get_deezer_factory(client_factory)
with _client_cache_lock:
client = _client_cache.get(cache_key)
if client is None:
client = factory()
_client_cache[cache_key] = client
return client
def get_itunes_client(client_factory: Optional[MetadataClientFactory] = None):
"""Get cached iTunes client."""
cache_key = "itunes"
factory = _get_itunes_factory(client_factory)
with _client_cache_lock:
client = _client_cache.get(cache_key)
if client is None:
client = factory()
_client_cache[cache_key] = client
return client
def get_discogs_client(
token: Optional[str] = None,
client_factory: Optional[MetadataClientFactory] = None,
):
"""Get cached Discogs client keyed by token."""
if token is None:
current_token = _get_config_value("discogs.token", "") or ""
else:
current_token = token or ""
cache_key = f"discogs::{current_token}"
factory = _get_discogs_factory(client_factory)
with _client_cache_lock:
client = _client_cache.get(cache_key)
if client is None:
client = factory(token=current_token or None) # type: ignore[misc]
_client_cache[cache_key] = client
return client
def is_hydrabase_enabled() -> bool:
"""Return True when Hydrabase is connected and app-enabled."""
try:
client = get_registered_runtime_client("hydrabase")
if not client or not client.is_connected():
return False
return bool(_dev_mode_enabled_provider())
except Exception:
return False
def get_hydrabase_client(allow_fallback: bool = True, require_enabled: bool = True):
"""Return registered Hydrabase client or iTunes fallback."""
try:
client = get_registered_runtime_client("hydrabase")
if client and client.is_connected():
if not require_enabled or bool(_dev_mode_enabled_provider()):
return client
except Exception:
pass
if allow_fallback:
return get_itunes_client()
return None
def get_primary_source(spotify_client_factory: Optional[MetadataClientFactory] = None) -> str:
"""Return configured primary metadata source."""
source = _get_config_value("metadata.fallback_source", "deezer") or "deezer"
if source == "spotify":
try:
spotify = get_spotify_client(client_factory=spotify_client_factory)
if not spotify or not spotify.is_spotify_authenticated():
return "deezer"
except Exception:
return "deezer"
return source
def get_source_priority(preferred_source: str):
"""Return source priority with preferred source first."""
ordered = []
if preferred_source in METADATA_SOURCE_PRIORITY:
ordered.append(preferred_source)
for source in METADATA_SOURCE_PRIORITY:
if source not in ordered:
ordered.append(source)
return ordered
def get_primary_client(
*,
spotify_client_factory: Optional[MetadataClientFactory] = None,
itunes_client_factory: Optional[MetadataClientFactory] = None,
deezer_client_factory: Optional[MetadataClientFactory] = None,
discogs_client_factory: Optional[MetadataClientFactory] = None,
):
"""Return client for configured primary source."""
return get_client_for_source(
get_primary_source(spotify_client_factory=spotify_client_factory),
spotify_client_factory=spotify_client_factory,
itunes_client_factory=itunes_client_factory,
deezer_client_factory=deezer_client_factory,
discogs_client_factory=discogs_client_factory,
)
def get_client_for_source(
source: str,
*,
spotify_client_factory: Optional[MetadataClientFactory] = None,
itunes_client_factory: Optional[MetadataClientFactory] = None,
deezer_client_factory: Optional[MetadataClientFactory] = None,
discogs_client_factory: Optional[MetadataClientFactory] = None,
):
"""Return exact client for a source, or None if unavailable."""
if source == "spotify":
try:
client = get_spotify_client(client_factory=spotify_client_factory)
if client and client.is_spotify_authenticated():
return client
except Exception:
pass
return None
if source == "deezer":
return get_deezer_client(client_factory=deezer_client_factory)
if source == "discogs":
return get_discogs_client(client_factory=discogs_client_factory)
if source == "hydrabase":
return get_hydrabase_client(allow_fallback=False)
if source == "itunes":
return get_itunes_client(client_factory=itunes_client_factory)
return None

174
core/metadata/service.py Normal file
View file

@ -0,0 +1,174 @@
"""Compatibility metadata service facade.
The modern lookup code prefers standalone functions and shared registry
helpers, but the legacy `MetadataService` wrapper remains available for
call sites that still expect an object.
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional, Literal
from core.metadata.registry import (
get_client_for_source,
get_primary_source,
get_spotify_client,
)
from utils.logging_config import get_logger
logger = get_logger("metadata_service")
MetadataProvider = Literal["spotify", "itunes", "auto"]
class MetadataService:
"""
Unified metadata service that seamlessly switches between Spotify and
the configured fallback source.
"""
def __init__(self, preferred_provider: MetadataProvider = "auto"):
self.preferred_provider = preferred_provider
try:
self.spotify = get_spotify_client()
except Exception:
self.spotify = None
self._fallback_source = get_primary_source()
try:
self.itunes = get_client_for_source(self._fallback_source)
except Exception:
self.itunes = None
self._log_initialization()
def _log_initialization(self):
spotify_status = "Authenticated" if self.spotify and self.spotify.is_spotify_authenticated() else "Not authenticated"
fallback_status = "Available" if self.itunes and getattr(self.itunes, "is_authenticated", lambda: False)() else "Not available"
logger.info(
"MetadataService initialized - Spotify: %s, %s: %s",
spotify_status,
self._fallback_source.capitalize(),
fallback_status,
)
logger.info("Preferred provider: %s", self.preferred_provider)
def get_active_provider(self) -> str:
if self.preferred_provider == "spotify":
return "spotify"
if self.preferred_provider == "itunes":
return self._fallback_source
return get_primary_source()
def _get_client(self):
provider = self.get_active_provider()
if provider == "spotify":
if not self.spotify or not self.spotify.is_spotify_authenticated():
logger.warning(
"Spotify requested but not authenticated, falling back to %s",
self._fallback_source,
)
return self.itunes
return self.spotify
return self.itunes
def search_tracks(self, query: str, limit: int = 20) -> List:
client = self._get_client()
provider = self.get_active_provider()
logger.debug("Searching tracks with %s: %r", provider, query)
return client.search_tracks(query, limit)
def search_artists(self, query: str, limit: int = 20) -> List:
client = self._get_client()
provider = self.get_active_provider()
logger.debug("Searching artists with %s: %r", provider, query)
return client.search_artists(query, limit)
def search_albums(self, query: str, limit: int = 20) -> List:
client = self._get_client()
provider = self.get_active_provider()
logger.debug("Searching albums with %s: %r", provider, query)
return client.search_albums(query, limit)
def get_track_details(self, track_id: str) -> Optional[Dict[str, Any]]:
client = self._get_client()
return client.get_track_details(track_id)
def get_album(self, album_id: str) -> Optional[Dict[str, Any]]:
client = self._get_client()
return client.get_album(album_id)
def get_album_tracks(self, album_id: str) -> Optional[Dict[str, Any]]:
client = self._get_client()
provider = self.get_active_provider()
logger.debug("Fetching album tracks with %s: %s", provider, album_id)
return client.get_album_tracks(album_id)
def get_artist(self, artist_id: str) -> Optional[Dict[str, Any]]:
client = self._get_client()
return client.get_artist(artist_id)
def get_artist_albums(self, artist_id: str, album_type: str = "album,single", limit: int = 50) -> List:
client = self._get_client()
provider = self.get_active_provider()
logger.debug("Fetching artist albums with %s: %s", provider, artist_id)
return client.get_artist_albums(artist_id, album_type, limit)
def get_track_features(self, track_id: str) -> Optional[Dict[str, Any]]:
client = self._get_client()
return client.get_track_features(track_id)
def get_user_playlists(self) -> List:
if self.spotify and self.spotify.is_spotify_authenticated():
return self.spotify.get_user_playlists()
logger.warning("User playlists only available with Spotify authentication")
return []
def get_saved_tracks(self) -> List:
if self.spotify and self.spotify.is_spotify_authenticated():
return self.spotify.get_saved_tracks()
logger.warning("Saved tracks only available with Spotify authentication")
return []
def get_saved_tracks_count(self) -> int:
if self.spotify and self.spotify.is_spotify_authenticated():
return self.spotify.get_saved_tracks_count()
return 0
def is_authenticated(self) -> bool:
return bool(self.spotify and self.spotify.is_spotify_authenticated()) or bool(
self.itunes and getattr(self.itunes, "is_authenticated", lambda: False)()
)
def get_provider_info(self) -> Dict[str, Any]:
spotify_authenticated = bool(self.spotify and self.spotify.is_spotify_authenticated())
itunes_available = bool(self.itunes and getattr(self.itunes, "is_authenticated", lambda: False)())
return {
"active_provider": self.get_active_provider(),
"spotify_authenticated": spotify_authenticated,
"itunes_available": itunes_available,
"fallback_source": self._fallback_source,
"preferred_provider": self.preferred_provider,
"can_access_user_data": spotify_authenticated,
}
def reload_config(self):
logger.info("Reloading metadata service configuration")
if self.spotify and hasattr(self.spotify, "reload_config"):
self.spotify.reload_config()
new_source = get_primary_source()
self._fallback_source = new_source
try:
self.itunes = get_client_for_source(new_source)
except Exception:
self.itunes = None
self._log_initialization()
_metadata_service_instance: Optional[MetadataService] = None
def get_metadata_service() -> MetadataService:
global _metadata_service_instance
if _metadata_service_instance is None:
_metadata_service_instance = MetadataService()
return _metadata_service_instance

View file

@ -0,0 +1,342 @@
"""MusicMap similar-artist helpers for metadata API."""
from __future__ import annotations
from typing import Any, Dict, List, Optional
import requests
from core.metadata import registry as metadata_registry
from core.metadata.artist_image import _extract_artist_image_url
from core.metadata.discography import (
_extract_lookup_value,
_normalize_artist_name,
_pick_best_artist_match,
_search_artists_for_source,
)
from core.metadata.lookup import MetadataLookupOptions
from utils.logging_config import get_logger
logger = get_logger("metadata.similar_artists")
__all__ = [
"get_musicmap_similar_artists",
"iter_musicmap_similar_artist_events",
]
def _get_source_chain_for_lookup(options: MetadataLookupOptions) -> List[str]:
primary_source = metadata_registry.get_primary_source()
source_chain = list(metadata_registry.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 _fetch_musicmap_similar_artist_names(artist_name: str) -> List[str]:
"""Fetch similar artist names from MusicMap."""
if not (artist_name or '').strip():
raise ValueError('Artist name is required')
from bs4 import BeautifulSoup
from urllib.parse import quote_plus
url_artist = quote_plus(artist_name.strip())
musicmap_url = f'https://www.music-map.com/{url_artist}'
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.5',
}
logger.debug("Fetching MusicMap: %s", musicmap_url)
response = requests.get(musicmap_url, headers=headers, timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.text, 'html.parser')
gnod_map = soup.find(id='gnodMap')
if not gnod_map:
raise ValueError('Could not find artist map on MusicMap')
searched_artist_lower = _normalize_artist_name(artist_name)
similar_artist_names: List[str] = []
seen_names = set()
for anchor in gnod_map.find_all('a'):
artist_text = anchor.get_text(strip=True)
normalized_name = _normalize_artist_name(artist_text)
if not normalized_name or normalized_name == searched_artist_lower or normalized_name in seen_names:
continue
seen_names.add(normalized_name)
similar_artist_names.append(artist_text)
logger.debug("Found %s similar artists from MusicMap", len(similar_artist_names))
return similar_artist_names
def _build_similar_artist_payload(artist_data: Any, source: str) -> Optional[Dict[str, Any]]:
artist_id = _extract_lookup_value(artist_data, 'id', 'artist_id', 'spotify_id', 'itunes_id', 'deezer_id')
if not artist_id:
return None
if isinstance(artist_data, dict):
name = artist_data.get('name') or artist_data.get('artist_name') or artist_data.get('title')
genres = artist_data.get('genres') or []
popularity = artist_data.get('popularity') or artist_data.get('rank') or 0
else:
name = (
getattr(artist_data, 'name', None)
or getattr(artist_data, 'artist_name', None)
or getattr(artist_data, 'title', None)
)
genres = getattr(artist_data, 'genres', None) or []
popularity = getattr(artist_data, 'popularity', None) or getattr(artist_data, 'rank', None) or 0
if isinstance(genres, str):
genres = [genres]
elif not isinstance(genres, list):
try:
genres = list(genres)
except TypeError:
genres = []
try:
popularity = int(popularity or 0)
except Exception:
popularity = 0
return {
'id': str(artist_id),
'name': str(name or artist_id),
'image_url': _extract_artist_image_url(artist_data),
'genres': genres,
'popularity': popularity,
'source': source,
}
def _resolve_musicmap_artist_source_ids(artist_name: str, source_chain: List[str]) -> Dict[str, Optional[str]]:
searched_source_ids: Dict[str, Optional[str]] = {}
for source in source_chain:
client = metadata_registry.get_client_for_source(source)
if not client:
searched_source_ids[source] = None
continue
search_results = _search_artists_for_source(source, client, artist_name, limit=1)
searched_source_ids[source] = _extract_lookup_value(search_results[0], 'id', 'artist_id') if search_results else None
return searched_source_ids
def _match_musicmap_similar_artist(
candidate_name: str,
source_chain: List[str],
searched_artist_name: str,
searched_source_ids: Dict[str, Optional[str]],
) -> tuple[Optional[str], Optional[Dict[str, Any]]]:
target_name = _normalize_artist_name(candidate_name)
searched_name = _normalize_artist_name(searched_artist_name)
for source in source_chain:
client = metadata_registry.get_client_for_source(source)
if not client:
continue
search_results = _search_artists_for_source(source, client, candidate_name, limit=1)
if not search_results:
continue
matched_artist = _pick_best_artist_match(search_results, candidate_name)
if not matched_artist:
continue
matched_name = _normalize_artist_name(
_extract_lookup_value(matched_artist, 'name', 'artist_name', 'title')
)
if matched_name and matched_name == searched_name:
continue
matched_id = _extract_lookup_value(matched_artist, 'id', 'artist_id')
if not matched_id:
continue
if str(matched_id) == str(searched_source_ids.get(source) or ''):
continue
payload = _build_similar_artist_payload(matched_artist, source)
if not payload:
continue
if source == 'itunes' and not payload.get('image_url') and hasattr(client, 'get_artist'):
try:
full_artist = client.get_artist(str(matched_id))
image_url = _extract_artist_image_url(full_artist)
if image_url:
payload['image_url'] = image_url
elif hasattr(client, '_get_artist_image_from_albums'):
album_image_url = client._get_artist_image_from_albums(str(matched_id))
if album_image_url:
payload['image_url'] = album_image_url
except Exception as exc:
logger.debug("Could not enrich iTunes image for %s: %s", matched_id, exc)
if target_name and _normalize_artist_name(payload['name']) == searched_name:
continue
return source, payload
return None, None
def iter_musicmap_similar_artist_events(
artist_name: str,
limit: int = 20,
source_override: Optional[str] = None,
):
"""Yield MusicMap similar-artist events using source priority."""
try:
source_chain = _get_source_chain_for_lookup(
MetadataLookupOptions(source_override=source_override, allow_fallback=True)
)
available_sources = [source for source in source_chain if metadata_registry.get_client_for_source(source)]
if not available_sources:
yield {
'type': 'error',
'error': 'No metadata providers available for similar artist matching',
'status_code': 503,
}
return
similar_artist_names = _fetch_musicmap_similar_artist_names(artist_name)
searched_source_ids = _resolve_musicmap_artist_source_ids(artist_name, source_chain)
yield {
'type': 'start',
'artist_name': artist_name,
'total_found': len(similar_artist_names),
'source_priority': source_chain,
}
matched_count = 0
seen_names = set()
seen_ids = set()
for candidate_name in similar_artist_names[:limit]:
normalized_candidate = _normalize_artist_name(candidate_name)
if not normalized_candidate or normalized_candidate in seen_names:
continue
source, payload = _match_musicmap_similar_artist(
candidate_name,
source_chain,
artist_name,
searched_source_ids,
)
if not payload:
continue
payload_id = str(payload.get('id') or '')
if payload_id in seen_ids:
continue
seen_names.add(normalized_candidate)
seen_ids.add(payload_id)
matched_count += 1
yield {
'type': 'artist',
'artist': payload,
'source': source,
}
yield {
'type': 'complete',
'complete': True,
'total': matched_count,
'total_found': len(similar_artist_names),
'artist_name': artist_name,
'source_priority': source_chain,
}
except requests.exceptions.RequestException as exc:
logger.debug("Error fetching MusicMap for %s: %s", artist_name, exc)
yield {
'type': 'error',
'error': f'Failed to fetch from MusicMap: {exc}',
'status_code': 502,
}
except ValueError as exc:
status_code = 404 if 'Could not find artist map on MusicMap' in str(exc) else 400
yield {
'type': 'error',
'error': str(exc),
'status_code': status_code,
}
except Exception as exc:
logger.error("Error streaming similar artists for %s: %s", artist_name, exc)
yield {
'type': 'error',
'error': str(exc),
'status_code': 500,
}
def get_musicmap_similar_artists(
artist_name: str,
limit: int = 20,
source_override: Optional[str] = None,
) -> Dict[str, Any]:
"""Return matched MusicMap similar artists as a single payload."""
artists: List[Dict[str, Any]] = []
total_found = 0
error_message = None
status_code = 500
source_priority: List[str] = []
for event in iter_musicmap_similar_artist_events(
artist_name,
limit=limit,
source_override=source_override,
):
if event.get('type') == 'start':
total_found = event.get('total_found', 0)
source_priority = event.get('source_priority', [])
elif event.get('type') == 'artist' and event.get('artist'):
artists.append(event['artist'])
elif event.get('type') == 'complete':
total_found = event.get('total_found', total_found)
source_priority = event.get('source_priority', source_priority)
elif event.get('type') == 'error':
error_message = event.get('error', 'Unknown error')
status_code = int(event.get('status_code') or status_code or 500)
break
if error_message:
return {
'success': False,
'error': error_message,
'status_code': status_code,
'artist': artist_name,
'similar_artists': [],
'total_found': total_found,
'total_matched': 0,
'source_priority': source_priority,
}
return {
'success': True,
'artist': artist_name,
'similar_artists': artists,
'total_found': total_found,
'total_matched': len(artists),
'source_priority': source_priority,
}

View file

@ -23,7 +23,7 @@ from core.imports.context import (
get_source_tag_names, get_source_tag_names,
normalize_import_context, normalize_import_context,
) )
from core.metadata_service import get_itunes_client from core.metadata.registry import get_itunes_client
from database.music_database import get_database from database.music_database import get_database
from core.metadata.common import ( from core.metadata.common import (
get_config_manager, get_config_manager,

File diff suppressed because it is too large Load diff

View file

@ -196,7 +196,7 @@ class RepairWorker:
def metadata_cache(self): def metadata_cache(self):
if self._metadata_cache is None: if self._metadata_cache is None:
try: try:
from core.metadata_cache import get_metadata_cache from core.metadata.cache import get_metadata_cache
self._metadata_cache = get_metadata_cache() self._metadata_cache = get_metadata_cache()
except Exception as e: except Exception as e:
logger.error("Failed to get metadata cache: %s", e) logger.error("Failed to get metadata cache: %s", e)

View file

@ -8,7 +8,7 @@ from functools import wraps
from dataclasses import dataclass from dataclasses import dataclass
from utils.logging_config import get_logger from utils.logging_config import get_logger
from config.settings import config_manager from config.settings import config_manager
from core.metadata_cache import get_metadata_cache from core.metadata.cache import get_metadata_cache
logger = get_logger("spotify_client") logger = get_logger("spotify_client")

View file

@ -358,7 +358,7 @@ class WatchlistScanner:
def metadata_service(self): def metadata_service(self):
"""Get or create MetadataService instance (lazy loading)""" """Get or create MetadataService instance (lazy loading)"""
if self._metadata_service is None: if self._metadata_service is None:
from core.metadata_service import MetadataService from core.metadata.service import MetadataService
self._metadata_service = MetadataService() self._metadata_service = MetadataService()
return self._metadata_service return self._metadata_service
@ -1642,7 +1642,7 @@ class WatchlistScanner:
return self._best_artist_match(results, artist_name) return self._best_artist_match(results, artist_name)
# Fallback: use cached Deezer client # Fallback: use cached Deezer client
from core.metadata_service import get_deezer_client from core.metadata.registry import get_deezer_client
client = get_deezer_client() client = get_deezer_client()
results = client.search_artists(artist_name, limit=5) results = client.search_artists(artist_name, limit=5)
return self._best_artist_match(results, artist_name) return self._best_artist_match(results, artist_name)
@ -1653,7 +1653,7 @@ class WatchlistScanner:
def _match_to_discogs(self, artist_name: str) -> Optional[str]: def _match_to_discogs(self, artist_name: str) -> Optional[str]:
"""Match artist name to Discogs ID using fuzzy name comparison.""" """Match artist name to Discogs ID using fuzzy name comparison."""
try: try:
from core.metadata_service import get_discogs_client from core.metadata.registry import get_discogs_client
client = get_discogs_client() client = get_discogs_client()
results = client.search_artists(artist_name, limit=5) results = client.search_artists(artist_name, limit=5)
return self._best_artist_match(results, artist_name) return self._best_artist_match(results, artist_name)

View file

@ -1,6 +1,6 @@
from types import SimpleNamespace from types import SimpleNamespace
from core import metadata_service from core.metadata import registry as metadata_registry
from core.imports import resolution from core.imports import resolution
@ -67,12 +67,12 @@ def test_get_single_track_import_context_uses_primary_source_priority(monkeypatc
) )
spotify_client = FakeClient() spotify_client = FakeClient()
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr( monkeypatch.setattr(
metadata_service, metadata_registry,
"get_client_for_source", "get_client_for_source",
lambda source: {"deezer": deezer_client, "spotify": spotify_client, "itunes": None}.get(source), lambda source, **kwargs: {"deezer": deezer_client, "spotify": spotify_client, "itunes": None}.get(source),
) )
result = resolution.get_single_track_import_context("Song One", "Artist One") result = resolution.get_single_track_import_context("Song One", "Artist One")
@ -101,12 +101,12 @@ def test_get_single_track_import_context_falls_back_to_next_source(monkeypatch):
artist_details={"spotify-artist-1": {"id": "spotify-artist-1", "genres": ["indie"]}}, artist_details={"spotify-artist-1": {"id": "spotify-artist-1", "genres": ["indie"]}},
) )
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr( monkeypatch.setattr(
metadata_service, metadata_registry,
"get_client_for_source", "get_client_for_source",
lambda source: {"deezer": deezer_client, "spotify": spotify_client, "itunes": None}.get(source), lambda source, **kwargs: {"deezer": deezer_client, "spotify": spotify_client, "itunes": None}.get(source),
) )
result = resolution.get_single_track_import_context("Song Two", "Artist Two") result = resolution.get_single_track_import_context("Song Two", "Artist Two")
@ -134,12 +134,12 @@ def test_get_single_track_import_context_uses_explicit_override_first(monkeypatc
) )
deezer_client = FakeClient() deezer_client = FakeClient()
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr( monkeypatch.setattr(
metadata_service, metadata_registry,
"get_client_for_source", "get_client_for_source",
lambda source: {"deezer": deezer_client, "spotify": spotify_client, "itunes": None}.get(source), lambda source, **kwargs: {"deezer": deezer_client, "spotify": spotify_client, "itunes": None}.get(source),
) )
result = resolution.get_single_track_import_context( result = resolution.get_single_track_import_context(
@ -167,12 +167,12 @@ def test_get_single_track_import_context_uses_explicit_override_source(monkeypat
spotify_client = FakeClient() spotify_client = FakeClient()
deezer_client = FakeClient() deezer_client = FakeClient()
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr( monkeypatch.setattr(
metadata_service, metadata_registry,
"get_client_for_source", "get_client_for_source",
lambda source: {"deezer": deezer_client, "spotify": spotify_client, "itunes": itunes_client}.get(source), lambda source, **kwargs: {"deezer": deezer_client, "spotify": spotify_client, "itunes": itunes_client}.get(source),
) )
result = resolution.get_single_track_import_context( result = resolution.get_single_track_import_context(
@ -199,12 +199,12 @@ def test_get_single_track_import_context_returns_fallback_payload_when_no_source
spotify_client = FakeClient() spotify_client = FakeClient()
itunes_client = FakeClient() itunes_client = FakeClient()
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr( monkeypatch.setattr(
metadata_service, metadata_registry,
"get_client_for_source", "get_client_for_source",
lambda source: {"deezer": deezer_client, "spotify": spotify_client, "itunes": itunes_client}.get(source), lambda source, **kwargs: {"deezer": deezer_client, "spotify": spotify_client, "itunes": itunes_client}.get(source),
) )
result = resolution.get_single_track_import_context("Missing Song", "Missing Artist") result = resolution.get_single_track_import_context("Missing Song", "Missing Artist")

View file

@ -18,7 +18,7 @@ from core.artist_source_detail import build_source_only_artist_detail
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Fixtures — stubs for the metadata_service helpers the function calls # Fixtures — stubs for the metadata helpers the function calls
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
def _success_discography(**overrides): def _success_discography(**overrides):
@ -41,13 +41,14 @@ def _empty_discography():
@pytest.fixture @pytest.fixture
def _stub_metadata(monkeypatch): def _stub_metadata(monkeypatch):
"""Replace the metadata_service imports with controllable stubs. """Replace the metadata imports with controllable stubs.
The function imports ``get_artist_image_url`` and The function imports ``get_artist_image_url`` and
``get_artist_detail_discography`` lazily inside its body (deferred import), ``get_artist_detail_discography`` from the concrete metadata modules, so we
so we patch on the metadata_service module directly. patch those modules directly.
""" """
from core import metadata_service from core.metadata import artist_image as metadata_artist_image
from core.metadata import discography as metadata_discography
state = { state = {
"image_url": None, "image_url": None,
@ -64,8 +65,8 @@ def _stub_metadata(monkeypatch):
state["last_discog_call"] = (artist_id, artist_name) state["last_discog_call"] = (artist_id, artist_name)
return state["discography"] return state["discography"]
monkeypatch.setattr(metadata_service, "get_artist_image_url", fake_get_artist_image_url) monkeypatch.setattr(metadata_artist_image, "get_artist_image_url", fake_get_artist_image_url)
monkeypatch.setattr(metadata_service, "get_artist_detail_discography", fake_get_artist_detail_discography) monkeypatch.setattr(metadata_discography, "get_artist_detail_discography", fake_get_artist_detail_discography)
return state return state
@ -156,7 +157,7 @@ class TestPerSourceEnrichment:
) )
assert payload["artist"]["genres"] == ["alt rock", "emo"] assert payload["artist"]["genres"] == ["alt rock", "emo"]
assert payload["artist"]["followers"] == 12345 assert payload["artist"]["followers"] == 12345
# image_url falls back to Spotify's image when metadata_service returned None # image_url falls back to Spotify's image when metadata returned None
assert payload["artist"]["image_url"] == "https://sp/img.jpg" assert payload["artist"]["image_url"] == "https://sp/img.jpg"
def test_deezer_extracts_genres_and_followers(self, _stub_metadata): def test_deezer_extracts_genres_and_followers(self, _stub_metadata):

View file

@ -41,14 +41,15 @@ if "config.settings" not in sys.modules:
sys.modules["config"] = config_pkg sys.modules["config"] = config_pkg
sys.modules["config.settings"] = settings_mod sys.modules["config.settings"] = settings_mod
from core import metadata_service from core.metadata import album_tracks as metadata_album_tracks
from core.metadata import registry as metadata_registry
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def _clear_metadata_client_cache(): def _clear_metadata_client_cache():
metadata_service.clear_cached_metadata_clients() metadata_registry.clear_cached_metadata_clients()
yield yield
metadata_service.clear_cached_metadata_clients() metadata_registry.clear_cached_metadata_clients()
def _album(album_id="album-1", name="Album One", album_type="album"): def _album(album_id="album-1", name="Album One", album_type="album"):
@ -80,9 +81,9 @@ def _track(track_id="track-1", name="Track One"):
def test_get_artist_album_tracks_uses_primary_source_priority(monkeypatch): def test_get_artist_album_tracks_uses_primary_source_priority(monkeypatch):
calls = [] calls = []
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: object()) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: object())
def fake_get_album_for_source(source, album_id): def fake_get_album_for_source(source, album_id):
calls.append(("album", source, album_id)) calls.append(("album", source, album_id))
@ -92,10 +93,10 @@ def test_get_artist_album_tracks_uses_primary_source_priority(monkeypatch):
calls.append(("tracks", source, album_id)) calls.append(("tracks", source, album_id))
return {"items": [_track()]} if source == "deezer" and album_id == "album-1" else None return {"items": [_track()]} if source == "deezer" and album_id == "album-1" else None
monkeypatch.setattr(metadata_service, "get_album_for_source", fake_get_album_for_source) monkeypatch.setattr("core.metadata.album_tracks.get_album_for_source", fake_get_album_for_source)
monkeypatch.setattr(metadata_service, "get_album_tracks_for_source", fake_get_album_tracks_for_source) monkeypatch.setattr("core.metadata.album_tracks.get_album_tracks_for_source", fake_get_album_tracks_for_source)
result = metadata_service.get_artist_album_tracks( result = metadata_album_tracks.get_artist_album_tracks(
"album-1", "album-1",
artist_name="Artist One", artist_name="Artist One",
album_name="Album One", album_name="Album One",
@ -114,9 +115,9 @@ def test_get_artist_album_tracks_uses_primary_source_priority(monkeypatch):
def test_get_artist_album_tracks_resolves_database_album_reference(monkeypatch): def test_get_artist_album_tracks_resolves_database_album_reference(monkeypatch):
calls = [] calls = []
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: object()) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: object())
def fake_get_album_for_source(source, album_id): def fake_get_album_for_source(source, album_id):
calls.append(("album", source, album_id)) calls.append(("album", source, album_id))
@ -135,11 +136,11 @@ def test_get_artist_album_tracks_resolves_database_album_reference(monkeypatch):
assert preferred_source == "itunes" assert preferred_source == "itunes"
return "itunes-123", "itunes" return "itunes-123", "itunes"
monkeypatch.setattr(metadata_service, "get_album_for_source", fake_get_album_for_source) monkeypatch.setattr("core.metadata.album_tracks.get_album_for_source", fake_get_album_for_source)
monkeypatch.setattr(metadata_service, "get_album_tracks_for_source", fake_get_album_tracks_for_source) monkeypatch.setattr("core.metadata.album_tracks.get_album_tracks_for_source", fake_get_album_tracks_for_source)
monkeypatch.setattr(metadata_service, "resolve_album_reference", fake_resolve_album_reference) monkeypatch.setattr("core.metadata.album_tracks.resolve_album_reference", fake_resolve_album_reference)
result = metadata_service.get_artist_album_tracks( result = metadata_album_tracks.get_artist_album_tracks(
"db-1", "db-1",
artist_name="Artist One", artist_name="Artist One",
album_name="Album One", album_name="Album One",
@ -189,10 +190,10 @@ def test_resolve_album_reference_prefers_stored_external_id(monkeypatch):
return conn return conn
monkeypatch.setattr("database.music_database.get_database", lambda: _FakeDatabase()) monkeypatch.setattr("database.music_database.get_database", lambda: _FakeDatabase())
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify"])
resolved_id, resolved_source = metadata_service.resolve_album_reference("1", preferred_source="deezer") resolved_id, resolved_source = metadata_album_tracks.resolve_album_reference("1", preferred_source="deezer")
assert resolved_id == "deezer-abc" assert resolved_id == "deezer-abc"
assert resolved_source == "deezer" assert resolved_source == "deezer"
@ -237,11 +238,11 @@ def test_resolve_album_reference_searches_by_name_when_no_external_id_exists(mon
fake_client = _FakeSearchClient() fake_client = _FakeSearchClient()
monkeypatch.setattr("database.music_database.get_database", lambda: _FakeDatabase()) monkeypatch.setattr("database.music_database.get_database", lambda: _FakeDatabase())
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify"])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: fake_client if source == "deezer" else None) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: fake_client if source == "deezer" else None)
resolved_id, resolved_source = metadata_service.resolve_album_reference("1", preferred_source="deezer") resolved_id, resolved_source = metadata_album_tracks.resolve_album_reference("1", preferred_source="deezer")
assert resolved_id == "searched-123" assert resolved_id == "searched-123"
assert resolved_source == "deezer" assert resolved_source == "deezer"

View file

@ -38,7 +38,8 @@ if "config.settings" not in sys.modules:
sys.modules["config"] = config_pkg sys.modules["config"] = config_pkg
sys.modules["config.settings"] = settings_mod sys.modules["config.settings"] = settings_mod
from core import metadata_service from core.metadata import artist_image as metadata_artist_image
from core.metadata import registry as metadata_registry
class _FakeSpotifyClient: class _FakeSpotifyClient:
@ -101,15 +102,15 @@ def test_get_artist_image_url_uses_primary_source_priority(monkeypatch):
deezer = _FakeDeezerClient() deezer = _FakeDeezerClient()
spotify = _FakeSpotifyClient() spotify = _FakeSpotifyClient()
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify"])
monkeypatch.setattr( monkeypatch.setattr(
metadata_service, metadata_registry,
"get_client_for_source", "get_client_for_source",
lambda source: {"deezer": deezer, "spotify": spotify}.get(source), lambda source, **kwargs: {"deezer": deezer, "spotify": spotify}.get(source),
) )
image_url = metadata_service.get_artist_image_url("artist-1") image_url = metadata_artist_image.get_artist_image_url("artist-1")
assert image_url == "https://deezer.example/artist.jpg" assert image_url == "https://deezer.example/artist.jpg"
assert deezer.calls == ["artist-1"] assert deezer.calls == ["artist-1"]
@ -120,15 +121,15 @@ def test_get_artist_image_url_uses_itunes_album_art_for_explicit_override(monkey
itunes = _FakeItunesClient() itunes = _FakeItunesClient()
spotify = _FakeSpotifyClient() spotify = _FakeSpotifyClient()
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "spotify") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "spotify")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "itunes"])
monkeypatch.setattr( monkeypatch.setattr(
metadata_service, metadata_registry,
"get_client_for_source", "get_client_for_source",
lambda source: {"itunes": itunes, "spotify": spotify}.get(source), lambda source, **kwargs: {"itunes": itunes, "spotify": spotify}.get(source),
) )
image_url = metadata_service.get_artist_image_url("12345", source_override="itunes") image_url = metadata_artist_image.get_artist_image_url("12345", source_override="itunes")
assert image_url == "https://itunes.example/artist.jpg" assert image_url == "https://itunes.example/artist.jpg"
assert itunes.calls == ["12345"] assert itunes.calls == ["12345"]
@ -140,15 +141,15 @@ def test_get_artist_image_url_handles_hydrabase_plugin(monkeypatch):
deezer = _FakeDeezerClient("https://deezer.example/hydra.jpg") deezer = _FakeDeezerClient("https://deezer.example/hydra.jpg")
spotify = _FakeSpotifyClient() spotify = _FakeSpotifyClient()
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "spotify") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "spotify")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "deezer"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "deezer"])
monkeypatch.setattr( monkeypatch.setattr(
metadata_service, metadata_registry,
"get_client_for_source", "get_client_for_source",
lambda source: {"deezer": deezer, "spotify": spotify}.get(source), lambda source, **kwargs: {"deezer": deezer, "spotify": spotify}.get(source),
) )
image_url = metadata_service.get_artist_image_url("artist-1", source_override="hydrabase", plugin="deezer") image_url = metadata_artist_image.get_artist_image_url("artist-1", source_override="hydrabase", plugin="deezer")
assert image_url == "https://deezer.example/hydra.jpg" assert image_url == "https://deezer.example/hydra.jpg"
assert deezer.calls == ["artist-1"] assert deezer.calls == ["artist-1"]

View file

@ -0,0 +1,234 @@
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
def get_active_media_server(self):
return "plex"
settings_mod.config_manager = _DummyConfigManager()
config_pkg.settings = settings_mod
sys.modules["config"] = config_pkg
sys.modules["config.settings"] = settings_mod
from core.metadata import registry as metadata_registry
from config.settings import config_manager
@pytest.fixture(autouse=True)
def _clear_metadata_client_cache():
metadata_registry.clear_cached_metadata_clients()
metadata_registry.register_profile_spotify_credentials_provider(lambda profile_id: None)
yield
metadata_registry.clear_cached_metadata_clients()
metadata_registry.register_profile_spotify_credentials_provider(lambda profile_id: None)
def test_primary_client_is_cached_for_same_source(monkeypatch):
calls = {"deezer": 0}
class FakeDeezerClient:
def __init__(self):
calls["deezer"] += 1
monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr("core.deezer_client.DeezerClient", FakeDeezerClient)
first = metadata_registry.get_primary_client()
second = metadata_registry.get_primary_client()
assert first is second
assert calls["deezer"] == 1
def test_primary_client_switches_cache_by_source(monkeypatch):
calls = {"deezer": 0, "itunes": 0}
sources = iter(["deezer", "itunes"])
class FakeDeezerClient:
def __init__(self):
calls["deezer"] += 1
class FakeITunesClient:
def __init__(self):
calls["itunes"] += 1
monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: next(sources))
monkeypatch.setattr("core.deezer_client.DeezerClient", FakeDeezerClient)
monkeypatch.setattr("core.itunes_client.iTunesClient", FakeITunesClient)
deezer_client = metadata_registry.get_primary_client()
itunes_client = metadata_registry.get_primary_client()
assert deezer_client is not itunes_client
assert calls["deezer"] == 1
assert calls["itunes"] == 1
def test_deezer_client_cache_tracks_token(monkeypatch):
tokens = iter(["token-a", "token-b"])
calls = {"deezer": 0}
class FakeDeezerClient:
def __init__(self):
calls["deezer"] += 1
monkeypatch.setattr("core.deezer_client.DeezerClient", FakeDeezerClient)
monkeypatch.setattr(config_manager, "get", lambda key, default=None: next(tokens) if key == "deezer.access_token" else default)
first = metadata_registry.get_deezer_client()
second = metadata_registry.get_deezer_client()
assert first is not second
assert calls["deezer"] == 2
def test_profile_spotify_client_is_cached_per_profile(monkeypatch):
calls = {"spotify": 0, "oauth": 0}
creds_by_profile = {
2: {"client_id": "cid-a", "client_secret": "sec-a", "redirect_uri": "uri-a"},
3: {"client_id": "cid-b", "client_secret": "sec-b", "redirect_uri": "uri-b"},
}
class FakeSpotifyClient:
def __init__(self):
calls["spotify"] += 1
self.sp = None
self.user_id = None
class FakeOAuth:
def __init__(self, *args, **kwargs):
calls["oauth"] += 1
class FakeSpotify:
def __init__(self, *args, **kwargs):
self.auth_manager = kwargs.get("auth_manager")
monkeypatch.setattr("core.spotify_client.SpotifyClient", FakeSpotifyClient)
monkeypatch.setattr(sys.modules["spotipy"].oauth2, "SpotifyOAuth", FakeOAuth)
monkeypatch.setattr(sys.modules["spotipy"], "Spotify", FakeSpotify)
metadata_registry.register_profile_spotify_credentials_provider(lambda profile_id: creds_by_profile.get(profile_id))
first = metadata_registry.get_spotify_client_for_profile(2)
second = metadata_registry.get_spotify_client_for_profile(2)
third = metadata_registry.get_spotify_client_for_profile(3)
assert first is second
assert first is not third
assert calls["spotify"] == 2
assert calls["oauth"] == 2
def test_clear_cached_profile_spotify_client_only_affects_one_profile(monkeypatch):
calls = {"spotify": 0}
creds_by_profile = {
2: {"client_id": "cid-a", "client_secret": "sec-a", "redirect_uri": "uri-a"},
3: {"client_id": "cid-b", "client_secret": "sec-b", "redirect_uri": "uri-b"},
}
class FakeSpotifyClient:
def __init__(self):
calls["spotify"] += 1
self.sp = None
self.user_id = None
class FakeOAuth:
def __init__(self, *args, **kwargs):
pass
class FakeSpotify:
def __init__(self, *args, **kwargs):
pass
monkeypatch.setattr("core.spotify_client.SpotifyClient", FakeSpotifyClient)
monkeypatch.setattr(sys.modules["spotipy"].oauth2, "SpotifyOAuth", FakeOAuth)
monkeypatch.setattr(sys.modules["spotipy"], "Spotify", FakeSpotify)
metadata_registry.register_profile_spotify_credentials_provider(lambda profile_id: creds_by_profile.get(profile_id))
first_profile = metadata_registry.get_spotify_client_for_profile(2)
other_profile = metadata_registry.get_spotify_client_for_profile(3)
metadata_registry.clear_cached_profile_spotify_client(2)
refreshed_profile = metadata_registry.get_spotify_client_for_profile(2)
same_other_profile = metadata_registry.get_spotify_client_for_profile(3)
assert refreshed_profile is not first_profile
assert same_other_profile is other_profile
assert calls["spotify"] == 3
def test_profile_spotify_client_falls_back_to_global_when_no_credentials(monkeypatch):
global_client = object()
monkeypatch.setattr(metadata_registry, "get_spotify_client", lambda client_factory=None: global_client)
metadata_registry.register_profile_spotify_credentials_provider(lambda profile_id: None)
assert metadata_registry.get_spotify_client_for_profile(2) is global_client
class _FakeHydrabaseClient:
def __init__(self, connected=True):
self._connected = connected
def is_connected(self):
return self._connected
def test_hydrabase_enabled_requires_connection_and_dev_mode(monkeypatch):
metadata_registry.register_runtime_clients(
hydrabase_client=_FakeHydrabaseClient(connected=True),
dev_mode_enabled_provider=lambda: True,
)
assert metadata_registry.is_hydrabase_enabled() is True
metadata_registry.register_runtime_clients(dev_mode_enabled_provider=lambda: False)
assert metadata_registry.is_hydrabase_enabled() is False
metadata_registry.register_runtime_clients(
hydrabase_client=_FakeHydrabaseClient(connected=False),
dev_mode_enabled_provider=lambda: True,
)
assert metadata_registry.is_hydrabase_enabled() is False
def test_get_client_for_source_hydrabase_requires_enablement(monkeypatch):
metadata_registry.register_runtime_clients(
hydrabase_client=_FakeHydrabaseClient(connected=True),
dev_mode_enabled_provider=lambda: False,
)
assert metadata_registry.get_client_for_source("hydrabase") is None
metadata_registry.register_runtime_clients(dev_mode_enabled_provider=lambda: True)
assert metadata_registry.get_client_for_source("hydrabase") is metadata_registry.get_registered_runtime_client("hydrabase")

View file

@ -14,7 +14,7 @@ from unittest.mock import MagicMock
import pytest import pytest
from core.metadata_cache import MetadataCache from core.metadata.cache import MetadataCache
@pytest.fixture @pytest.fixture

View file

@ -41,16 +41,18 @@ if "config.settings" not in sys.modules:
sys.modules["config"] = config_pkg sys.modules["config"] = config_pkg
sys.modules["config.settings"] = settings_mod sys.modules["config.settings"] = settings_mod
from core import metadata_service from core.metadata import registry as metadata_registry
from core.metadata_service import MetadataLookupOptions from core.metadata import completion as metadata_completion
from core.metadata import discography as metadata_discography
from core.metadata.lookup import MetadataLookupOptions
from database.music_database import MusicDatabase from database.music_database import MusicDatabase
@pytest.fixture(autouse=True) @pytest.fixture(autouse=True)
def _clear_metadata_client_cache(): def _clear_metadata_client_cache():
metadata_service.clear_cached_metadata_clients() metadata_registry.clear_cached_metadata_clients()
yield yield
metadata_service.clear_cached_metadata_clients() metadata_registry.clear_cached_metadata_clients()
class _FakeSourceClient: class _FakeSourceClient:
@ -117,11 +119,11 @@ def test_get_artist_discography_uses_primary_then_fallback(monkeypatch):
"itunes": itunes, "itunes": itunes,
} }
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: clients.get(source)) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: clients.get(source))
result = metadata_service.get_artist_discography("artist-1", "Artist One", MetadataLookupOptions()) result = metadata_discography.get_artist_discography("artist-1", "Artist One", MetadataLookupOptions())
assert result["source"] == "spotify" assert result["source"] == "spotify"
assert result["source_priority"] == ["deezer", "spotify", "itunes"] assert result["source_priority"] == ["deezer", "spotify", "itunes"]
@ -152,11 +154,11 @@ def test_get_artist_discography_uses_name_search_when_direct_lookup_missing(monk
) )
clients = {"deezer": deezer} clients = {"deezer": deezer}
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: clients.get(source)) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: clients.get(source))
result = metadata_service.get_artist_discography("artist-1", "Artist One", MetadataLookupOptions()) result = metadata_discography.get_artist_discography("artist-1", "Artist One", MetadataLookupOptions())
assert result["source"] == "deezer" assert result["source"] == "deezer"
assert [album["id"] for album in result["albums"]] == ["deezer-album-1"] assert [album["id"] for album in result["albums"]] == ["deezer-album-1"]
@ -189,11 +191,11 @@ def test_get_artist_discography_respects_source_override_without_fallback(monkey
"spotify": spotify, "spotify": spotify,
} }
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: clients.get(source)) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: clients.get(source))
result = metadata_service.get_artist_discography( result = metadata_discography.get_artist_discography(
"artist-1", "artist-1",
"Artist One", "Artist One",
MetadataLookupOptions(source_override="itunes", allow_fallback=False), MetadataLookupOptions(source_override="itunes", allow_fallback=False),
@ -231,16 +233,16 @@ def test_get_artist_discography_uses_hydrabase_fast_path_when_active(monkeypatch
) )
clients = {"deezer": None, "spotify": None, "itunes": None} clients = {"deezer": None, "spotify": None, "itunes": None}
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes", "hydrabase"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify", "itunes", "hydrabase"])
def fake_get_client_for_source(source): def fake_get_client_for_source(source):
if source == "hydrabase": if source == "hydrabase":
return hydrabase return hydrabase
return clients.get(source) return clients.get(source)
monkeypatch.setattr(metadata_service, "get_client_for_source", fake_get_client_for_source) monkeypatch.setattr(metadata_registry, "get_client_for_source", fake_get_client_for_source)
result = metadata_service.get_artist_discography("artist-1", "Artist One", MetadataLookupOptions()) result = metadata_discography.get_artist_discography("artist-1", "Artist One", MetadataLookupOptions())
assert result["source"] == "hydrabase" assert result["source"] == "hydrabase"
assert [album["id"] for album in result["albums"]] == ["hydrabase-album-1"] assert [album["id"] for album in result["albums"]] == ["hydrabase-album-1"]
@ -298,12 +300,12 @@ def test_iter_artist_discography_completion_uses_primary_source_first(monkeypatc
"itunes": itunes, "itunes": itunes,
} }
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: clients.get(source)) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: clients.get(source))
db = _CompletionFakeDB(owned_tracks=1, expected_tracks=2) db = _CompletionFakeDB(owned_tracks=1, expected_tracks=2)
events = list(metadata_service.iter_artist_discography_completion_events( events = list(metadata_completion.iter_artist_discography_completion_events(
{ {
"albums": [{"id": "release-1", "name": "Album One", "total_tracks": 0}], "albums": [{"id": "release-1", "name": "Album One", "total_tracks": 0}],
"singles": [], "singles": [],
@ -337,12 +339,12 @@ def test_iter_artist_discography_completion_respects_source_override(monkeypatch
"itunes": itunes, "itunes": itunes,
} }
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "spotify", "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "spotify", "itunes"])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: clients.get(source)) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: clients.get(source))
db = _CompletionFakeDB(owned_tracks=1, expected_tracks=3) db = _CompletionFakeDB(owned_tracks=1, expected_tracks=3)
events = list(metadata_service.iter_artist_discography_completion_events( events = list(metadata_completion.iter_artist_discography_completion_events(
{ {
"albums": [{"id": "release-2", "name": "Album Two", "total_tracks": 0}], "albums": [{"id": "release-2", "name": "Album Two", "total_tracks": 0}],
"singles": [], "singles": [],
@ -362,12 +364,12 @@ def test_iter_artist_discography_completion_uses_release_artist_metadata(monkeyp
source = _FakeSourceClient() source = _FakeSourceClient()
clients = {"deezer": source} clients = {"deezer": source}
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source_name: clients.get(source_name)) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source_name, **kwargs: clients.get(source_name))
db = _CompletionFakeDB(owned_tracks=1, expected_tracks=2) db = _CompletionFakeDB(owned_tracks=1, expected_tracks=2)
events = list(metadata_service.iter_artist_discography_completion_events( events = list(metadata_completion.iter_artist_discography_completion_events(
{ {
"albums": [{ "albums": [{
"id": "release-3", "id": "release-3",
@ -389,8 +391,7 @@ def test_iter_artist_discography_completion_uses_release_artist_metadata(monkeyp
def test_get_artist_detail_discography_classifies_release_types(monkeypatch): def test_get_artist_detail_discography_classifies_release_types(monkeypatch):
monkeypatch.setattr( monkeypatch.setattr(
metadata_service, "core.metadata.discography.get_artist_discography",
"get_artist_discography",
lambda artist_id, artist_name='', options=None: { lambda artist_id, artist_name='', options=None: {
"albums": [ "albums": [
{ {
@ -425,7 +426,7 @@ def test_get_artist_detail_discography_classifies_release_types(monkeypatch):
}, },
) )
result = metadata_service.get_artist_detail_discography("artist-1", "Artist One", MetadataLookupOptions()) result = metadata_discography.get_artist_detail_discography("artist-1", "Artist One", MetadataLookupOptions())
assert result["success"] is True assert result["success"] is True
assert result["source"] == "deezer" assert result["source"] == "deezer"
@ -440,8 +441,7 @@ def test_get_artist_detail_discography_classifies_release_types(monkeypatch):
def test_get_artist_detail_discography_dedups_variant_releases(monkeypatch): def test_get_artist_detail_discography_dedups_variant_releases(monkeypatch):
monkeypatch.setattr( monkeypatch.setattr(
metadata_service, "core.metadata.discography.get_artist_discography",
"get_artist_discography",
lambda artist_id, artist_name='', options=None: { lambda artist_id, artist_name='', options=None: {
"albums": [ "albums": [
{ {
@ -475,7 +475,7 @@ def test_get_artist_detail_discography_dedups_variant_releases(monkeypatch):
}, },
) )
result = metadata_service.get_artist_detail_discography("artist-1", "Artist One", MetadataLookupOptions()) result = metadata_discography.get_artist_detail_discography("artist-1", "Artist One", MetadataLookupOptions())
assert result["success"] is True assert result["success"] is True
assert [album["id"] for album in result["albums"]] == ["album-standard"] assert [album["id"] for album in result["albums"]] == ["album-standard"]
@ -489,8 +489,7 @@ def test_get_artist_detail_discography_keeps_variants_when_dedup_disabled(monkey
every release the source returns (matching the retired inline Artists every release the source returns (matching the retired inline Artists
page behaviour).""" page behaviour)."""
monkeypatch.setattr( monkeypatch.setattr(
metadata_service, "core.metadata.discography.get_artist_discography",
"get_artist_discography",
lambda artist_id, artist_name='', options=None: { lambda artist_id, artist_name='', options=None: {
"albums": [ "albums": [
{ {
@ -524,7 +523,7 @@ def test_get_artist_detail_discography_keeps_variants_when_dedup_disabled(monkey
}, },
) )
result = metadata_service.get_artist_detail_discography( result = metadata_discography.get_artist_detail_discography(
"artist-1", "artist-1",
"Artist One", "Artist One",
MetadataLookupOptions(dedup_variants=False), MetadataLookupOptions(dedup_variants=False),
@ -558,11 +557,11 @@ def test_get_artist_discography_keeps_provider_artist_ids(monkeypatch):
spotify = _SpotifyArtistIdClient() spotify = _SpotifyArtistIdClient()
clients = {"spotify": spotify} clients = {"spotify": spotify}
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "spotify") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "spotify")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: clients.get(source)) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: clients.get(source))
result = metadata_service.get_artist_discography("364555966", "Amarok", MetadataLookupOptions()) result = metadata_discography.get_artist_discography("364555966", "Amarok", MetadataLookupOptions())
assert result["source"] == "spotify" assert result["source"] == "spotify"
assert [album["id"] for album in result["albums"]] == ["spotify-release-1"] assert [album["id"] for album in result["albums"]] == ["spotify-release-1"]
@ -601,11 +600,11 @@ def test_get_artist_discography_prefers_source_specific_artist_ids(monkeypatch):
"deezer": deezer, "deezer": deezer,
} }
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "spotify") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "spotify")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "deezer"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "deezer"])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: clients.get(source)) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: clients.get(source))
result = metadata_service.get_artist_discography( result = metadata_discography.get_artist_discography(
"artist-1", "artist-1",
"Artist One", "Artist One",
MetadataLookupOptions( MetadataLookupOptions(

View file

@ -3,7 +3,7 @@ import sys
import types import types
from types import SimpleNamespace from types import SimpleNamespace
# Stub optional Spotify dependency so metadata_service can import in tests. # Stub optional Spotify dependency so the metadata package can import in tests.
if 'spotipy' not in sys.modules: if 'spotipy' not in sys.modules:
spotipy = types.ModuleType('spotipy') spotipy = types.ModuleType('spotipy')
oauth2 = types.ModuleType('spotipy.oauth2') oauth2 = types.ModuleType('spotipy.oauth2')

View file

@ -40,7 +40,8 @@ if "config.settings" not in sys.modules:
import types as pytypes import types as pytypes
from core import metadata_service from core.metadata import registry as metadata_registry
from core.metadata import similar_artists as metadata_similar_artists
class _FakeMusicMapResponse: class _FakeMusicMapResponse:
@ -91,16 +92,16 @@ def test_iter_musicmap_similar_artist_events_uses_source_priority(monkeypatch):
}) })
spotify = _FakeSourceClient({}) spotify = _FakeSourceClient({})
monkeypatch.setattr(metadata_service.requests, "get", lambda *args, **kwargs: _FakeMusicMapResponse(html)) monkeypatch.setattr(metadata_similar_artists.requests, "get", lambda *args, **kwargs: _FakeMusicMapResponse(html))
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "itunes", "spotify"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "itunes", "spotify"])
monkeypatch.setattr( monkeypatch.setattr(
metadata_service, metadata_registry,
"get_client_for_source", "get_client_for_source",
lambda source: {"deezer": deezer, "itunes": itunes, "spotify": spotify}.get(source), lambda source, **kwargs: {"deezer": deezer, "itunes": itunes, "spotify": spotify}.get(source),
) )
events = list(metadata_service.iter_musicmap_similar_artist_events("Artist One", limit=5)) events = list(metadata_similar_artists.iter_musicmap_similar_artist_events("Artist One", limit=5))
assert events[0]["type"] == "start" assert events[0]["type"] == "start"
assert events[0]["source_priority"] == ["deezer", "itunes", "spotify"] assert events[0]["source_priority"] == ["deezer", "itunes", "spotify"]
@ -151,12 +152,12 @@ def test_iter_musicmap_similar_artist_events_enriches_itunes_images(monkeypatch)
} }
itunes = _ItunesClient() itunes = _ItunesClient()
monkeypatch.setattr(metadata_service.requests, "get", lambda *args, **kwargs: _FakeMusicMapResponse(html)) monkeypatch.setattr(metadata_similar_artists.requests, "get", lambda *args, **kwargs: _FakeMusicMapResponse(html))
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "itunes") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "itunes")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: itunes if source == "itunes" else None) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: itunes if source == "itunes" else None)
events = list(metadata_service.iter_musicmap_similar_artist_events("Artist One", limit=5)) events = list(metadata_similar_artists.iter_musicmap_similar_artist_events("Artist One", limit=5))
artist_events = [event for event in events if event.get("type") == "artist"] artist_events = [event for event in events if event.get("type") == "artist"]
assert len(artist_events) == 1 assert len(artist_events) == 1
@ -199,12 +200,12 @@ def test_iter_musicmap_similar_artist_events_falls_back_to_itunes_album_art(monk
return "https://itunes.example/album-art.jpg" return "https://itunes.example/album-art.jpg"
itunes = _ItunesClient() itunes = _ItunesClient()
monkeypatch.setattr(metadata_service.requests, "get", lambda *args, **kwargs: _FakeMusicMapResponse(html)) monkeypatch.setattr(metadata_similar_artists.requests, "get", lambda *args, **kwargs: _FakeMusicMapResponse(html))
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "itunes") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "itunes")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: itunes if source == "itunes" else None) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: itunes if source == "itunes" else None)
events = list(metadata_service.iter_musicmap_similar_artist_events("Artist One", limit=5)) events = list(metadata_similar_artists.iter_musicmap_similar_artist_events("Artist One", limit=5))
artist_events = [event for event in events if event.get("type") == "artist"] artist_events = [event for event in events if event.get("type") == "artist"]
assert len(artist_events) == 1 assert len(artist_events) == 1
@ -222,12 +223,12 @@ def test_get_musicmap_similar_artists_returns_not_found_when_musicmap_missing(mo
</html> </html>
""" """
monkeypatch.setattr(metadata_service.requests, "get", lambda *args, **kwargs: _FakeMusicMapResponse(html)) monkeypatch.setattr(metadata_similar_artists.requests, "get", lambda *args, **kwargs: _FakeMusicMapResponse(html))
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer") monkeypatch.setattr(metadata_registry, "get_primary_source", lambda spotify_client_factory=None: "deezer")
monkeypatch.setattr(metadata_service, "get_source_priority", lambda primary: [primary, "itunes"]) monkeypatch.setattr(metadata_registry, "get_source_priority", lambda primary: [primary, "itunes"])
monkeypatch.setattr(metadata_service, "get_client_for_source", lambda source: object()) monkeypatch.setattr(metadata_registry, "get_client_for_source", lambda source, **kwargs: object())
result = metadata_service.get_musicmap_similar_artists("Artist One", limit=5) result = metadata_similar_artists.get_musicmap_similar_artists("Artist One", limit=5)
assert result["success"] is False assert result["success"] is False
assert result["status_code"] == 404 assert result["status_code"] == 404

View file

@ -1,146 +0,0 @@
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
def get_active_media_server(self):
return "plex"
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 config.settings import config_manager
@pytest.fixture(autouse=True)
def _clear_metadata_client_cache():
metadata_service.clear_cached_metadata_clients()
yield
metadata_service.clear_cached_metadata_clients()
def test_primary_client_is_cached_for_same_source(monkeypatch):
calls = {"deezer": 0}
class FakeDeezerClient:
def __init__(self):
calls["deezer"] += 1
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: "deezer")
monkeypatch.setattr("core.deezer_client.DeezerClient", FakeDeezerClient)
first = metadata_service.get_primary_client()
second = metadata_service.get_primary_client()
assert first is second
assert calls["deezer"] == 1
def test_primary_client_switches_cache_by_source(monkeypatch):
calls = {"deezer": 0, "itunes": 0}
sources = iter(["deezer", "itunes"])
class FakeDeezerClient:
def __init__(self):
calls["deezer"] += 1
class FakeITunesClient:
def __init__(self):
calls["itunes"] += 1
monkeypatch.setattr(metadata_service, "get_primary_source", lambda: next(sources))
monkeypatch.setattr("core.deezer_client.DeezerClient", FakeDeezerClient)
monkeypatch.setattr(metadata_service, "iTunesClient", FakeITunesClient)
deezer_client = metadata_service.get_primary_client()
itunes_client = metadata_service.get_primary_client()
assert deezer_client is not itunes_client
assert calls["deezer"] == 1
assert calls["itunes"] == 1
def test_deezer_client_cache_tracks_token(monkeypatch):
tokens = iter(["token-a", "token-b"])
calls = {"deezer": 0}
class FakeDeezerClient:
def __init__(self):
calls["deezer"] += 1
monkeypatch.setattr("core.deezer_client.DeezerClient", FakeDeezerClient)
monkeypatch.setattr(config_manager, "get", lambda key, default=None: next(tokens) if key == "deezer.access_token" else default)
first = metadata_service.get_deezer_client()
second = metadata_service.get_deezer_client()
assert first is not second
assert calls["deezer"] == 2
class _FakeHydrabaseClient:
def __init__(self, connected=True):
self._connected = connected
def is_connected(self):
return self._connected
def test_hydrabase_enabled_requires_connection_and_dev_mode(monkeypatch):
fake_ws = types.ModuleType("web_server")
fake_ws.hydrabase_client = _FakeHydrabaseClient(connected=True)
fake_ws.dev_mode_enabled = True
monkeypatch.setitem(sys.modules, "web_server", fake_ws)
assert metadata_service.is_hydrabase_enabled() is True
fake_ws.dev_mode_enabled = False
assert metadata_service.is_hydrabase_enabled() is False
fake_ws.dev_mode_enabled = True
fake_ws.hydrabase_client = _FakeHydrabaseClient(connected=False)
assert metadata_service.is_hydrabase_enabled() is False
def test_get_client_for_source_hydrabase_requires_enablement(monkeypatch):
fake_ws = types.ModuleType("web_server")
fake_ws.hydrabase_client = _FakeHydrabaseClient(connected=True)
fake_ws.dev_mode_enabled = False
monkeypatch.setitem(sys.modules, "web_server", fake_ws)
assert metadata_service.get_client_for_source("hydrabase") is None
fake_ws.dev_mode_enabled = True
assert metadata_service.get_client_for_source("hydrabase") is fake_ws.hydrabase_client

View file

@ -94,7 +94,13 @@ from core.tidal_client import TidalClient # Added import for Tidal
from core.matching_engine import MusicMatchingEngine from core.matching_engine import MusicMatchingEngine
from core.database_update_worker import DatabaseUpdateWorker from core.database_update_worker import DatabaseUpdateWorker
from core.web_scan_manager import WebScanManager from core.web_scan_manager import WebScanManager
from core.metadata_cache import get_metadata_cache from core.metadata.cache import get_metadata_cache
from core.metadata import registry as metadata_registry
from core.metadata.registry import (
clear_cached_metadata_client,
get_spotify_client,
register_runtime_clients,
)
from core.imports.context import ( from core.imports.context import (
get_import_clean_album, get_import_clean_album,
get_import_clean_title, get_import_clean_title,
@ -477,10 +483,6 @@ def admin_only(view_fn):
return view_fn(*args, **kwargs) return view_fn(*args, **kwargs)
return wrapper return wrapper
# ── Per-profile Spotify client cache ──
_profile_spotify_clients = {} # profile_id -> SpotifyClient
_profile_spotify_lock = threading.Lock()
def get_spotify_client_for_profile(profile_id=None): def get_spotify_client_for_profile(profile_id=None):
"""Get the Spotify client for the current profile. """Get the Spotify client for the current profile.
@ -490,53 +492,7 @@ def get_spotify_client_for_profile(profile_id=None):
""" """
if profile_id is None: if profile_id is None:
profile_id = get_current_profile_id() profile_id = get_current_profile_id()
return metadata_registry.get_spotify_client_for_profile(profile_id)
# Admin (profile 1) always uses global client
if profile_id == 1:
return spotify_client
# Check if this profile has custom Spotify credentials
try:
db = get_database()
creds = db.get_profile_spotify(profile_id)
if not creds or not creds.get('client_id'):
return spotify_client # No custom creds — use global
except Exception:
return spotify_client
# Check cache (don't hold lock during auth check — could block on network)
with _profile_spotify_lock:
cached = _profile_spotify_clients.get(profile_id)
if cached and cached.sp is not None:
return cached
# Create a new SpotifyClient for this profile
try:
from spotipy.oauth2 import SpotifyOAuth
import spotipy
auth_manager = SpotifyOAuth(
client_id=creds['client_id'],
client_secret=creds['client_secret'],
redirect_uri=creds.get('redirect_uri', 'http://127.0.0.1:8888/callback'),
scope="user-library-read user-read-private playlist-read-private playlist-read-collaborative user-read-email user-follow-read",
cache_path=f'config/.spotify_cache_profile_{profile_id}'
)
# Create a bare SpotifyClient and immediately set the profile-specific
# spotipy instance (overwrites the global-config one from __init__)
profile_client = SpotifyClient()
profile_client.sp = spotipy.Spotify(auth_manager=auth_manager, retries=0, requests_timeout=15)
profile_client.user_id = None # Will be fetched lazily
with _profile_spotify_lock:
_profile_spotify_clients[profile_id] = profile_client
logger.info(f"Created per-profile Spotify client for profile {profile_id}")
return profile_client
except Exception as e:
logger.error(f"Failed to create per-profile Spotify client for profile {profile_id}: {e}")
return spotify_client # Fall back to global
# Valid page IDs for profile permission validation # Valid page IDs for profile permission validation
VALID_PAGE_IDS = {'dashboard', 'sync', 'search', 'downloads', 'discover', 'artists', 'automations', 'library', 'import', 'settings', 'help'} VALID_PAGE_IDS = {'dashboard', 'sync', 'search', 'downloads', 'discover', 'artists', 'automations', 'library', 'import', 'settings', 'help'}
@ -608,8 +564,8 @@ logger.info("Initializing SoulSync services for Web UI...")
spotify_client = plex_client = jellyfin_client = navidrome_client = soulsync_library_client = soulseek_client = tidal_client = matching_engine = sync_service = web_scan_manager = None spotify_client = plex_client = jellyfin_client = navidrome_client = soulsync_library_client = soulseek_client = tidal_client = matching_engine = sync_service = web_scan_manager = None
try: try:
spotify_client = SpotifyClient() spotify_client = get_spotify_client()
logger.info(" Spotify client initialized") logger.info(" Spotify client initialized via metadata registry")
except Exception as e: except Exception as e:
logger.error(f" Spotify client failed to initialize: {e}") logger.error(f" Spotify client failed to initialize: {e}")
@ -5810,7 +5766,7 @@ _comparison_lock = threading.Lock()
def _is_hydrabase_active(): def _is_hydrabase_active():
"""Check if Hydrabase is connected and enabled for metadata use.""" """Check if Hydrabase is connected and enabled for metadata use."""
try: try:
from core.metadata_service import is_hydrabase_enabled from core.metadata.registry import is_hydrabase_enabled
return is_hydrabase_enabled() return is_hydrabase_enabled()
except Exception: except Exception:
return False return False
@ -7061,7 +7017,7 @@ def auth_spotify():
logger.error(f"Per-profile Spotify auth failed, falling back to global: {e}") logger.error(f"Per-profile Spotify auth failed, falling back to global: {e}")
# Global auth (admin or fallback) # Global auth (admin or fallback)
temp_spotify_client = SpotifyClient() temp_spotify_client = get_spotify_client()
if temp_spotify_client.sp and temp_spotify_client.sp.auth_manager: if temp_spotify_client.sp and temp_spotify_client.sp.auth_manager:
# Get the authorization URL # Get the authorization URL
auth_url = temp_spotify_client.sp.auth_manager.get_authorize_url() auth_url = temp_spotify_client.sp.auth_manager.get_authorize_url()
@ -7385,8 +7341,7 @@ def spotify_callback():
token_info = auth_manager.get_access_token(auth_code) token_info = auth_manager.get_access_token(auth_code)
if token_info: if token_info:
# Invalidate cached profile client so it gets recreated with new tokens # Invalidate cached profile client so it gets recreated with new tokens
with _profile_spotify_lock: metadata_registry.clear_cached_profile_spotify_client(profile_id_from_state)
_profile_spotify_clients.pop(profile_id_from_state, None)
add_activity_item("", "Spotify Auth Complete", f"Profile {profile_id_from_state} authenticated with Spotify", "Now") add_activity_item("", "Spotify Auth Complete", f"Profile {profile_id_from_state} authenticated with Spotify", "Now")
return "<h1>Spotify Authentication Successful!</h1><p>Your personal Spotify account is now connected. You can close this window.</p>" return "<h1>Spotify Authentication Successful!</h1><p>Your personal Spotify account is now connected. You can close this window.</p>"
else: else:
@ -7410,7 +7365,8 @@ def spotify_callback():
if token_info: if token_info:
# CRITICAL: update the GLOBAL spotify_client, not a local variable # CRITICAL: update the GLOBAL spotify_client, not a local variable
global spotify_client global spotify_client
spotify_client = SpotifyClient() clear_cached_metadata_client("spotify")
spotify_client = get_spotify_client()
if spotify_client.is_spotify_authenticated(): if spotify_client.is_spotify_authenticated():
# Clear any active rate limit ban and post-ban cooldown # Clear any active rate limit ban and post-ban cooldown
# so Spotify is immediately usable after re-auth # so Spotify is immediately usable after re-auth
@ -10045,7 +10001,8 @@ def get_artist_detail(artist_id):
# Get source-priority discography for proper categorization and missing releases # Get source-priority discography for proper categorization and missing releases
artist_detail_discography = None artist_detail_discography = None
try: try:
from core.metadata_service import MetadataLookupOptions, get_artist_detail_discography as _get_artist_detail_discography from core.metadata.lookup import MetadataLookupOptions
from core.metadata_service import get_artist_detail_discography as _get_artist_detail_discography
artist_source_ids = { artist_source_ids = {
'spotify': artist_info.get('spotify_artist_id'), 'spotify': artist_info.get('spotify_artist_id'),
@ -10281,7 +10238,8 @@ def get_artist_discography(artist_id):
else: else:
effective_override_source = 'spotify' effective_override_source = 'spotify'
from core.metadata_service import MetadataLookupOptions, get_artist_discography as _get_artist_discography from core.metadata.lookup import MetadataLookupOptions
from core.metadata_service import get_artist_discography as _get_artist_discography
discography = _get_artist_discography( discography = _get_artist_discography(
artist_id, artist_id,
@ -23762,38 +23720,38 @@ deezer_discovery_executor = ThreadPoolExecutor(max_workers=3, thread_name_prefix
def _get_deezer_client(): def _get_deezer_client():
"""Get cached Deezer client.""" """Get cached Deezer client."""
from core.metadata_service import get_deezer_client from core.metadata.registry import get_deezer_client
return get_deezer_client() return get_deezer_client()
def _get_itunes_client(): def _get_itunes_client():
"""Get cached iTunes client.""" """Get cached iTunes client."""
from core.metadata_service import get_itunes_client from core.metadata.registry import get_itunes_client
return get_itunes_client() return get_itunes_client()
def _get_discogs_client(token=None): def _get_discogs_client(token=None):
"""Get cached Discogs client.""" """Get cached Discogs client."""
from core.metadata_service import get_discogs_client from core.metadata.registry import get_discogs_client
return get_discogs_client(token) return get_discogs_client(token)
def _get_metadata_fallback_source(): def _get_metadata_fallback_source():
"""Get the configured primary metadata source. """Get the configured primary metadata source.
Returns 'spotify', 'itunes', 'deezer', 'discogs', or 'hydrabase'. Returns 'spotify', 'itunes', 'deezer', 'discogs', or 'hydrabase'.
NOTE: This is a thin wrapper canonical logic lives in core.metadata_service.get_primary_source(). NOTE: This is a thin wrapper canonical logic lives in core.metadata.registry.get_primary_source().
Kept as a local function because 70+ callers reference it by name.""" Kept as a local function because 70+ callers reference it by name."""
from core.metadata_service import get_primary_source from core.metadata.registry import get_primary_source
return get_primary_source() return get_primary_source()
def _get_metadata_fallback_client(): def _get_metadata_fallback_client():
"""Get the active metadata client based on settings. """Get the active metadata client based on settings.
Returns a SpotifyClient, iTunesClient, DeezerClient, DiscogsClient, or HydrabaseClient instance.""" Returns a SpotifyClient, iTunesClient, DeezerClient, DiscogsClient, or HydrabaseClient instance."""
source = _get_metadata_fallback_source() source = _get_metadata_fallback_source()
from core.metadata.registry import get_client_for_source
client = get_client_for_source(source)
if client is not None:
return client
if source == 'spotify': if source == 'spotify':
if spotify_client and spotify_client.is_spotify_authenticated():
return spotify_client
# Spotify selected but not authed — fall back to deezer
return _get_deezer_client()
if source == 'deezer':
return _get_deezer_client() return _get_deezer_client()
if source == 'discogs': if source == 'discogs':
token = config_manager.get('discogs.token', '') token = config_manager.get('discogs.token', '')
@ -27208,6 +27166,7 @@ def save_profile_spotify_creds():
success = db.set_profile_spotify(profile_id, client_id, client_secret, redirect_uri) success = db.set_profile_spotify(profile_id, client_id, client_secret, redirect_uri)
if success: if success:
metadata_registry.clear_cached_profile_spotify_client(profile_id)
return jsonify({'success': True}) return jsonify({'success': True})
return jsonify({'success': False, 'error': 'Failed to save credentials'}), 500 return jsonify({'success': False, 'error': 'Failed to save credentials'}), 500
except Exception as e: except Exception as e:
@ -27229,6 +27188,7 @@ def delete_profile_spotify_creds():
WHERE id = ? WHERE id = ?
""", (profile_id,)) """, (profile_id,))
conn.commit() conn.commit()
metadata_registry.clear_cached_profile_spotify_client(profile_id)
return jsonify({'success': True}) return jsonify({'success': True})
except Exception as e: except Exception as e:
return jsonify({'success': False, 'error': str(e)}), 500 return jsonify({'success': False, 'error': str(e)}), 500
@ -27734,7 +27694,7 @@ def start_watchlist_scan():
"""Start a watchlist scan for new releases""" """Start a watchlist scan for new releases"""
try: try:
# Check if MetadataService can provide a working client (Spotify OR fallback) # Check if MetadataService can provide a working client (Spotify OR fallback)
from core.metadata_service import MetadataService from core.metadata.service import MetadataService
metadata_service = MetadataService() metadata_service = MetadataService()
# Get active provider - will be spotify or the configured fallback # Get active provider - will be spotify or the configured fallback
@ -28221,7 +28181,7 @@ def watchlist_artist_config(artist_id):
'preferred_metadata_source': result[17] if len(result) > 17 else None, 'preferred_metadata_source': result[17] if len(result) > 17 else None,
} }
from core.metadata_service import get_primary_source from core.metadata.registry import get_primary_source
return jsonify({ return jsonify({
"success": True, "success": True,
"config": config, "config": config,
@ -28638,7 +28598,7 @@ def _get_active_discovery_source():
NOTE: Thin wrapper canonical logic lives in core.metadata_service.get_primary_source(). NOTE: Thin wrapper canonical logic lives in core.metadata_service.get_primary_source().
""" """
from core.metadata_service import get_primary_source from core.metadata.registry import get_primary_source
return get_primary_source() return get_primary_source()
@ -34427,7 +34387,6 @@ def enrich_beatport_tracks():
uncached_tracks = [] uncached_tracks = []
uncached_indices = [] uncached_indices = []
from core.metadata_cache import get_metadata_cache
mcache = get_metadata_cache() mcache = get_metadata_cache()
for i, track in enumerate(tracks): for i, track in enumerate(tracks):
@ -36640,7 +36599,6 @@ def start_oauth_callback_servers():
# Manually trigger the token exchange using spotipy's auth manager # Manually trigger the token exchange using spotipy's auth manager
try: try:
from core.spotify_client import SpotifyClient
from spotipy.oauth2 import SpotifyOAuth from spotipy.oauth2 import SpotifyOAuth
from config.settings import config_manager from config.settings import config_manager
@ -36664,7 +36622,8 @@ def start_oauth_callback_servers():
if token_info: if token_info:
# Reinitialize the global client with new tokens # Reinitialize the global client with new tokens
global spotify_client global spotify_client
spotify_client = SpotifyClient() clear_cached_metadata_client("spotify")
spotify_client = get_spotify_client()
if spotify_client.is_spotify_authenticated(): if spotify_client.is_spotify_authenticated():
# Clear rate limit ban + post-ban cooldown so Spotify is usable immediately # Clear rate limit ban + post-ban cooldown so Spotify is usable immediately
@ -37719,6 +37678,14 @@ except Exception as e:
hydrabase_worker = None hydrabase_worker = None
hydrabase_client = None hydrabase_client = None
register_runtime_clients(
hydrabase_client=hydrabase_client,
dev_mode_enabled_provider=lambda: dev_mode_enabled,
)
metadata_registry.register_profile_spotify_credentials_provider(
lambda profile_id: get_database().get_profile_spotify(profile_id)
)
# --- Hydrabase Auto-Reconnect --- # --- Hydrabase Auto-Reconnect ---
try: try:
_hydra_cfg = config_manager.get_hydrabase_config() _hydra_cfg = config_manager.get_hydrabase_config()
@ -38237,7 +38204,6 @@ def repair_findings_counts():
def repair_cache_health(): def repair_cache_health():
"""Get metadata cache health stats for the repair dashboard""" """Get metadata cache health stats for the repair dashboard"""
try: try:
from core.metadata_cache import get_metadata_cache
cache = get_metadata_cache() cache = get_metadata_cache()
return jsonify(cache.get_health_stats()), 200 return jsonify(cache.get_health_stats()), 200
except Exception as e: except Exception as e:
@ -38576,7 +38542,7 @@ def import_search_albums():
return jsonify({'success': False, 'error': 'Missing query parameter'}), 400 return jsonify({'success': False, 'error': 'Missing query parameter'}), 400
limit = min(int(request.args.get('limit', 12)), 50) limit = min(int(request.args.get('limit', 12)), 50)
from core.metadata_service import get_primary_source from core.metadata.registry import get_primary_source
if get_primary_source() == 'hydrabase' and hydrabase_worker and dev_mode_enabled: if get_primary_source() == 'hydrabase' and hydrabase_worker and dev_mode_enabled:
hydrabase_worker.enqueue(query, 'albums') hydrabase_worker.enqueue(query, 'albums')