/
/
/
1"""Last.fm Recommendations music provider for Music Assistant."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import TYPE_CHECKING
7
8from music_assistant_models.background_task import TaskSchedule
9from music_assistant_models.config_entries import (
10 ConfigActionResult,
11 ConfigEntry,
12 ConfigValueOption,
13)
14from music_assistant_models.enums import ConfigEntryType, ExternalID, ProviderFeature
15from music_assistant_models.errors import (
16 AuthenticationFailed,
17 InvalidToken,
18 MusicAssistantError,
19 ResourceTemporarilyUnavailable,
20)
21from music_assistant_models.media_items import RecommendationFolder, Track, UniqueList
22
23from music_assistant.constants import CONF_USERNAME
24from music_assistant.controllers.cache import use_cache
25from music_assistant.models.metadata_provider import MetadataProvider
26from music_assistant.providers.lastfm_recommendations.api_client import LastFMAPIClient
27from music_assistant.providers.lastfm_recommendations.constants import (
28 CACHE_CATEGORY_RESOLVED_ITEMS,
29 CACHE_EXPIRATION_SECONDS,
30 CONF_ACTION_CLEAR_CACHE,
31 CONF_API_KEY,
32 CONF_ENABLE_GENRE,
33 CONF_ENABLE_GEO,
34 CONF_ENABLE_GLOBAL_CHARTS,
35 CONF_ENABLE_PERSONALIZED,
36 CONF_GEO_COUNTRY,
37 GEO_COUNTRIES,
38 REFRESH_TASK_ID,
39)
40from music_assistant.providers.lastfm_recommendations.recommendations import (
41 LastFMRecommendationManager,
42)
43
44if TYPE_CHECKING:
45 from music_assistant_models.config_entries import ProviderConfig
46 from music_assistant_models.media_items import (
47 Artist,
48 BrowseFolder,
49 ItemMapping,
50 MediaItemType,
51 )
52 from music_assistant_models.provider import ProviderManifest
53
54 from music_assistant.mass import MusicAssistant
55
56
57SUPPORTED_FEATURES = {
58 ProviderFeature.RECOMMENDATIONS,
59 ProviderFeature.SIMILAR_ARTISTS,
60 ProviderFeature.SIMILAR_TRACKS,
61 ProviderFeature.ARTIST_TOPTRACKS,
62}
63
64
65async def setup(
66 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
67) -> LastFMRecommendationsProvider:
68 """Initialize provider(instance) with given configuration."""
69 return LastFMRecommendationsProvider(mass, manifest, config, SUPPORTED_FEATURES)
70
71
72class LastFMRecommendationsProvider(MetadataProvider):
73 """Last.fm Recommendations Provider for Music Assistant."""
74
75 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
76 """Return Config entries to configure this provider."""
77 return (
78 ConfigEntry(
79 key=CONF_API_KEY,
80 type=ConfigEntryType.SECURE_STRING,
81 required=False,
82 advanced=True,
83 ),
84 ConfigEntry(
85 key=CONF_USERNAME,
86 type=ConfigEntryType.STRING,
87 required=False,
88 ),
89 ConfigEntry(
90 key=CONF_ENABLE_PERSONALIZED,
91 type=ConfigEntryType.BOOLEAN,
92 default_value=False,
93 category="recommendations",
94 ),
95 ConfigEntry(
96 key=CONF_ENABLE_GLOBAL_CHARTS,
97 type=ConfigEntryType.BOOLEAN,
98 default_value=False,
99 category="recommendations",
100 ),
101 ConfigEntry(
102 key=CONF_ENABLE_GENRE,
103 type=ConfigEntryType.BOOLEAN,
104 default_value=False,
105 category="recommendations",
106 ),
107 ConfigEntry(
108 key=CONF_ENABLE_GEO,
109 type=ConfigEntryType.BOOLEAN,
110 default_value=False,
111 category="recommendations",
112 ),
113 ConfigEntry(
114 key=CONF_GEO_COUNTRY,
115 type=ConfigEntryType.STRING,
116 default_value="Argentina",
117 options=[ConfigValueOption(country, title=country) for country in GEO_COUNTRIES],
118 category="recommendations",
119 ),
120 ConfigEntry(
121 key=CONF_ACTION_CLEAR_CACHE,
122 type=ConfigEntryType.ACTION,
123 action=CONF_ACTION_CLEAR_CACHE,
124 category="recommendations",
125 advanced=True,
126 required=False,
127 ),
128 )
129
130 async def handle_config_action(
131 self, action: str
132 ) -> tuple[ConfigEntry, ...] | ConfigActionResult | None:
133 """Handle a one-shot config action button press."""
134 if action == CONF_ACTION_CLEAR_CACHE:
135 await self.recommendations_manager.clear_cache()
136 self.mass.create_task(self._refresh_recommendations())
137 return None
138 return await super().handle_config_action(action)
139
140 async def handle_async_init(self) -> None:
141 """Handle async initialization of the provider."""
142 self.api = LastFMAPIClient(self)
143 self.recommendations_manager = LastFMRecommendationManager(self)
144
145 self._recommendation_folders: list[RecommendationFolder] = []
146
147 # Register recurring refresh task (runs every 6 hours).
148 self.mass.tasks.register_scheduled_task(
149 task_id=f"{REFRESH_TASK_ID}_{self.instance_id}",
150 name="Refresh Last.fm recommendations",
151 handler=self._refresh_recommendations,
152 schedule=TaskSchedule.hourly(every=6),
153 translation_key="refresh_lastfm_recommendations",
154 translation_owner=self.translation_owner,
155 )
156
157 # Populate on every startup so the UI isn't empty until the next scheduled refresh.
158 # Delayed 20s to let streaming providers finish loading first.
159 self.mass.call_later(
160 20,
161 self._refresh_recommendations,
162 task_id=f"{REFRESH_TASK_ID}_initial_{self.instance_id}",
163 )
164
165 async def unload(self, is_removed: bool = False) -> None:
166 """Unload the provider."""
167 self.mass.tasks.unregister_scheduled_task(
168 f"{REFRESH_TASK_ID}_{self.instance_id}",
169 clear_persisted_state=is_removed,
170 )
171
172 async def get_recommendations(self) -> list[RecommendationFolder]:
173 """Get this provider's available recommendation rows, without items."""
174 # rows come from the precomputed in-memory folders: no backend I/O
175 return [
176 RecommendationFolder(
177 item_id=folder.item_id,
178 provider=folder.provider,
179 name=folder.name,
180 translation_key=folder.translation_key,
181 translation_params=folder.translation_params,
182 icon=folder.icon,
183 subtitle=folder.subtitle,
184 )
185 for folder in self._recommendation_folders
186 ]
187
188 async def get_recommendation_items(
189 self, item_id: str
190 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
191 """
192 Get the items for a single recommendation row.
193
194 :param item_id: The item_id of the row, as returned by get_recommendations.
195 """
196 for folder in self._recommendation_folders:
197 if folder.item_id == item_id:
198 return folder.items
199 return UniqueList()
200
201 async def _refresh_recommendations(self) -> None:
202 """Rebuild recommendation folders."""
203 # Build into a local list and swap it in atomically at the end, so a slow,
204 # rate-limited rebuild keeps serving the previous generation's rows instead of
205 # returning empty for folders that haven't been rebuilt yet.
206 new_folders: list[RecommendationFolder] = []
207
208 try:
209 self.logger.info("Building Last.fm recommendations")
210 async for folder in self.recommendations_manager.build_recommendation_folders():
211 new_folders.append(folder)
212 self._recommendation_folders = new_folders
213 self.logger.info(
214 "Last.fm recommendations built (%d folders)",
215 len(self._recommendation_folders),
216 )
217 except (AuthenticationFailed, InvalidToken) as err:
218 self.logger.error(
219 "Last.fm authentication failed â check your API key in the provider settings: %s",
220 err,
221 )
222 except ResourceTemporarilyUnavailable as err:
223 self.logger.warning("Last.fm rate-limited the refresh, will retry later: %s", err)
224 except MusicAssistantError as err:
225 self.logger.warning("Failed to build recommendations: %s", err)
226
227 async def get_similar_artists(self, artist: Artist, limit: int = 25) -> list[Artist]:
228 """
229 Retrieve similar artists from Last.fm.
230
231 :param artist: The reference artist.
232 :param limit: Maximum number of similar artists to return.
233 """
234 artist_mbid = artist.get_external_id(ExternalID.MB_ARTIST)
235 similar_raw = await self.api.get_similar_artists(artist.name, artist_mbid, limit)
236 if not similar_raw:
237 return []
238
239 resolved = await asyncio.gather(
240 *[self.recommendations_manager.get_or_resolve_artist(raw) for raw in similar_raw]
241 )
242 return [a for a in resolved if a is not None]
243
244 async def get_similar_tracks(self, track: Track, limit: int = 25) -> list[Track]:
245 """
246 Retrieve similar tracks from Last.fm.
247
248 :param track: The reference track.
249 :param limit: Maximum number of similar tracks to return.
250 """
251 artist_name = track.artists[0].name if track.artists else "Unknown Artist"
252 track_mbid = track.get_external_id(ExternalID.MB_RECORDING)
253 similar_raw = await self.api.get_similar_tracks(artist_name, track.name, track_mbid, limit)
254 if not similar_raw:
255 return []
256
257 resolved = await asyncio.gather(
258 *[self.recommendations_manager.get_or_resolve_track(raw) for raw in similar_raw]
259 )
260 return [t for t in resolved if t is not None]
261
262 async def get_artist_toptracks(self, artist: Artist, limit: int = 25) -> list[Track]:
263 """
264 Retrieve an artist's top tracks from Last.fm.
265
266 :param artist: The reference artist.
267 :param limit: Maximum number of top tracks to return.
268 """
269 artist_mbid = artist.get_external_id(ExternalID.MB_ARTIST)
270 return await self._get_artist_toptracks(artist.name, artist_mbid, limit)
271
272 @use_cache(
273 CACHE_EXPIRATION_SECONDS,
274 category=CACHE_CATEGORY_RESOLVED_ITEMS,
275 allow_expired_cache=True,
276 )
277 async def _get_artist_toptracks(
278 self, artist_name: str, artist_mbid: str | None, limit: int
279 ) -> list[Track]:
280 """Fetch and resolve an artist's top tracks, keyed by name/mbid for caching."""
281 top_raw = await self.api.get_artist_top_tracks(artist_name, artist_mbid, limit)
282 if not top_raw:
283 return []
284
285 # Tolerate individual resolution failures (e.g. a rate-limited lookup) so one bad
286 # track can't sink the whole listing.
287 resolved = await asyncio.gather(
288 *[self.recommendations_manager.get_or_resolve_track(raw) for raw in top_raw],
289 return_exceptions=True,
290 )
291 tracks = [t for t in resolved if isinstance(t, Track)]
292 self.logger.debug(
293 "Resolved %d/%d top tracks to playable items for '%s'",
294 len(tracks),
295 len(top_raw),
296 artist_name,
297 )
298 return tracks
299