/
/
/
1"""Allows scrobbling of tracks with the help of PyLast."""
2
3import asyncio
4import enum
5import logging
6import time
7from collections.abc import Callable, Mapping
8from typing import TYPE_CHECKING, ClassVar, Final, cast
9
10import pylast
11from music_assistant_models.constants import SECURE_STRING_SUBSTITUTE
12from music_assistant_models.enums import EventType, MediaType, ProviderFeature
13from music_assistant_models.errors import SetupFailedError
14
15from music_assistant.helpers.app_vars import app_var
16from music_assistant.helpers.scrobbler import ScrobblerConfig, ScrobblerHelper
17from music_assistant.mass import MusicAssistant
18from music_assistant.models import ProviderInstanceType
19from music_assistant.models.plugin import PluginProvider
20
21if TYPE_CHECKING:
22 from music_assistant_models.config_entries import ConfigEntry, ConfigValueType, ProviderConfig
23 from music_assistant_models.playback_progress_report import MediaItemPlaybackProgressReport
24 from music_assistant_models.provider import ProviderManifest
25
26# Built-in Last.fm API credentials (not available for Libre.fm)
27_DEFAULT_API_KEY: str = app_var("lastfm_api_key")
28_DEFAULT_API_SECRET: str = app_var("lastfm_api_secret")
29
30
31# we don't have any special supported features (yet)
32# TODO(@anyone): this really should be a frozenset, but that requires
33# updating the PluginProvider base class
34# as well as other similar classes that also use set[ProviderFeature].
35SUPPORTED_FEATURES: Final[set[ProviderFeature]] = set()
36SUPPORTED_SCROBBLE_MEDIA_TYPES: Final[frozenset[MediaType]] = frozenset({MediaType.TRACK})
37
38# Configuration keys
39CONF_API_KEY: Final[str] = "_api_key"
40CONF_API_SECRET: Final[str] = "_api_secret"
41CONF_SESSION_KEY: Final[str] = "_api_session_key"
42CONF_USERNAME: Final[str] = "_username"
43CONF_PROVIDER: Final[str] = "_provider"
44
45
46class _NetworkType(enum.Enum):
47 """
48 Available scrobbling network provider types.
49
50 This is a plain Enum class with string values.
51 Use ``.value`` when passing to ``ConfigEntry`` or ``ConfigValueOption``
52 which require raw strings.
53 """
54
55 LASTFM = "lastfm"
56 LIBREFM = "librefm"
57
58
59def _resolve_credentials(
60 values: Mapping[str, ConfigValueType],
61 network_type: _NetworkType = _NetworkType.LASTFM,
62) -> tuple[str, str]:
63 """
64 Resolve the effective API key and secret.
65
66 Uses user-provided values if present, otherwise falls back to the
67 built-in Last.fm credentials. Libre.fm always requires user-provided values.
68
69 :param values: Config values dict that may contain user-provided key/secret.
70 :param network_type: The network provider type.
71 :returns: A tuple of (api_key, api_secret) strings.
72 :raises SetupFailedError: If credentials cannot be resolved.
73 """
74 key = cast("str | None", values.get(CONF_API_KEY))
75 secret = cast("str | None", values.get(CONF_API_SECRET))
76
77 has_custom_key = bool(key and key != SECURE_STRING_SUBSTITUTE)
78 has_custom_secret = bool(secret and secret != SECURE_STRING_SUBSTITUTE)
79
80 if has_custom_key and has_custom_secret:
81 return str(key), str(secret)
82 if has_custom_key or has_custom_secret:
83 err_msg = "Both API Key and Shared Secret are required (only one provided)."
84 raise SetupFailedError(err_msg)
85
86 match network_type:
87 case _NetworkType.LASTFM:
88 return str(_DEFAULT_API_KEY), str(_DEFAULT_API_SECRET)
89 case _:
90 err_msg = f"API Key and Secret are required for {network_type.value}. "
91 raise SetupFailedError(err_msg)
92
93
94async def setup(
95 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
96) -> ProviderInstanceType:
97 """
98 Initialize provider(instance) with given configuration.
99
100 :returns: A configured LastFMScrobbleProvider instance.
101 """
102 provider = LastFMScrobbleProvider(mass, manifest, config, SUPPORTED_FEATURES)
103 pylast.logger.setLevel(provider.logger.level)
104
105 # httpcore is very spammy on debug without providing useful information 99% of the time
106 if provider.logger.level == logging.DEBUG:
107 logging.getLogger("httpcore").setLevel(logging.INFO)
108 else:
109 logging.getLogger("httpcore").setLevel(logging.WARNING)
110
111 return provider
112
113
114class LastFMScrobbleProvider(PluginProvider):
115 """Plugin provider to support scrobbling of tracks."""
116
117 _network: pylast._Network | None
118 _on_unload: list[Callable[[], None]]
119
120 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
121 """
122 Return config entries to configure this provider.
123
124 Authentication (network, credentials and session key) is handled by the setup
125 flow (see setup_flow.py); only the genuine scrobble-filter options are
126 configurable here.
127 """
128 return tuple(await ScrobblerConfig.get_shared_config_entries(self.mass, None))
129
130 async def handle_async_init(self) -> None:
131 """Handle async setup."""
132 self._on_unload: list[Callable[[], None]] = []
133 self._network = None
134
135 if not self.get_setup_value(CONF_SESSION_KEY):
136 self.logger.info("No session key available, don't forget to authenticate!")
137 return
138 # creating the network instance is (potentially) blocking IO
139 # so run it in an executor thread to be safe
140 self._network = await asyncio.to_thread(get_network, self._get_network_config())
141
142 async def loaded_in_mass(self) -> None:
143 """Call after the provider has been loaded."""
144 await super().loaded_in_mass()
145
146 if self._network is None:
147 return
148
149 # subscribe to media_item_played event
150 handler = LastFMEventHandler(self._network, self.logger, self.config)
151 self._on_unload.append(
152 self.mass.subscribe(handler._on_mass_media_item_played, EventType.MEDIA_ITEM_PLAYED)
153 )
154
155 async def unload(self, is_removed: bool = False) -> None:
156 """
157 Handle unload/close of the provider.
158
159 Called when provider is deregistered (e.g. MA exiting or config reloading).
160 """
161 for unload_cb in self._on_unload:
162 unload_cb()
163
164 def _get_network_config(self) -> dict[str, ConfigValueType]:
165 """
166 Build the network configuration dict from the provider's setup data.
167
168 :returns: Dict of config keys to their current stored values.
169 """
170 return {
171 CONF_API_KEY: self.get_setup_value(CONF_API_KEY),
172 CONF_API_SECRET: self.get_setup_value(CONF_API_SECRET),
173 CONF_PROVIDER: self.get_setup_value(CONF_PROVIDER),
174 CONF_USERNAME: self.get_setup_value(CONF_USERNAME),
175 CONF_SESSION_KEY: self.get_setup_value(CONF_SESSION_KEY),
176 }
177
178
179class LastFMEventHandler(ScrobblerHelper):
180 """Handle Last.fm event processing for scrobbling and now-playing updates."""
181
182 # pylast wraps every failure â including network errors â in PyLastError.
183 scrobble_exceptions: ClassVar[tuple[type[Exception], ...]] = (pylast.PyLastError,)
184
185 def __init__(
186 self, network: pylast._Network, logger: logging.Logger, config: ProviderConfig
187 ) -> None:
188 """Initialize."""
189 super().__init__(
190 logger,
191 ScrobblerConfig.create_from_config(config),
192 SUPPORTED_SCROBBLE_MEDIA_TYPES,
193 )
194 self._network = network
195
196 async def _update_now_playing(self, report: MediaItemPlaybackProgressReport) -> None:
197 """Send a now-playing update to Last.fm."""
198 # the lastfm client is not async friendly,
199 # so we need to run it in a executor thread
200 await asyncio.to_thread(
201 self._network.update_now_playing,
202 report.artist,
203 self.get_name(report),
204 report.album,
205 duration=report.duration,
206 mbid=report.mbid,
207 )
208
209 async def _scrobble(self, report: MediaItemPlaybackProgressReport) -> None:
210 """Scrobble a track to Last.fm."""
211 # the listenbrainz client is not async friendly,
212 # so we need to run it in a executor thread
213 # NOTE: album artist and track number are not available without an extra API call
214 # so they won't be scrobbled
215 await asyncio.to_thread(
216 self._network.scrobble,
217 report.artist or "unknown artist",
218 self.get_name(report),
219 int(time.time()),
220 report.album,
221 duration=report.duration,
222 mbid=report.mbid,
223 )
224
225
226def get_network(config: dict[str, ConfigValueType]) -> pylast._Network:
227 """
228 Create a pylast network instance with resolved credentials.
229
230 Called in two contexts:
231 1. during the setup flow (from ``setup_flow.run_setup``)
232 to build the authorization URL before any session exists
233 2. during provider startup (from ``handle_async_init``)
234 for scrobbling with a stored session.
235
236 Session key and username default to empty strings
237 because the auth flow legitimately needs a network without them.
238
239 :param config: Config values dict containing provider type, credentials, etc.
240 :returns: A pylast LastFMNetwork or LibreFMNetwork instance.
241 :raises SetupFailedError: If the provider is unknown or credentials cannot be resolved.
242 """
243 network_type = _NetworkType(str(config.get(CONF_PROVIDER, _NetworkType.LASTFM.value)))
244 key, secret = _resolve_credentials(config, network_type)
245 session_key = str(config.get(CONF_SESSION_KEY) or "")
246 username = str(config.get(CONF_USERNAME) or "")
247
248 match network_type:
249 case _NetworkType.LASTFM:
250 return pylast.LastFMNetwork(key, secret, username=username, session_key=session_key)
251 case _NetworkType.LIBREFM:
252 return pylast.LibreFMNetwork(key, secret, username=username, session_key=session_key)
253