cca396e7bd
Move Hydrabase availability checks into metadata_service so source resolution owns the policy. Keep web_server delegating to the centralized helper and add tests for the enabled/disabled cases.
748 lines
25 KiB
Python
748 lines
25 KiB
Python
"""
|
|
Metadata Service - Centralized metadata source selection
|
|
|
|
ALL metadata source decisions flow through this module. Other files import
|
|
get_primary_source() and get_primary_client() instead of reimplementing
|
|
the logic. This prevents bugs where different files have different defaults,
|
|
auth checks, or source-fallback behavior.
|
|
"""
|
|
|
|
import threading
|
|
from dataclasses import dataclass
|
|
from typing import List, Optional, Dict, Any, Literal
|
|
from core.spotify_client import SpotifyClient
|
|
from core.itunes_client import iTunesClient
|
|
from utils.logging_config import get_logger
|
|
|
|
logger = get_logger("metadata_service")
|
|
|
|
MetadataProvider = Literal["spotify", "itunes", "auto"]
|
|
|
|
# Ordered by fallback preference. Higher-priority sources appear earlier.
|
|
METADATA_SOURCE_PRIORITY = ('deezer', 'itunes', 'spotify', 'discogs', 'hydrabase')
|
|
|
|
_client_cache_lock = threading.RLock()
|
|
_client_cache: Dict[str, Any] = {}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class MetadataLookupOptions:
|
|
"""Generic metadata lookup policy shared by metadata services."""
|
|
|
|
source_override: Optional[str] = None
|
|
allow_fallback: bool = True
|
|
skip_cache: bool = False
|
|
max_pages: int = 0
|
|
limit: int = 50
|
|
|
|
|
|
# =============================================================================
|
|
# CANONICAL SOURCE SELECTION — all code should use these two functions
|
|
# =============================================================================
|
|
|
|
def get_primary_source() -> str:
|
|
"""Get the user's configured primary metadata source.
|
|
|
|
Returns 'spotify', 'deezer', 'itunes', 'discogs', or 'hydrabase'.
|
|
If the user selected Spotify but it's not authenticated, falls back to 'deezer'.
|
|
|
|
This is THE single source of truth for "which metadata source should I use?"
|
|
All other modules should import this function instead of reading config directly.
|
|
"""
|
|
try:
|
|
from config.settings import config_manager
|
|
source = config_manager.get('metadata.fallback_source', 'deezer') or 'deezer'
|
|
except Exception:
|
|
return 'deezer'
|
|
|
|
# Validate Spotify selection — can't use it if not authenticated
|
|
if source == 'spotify':
|
|
try:
|
|
import importlib
|
|
ws = importlib.import_module('web_server')
|
|
sc = getattr(ws, 'spotify_client', None)
|
|
if not sc or not sc.is_spotify_authenticated():
|
|
return 'deezer'
|
|
except Exception:
|
|
return 'deezer'
|
|
|
|
return source
|
|
|
|
|
|
def get_primary_client():
|
|
"""Get the client object for the user's configured primary metadata source.
|
|
|
|
Returns a SpotifyClient, DeezerClient, iTunesClient, DiscogsClient,
|
|
or HydrabaseClient instance.
|
|
|
|
This is THE single source of truth for "which client should I call?"
|
|
"""
|
|
return get_client_for_source(get_primary_source())
|
|
|
|
|
|
def get_source_priority(preferred_source: str):
|
|
"""Return supported sources with the 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_client_for_source(source: str):
|
|
"""Get the client object for an exact metadata source.
|
|
|
|
Returns the matching client or None if that source is unavailable.
|
|
No fallback swaps.
|
|
"""
|
|
if source == 'spotify':
|
|
try:
|
|
import importlib
|
|
ws = importlib.import_module('web_server')
|
|
sc = getattr(ws, 'spotify_client', None)
|
|
if sc and sc.is_spotify_authenticated():
|
|
return sc
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
if source == 'deezer':
|
|
return get_deezer_client()
|
|
|
|
if source == 'discogs':
|
|
return get_discogs_client()
|
|
|
|
if source == 'hydrabase':
|
|
return get_hydrabase_client(allow_fallback=False)
|
|
|
|
if source == 'itunes':
|
|
return get_itunes_client()
|
|
|
|
return None
|
|
|
|
|
|
def get_album_tracks_for_source(source: str, album_id: str):
|
|
"""Get album tracks for an exact source.
|
|
|
|
Returns Spotify-compatible dict/list data or None.
|
|
No fallback swaps.
|
|
"""
|
|
client = 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_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.
|
|
|
|
Returns a provider-native album list or None if the source is unavailable.
|
|
Tries the requested artist ID first, then falls back to artist-name
|
|
search using the same flow for every provider when artist_name is provided.
|
|
|
|
Set skip_cache=True only for freshness-sensitive flows that need newly
|
|
released albums to show up immediately.
|
|
"""
|
|
client = get_client_for_source(source)
|
|
if not client or not artist_id 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:
|
|
albums = _fetch_for_artist(artist_id) or []
|
|
if albums:
|
|
return albums
|
|
|
|
if not artist_name:
|
|
return albums
|
|
|
|
search_results = _search_artists_for_source(source, client, artist_name, limit=5)
|
|
if not search_results:
|
|
return albums
|
|
|
|
best = _pick_best_artist_match(search_results, artist_name)
|
|
if not best:
|
|
return albums
|
|
|
|
found_artist_id = _extract_lookup_value(best, 'id', 'artist_id')
|
|
if not found_artist_id:
|
|
return albums
|
|
|
|
resolved = _fetch_for_artist(found_artist_id) or []
|
|
if resolved:
|
|
logger.debug("Found %s artist '%s' (id=%s)", source, _extract_lookup_value(best, 'name', 'artist_name', 'title'), found_artist_id)
|
|
return resolved
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _get_source_chain_for_lookup(options: MetadataLookupOptions) -> List[str]:
|
|
primary_source = get_primary_source()
|
|
source_chain = list(get_source_priority(primary_source))
|
|
override = (options.source_override or '').strip().lower()
|
|
|
|
if override:
|
|
source_chain = [override] + [source for source in source_chain if source != override]
|
|
|
|
if not options.allow_fallback:
|
|
source_chain = source_chain[:1]
|
|
|
|
return source_chain
|
|
|
|
|
|
def _extract_lookup_value(value: Any, *names: str, default: Any = None) -> Any:
|
|
if value is None:
|
|
return default
|
|
|
|
for name in names:
|
|
if isinstance(value, dict):
|
|
if name in value and value[name] is not None:
|
|
return value[name]
|
|
else:
|
|
candidate = getattr(value, name, None)
|
|
if candidate is not None:
|
|
return candidate
|
|
return default
|
|
|
|
|
|
def _normalize_artist_name(value: Any) -> str:
|
|
return (value or '').strip().casefold()
|
|
|
|
|
|
def _search_artists_for_source(source: str, client: Any, artist_name: str, limit: int = 5) -> List[Any]:
|
|
if not client or not hasattr(client, 'search_artists'):
|
|
return []
|
|
|
|
try:
|
|
kwargs = {'limit': limit}
|
|
if source == 'spotify':
|
|
kwargs['allow_fallback'] = False
|
|
return client.search_artists(artist_name, **kwargs) or []
|
|
except Exception as exc:
|
|
logger.debug("Could not search %s for %s: %s", source, artist_name, exc)
|
|
return []
|
|
|
|
|
|
def _pick_best_artist_match(search_results: List[Any], artist_name: str) -> Optional[Any]:
|
|
"""Prefer an exact artist-name match, otherwise use the first result."""
|
|
if not search_results:
|
|
return None
|
|
|
|
target_name = _normalize_artist_name(artist_name)
|
|
for artist in search_results:
|
|
candidate_name = _normalize_artist_name(
|
|
_extract_lookup_value(artist, 'name', 'artist_name', 'title')
|
|
)
|
|
if candidate_name == target_name:
|
|
return artist
|
|
|
|
return search_results[0]
|
|
|
|
|
|
def _build_discography_release_dict(release: Any, artist_id: str) -> Optional[Dict[str, Any]]:
|
|
release_id = _extract_lookup_value(release, 'id', 'album_id', 'release_id')
|
|
if not release_id:
|
|
return None
|
|
|
|
artist_ids = _extract_lookup_value(release, 'artist_ids') or []
|
|
if isinstance(artist_ids, (str, bytes)):
|
|
artist_ids = [artist_ids]
|
|
else:
|
|
try:
|
|
artist_ids = list(artist_ids)
|
|
except TypeError:
|
|
artist_ids = [artist_ids]
|
|
|
|
if artist_ids and str(artist_ids[0]) != str(artist_id):
|
|
return None
|
|
|
|
album_type = _extract_lookup_value(release, 'album_type', default='album') or 'album'
|
|
release_date = _extract_lookup_value(release, 'release_date')
|
|
|
|
return {
|
|
'id': release_id,
|
|
'name': _extract_lookup_value(release, 'name', 'title', default=release_id),
|
|
'release_date': release_date,
|
|
'album_type': album_type,
|
|
'image_url': _extract_lookup_value(release, 'image_url', 'thumb_url', 'cover_image'),
|
|
'total_tracks': _extract_lookup_value(release, 'total_tracks', default=0) or 0,
|
|
'external_urls': _extract_lookup_value(release, 'external_urls', default={}) or {},
|
|
}
|
|
|
|
|
|
def _sort_discography_releases(releases: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
def get_release_year(item):
|
|
if item.get('release_date'):
|
|
try:
|
|
return int(str(item['release_date'])[:4])
|
|
except (ValueError, IndexError, TypeError):
|
|
return 0
|
|
return 0
|
|
|
|
return sorted(releases, key=get_release_year, reverse=True)
|
|
|
|
|
|
def get_artist_discography(
|
|
artist_id: str,
|
|
artist_name: str = '',
|
|
options: Optional[MetadataLookupOptions] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Get a normalized artist discography with source resolution and fallback.
|
|
|
|
Each provider uses the same lookup flow:
|
|
1. try the requested artist ID
|
|
2. if that misses, search by artist name
|
|
3. retry with the provider-specific artist ID from the search result
|
|
"""
|
|
options = options or MetadataLookupOptions()
|
|
source_priority = _get_source_chain_for_lookup(options)
|
|
|
|
albums: List[Any] = []
|
|
active_source: Optional[str] = None
|
|
|
|
if not albums:
|
|
for source in source_priority:
|
|
client = get_client_for_source(source)
|
|
if not client:
|
|
continue
|
|
|
|
try:
|
|
albums = get_artist_albums_for_source(
|
|
source,
|
|
artist_id,
|
|
artist_name=artist_name,
|
|
limit=options.limit,
|
|
skip_cache=options.skip_cache,
|
|
max_pages=options.max_pages,
|
|
) or []
|
|
except Exception as exc:
|
|
logger.debug("%s direct lookup failed for artist %s: %s", source, artist_id, exc)
|
|
albums = []
|
|
|
|
if albums:
|
|
active_source = source
|
|
logger.info("Got %s albums from %s for artist %s", len(albums), source, artist_id)
|
|
break
|
|
|
|
album_list: List[Dict[str, Any]] = []
|
|
singles_list: List[Dict[str, Any]] = []
|
|
seen_albums = set()
|
|
|
|
for release in albums or []:
|
|
release_data = _build_discography_release_dict(release, artist_id)
|
|
if not release_data:
|
|
continue
|
|
|
|
release_id = release_data['id']
|
|
if release_id in seen_albums:
|
|
continue
|
|
seen_albums.add(release_id)
|
|
|
|
album_type = release_data.get('album_type') or 'album'
|
|
if album_type in ['single', 'ep']:
|
|
singles_list.append(release_data)
|
|
else:
|
|
album_list.append(release_data)
|
|
|
|
album_list = _sort_discography_releases(album_list)
|
|
singles_list = _sort_discography_releases(singles_list)
|
|
|
|
logger.debug(
|
|
"Total albums returned for artist %s: %s (source=%s)",
|
|
artist_id,
|
|
len(album_list) + len(singles_list),
|
|
active_source,
|
|
)
|
|
|
|
return {
|
|
'albums': album_list,
|
|
'singles': singles_list,
|
|
'source': active_source or (source_priority[0] if source_priority else 'unknown'),
|
|
'source_priority': source_priority,
|
|
}
|
|
|
|
|
|
def get_deezer_client():
|
|
"""Get cached Deezer client.
|
|
|
|
Deezer client is safe to reuse across requests because it owns no
|
|
request-specific state beyond the current access token.
|
|
"""
|
|
from core.deezer_client import DeezerClient
|
|
try:
|
|
from config.settings import config_manager
|
|
current_token = config_manager.get('deezer.access_token', None)
|
|
except Exception:
|
|
current_token = None
|
|
|
|
cache_key = f"deezer::{current_token or ''}"
|
|
with _client_cache_lock:
|
|
client = _client_cache.get(cache_key)
|
|
if client is None:
|
|
client = DeezerClient()
|
|
_client_cache[cache_key] = client
|
|
return client
|
|
|
|
|
|
def get_itunes_client():
|
|
"""Get cached iTunes client."""
|
|
with _client_cache_lock:
|
|
client = _client_cache.get("itunes")
|
|
if client is None:
|
|
client = iTunesClient()
|
|
_client_cache["itunes"] = client
|
|
return client
|
|
|
|
|
|
def get_discogs_client(token: Optional[str] = None):
|
|
"""Get cached Discogs client.
|
|
|
|
Discogs auth changes are token-driven, so the cache key tracks the
|
|
current configured token.
|
|
"""
|
|
if token is None:
|
|
try:
|
|
from config.settings import config_manager
|
|
current_token = config_manager.get('discogs.token', '') or ''
|
|
except Exception:
|
|
current_token = ''
|
|
else:
|
|
current_token = token or ''
|
|
|
|
cache_key = f"discogs::{current_token}"
|
|
with _client_cache_lock:
|
|
client = _client_cache.get(cache_key)
|
|
if client is None:
|
|
from core.discogs_client import DiscogsClient
|
|
client = DiscogsClient(token=current_token or None)
|
|
_client_cache[cache_key] = client
|
|
return client
|
|
|
|
|
|
def is_hydrabase_enabled() -> bool:
|
|
"""Return True when Hydrabase is connected and allowed for metadata use."""
|
|
try:
|
|
import importlib
|
|
ws = importlib.import_module('web_server')
|
|
client = getattr(ws, 'hydrabase_client', None)
|
|
if not client or not client.is_connected():
|
|
return False
|
|
return bool(getattr(ws, 'dev_mode_enabled', False))
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def get_hydrabase_client(allow_fallback: bool = True, require_enabled: bool = True):
|
|
"""Return current Hydrabase client if connected and enabled.
|
|
|
|
If allow_fallback is True, return iTunes fallback when Hydrabase is not
|
|
connected or not enabled. If False, return None instead.
|
|
"""
|
|
try:
|
|
import importlib
|
|
ws = importlib.import_module('web_server')
|
|
client = getattr(ws, 'hydrabase_client', None)
|
|
if client and client.is_connected():
|
|
if not require_enabled or bool(getattr(ws, 'dev_mode_enabled', False)):
|
|
return client
|
|
except Exception:
|
|
pass
|
|
if allow_fallback:
|
|
return get_itunes_client()
|
|
return None
|
|
|
|
|
|
def clear_cached_metadata_clients():
|
|
"""Clear cached metadata clients.
|
|
|
|
Useful for tests and config reload flows.
|
|
"""
|
|
with _client_cache_lock:
|
|
_client_cache.clear()
|
|
|
|
|
|
def _get_client_for_source(source: str):
|
|
if source == 'spotify':
|
|
try:
|
|
import importlib
|
|
ws = importlib.import_module('web_server')
|
|
sc = getattr(ws, 'spotify_client', None)
|
|
if sc and sc.is_spotify_authenticated():
|
|
return sc
|
|
except Exception:
|
|
pass
|
|
return get_deezer_client()
|
|
|
|
if source == 'deezer':
|
|
return get_deezer_client()
|
|
|
|
if source == 'discogs':
|
|
return get_discogs_client()
|
|
|
|
if source == 'hydrabase':
|
|
return get_hydrabase_client()
|
|
|
|
return get_itunes_client()
|
|
|
|
|
|
# =============================================================================
|
|
# LEGACY ALIASES — kept for backward compatibility, delegate to canonical funcs
|
|
# =============================================================================
|
|
|
|
def _get_configured_fallback_source():
|
|
"""Legacy alias for get_primary_source(). Use get_primary_source() instead."""
|
|
return get_primary_source()
|
|
|
|
|
|
def _create_fallback_client():
|
|
"""Legacy alias for get_primary_client(). Use get_primary_client() instead."""
|
|
return get_primary_client()
|
|
|
|
|
|
class MetadataService:
|
|
"""
|
|
Unified metadata service that seamlessly switches between Spotify and
|
|
the configured fallback source (iTunes or Deezer).
|
|
|
|
Usage:
|
|
service = MetadataService()
|
|
tracks = service.search_tracks("Radiohead OK Computer")
|
|
# Uses Spotify if authenticated, otherwise configured fallback
|
|
"""
|
|
|
|
def __init__(self, preferred_provider: MetadataProvider = "auto"):
|
|
"""
|
|
Initialize metadata service.
|
|
|
|
Args:
|
|
preferred_provider: "spotify", "itunes", or "auto" (default)
|
|
- "auto": Use Spotify if authenticated, else configured fallback
|
|
- "spotify": Always use Spotify (may fail if not authenticated)
|
|
- "itunes": Always use configured fallback source
|
|
"""
|
|
self.preferred_provider = preferred_provider
|
|
self.spotify = SpotifyClient()
|
|
self._fallback_source = get_primary_source()
|
|
self.itunes = get_client_for_source(self._fallback_source)
|
|
|
|
self._log_initialization()
|
|
|
|
def _log_initialization(self):
|
|
"""Log initialization status"""
|
|
spotify_status = "Authenticated" if self.spotify.is_spotify_authenticated() else "Not authenticated"
|
|
fallback_status = "Available" if self.itunes.is_authenticated() else "Not available"
|
|
|
|
logger.info(f"MetadataService initialized - Spotify: {spotify_status}, {self._fallback_source.capitalize()}: {fallback_status}")
|
|
logger.info(f"Preferred provider: {self.preferred_provider}")
|
|
|
|
def get_active_provider(self) -> str:
|
|
"""
|
|
Get the currently active metadata provider.
|
|
|
|
Returns:
|
|
"spotify" or the configured fallback source name
|
|
"""
|
|
if self.preferred_provider == "spotify":
|
|
return "spotify"
|
|
elif self.preferred_provider == "itunes":
|
|
return self._fallback_source
|
|
else: # auto — use the centralized source selection
|
|
return get_primary_source()
|
|
|
|
def _get_client(self):
|
|
"""Get the appropriate client based on provider selection"""
|
|
provider = self.get_active_provider()
|
|
|
|
if provider == "spotify":
|
|
if not self.spotify.is_spotify_authenticated():
|
|
logger.warning(f"Spotify requested but not authenticated, falling back to {self._fallback_source}")
|
|
return self.itunes
|
|
return self.spotify
|
|
else:
|
|
return self.itunes
|
|
|
|
# ==================== Search Methods ====================
|
|
|
|
def search_tracks(self, query: str, limit: int = 20) -> List:
|
|
"""
|
|
Search for tracks using active provider.
|
|
|
|
Args:
|
|
query: Search query
|
|
limit: Maximum results
|
|
|
|
Returns:
|
|
List of Track objects
|
|
"""
|
|
client = self._get_client()
|
|
provider = self.get_active_provider()
|
|
logger.debug(f"Searching tracks with {provider}: '{query}'")
|
|
return client.search_tracks(query, limit)
|
|
|
|
def search_artists(self, query: str, limit: int = 20) -> List:
|
|
"""
|
|
Search for artists using active provider.
|
|
|
|
Args:
|
|
query: Search query
|
|
limit: Maximum results
|
|
|
|
Returns:
|
|
List of Artist objects
|
|
"""
|
|
client = self._get_client()
|
|
provider = self.get_active_provider()
|
|
logger.debug(f"Searching artists with {provider}: '{query}'")
|
|
return client.search_artists(query, limit)
|
|
|
|
def search_albums(self, query: str, limit: int = 20) -> List:
|
|
"""
|
|
Search for albums using active provider.
|
|
|
|
Args:
|
|
query: Search query
|
|
limit: Maximum results
|
|
|
|
Returns:
|
|
List of Album objects
|
|
"""
|
|
client = self._get_client()
|
|
provider = self.get_active_provider()
|
|
logger.debug(f"Searching albums with {provider}: '{query}'")
|
|
return client.search_albums(query, limit)
|
|
|
|
# ==================== Detail Fetching ====================
|
|
|
|
def get_track_details(self, track_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Get detailed track information"""
|
|
client = self._get_client()
|
|
return client.get_track_details(track_id)
|
|
|
|
def get_album(self, album_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Get album information"""
|
|
client = self._get_client()
|
|
return client.get_album(album_id)
|
|
|
|
def get_album_tracks(self, album_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Get all tracks from an album"""
|
|
client = self._get_client()
|
|
provider = self.get_active_provider()
|
|
logger.debug(f"Fetching album tracks with {provider}: {album_id}")
|
|
return client.get_album_tracks(album_id)
|
|
|
|
def get_artist(self, artist_id: str) -> Optional[Dict[str, Any]]:
|
|
"""Get artist information"""
|
|
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:
|
|
"""Get artist's albums/discography"""
|
|
client = self._get_client()
|
|
provider = self.get_active_provider()
|
|
logger.debug(f"Fetching artist albums with {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]]:
|
|
"""
|
|
Get track audio features (Spotify only).
|
|
Returns None for iTunes.
|
|
"""
|
|
client = self._get_client()
|
|
return client.get_track_features(track_id)
|
|
|
|
# ==================== User Library (Spotify only) ====================
|
|
|
|
def get_user_playlists(self) -> List:
|
|
"""Get user playlists (Spotify only)"""
|
|
if 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:
|
|
"""Get user's saved/liked tracks (Spotify only)"""
|
|
if 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:
|
|
"""Get count of user's saved tracks (Spotify only)"""
|
|
if self.spotify.is_spotify_authenticated():
|
|
return self.spotify.get_saved_tracks_count()
|
|
return 0
|
|
|
|
# ==================== Utility Methods ====================
|
|
|
|
def is_authenticated(self) -> bool:
|
|
"""Check if any provider is available"""
|
|
return self.spotify.is_spotify_authenticated() or self.itunes.is_authenticated()
|
|
|
|
def get_provider_info(self) -> Dict[str, Any]:
|
|
"""Get information about available providers"""
|
|
return {
|
|
"active_provider": self.get_active_provider(),
|
|
"spotify_authenticated": self.spotify.is_spotify_authenticated(),
|
|
"itunes_available": self.itunes.is_authenticated(),
|
|
"fallback_source": self._fallback_source,
|
|
"preferred_provider": self.preferred_provider,
|
|
"can_access_user_data": self.spotify.is_spotify_authenticated(),
|
|
}
|
|
|
|
def reload_config(self):
|
|
"""Reload configuration for both clients"""
|
|
logger.info("Reloading metadata service configuration")
|
|
self.spotify.reload_config()
|
|
new_source = get_primary_source()
|
|
self._fallback_source = new_source
|
|
self.itunes = get_client_for_source(new_source)
|
|
self._log_initialization()
|
|
|
|
|
|
# Convenience singleton instance
|
|
_metadata_service_instance: Optional[MetadataService] = None
|
|
|
|
|
|
def get_metadata_service() -> MetadataService:
|
|
"""
|
|
Get global metadata service instance (singleton pattern).
|
|
|
|
Returns:
|
|
MetadataService instance
|
|
"""
|
|
global _metadata_service_instance
|
|
if _metadata_service_instance is None:
|
|
_metadata_service_instance = MetadataService()
|
|
return _metadata_service_instance
|