Harvey
7fd6bb83bd
* MUS-42 Discover page improvements + playlist/album changes + various discovery fixes + tanstack query stuuff + placeholder image fix * fix formatter truncate * fix tests * make lint * fix make ci errors * address copilot
91 lines
3.7 KiB
Python
91 lines
3.7 KiB
Python
import logging
|
|
|
|
from api.v1.schemas.discover import (
|
|
DiscoverIntegrationStatus,
|
|
QueueSettings,
|
|
)
|
|
from api.v1.schemas.settings import HomeSettings
|
|
from services.preferences_service import PreferencesService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DISCOVER_CACHE_KEY = "discover_response"
|
|
|
|
|
|
class IntegrationHelpers:
|
|
def __init__(self, preferences_service: PreferencesService) -> None:
|
|
self._preferences = preferences_service
|
|
|
|
def is_listenbrainz_enabled(self) -> bool:
|
|
lb_settings = self._preferences.get_listenbrainz_connection()
|
|
return lb_settings.enabled and bool(lb_settings.username)
|
|
|
|
def is_jellyfin_enabled(self) -> bool:
|
|
jf_settings = self._preferences.get_jellyfin_connection()
|
|
return jf_settings.enabled and bool(jf_settings.jellyfin_url) and bool(jf_settings.api_key)
|
|
|
|
def is_lidarr_configured(self) -> bool:
|
|
lidarr_connection = self._preferences.get_lidarr_connection()
|
|
return bool(lidarr_connection.lidarr_url) and bool(lidarr_connection.lidarr_api_key)
|
|
|
|
def is_youtube_api_enabled(self) -> bool:
|
|
yt_settings = self._preferences.get_youtube_connection()
|
|
return yt_settings.enabled and yt_settings.api_enabled and yt_settings.has_valid_api_key()
|
|
|
|
def is_lastfm_enabled(self) -> bool:
|
|
return self._preferences.is_lastfm_enabled()
|
|
|
|
def get_listenbrainz_username(self) -> str | None:
|
|
lb_settings = self._preferences.get_listenbrainz_connection()
|
|
return lb_settings.username if lb_settings.enabled else None
|
|
|
|
def get_lastfm_username(self) -> str | None:
|
|
lf_settings = self._preferences.get_lastfm_connection()
|
|
return lf_settings.username if lf_settings.enabled else None
|
|
|
|
def resolve_source(self, source: str | None) -> str:
|
|
if source in ("listenbrainz", "lastfm"):
|
|
resolved = source
|
|
else:
|
|
resolved = self._preferences.get_primary_music_source().source
|
|
lb_enabled = self.is_listenbrainz_enabled()
|
|
lfm_enabled = self.is_lastfm_enabled()
|
|
if resolved == "listenbrainz" and not lb_enabled and lfm_enabled:
|
|
return "lastfm"
|
|
if resolved == "lastfm" and not lfm_enabled and lb_enabled:
|
|
return "listenbrainz"
|
|
return resolved
|
|
|
|
def get_queue_settings(self) -> QueueSettings:
|
|
adv = self._preferences.get_advanced_settings()
|
|
return QueueSettings(
|
|
queue_size=adv.discover_queue_size,
|
|
queue_ttl=adv.discover_queue_ttl,
|
|
seed_artists=adv.discover_queue_seed_artists,
|
|
wildcard_slots=adv.discover_queue_wildcard_slots,
|
|
similar_artists_limit=adv.discover_queue_similar_artists_limit,
|
|
albums_per_similar=adv.discover_queue_albums_per_similar,
|
|
enrich_ttl=adv.discover_queue_enrich_ttl,
|
|
lastfm_mbid_max_lookups=adv.discover_queue_lastfm_mbid_max_lookups,
|
|
)
|
|
|
|
def get_discover_cache_key(self, source: str | None = None) -> str:
|
|
resolved = self.resolve_source(source)
|
|
return f"{DISCOVER_CACHE_KEY}:{resolved}"
|
|
|
|
def get_home_settings(self) -> HomeSettings:
|
|
return self._preferences.get_home_settings()
|
|
|
|
def get_integration_status(self) -> DiscoverIntegrationStatus:
|
|
return DiscoverIntegrationStatus(
|
|
listenbrainz=self.is_listenbrainz_enabled(),
|
|
jellyfin=self.is_jellyfin_enabled(),
|
|
lidarr=self.is_lidarr_configured(),
|
|
youtube=self.is_youtube_api_enabled(),
|
|
lastfm=self.is_lastfm_enabled(),
|
|
)
|
|
|
|
def get_discover_picks_settings(self) -> tuple[float, int]:
|
|
adv = self._preferences.get_advanced_settings()
|
|
return adv.discover_picks_genre_affinity_weight, adv.discover_picks_count
|