/
/
/
1"""Recommendation logic for Last.fm."""
2
3from __future__ import annotations
4
5import asyncio
6import random
7from collections.abc import AsyncIterator
8from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar
9
10from music_assistant_models.enums import ExternalID, MediaType
11from music_assistant_models.media_items import (
12 Album,
13 Artist,
14 RecommendationFolder,
15 Track,
16 UniqueList,
17)
18
19from music_assistant.constants import CONF_USERNAME, VARIOUS_ARTISTS_NAME
20from music_assistant.helpers.compare import compare_strings
21from music_assistant.helpers.datetime import utc, utc_timestamp
22from music_assistant.helpers.util import parse_title_and_version
23from music_assistant.providers.lastfm_recommendations.constants import (
24 CACHE_CATEGORY_RESOLVED_ITEMS,
25 CACHE_CATEGORY_TOP_GENRES,
26 CACHE_EXPIRATION_SECONDS,
27 CONF_ENABLE_GENRE,
28 CONF_ENABLE_GEO,
29 CONF_ENABLE_GLOBAL_CHARTS,
30 CONF_ENABLE_PERSONALIZED,
31 CONF_GEO_COUNTRY,
32 GENRE_ARTISTS_LIMIT,
33 GENRE_ARTISTS_PERIOD,
34 LIBRARY_MATCH_SCAN_LIMIT,
35 RECENT_PLAYS_SCAN_LIMIT,
36 RECENT_PLAYS_WINDOW_DAYS,
37 RESOLUTION_BUFFER_LARGE,
38 RESOLUTION_BUFFER_SMALL,
39 SIMILAR_ITEMS_BUFFER,
40 SIMILAR_ITEMS_PER_SEED,
41 SIMILAR_TRACKS_BUFFER,
42 TAGS_PER_ARTIST,
43 TARGET_ITEM_COUNT,
44 TOP_ARTISTS_LIMIT,
45 TOP_GENRES_CACHE_EXPIRATION_SECONDS,
46 TOP_GENRES_LIMIT,
47 TOP_ITEMS_TO_TAKE,
48 TOP_TRACKS_LIMIT,
49)
50from music_assistant.providers.lastfm_recommendations.parsers import (
51 parse_album,
52 parse_artist,
53 parse_track,
54)
55
56if TYPE_CHECKING:
57 import logging
58
59 from music_assistant.providers.lastfm_recommendations import LastFMRecommendationsProvider
60
61_MediaItemT = TypeVar("_MediaItemT", Artist, Album, Track)
62
63
64class _SeedTrack(NamedTuple):
65 """A recently played track used to seed similar-artist/track lookups."""
66
67 artist: str
68 name: str
69
70
71class LastFMRecommendationManager:
72 """Manages Last.fm recommendations."""
73
74 def __init__(self, provider: LastFMRecommendationsProvider) -> None:
75 """
76 Initialize recommendation manager.
77
78 :param provider: The Last.fm recommendations provider instance.
79 """
80 self.provider = provider
81 self.api = provider.api
82 self.mass = provider.mass
83
84 # Resolved items keyed by MBID (preferred) or name to avoid re-resolving.
85 self._resolved_cache: dict[str, Artist | Album | Track] = {}
86
87 @property
88 def logger(self) -> logging.Logger:
89 """Return the provider's active logger."""
90 return self.provider.logger
91
92 async def clear_cache(self) -> None:
93 """Clear in-memory and persistent recommendation caches."""
94 self._resolved_cache.clear()
95
96 await self.mass.cache.clear(
97 category_filter=CACHE_CATEGORY_RESOLVED_ITEMS,
98 provider_filter=self.provider.instance_id,
99 )
100 await self.mass.cache.clear(
101 category_filter=CACHE_CATEGORY_TOP_GENRES,
102 provider_filter=self.provider.instance_id,
103 )
104
105 self.provider._recommendation_folders.clear()
106
107 self.logger.info("Cleared all recommendation caches (in-memory and persistent)")
108
109 async def build_recommendation_folders(self) -> AsyncIterator[RecommendationFolder]:
110 """Yield recommendation folders across all enabled categories."""
111 async for folder in self._yield_and_count(
112 self._get_personalized_recommendations(), "personalized"
113 ):
114 yield folder
115
116 async for folder in self._yield_and_count(self._get_global_recommendations(), "global"):
117 yield folder
118
119 async for folder in self._yield_and_count(
120 self._get_genre_based_recommendations(), "genre-based"
121 ):
122 yield folder
123
124 async for folder in self._yield_and_count(
125 self._get_geo_based_recommendations(), "geography-based"
126 ):
127 yield folder
128
129 async def _yield_and_count(
130 self, source: AsyncIterator[RecommendationFolder], category_label: str
131 ) -> AsyncIterator[RecommendationFolder]:
132 """Yield folders from a single category."""
133 count = 0
134 async for folder in source:
135 count += 1
136 yield folder
137 if count:
138 self.logger.debug("Added %d %s recommendation folder(s)", count, category_label)
139
140 async def _is_in_library(self, item_data: dict[str, Any], media_type: MediaType) -> bool:
141 """
142 Return True if the Last.fm item already exists in the MA library.
143
144 :param item_data: Raw Last.fm item data (artist, album, or track dict).
145 :param media_type: Type of media item to check.
146 """
147 name = item_data.get("name", "")
148 # MBID lookup is the most reliable; fall back to name search for items without MBID.
149 mbid = item_data.get("mbid")
150 if mbid:
151 if media_type == MediaType.ARTIST:
152 if await self.mass.music.artists.get_library_item_by_external_id(
153 mbid, ExternalID.MB_ARTIST
154 ):
155 self.logger.debug("Filtered artist '%s' (MBID match: %s)", name, mbid)
156 return True
157 elif media_type == MediaType.ALBUM:
158 if await self.mass.music.albums.get_library_item_by_external_id(
159 mbid, ExternalID.MB_ALBUM
160 ):
161 self.logger.debug("Filtered album '%s' (MBID match: %s)", name, mbid)
162 return True
163 elif media_type == MediaType.TRACK:
164 if await self.mass.music.tracks.get_library_item_by_external_id(
165 mbid, ExternalID.MB_RECORDING
166 ):
167 self.logger.debug("Filtered track '%s' (MBID match: %s)", name, mbid)
168 return True
169
170 # Library search is fuzzy and always returns a best-effort row, so verify the hit's
171 # name actually matches before treating the item as owned. Lenient (strict=False) so
172 # tagging variants still count, erring towards filtering over showing owned items.
173 if media_type == MediaType.ARTIST:
174 if name:
175 artist_results = await self.mass.music.artists.library_items(
176 search=name, limit=LIBRARY_MATCH_SCAN_LIMIT, summary=False
177 )
178 artist_match = next(
179 (a for a in artist_results if compare_strings(name, a.name, strict=False)),
180 None,
181 )
182 if artist_match:
183 self.logger.debug(
184 "Filtered artist '%s' (name match: '%s')", name, artist_match.name
185 )
186 return True
187
188 elif media_type == MediaType.ALBUM:
189 if name:
190 album_results = await self.mass.music.albums.library_items(
191 search=name, limit=LIBRARY_MATCH_SCAN_LIMIT, summary=False
192 )
193 album_match = next(
194 (a for a in album_results if compare_strings(name, a.name, strict=False)),
195 None,
196 )
197 if album_match:
198 self.logger.debug(
199 "Filtered album '%s' (name match: '%s')", name, album_match.name
200 )
201 return True
202
203 elif media_type == MediaType.TRACK:
204 artist_info = item_data.get("artist", {})
205 artist_name = (
206 artist_info if isinstance(artist_info, str) else artist_info.get("name", "")
207 )
208 if name and artist_name:
209 # Last.fm track names carry scrobbled version suffixes ("- 2006 Remaster");
210 # strip those so variants of an owned track still match
211 clean_name, _ = parse_title_and_version(name, strip_for_search=True)
212 # "Artist - Title" format so the tracks controller searches both fields
213 search_query = f"{artist_name} - {clean_name}"
214 track_results = await self.mass.music.tracks.library_items(
215 search=search_query, limit=LIBRARY_MATCH_SCAN_LIMIT, summary=False
216 )
217 # Match both title and artist; a title-only check would treat a same-named
218 # track by a different artist as owned. Differing recording MBIDs identify
219 # genuinely different tracks, so those never count as a match.
220 track_match = next(
221 (
222 track
223 for track in track_results
224 if compare_strings(clean_name, track.name, strict=False)
225 and any(
226 compare_strings(artist_name, track_artist.name, strict=False)
227 for track_artist in track.artists
228 )
229 and not (mbid and track.mbid and track.mbid != mbid)
230 ),
231 None,
232 )
233 if track_match:
234 self.logger.debug(
235 "Filtered track '%s - %s' (name match: '%s')",
236 artist_name,
237 name,
238 track_match.name,
239 )
240 return True
241
242 return False
243
244 @staticmethod
245 def _exclude_owned(items: list[_MediaItemT]) -> list[_MediaItemT]:
246 """
247 Drop items that resolved to the user's own library copy from a discovery row.
248
249 :param items: Resolved discovery items to filter.
250 """
251 # Resolution returns the library copy when a recommendation matches on ISRC/MBID,
252 # which the cheaper pre-filter can miss; exclude those so Discover rows stay discovery.
253 return [item for item in items if item.provider != "library"]
254
255 def _sample_items(
256 self, items: list[dict[str, Any]], seed_suffix: str, target_count: int = TARGET_ITEM_COUNT
257 ) -> list[dict[str, Any]]:
258 """
259 Sample items using a 'top N + random remainder' strategy with an hourly seed.
260
261 :param items: List of items to sample from (already filtered).
262 :param seed_suffix: Unique suffix for random seed (to vary between recommendation types).
263 :param target_count: Target number of items to return.
264 """
265 if len(items) <= target_count:
266 return items
267
268 top_items = items[:TOP_ITEMS_TO_TAKE]
269
270 remaining = items[TOP_ITEMS_TO_TAKE:]
271 random_count = target_count - TOP_ITEMS_TO_TAKE
272
273 # Hourly seed keeps the sampled remainder stable within the hour and rotates it each hour.
274 now = utc()
275 seed = f"{now.date().isoformat()}_{now.hour}_{seed_suffix}"
276 rng = random.Random(seed)
277 random_items = rng.sample(remaining, min(random_count, len(remaining)))
278
279 return top_items + random_items
280
281 async def get_or_resolve_artist(self, lastfm_artist: dict[str, Any]) -> Artist | None:
282 """
283 Return an Artist from cache (in-memory or persistent) or resolve and cache it.
284
285 :param lastfm_artist: Raw Last.fm artist dict.
286 """
287 cache_key = lastfm_artist.get("mbid") or lastfm_artist.get("name", "")
288 if not cache_key:
289 return None
290
291 if cache_key in self._resolved_cache:
292 cached = self._resolved_cache[cache_key]
293 if isinstance(cached, Artist):
294 return cached
295
296 persistent_cache_key = f"artist_{cache_key}"
297 cached_artist = await self.mass.cache.get(
298 key=persistent_cache_key,
299 category=CACHE_CATEGORY_RESOLVED_ITEMS,
300 provider=self.provider.instance_id,
301 base_class=Artist,
302 )
303 if isinstance(cached_artist, Artist):
304 self._resolved_cache[cache_key] = cached_artist
305 return cached_artist
306
307 artist = await parse_artist(lastfm_artist, self.mass, self.provider.instance_id)
308 if artist:
309 self._resolved_cache[cache_key] = artist
310 await self.mass.cache.set(
311 persistent_cache_key,
312 artist.to_dict(),
313 category=CACHE_CATEGORY_RESOLVED_ITEMS,
314 provider=self.provider.instance_id,
315 expiration=CACHE_EXPIRATION_SECONDS,
316 )
317 return artist
318
319 async def get_or_resolve_track(self, lastfm_track: dict[str, Any]) -> Track | None:
320 """
321 Return a Track from cache (in-memory or persistent) or resolve and cache it.
322
323 :param lastfm_track: Raw Last.fm track dict.
324 """
325 cache_key = lastfm_track.get("mbid")
326 if not cache_key:
327 artist_data = lastfm_track.get("artist", {})
328 artist_name = (
329 artist_data if isinstance(artist_data, str) else artist_data.get("name", "")
330 )
331 track_name = lastfm_track.get("name", "")
332 cache_key = f"{artist_name}_{track_name}" if artist_name and track_name else ""
333
334 if not cache_key:
335 return None
336
337 if cache_key in self._resolved_cache:
338 cached = self._resolved_cache[cache_key]
339 if isinstance(cached, Track):
340 return cached
341
342 persistent_cache_key = f"track_{cache_key}"
343 cached_track = await self.mass.cache.get(
344 key=persistent_cache_key,
345 category=CACHE_CATEGORY_RESOLVED_ITEMS,
346 provider=self.provider.instance_id,
347 base_class=Track,
348 )
349 if isinstance(cached_track, Track):
350 self._resolved_cache[cache_key] = cached_track
351 return cached_track
352
353 track = await parse_track(lastfm_track, self.mass, self.provider.instance_id)
354 if track:
355 self._resolved_cache[cache_key] = track
356 await self.mass.cache.set(
357 persistent_cache_key,
358 track.to_dict(),
359 category=CACHE_CATEGORY_RESOLVED_ITEMS,
360 provider=self.provider.instance_id,
361 expiration=CACHE_EXPIRATION_SECONDS,
362 )
363 return track
364
365 async def _get_or_resolve_album(self, lastfm_album: dict[str, Any]) -> Album | None:
366 """
367 Return an Album from cache (in-memory or persistent) or resolve and cache it.
368
369 :param lastfm_album: Raw Last.fm album dict.
370 """
371 cache_key = lastfm_album.get("mbid")
372 if not cache_key:
373 artist_data = lastfm_album.get("artist", {})
374 artist_name = (
375 artist_data if isinstance(artist_data, str) else artist_data.get("name", "")
376 )
377 album_name = lastfm_album.get("name", "")
378 cache_key = f"{artist_name}_{album_name}" if artist_name and album_name else ""
379
380 if not cache_key:
381 return None
382
383 if cache_key in self._resolved_cache:
384 cached = self._resolved_cache[cache_key]
385 if isinstance(cached, Album):
386 return cached
387
388 persistent_cache_key = f"album_{cache_key}"
389 cached_album = await self.mass.cache.get(
390 key=persistent_cache_key,
391 category=CACHE_CATEGORY_RESOLVED_ITEMS,
392 provider=self.provider.instance_id,
393 base_class=Album,
394 )
395 if isinstance(cached_album, Album):
396 self._resolved_cache[cache_key] = cached_album
397 return cached_album
398
399 album = await parse_album(lastfm_album, self.mass, self.provider.instance_id)
400 if album:
401 self._resolved_cache[cache_key] = album
402 await self.mass.cache.set(
403 persistent_cache_key,
404 album.to_dict(),
405 category=CACHE_CATEGORY_RESOLVED_ITEMS,
406 provider=self.provider.instance_id,
407 expiration=CACHE_EXPIRATION_SECONDS,
408 )
409 return album
410
411 async def _get_personalized_recommendations(self) -> AsyncIterator[RecommendationFolder]:
412 """Yield personalized recommendation folders based on the user's listening history."""
413 if not self.provider.config.get_value(CONF_ENABLE_PERSONALIZED):
414 return
415
416 # Both seed rows derive from the same recent play events, so scan the playlog once.
417 recent_plays = await self._get_recent_play_seeds()
418
419 top_artists = self._rank_seed_artists(recent_plays)
420
421 if top_artists:
422 similar_artists = await self._get_similar_artists_from_seeds(top_artists)
423
424 if similar_artists:
425 yield RecommendationFolder(
426 item_id=f"{self.provider.instance_id}_similar_artists",
427 name="Discover Similar Artists",
428 translation_key="discover_similar_artists",
429 translation_params=[str(len(top_artists))],
430 provider=self.provider.instance_id,
431 items=UniqueList(similar_artists[:TARGET_ITEM_COUNT]),
432 subtitle=f"Based on your top {len(top_artists)} artists",
433 icon="mdi-account-music-outline",
434 )
435
436 top_tracks = self._rank_seed_tracks(recent_plays)[:TOP_TRACKS_LIMIT]
437
438 if top_tracks:
439 similar_tracks = await self._get_similar_tracks_from_seeds(top_tracks)
440
441 if similar_tracks:
442 yield RecommendationFolder(
443 item_id=f"{self.provider.instance_id}_similar_tracks",
444 name="Discover Similar Tracks",
445 translation_key="discover_similar_tracks",
446 translation_params=[str(len(top_tracks))],
447 provider=self.provider.instance_id,
448 items=UniqueList(similar_tracks[:TARGET_ITEM_COUNT]),
449 subtitle=f"Based on your top {len(top_tracks)} tracks",
450 icon="mdi-music-note-outline",
451 )
452
453 async def _get_global_recommendations(self) -> AsyncIterator[RecommendationFolder]:
454 """Yield global chart recommendation folders (worldwide top artists and tracks)."""
455 if not self.provider.config.get_value(CONF_ENABLE_GLOBAL_CHARTS):
456 return
457
458 # Over-fetch so deduplication and resolution failures still leave TARGET_ITEM_COUNT.
459 top_artists_raw = await self.api.get_chart_top_artists(limit=RESOLUTION_BUFFER_SMALL)
460 if top_artists_raw:
461 resolved_artists = await asyncio.gather(
462 *[self.get_or_resolve_artist(artist_data) for artist_data in top_artists_raw]
463 )
464 all_resolved = [a for a in resolved_artists if a is not None]
465 top_artists = list(UniqueList(all_resolved))[:TARGET_ITEM_COUNT]
466
467 if top_artists:
468 yield RecommendationFolder(
469 item_id=f"{self.provider.instance_id}_chart_top_artists",
470 name="Global Top Artists",
471 translation_key="global_top_artists",
472 provider=self.provider.instance_id,
473 items=UniqueList(top_artists),
474 subtitle="Most popular artists worldwide",
475 icon="mdi-chart-line",
476 )
477
478 top_tracks_raw = await self.api.get_chart_top_tracks(limit=RESOLUTION_BUFFER_SMALL)
479 if top_tracks_raw:
480 resolved_tracks = await asyncio.gather(
481 *[self.get_or_resolve_track(track_data) for track_data in top_tracks_raw]
482 )
483 all_resolved_tracks = [t for t in resolved_tracks if t is not None]
484 top_tracks = list(UniqueList(all_resolved_tracks))[:TARGET_ITEM_COUNT]
485
486 if top_tracks:
487 yield RecommendationFolder(
488 item_id=f"{self.provider.instance_id}_chart_top_tracks",
489 name="Global Top Tracks",
490 translation_key="global_top_tracks",
491 provider=self.provider.instance_id,
492 items=UniqueList(top_tracks),
493 subtitle="Most popular tracks worldwide",
494 icon="mdi-chart-box",
495 )
496
497 async def _get_genre_based_recommendations(self) -> AsyncIterator[RecommendationFolder]:
498 """
499 Yield genre-based recommendation folders derived from the user's top genres.
500
501 Requires a username to be configured.
502 """
503 if not self.provider.config.get_value(CONF_ENABLE_GENRE):
504 return
505
506 top_genres = await self._get_top_genres()
507 if not top_genres:
508 return
509
510 # cycle through the user's top genres day by day so the genre rows vary
511 day_index = utc().date().toordinal()
512 tag_name = top_genres[day_index % len(top_genres)]
513
514 # Over-fetch so there's enough left after library filtering and resolution failures.
515 genre_artists_raw = await self.api.get_tag_top_artists(
516 tag_name, limit=RESOLUTION_BUFFER_LARGE
517 )
518 if genre_artists_raw:
519 # Drop items already in the library using a cheap DB lookup, before the
520 # expensive provider resolution step.
521 non_library_artists_raw = [
522 artist_data
523 for artist_data in genre_artists_raw
524 if not await self._is_in_library(artist_data, MediaType.ARTIST)
525 ]
526
527 sampled_artists_raw = self._sample_items(
528 non_library_artists_raw,
529 seed_suffix="genre_artists",
530 target_count=RESOLUTION_BUFFER_SMALL,
531 )
532
533 resolved_artists = await asyncio.gather(
534 *[self.get_or_resolve_artist(artist_data) for artist_data in sampled_artists_raw]
535 )
536 all_resolved = self._exclude_owned([a for a in resolved_artists if a is not None])
537 genre_artists = list(UniqueList(all_resolved))[:TARGET_ITEM_COUNT]
538
539 if genre_artists:
540 yield RecommendationFolder(
541 item_id=f"{self.provider.instance_id}_genre_artists",
542 name=f"Discover {tag_name.title()} Artists",
543 translation_key="genre_artists",
544 translation_params=[tag_name.title()],
545 provider=self.provider.instance_id,
546 items=UniqueList(genre_artists),
547 subtitle="Top artists in your top genres",
548 icon="mdi-account-music",
549 )
550
551 genre_albums_raw = await self.api.get_tag_top_albums(
552 tag_name, limit=RESOLUTION_BUFFER_LARGE
553 )
554 if genre_albums_raw:
555 non_library_albums_raw = [
556 album_data
557 for album_data in genre_albums_raw
558 if not await self._is_in_library(album_data, MediaType.ALBUM)
559 ]
560
561 sampled_albums_raw = self._sample_items(
562 non_library_albums_raw,
563 seed_suffix="genre_albums",
564 target_count=RESOLUTION_BUFFER_SMALL,
565 )
566
567 resolved_albums = await asyncio.gather(
568 *[self._get_or_resolve_album(album_data) for album_data in sampled_albums_raw]
569 )
570 all_resolved_albums = self._exclude_owned(
571 [album for album in resolved_albums if album is not None]
572 )
573 genre_albums = list(UniqueList(all_resolved_albums))[:TARGET_ITEM_COUNT]
574
575 if genre_albums:
576 yield RecommendationFolder(
577 item_id=f"{self.provider.instance_id}_genre_albums",
578 name=f"Discover {tag_name.title()} Albums",
579 translation_key="genre_albums",
580 translation_params=[tag_name.title()],
581 provider=self.provider.instance_id,
582 items=UniqueList(genre_albums),
583 subtitle="Top albums in your top genres",
584 icon="mdi-album",
585 )
586
587 genre_tracks_raw = await self.api.get_tag_top_tracks(
588 tag_name, limit=RESOLUTION_BUFFER_LARGE
589 )
590 if genre_tracks_raw:
591 non_library_tracks_raw = [
592 track_data
593 for track_data in genre_tracks_raw
594 if not await self._is_in_library(track_data, MediaType.TRACK)
595 ]
596
597 sampled_tracks_raw = self._sample_items(
598 non_library_tracks_raw,
599 seed_suffix="genre_tracks",
600 target_count=RESOLUTION_BUFFER_SMALL,
601 )
602
603 resolved_tracks = await asyncio.gather(
604 *[self.get_or_resolve_track(track_data) for track_data in sampled_tracks_raw]
605 )
606 all_resolved_genre_tracks = self._exclude_owned(
607 [track for track in resolved_tracks if track is not None]
608 )
609 genre_tracks = list(UniqueList(all_resolved_genre_tracks))[:TARGET_ITEM_COUNT]
610
611 if genre_tracks:
612 yield RecommendationFolder(
613 item_id=f"{self.provider.instance_id}_genre_tracks",
614 name=f"Discover {tag_name.title()} Tracks",
615 translation_key="genre_tracks",
616 translation_params=[tag_name.title()],
617 provider=self.provider.instance_id,
618 items=UniqueList(genre_tracks),
619 subtitle="Top tracks in your top genres",
620 icon="mdi-music",
621 )
622
623 async def _get_top_genres(self) -> list[str]:
624 """Return the user's top genres, most prominent first, derived from their listening."""
625 username = self.provider.config.get_value(CONF_USERNAME)
626 if not username or not isinstance(username, str):
627 return []
628
629 # Deriving genres costs a request per top artist, so cache the result.
630 cached = await self.mass.cache.get(
631 key="top_genres",
632 category=CACHE_CATEGORY_TOP_GENRES,
633 provider=self.provider.instance_id,
634 )
635 if isinstance(cached, list):
636 return cached
637
638 top_artists = await self.api.get_user_top_artists(
639 username, period=GENRE_ARTISTS_PERIOD, limit=GENRE_ARTISTS_LIMIT
640 )
641 if not top_artists:
642 return []
643
644 # Last.fm has no genre data, so an artist's top community tags stand in for its genre.
645 # The user's genres are those tags aggregated across their most played artists, weighted
646 # by how much they play each (playcount) and how strongly each tag applies (count 0-100).
647 tag_lists = await asyncio.gather(
648 *[
649 self.api.get_artist_top_tags(
650 artist.get("name", ""), artist.get("mbid"), limit=TAGS_PER_ARTIST
651 )
652 for artist in top_artists
653 ]
654 )
655
656 scores: dict[str, float] = {}
657 display_names: dict[str, str] = {}
658 for artist, tags in zip(top_artists, tag_lists, strict=True):
659 try:
660 artist_weight = float(artist.get("playcount", 0))
661 except TypeError, ValueError:
662 continue
663 if artist_weight <= 0:
664 continue
665 for tag in tags:
666 name = tag.get("name", "")
667 if not name:
668 continue
669 try:
670 tag_count = float(tag.get("count", 0))
671 except TypeError, ValueError:
672 continue
673 key = name.lower()
674 scores[key] = scores.get(key, 0.0) + artist_weight * tag_count / 100
675 display_names.setdefault(key, name)
676
677 ranked = sorted(scores, key=lambda key: scores[key], reverse=True)
678 top_genres = [display_names[key] for key in ranked[:TOP_GENRES_LIMIT]]
679
680 if top_genres:
681 await self.mass.cache.set(
682 "top_genres",
683 top_genres,
684 category=CACHE_CATEGORY_TOP_GENRES,
685 provider=self.provider.instance_id,
686 expiration=TOP_GENRES_CACHE_EXPIRATION_SECONDS,
687 )
688 return top_genres
689
690 async def _get_geo_based_recommendations(self) -> AsyncIterator[RecommendationFolder]:
691 """Yield geography-based recommendation folders for the configured country."""
692 if not self.provider.config.get_value(CONF_ENABLE_GEO):
693 return
694
695 country = self.provider.config.get_value(CONF_GEO_COUNTRY)
696 if not country or not isinstance(country, str):
697 return
698
699 geo_artists_raw = await self.api.get_geo_top_artists(country, limit=RESOLUTION_BUFFER_SMALL)
700 if geo_artists_raw:
701 resolved_artists = await asyncio.gather(
702 *[self.get_or_resolve_artist(artist_data) for artist_data in geo_artists_raw]
703 )
704 all_resolved = [artist for artist in resolved_artists if artist is not None]
705 geo_artists = list(UniqueList(all_resolved))[:TARGET_ITEM_COUNT]
706
707 if geo_artists:
708 yield RecommendationFolder(
709 item_id=f"{self.provider.instance_id}_geo_artists",
710 name=f"Top artists for {country}",
711 translation_key="geo_artists",
712 translation_params=[country],
713 provider=self.provider.instance_id,
714 items=UniqueList(geo_artists),
715 subtitle=f"Most popular artists in {country}",
716 icon="mdi-earth",
717 )
718
719 geo_tracks_raw = await self.api.get_geo_top_tracks(country, limit=RESOLUTION_BUFFER_SMALL)
720 if geo_tracks_raw:
721 resolved_tracks = await asyncio.gather(
722 *[self.get_or_resolve_track(track_data) for track_data in geo_tracks_raw]
723 )
724 all_resolved_geo_tracks = [track for track in resolved_tracks if track is not None]
725 geo_tracks = list(UniqueList(all_resolved_geo_tracks))[:TARGET_ITEM_COUNT]
726
727 if geo_tracks:
728 yield RecommendationFolder(
729 item_id=f"{self.provider.instance_id}_geo_tracks",
730 name=f"Top tracks for {country}",
731 translation_key="geo_tracks",
732 translation_params=[country],
733 provider=self.provider.instance_id,
734 items=UniqueList(geo_tracks),
735 subtitle=f"Most popular tracks in {country}",
736 icon="mdi-earth",
737 )
738
739 async def _get_recent_play_seeds(self) -> list[_SeedTrack]:
740 """
741 Return artist/track pairs for recently played tracks, most recent first.
742
743 Plays without a recorded artist are omitted.
744 """
745 # Rank by recent plays, not lifetime play_count, so seeds follow current listening.
746 # Artists are stored on the playlog at play time, so no provider lookup is needed.
747 cutoff = int(utc_timestamp()) - RECENT_PLAYS_WINDOW_DAYS * 24 * 60 * 60
748 plays = await self.mass.music.recently_played_tracks(
749 limit=RECENT_PLAYS_SCAN_LIMIT,
750 played_after_timestamp=cutoff,
751 )
752 # Skip rows predating the playlog artists column; they regain an artist on the next play.
753 # Various Artists is a compilation placeholder, not a usable seed.
754 return [
755 _SeedTrack(artist=play.artists[0].name, name=play.track.name)
756 for play in plays
757 if play.artists and play.artists[0].name != VARIOUS_ARTISTS_NAME
758 ]
759
760 def _rank_seed_tracks(self, recent_plays: list[_SeedTrack]) -> list[_SeedTrack]:
761 """
762 Return distinct recently played tracks, most recent first.
763
764 :param recent_plays: Recently played artist/track pairs, most recent first.
765 """
766 seen: set[tuple[str, str]] = set()
767 unique: list[_SeedTrack] = []
768 for seed in recent_plays:
769 key = (seed.artist.lower(), seed.name.lower())
770 if key not in seen:
771 seen.add(key)
772 unique.append(seed)
773 return unique
774
775 def _rank_seed_artists(self, recent_plays: list[_SeedTrack]) -> list[str]:
776 """
777 Rank artists by how many of their tracks were recently played, most first.
778
779 :param recent_plays: Recently played artist/track pairs, most recent first.
780 """
781 # The playlog has no per-play counts, so rank by distinct recently played tracks per
782 # artist; ties fall to the more recently played artist via the most-recent-first order.
783 counts: dict[str, int] = {}
784 names_by_key: dict[str, str] = {}
785 order: list[str] = []
786 for seed in self._rank_seed_tracks(recent_plays):
787 key = seed.artist.lower()
788 if key not in counts:
789 order.append(key)
790 names_by_key[key] = seed.artist
791 counts[key] = counts[key] + 1 if key in counts else 1
792
793 order.sort(key=lambda key: counts[key], reverse=True)
794 return [names_by_key[key] for key in order[:TOP_ARTISTS_LIMIT]]
795
796 async def _get_similar_artists_from_seeds(self, seed_artists: list[str]) -> list[Artist]:
797 """
798 Return resolved artists similar to the given seed artists.
799
800 :param seed_artists: Names of the user's recently played artists.
801 """
802 all_similar: list[dict[str, Any]] = []
803
804 # Seed names are tracked so seeds don't appear in their own recommendations.
805 seed_names = {name.lower() for name in seed_artists}
806
807 similar_lists = await asyncio.gather(
808 *[
809 self.api.get_similar_artists(
810 artist_name=name,
811 artist_mbid=None,
812 limit=SIMILAR_ITEMS_PER_SEED,
813 )
814 for name in seed_artists
815 ]
816 )
817 for similar in similar_lists:
818 all_similar.extend(similar)
819
820 # Deduplicate by MBID and by name: Last.fm sometimes returns the same artist twice,
821 # once with an MBID and once without.
822 seen_mbids = set()
823 seen_names = set()
824 unique_similar: list[dict[str, Any]] = []
825 for artist_data in all_similar:
826 mbid = artist_data.get("mbid")
827 name = artist_data.get("name", "").lower()
828
829 if name and name in seed_names:
830 continue
831
832 if mbid and mbid in seen_mbids:
833 continue
834 if name and name in seen_names:
835 continue
836
837 unique_similar.append(artist_data)
838 if mbid:
839 seen_mbids.add(mbid)
840 if name:
841 seen_names.add(name)
842
843 unique_similar.sort(key=lambda x: float(x.get("match", 0)), reverse=True)
844
845 resolved_artists = await asyncio.gather(
846 *[
847 self.get_or_resolve_artist(artist_data)
848 for artist_data in unique_similar[:SIMILAR_ITEMS_BUFFER]
849 ]
850 )
851 return self._exclude_owned([artist for artist in resolved_artists if artist is not None])
852
853 async def _get_similar_tracks_from_seeds(self, seed_tracks: list[_SeedTrack]) -> list[Track]:
854 """
855 Return resolved tracks similar to the given seed tracks.
856
857 :param seed_tracks: The user's recently played artist/track pairs.
858 """
859 all_similar: list[dict[str, Any]] = []
860
861 # Seed names are tracked so seeds don't appear in their own recommendations.
862 seed_name_keys = {f"{seed.artist}_{seed.name}".lower() for seed in seed_tracks}
863
864 similar_lists = await asyncio.gather(
865 *[
866 self.api.get_similar_tracks(
867 artist_name=seed.artist,
868 track_name=seed.name,
869 track_mbid=None,
870 limit=SIMILAR_ITEMS_PER_SEED,
871 )
872 for seed in seed_tracks
873 ]
874 )
875 for similar in similar_lists:
876 all_similar.extend(similar)
877
878 # Deduplicate by MBID and by artist+name: Last.fm sometimes returns the same track
879 # twice, once with a MBID and once without.
880 seen_mbids = set()
881 seen_names = set()
882 unique_similar: list[dict[str, Any]] = []
883 for track_data in all_similar:
884 mbid = track_data.get("mbid")
885
886 artist_info = track_data.get("artist", {})
887 if isinstance(artist_info, str):
888 artist_name = artist_info
889 else:
890 artist_name = artist_info.get("name", "")
891 track_name = track_data.get("name", "")
892 name_key = f"{artist_name}_{track_name}".lower() if artist_name and track_name else ""
893
894 if name_key and name_key in seed_name_keys:
895 continue
896
897 if mbid and mbid in seen_mbids:
898 continue
899 if name_key and name_key in seen_names:
900 continue
901
902 unique_similar.append(track_data)
903 if mbid:
904 seen_mbids.add(mbid)
905 if name_key:
906 seen_names.add(name_key)
907
908 unique_similar.sort(key=lambda x: float(x.get("match", 0)), reverse=True)
909
910 # Resolve a small buffer beyond the target so resolution failures and owned-copy
911 # exclusion still leave enough to fill the row, while capping provider searches.
912 top_tracks_data = unique_similar[:SIMILAR_TRACKS_BUFFER]
913
914 resolved_tracks = await asyncio.gather(
915 *[self.get_or_resolve_track(track_data) for track_data in top_tracks_data]
916 )
917 return self._exclude_owned([track for track in resolved_tracks if track is not None])
918