/
/
/
1"""Last.fm API client for recommendations."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, Any, ClassVar
6
7from aiohttp import ClientError
8from music_assistant_models.errors import (
9 AuthenticationFailed,
10 InvalidDataError,
11 InvalidToken,
12 MusicAssistantError,
13 ResourceTemporarilyUnavailable,
14)
15
16from music_assistant.helpers.app_vars import app_var
17from music_assistant.helpers.throttle_retry import ThrottlerManager
18from music_assistant.providers.lastfm_recommendations.constants import CONF_API_KEY
19
20# Built-in Last.fm API key (read-only access, no secret needed)
21_DEFAULT_API_KEY: str = app_var("lastfm_api_key")
22
23if TYPE_CHECKING:
24 import logging
25
26 from aiohttp import ClientSession
27
28 from music_assistant.providers.lastfm_recommendations import LastFMRecommendationsProvider
29
30
31class LastFMAPIClient:
32 """Last.fm API client for fetching recommendations."""
33
34 BASE_URL = "https://ws.audioscrobbler.com/2.0/"
35 throttler = ThrottlerManager(rate_limit=5, period=1) # 5 requests per second
36
37 # Map known Last.fm API error codes to MA exceptions. Unmapped codes raise
38 # InvalidDataError which callers treat as a recoverable empty response.
39 _ERROR_MAP: ClassVar[dict[int, type[MusicAssistantError]]] = {
40 4: AuthenticationFailed,
41 10: InvalidToken,
42 26: InvalidToken,
43 29: ResourceTemporarilyUnavailable,
44 }
45
46 def __init__(self, provider: LastFMRecommendationsProvider) -> None:
47 """
48 Initialize Last.fm API client.
49
50 :param provider: The Last.fm recommendations provider instance.
51 """
52 self.provider = provider
53 self.http_session: ClientSession = provider.mass.http_session
54
55 @property
56 def logger(self) -> logging.Logger:
57 """Return the provider's active logger."""
58 return self.provider.logger
59
60 async def _get_data(self, method: str, **params: Any) -> dict[str, Any]:
61 """
62 Make a request to the Last.fm API.
63
64 :param method: The Last.fm API method to call.
65 :param params: Additional query parameters.
66 """
67 async with self.throttler.acquire():
68 api_key = self.provider.config.get_value(CONF_API_KEY) or _DEFAULT_API_KEY
69 params.update(
70 {
71 "method": method,
72 "api_key": api_key,
73 "format": "json",
74 }
75 )
76
77 async with self.http_session.get(self.BASE_URL, params=params) as response:
78 response.raise_for_status()
79 data: dict[str, Any] = await response.json()
80
81 # Last.fm returns errors in the response body rather than as an HTTP status.
82 if "error" in data:
83 error_code = int(data.get("error", 0))
84 error_msg = data.get("message", "Unknown error")
85 msg = f"Last.fm API error {error_code}: {error_msg} (method: {method})"
86 exc_cls = self._ERROR_MAP.get(error_code, InvalidDataError)
87 raise exc_cls(msg)
88
89 return data
90
91 async def _get_list(
92 self,
93 method: str,
94 container: str,
95 item_key: str,
96 log_label: str,
97 primary_params: dict[str, Any],
98 fallback_params: dict[str, Any] | None = None,
99 ) -> list[dict[str, Any]]:
100 """
101 Fetch a list-type Last.fm response, retrying with the fallback params when empty.
102
103 :param method: Last.fm API method to call.
104 :param container: Top-level response key wrapping the result list.
105 :param item_key: Key of the result list within the container.
106 :param log_label: Human-readable label used in debug logging.
107 :param primary_params: Preferred request params (MBID when available).
108 :param fallback_params: Name-based params to retry with, or None for no retry.
109 """
110 # Last.fm's MBID index is sparse (recording MBIDs especially), so a missing MBID
111 # is retried by name rather than treated as "no similar items exist".
112 attempts = [primary_params]
113 if fallback_params is not None:
114 attempts.append(fallback_params)
115
116 for index, params in enumerate(attempts):
117 if index:
118 self.logger.debug("%s: MBID lookup empty, retrying by name", log_label)
119 try:
120 data = await self._get_data(method, **params)
121 except (TimeoutError, ClientError, InvalidDataError) as err:
122 self.logger.debug("%s request failed: %s", log_label, err)
123 continue
124
125 items: list[dict[str, Any]] | dict[str, Any] = data.get(container, {}).get(item_key, [])
126 # Last.fm returns a single dict when only one result is present.
127 if isinstance(items, dict):
128 return [items]
129 if items:
130 return items
131
132 return []
133
134 async def get_similar_artists(
135 self, artist_name: str, artist_mbid: str | None = None, limit: int = 10
136 ) -> list[dict[str, Any]]:
137 """
138 Get similar artists from Last.fm.
139
140 :param artist_name: Name of the artist.
141 :param artist_mbid: Optional MusicBrainz ID for more accurate matching.
142 :param limit: Maximum number of similar artists to return.
143 """
144 self.logger.debug(
145 "Fetching similar artists for: %s (MBID: %s)",
146 artist_name,
147 artist_mbid or "none",
148 )
149 name_params: dict[str, Any] | None = (
150 {"limit": limit, "artist": artist_name, "autocorrect": 1} if artist_name else None
151 )
152 primary_params: dict[str, Any] | None
153 fallback_params: dict[str, Any] | None
154 # Lead with the MBID (exact when Last.fm has it), fall back to the name lookup.
155 if artist_mbid:
156 primary_params = {"limit": limit, "mbid": artist_mbid}
157 fallback_params = name_params
158 else:
159 primary_params = name_params
160 fallback_params = None
161
162 if primary_params is None:
163 return []
164
165 return await self._get_list(
166 "artist.getSimilar",
167 "similarartists",
168 "artist",
169 "Similar artists",
170 primary_params,
171 fallback_params,
172 )
173
174 async def get_similar_tracks(
175 self,
176 artist_name: str,
177 track_name: str,
178 track_mbid: str | None = None,
179 limit: int = 10,
180 ) -> list[dict[str, Any]]:
181 """
182 Get similar tracks from Last.fm.
183
184 :param artist_name: Name of the track's artist.
185 :param track_name: Name of the track.
186 :param track_mbid: Optional MusicBrainz ID for more accurate matching.
187 :param limit: Maximum number of similar tracks to return.
188 """
189 self.logger.debug(
190 "Fetching similar tracks for: %s - %s (MBID: %s)",
191 artist_name,
192 track_name,
193 track_mbid or "none",
194 )
195 name_params: dict[str, Any] | None = (
196 {"limit": limit, "artist": artist_name, "track": track_name, "autocorrect": 1}
197 if artist_name and track_name
198 else None
199 )
200 primary_params: dict[str, Any] | None
201 fallback_params: dict[str, Any] | None
202 # Lead with the MBID (exact when Last.fm has it), fall back to the name lookup.
203 if track_mbid:
204 primary_params = {"limit": limit, "mbid": track_mbid}
205 fallback_params = name_params
206 else:
207 primary_params = name_params
208 fallback_params = None
209
210 if primary_params is None:
211 return []
212
213 return await self._get_list(
214 "track.getSimilar",
215 "similartracks",
216 "track",
217 "Similar tracks",
218 primary_params,
219 fallback_params,
220 )
221
222 async def get_artist_top_tracks(
223 self, artist_name: str, artist_mbid: str | None = None, limit: int = 10
224 ) -> list[dict[str, Any]]:
225 """
226 Get an artist's top tracks from Last.fm, ordered by popularity.
227
228 :param artist_name: Name of the artist.
229 :param artist_mbid: Optional MusicBrainz ID for more accurate matching.
230 :param limit: Maximum number of top tracks to return.
231 """
232 self.logger.debug(
233 "Fetching top tracks for artist: %s (MBID: %s)",
234 artist_name,
235 artist_mbid or "none",
236 )
237 name_params: dict[str, Any] | None = (
238 {"limit": limit, "artist": artist_name, "autocorrect": 1} if artist_name else None
239 )
240 primary_params: dict[str, Any] | None
241 fallback_params: dict[str, Any] | None
242 # Lead with the MBID (exact when Last.fm has it), fall back to the name lookup.
243 if artist_mbid:
244 primary_params = {"limit": limit, "mbid": artist_mbid}
245 fallback_params = name_params
246 else:
247 primary_params = name_params
248 fallback_params = None
249
250 if primary_params is None:
251 return []
252
253 return await self._get_list(
254 "artist.getTopTracks",
255 "toptracks",
256 "track",
257 "Artist top tracks",
258 primary_params,
259 fallback_params,
260 )
261
262 async def get_chart_top_artists(self, limit: int = 10) -> list[dict[str, Any]]:
263 """
264 Get global top artists chart from Last.fm.
265
266 :param limit: Maximum number of artists to return.
267 """
268 try:
269 data = await self._get_data("chart.getTopArtists", limit=limit)
270 except (TimeoutError, ClientError, InvalidDataError) as err:
271 self.logger.debug("Chart top artists request failed: %s", err)
272 return []
273
274 artists: list[dict[str, Any]] | dict[str, Any] = data.get("artists", {}).get("artist", [])
275
276 # Last.fm returns a single dict when only one result is present.
277 if isinstance(artists, dict):
278 return [artists]
279
280 return artists
281
282 async def get_chart_top_tracks(self, limit: int = 10) -> list[dict[str, Any]]:
283 """
284 Get global top tracks chart from Last.fm.
285
286 :param limit: Maximum number of tracks to return.
287 """
288 try:
289 data = await self._get_data("chart.getTopTracks", limit=limit)
290 except (TimeoutError, ClientError, InvalidDataError) as err:
291 self.logger.debug("Chart top tracks request failed: %s", err)
292 return []
293
294 tracks: list[dict[str, Any]] | dict[str, Any] = data.get("tracks", {}).get("track", [])
295
296 # Last.fm returns a single dict when only one result is present.
297 if isinstance(tracks, dict):
298 return [tracks]
299
300 return tracks
301
302 async def get_user_top_artists(
303 self, username: str, period: str = "overall", limit: int = 50
304 ) -> list[dict[str, Any]]:
305 """
306 Get a user's most played artists from Last.fm, most played first.
307
308 :param username: Last.fm username.
309 :param period: Span to rank over (overall, 7day, 1month, 3month, 6month, 12month).
310 :param limit: Maximum number of artists to return.
311 """
312 self.logger.debug("Fetching top artists for user: %s (period: %s)", username, period)
313 try:
314 data = await self._get_data(
315 "user.getTopArtists", user=username, period=period, limit=limit
316 )
317 except (TimeoutError, ClientError, InvalidDataError) as err:
318 self.logger.debug("User top artists request failed: %s", err)
319 return []
320
321 artists: list[dict[str, Any]] | dict[str, Any] = data.get("topartists", {}).get(
322 "artist", []
323 )
324
325 # Last.fm returns a single dict when only one result is present.
326 if isinstance(artists, dict):
327 return [artists]
328
329 return artists
330
331 async def get_artist_top_tags(
332 self, artist_name: str, artist_mbid: str | None = None, limit: int = 5
333 ) -> list[dict[str, Any]]:
334 """
335 Get an artist's top community tags from Last.fm, most agreed-upon first.
336
337 :param artist_name: Name of the artist.
338 :param artist_mbid: Optional MusicBrainz ID for more accurate matching.
339 :param limit: Maximum number of tags to return.
340 """
341 self.logger.debug(
342 "Fetching top tags for artist: %s (MBID: %s)", artist_name, artist_mbid or "none"
343 )
344 name_params: dict[str, Any] | None = (
345 {"artist": artist_name, "autocorrect": 1} if artist_name else None
346 )
347 primary_params: dict[str, Any] | None
348 fallback_params: dict[str, Any] | None
349 # Lead with the MBID (exact when Last.fm has it), fall back to the name lookup.
350 if artist_mbid:
351 primary_params = {"mbid": artist_mbid}
352 fallback_params = name_params
353 else:
354 primary_params = name_params
355 fallback_params = None
356
357 if primary_params is None:
358 return []
359
360 # artist.getTopTags has no limit parameter, so cap the (count-ordered) result ourselves.
361 tags = await self._get_list(
362 "artist.getTopTags",
363 "toptags",
364 "tag",
365 "Artist top tags",
366 primary_params,
367 fallback_params,
368 )
369 return tags[:limit]
370
371 async def get_tag_top_artists(self, tag: str, limit: int = 10) -> list[dict[str, Any]]:
372 """
373 Get top artists for a tag from Last.fm.
374
375 :param tag: Tag name (genre).
376 :param limit: Maximum number of artists to return.
377 """
378 self.logger.debug("Fetching top artists for tag: %s (limit: %d)", tag, limit)
379 try:
380 data = await self._get_data("tag.getTopArtists", tag=tag, limit=limit)
381 except (TimeoutError, ClientError, InvalidDataError) as err:
382 self.logger.debug("Tag top artists request failed: %s", err)
383 return []
384
385 artists: list[dict[str, Any]] | dict[str, Any] = data.get("topartists", {}).get(
386 "artist", []
387 )
388
389 # Last.fm returns a single dict when only one result is present.
390 if isinstance(artists, dict):
391 return [artists]
392
393 return artists
394
395 async def get_tag_top_albums(self, tag: str, limit: int = 10) -> list[dict[str, Any]]:
396 """
397 Get top albums for a tag from Last.fm.
398
399 :param tag: Tag name (genre).
400 :param limit: Maximum number of albums to return.
401 """
402 self.logger.debug("Fetching top albums for tag: %s (limit: %d)", tag, limit)
403 try:
404 data = await self._get_data("tag.getTopAlbums", tag=tag, limit=limit)
405 except (TimeoutError, ClientError, InvalidDataError) as err:
406 self.logger.debug("Tag top albums request failed: %s", err)
407 return []
408
409 albums: list[dict[str, Any]] | dict[str, Any] = data.get("albums", {}).get("album", [])
410
411 # Last.fm returns a single dict when only one result is present.
412 if isinstance(albums, dict):
413 return [albums]
414
415 return albums
416
417 async def get_tag_top_tracks(self, tag: str, limit: int = 10) -> list[dict[str, Any]]:
418 """
419 Get top tracks for a tag from Last.fm.
420
421 :param tag: Tag name (genre).
422 :param limit: Maximum number of tracks to return.
423 """
424 self.logger.debug("Fetching top tracks for tag: %s (limit: %d)", tag, limit)
425 try:
426 data = await self._get_data("tag.getTopTracks", tag=tag, limit=limit)
427 except (TimeoutError, ClientError, InvalidDataError) as err:
428 self.logger.debug("Tag top tracks request failed: %s", err)
429 return []
430
431 tracks: list[dict[str, Any]] | dict[str, Any] = data.get("tracks", {}).get("track", [])
432
433 # Last.fm returns a single dict when only one result is present.
434 if isinstance(tracks, dict):
435 return [tracks]
436
437 return tracks
438
439 async def get_geo_top_artists(self, country: str, limit: int = 10) -> list[dict[str, Any]]:
440 """
441 Get top artists for a country from Last.fm.
442
443 :param country: Country name (e.g., "United States", "Spain").
444 :param limit: Maximum number of artists to return.
445 """
446 self.logger.debug("Fetching geo top artists for country: %s (limit: %d)", country, limit)
447 try:
448 data = await self._get_data("geo.getTopArtists", country=country, limit=limit)
449 except (TimeoutError, ClientError, InvalidDataError) as err:
450 self.logger.debug("Geo top artists request failed: %s", err)
451 return []
452
453 artists: list[dict[str, Any]] | dict[str, Any] = data.get("topartists", {}).get(
454 "artist", []
455 )
456
457 # Last.fm returns a single dict when only one result is present.
458 if isinstance(artists, dict):
459 return [artists]
460
461 return artists
462
463 async def get_geo_top_tracks(self, country: str, limit: int = 10) -> list[dict[str, Any]]:
464 """
465 Get top tracks for a country from Last.fm.
466
467 :param country: Country name (e.g., "United States", "Spain").
468 :param limit: Maximum number of tracks to return.
469 """
470 self.logger.debug("Fetching geo top tracks for country: %s (limit: %d)", country, limit)
471 try:
472 data = await self._get_data("geo.getTopTracks", country=country, limit=limit)
473 except (TimeoutError, ClientError, InvalidDataError) as err:
474 self.logger.debug("Geo top tracks request failed: %s", err)
475 return []
476
477 tracks: list[dict[str, Any]] | dict[str, Any] = data.get("tracks", {}).get("track", [])
478
479 # Last.fm returns a single dict when only one result is present.
480 if isinstance(tracks, dict):
481 return [tracks]
482
483 return tracks
484