/
/
/
1"""
2Rich metadata enrichment for the Metadata Controller.
3
4Provides the MetadataEnrichmentMixin, mixed into the MetaDataController, with the
5per-mediatype updaters that merge metadata from the music and metadata providers
6into library items (artists, albums, tracks, playlists, audiobooks and podcasts).
7"""
8
9from __future__ import annotations
10
11import asyncio
12from contextlib import suppress
13from dataclasses import replace
14from time import time
15from typing import TYPE_CHECKING, cast
16
17from music_assistant_models.enums import AlbumType, MediaType, ProviderFeature
18from music_assistant_models.errors import MediaNotFoundError, MusicAssistantError
19from music_assistant_models.helpers import get_global_cache_value
20from music_assistant_models.media_items import Album, Artist, MediaItemImage, Track
21
22from music_assistant.constants import VARIOUS_ARTISTS_MBID, VARIOUS_ARTISTS_NAME
23from music_assistant.helpers.compare import compare_strings
24from music_assistant.models.music_provider import MusicProvider
25
26from .constants import CONF_ENABLE_ONLINE_METADATA, CONF_PREFER_LOCAL_GENRES, REFRESH_INTERVAL
27
28if TYPE_CHECKING:
29 import logging
30 from collections.abc import Sequence
31
32 from music_assistant_models.config_entries import CoreConfig
33 from music_assistant_models.media_items import Audiobook, Playlist, Podcast
34 from music_assistant_models.unique_list import UniqueList
35
36 from music_assistant import MusicAssistant
37 from music_assistant.models.metadata_provider import MetadataProvider
38 from music_assistant.providers.musicbrainz import MusicbrainzProvider
39
40
41class MetadataEnrichmentMixin:
42 """
43 Rich metadata enrichment functionality for the MetaDataController.
44
45 Expects to be mixed with a class providing ``mass``, ``logger``, ``config``,
46 the ``providers`` and ``preferred_language`` properties, the
47 ``create_collage_image`` method and the ``_collage_images_dir`` attribute.
48 """
49
50 if TYPE_CHECKING:
51 mass: MusicAssistant
52 logger: logging.Logger
53 config: CoreConfig
54 _collage_images_dir: str
55
56 @property
57 def preferred_language(self) -> str: ... # noqa: D102
58
59 @property
60 def providers(self) -> list[MetadataProvider]: ... # noqa: D102
61
62 async def create_collage_image( # noqa: D102
63 self,
64 images: list[MediaItemImage],
65 filename: str,
66 fanart: bool = False,
67 ) -> MediaItemImage | None: ...
68
69 async def _update_artist_metadata(self, artist: Artist, force_refresh: bool = False) -> None:
70 """Get/update rich metadata for an artist."""
71 # collect metadata from all (online) music + metadata providers
72 # NOTE: we only do/allow this every REFRESH_INTERVAL
73 needs_refresh = (time() - (artist.metadata.last_refresh or 0)) > REFRESH_INTERVAL
74 if not (force_refresh or needs_refresh):
75 return
76
77 self.logger.debug("Updating metadata for Artist %s", artist.name)
78 unique_keys: set[str] = set()
79
80 # The bio is re-derived from the providers on every refresh. Each provider's
81 # description is collected as a (language, text) candidate and excluded from the
82 # field merge; _select_description picks the winner below. Candidates are appended
83 # in priority order: music providers first, then metadata providers (TADB, Wikipedia).
84 prev_description = artist.metadata.description
85 prev_description_language = artist.metadata.description_language
86 description_candidates: list[tuple[str | None, str]] = []
87
88 # collect (local) metadata from all local providers
89 local_provs = get_global_cache_value("non_streaming_providers")
90 if TYPE_CHECKING:
91 local_provs = cast("set[str]", local_provs)
92
93 # collect metadata from all [music] providers
94 # note that we sort the providers by priority so that we always
95 # prefer local providers over online providers
96 for prov_mapping in sorted(
97 artist.provider_mappings, key=lambda x: x.priority, reverse=True
98 ):
99 prov = self.mass.get_provider(
100 prov_mapping.provider_instance, provider_type=MusicProvider
101 )
102 if prov is None:
103 continue
104 # prefer domain for streaming providers as the catalog is the same across instances
105 prov_key = prov.domain if prov.is_streaming_provider else prov.instance_id
106 if prov_key in unique_keys:
107 continue
108 unique_keys.add(prov_key)
109 with suppress(MediaNotFoundError):
110 prov_item = await self.mass.music.artists.get_provider_item(
111 prov_mapping.item_id, prov_mapping.provider_instance
112 )
113 if prov_item.metadata.description:
114 description_candidates.append(
115 (prov_item.metadata.description_language, prov_item.metadata.description)
116 )
117 artist.metadata.update(
118 replace(prov_item.metadata, description=None, description_language=None)
119 )
120
121 # The musicbrainz ID is mandatory for all metadata lookups
122 if not artist.mbid:
123 if mbid := await self._get_artist_mbid(artist):
124 artist.mbid = mbid
125
126 # don't merge online genres on top of source-supplied ones; propagation-derived
127 # genres also count as a local source so they survive metadata refreshes
128 prefer_local_genres = self.config.get_value(CONF_PREFER_LOCAL_GENRES) and (
129 bool(artist.metadata.genres)
130 or await self.mass.music.genres.has_derived_genre_mappings(
131 MediaType.ARTIST, artist.item_id
132 )
133 )
134
135 # collect metadata from all (online)[metadata] providers
136 # TODO: Utilize a global (cloud) cache for metadata lookups to save on API calls
137 if self.config.get_value(CONF_ENABLE_ONLINE_METADATA) and artist.mbid:
138 for provider in self.providers:
139 if ProviderFeature.ARTIST_METADATA not in provider.supported_features:
140 continue
141 try:
142 metadata = await provider.get_artist_metadata(artist)
143 except Exception as err:
144 self.logger.warning(
145 "Error fetching metadata for Artist %s from provider %s: %s",
146 artist.name,
147 provider.name,
148 err,
149 exc_info=err if self.logger.isEnabledFor(10) else None,
150 )
151 continue
152 if metadata:
153 if prefer_local_genres:
154 metadata = replace(metadata, genres=None)
155 if metadata.description:
156 description_candidates.append(
157 (metadata.description_language, metadata.description)
158 )
159 metadata = replace(metadata, description=None, description_language=None)
160 artist.metadata.update(metadata)
161 self.logger.debug(
162 "Fetched metadata for Artist %s on provider %s",
163 artist.name,
164 provider.name,
165 )
166 artist.metadata.description, artist.metadata.description_language = (
167 self._select_description(
168 description_candidates, prev_description, prev_description_language
169 )
170 )
171
172 # update final item in library database
173 # set timestamp, used to determine when this function was last called
174 artist.metadata.last_refresh = int(time())
175 await self.mass.music.artists.update_item_in_library(artist.item_id, artist)
176
177 def _select_description(
178 self,
179 candidates: Sequence[tuple[str | None, str]],
180 prev_description: str | None,
181 prev_description_language: str | None,
182 ) -> tuple[str | None, str | None]:
183 """
184 Return the chosen ``(description, language)`` for the artist this refresh.
185
186 :param candidates: ``(language, text)`` tuples in provider-priority order
187 (music providers first, then TADB, then Wikipedia).
188 :param prev_description: Bio stored before this refresh.
189 :param prev_description_language: Language of the bio stored before this refresh.
190 """
191 pref = self.preferred_language
192 # 1. first candidate in the user's preferred language
193 for lang, text in candidates:
194 if lang == pref:
195 return text, lang
196 # 2. keep a stored preferred-language bio rather than downgrade
197 if prev_description is not None and prev_description_language == pref:
198 return prev_description, prev_description_language
199 # 3. English fallback, same priority order
200 for lang, text in candidates:
201 if lang == "en":
202 return text, lang
203 # 4. last resort: highest-priority bio in any (incl. unknown) language
204 if candidates:
205 lang, text = candidates[0]
206 return text, lang
207 return prev_description, prev_description_language
208
209 async def _update_album_metadata(self, album: Album, force_refresh: bool = False) -> None:
210 """Get/update rich metadata for an album."""
211 # collect metadata from all (online) music + metadata providers
212 # NOTE: we only do/allow this every REFRESH_INTERVAL
213 needs_refresh = (time() - (album.metadata.last_refresh or 0)) > REFRESH_INTERVAL
214 if not (force_refresh or needs_refresh):
215 return
216
217 self.logger.debug("Updating metadata for Album %s", album.name)
218
219 # collect metadata from all [music] providers
220 # note that we sort the providers by priority so that we always
221 # prefer local providers over online providers
222 unique_keys: set[str] = set()
223 for prov_mapping in sorted(album.provider_mappings, key=lambda x: x.priority, reverse=True):
224 prov = self.mass.get_provider(
225 prov_mapping.provider_instance, provider_type=MusicProvider
226 )
227 if prov is None:
228 continue
229 # prefer domain for streaming providers as the catalog is the same across instances
230 prov_key = prov.domain if prov.is_streaming_provider else prov.instance_id
231 if prov_key in unique_keys:
232 continue
233 unique_keys.add(prov_key)
234 with suppress(MediaNotFoundError):
235 prov_item = await self.mass.music.albums.get_provider_item(
236 prov_mapping.item_id, prov_mapping.provider_instance
237 )
238 album.metadata.update(prov_item.metadata)
239 # backfill identifiers (e.g. barcode) missing on a sparse library add
240 album.external_ids.update(prov_item.external_ids)
241 if album.year is None and prov_item.year:
242 album.year = prov_item.year
243 if album.album_type == AlbumType.UNKNOWN:
244 album.album_type = prov_item.album_type
245
246 # don't merge online genres on top of source-supplied ones; propagation-derived
247 # genres also count as a local source so they survive metadata refreshes
248 prefer_local_genres = self.config.get_value(CONF_PREFER_LOCAL_GENRES) and (
249 bool(album.metadata.genres)
250 or await self.mass.music.genres.has_derived_genre_mappings(
251 MediaType.ALBUM, album.item_id
252 )
253 )
254
255 # collect metadata from all (online) [metadata] providers
256 # TODO: Utilize a global (cloud) cache for metadata lookups to save on API calls
257 if self.config.get_value(CONF_ENABLE_ONLINE_METADATA):
258 for provider in self.providers:
259 if ProviderFeature.ALBUM_METADATA not in provider.supported_features:
260 continue
261 try:
262 metadata = await provider.get_album_metadata(album)
263 except Exception as err:
264 self.logger.warning(
265 "Error fetching metadata for Album %s from provider %s: %s",
266 album.name,
267 provider.name,
268 err,
269 exc_info=err if self.logger.isEnabledFor(10) else None,
270 )
271 continue
272 if metadata:
273 if prefer_local_genres:
274 metadata = replace(metadata, genres=None)
275 album.metadata.update(metadata)
276 self.logger.debug(
277 "Fetched metadata for Album %s on provider %s",
278 album.name,
279 provider.name,
280 )
281 # update final item in library database
282 # set timestamp, used to determine when this function was last called
283 album.metadata.last_refresh = int(time())
284 await self.mass.music.albums.update_item_in_library(album.item_id, album)
285
286 async def _update_track_metadata(self, track: Track, force_refresh: bool = False) -> None:
287 """Get/update rich metadata for a track."""
288 # collect metadata from all (online) music + metadata providers
289 # NOTE: we only do/allow this every REFRESH_INTERVAL
290 needs_refresh = (time() - (track.metadata.last_refresh or 0)) > REFRESH_INTERVAL
291 if not (force_refresh or needs_refresh):
292 return
293
294 self.logger.debug("Updating metadata for Track %s", track.name)
295
296 # collect metadata from all [music] providers
297 # note that we sort the providers by priority so that we always
298 # prefer local providers over online providers
299 unique_keys: set[str] = set()
300 for prov_mapping in sorted(track.provider_mappings, key=lambda x: x.priority, reverse=True):
301 prov = self.mass.get_provider(
302 prov_mapping.provider_instance, provider_type=MusicProvider
303 )
304 if prov is None:
305 continue
306 # prefer domain for streaming providers as the catalog is the same across instances
307 prov_key = prov.domain if prov.is_streaming_provider else prov.instance_id
308 if prov_key in unique_keys:
309 continue
310 unique_keys.add(prov_key)
311 with suppress(MediaNotFoundError):
312 prov_item = await self.mass.music.tracks.get_provider_item(
313 prov_mapping.item_id, prov_mapping.provider_instance
314 )
315 track.metadata.update(prov_item.metadata)
316
317 # don't merge online genres on top of source-supplied ones
318 prefer_local_genres = self.config.get_value(CONF_PREFER_LOCAL_GENRES) and bool(
319 track.metadata.genres
320 )
321
322 # collect metadata from all [metadata] providers
323 # Only fetch metadata from these sources if force_refresh is set OR
324 # if the track needs a refresh (based on REFRESH_INTERVAL) AND
325 # online metadata is enabled.
326 if (force_refresh or needs_refresh) and self.config.get_value(CONF_ENABLE_ONLINE_METADATA):
327 for provider in self.providers:
328 if ProviderFeature.TRACK_METADATA not in provider.supported_features:
329 continue
330
331 try:
332 metadata = await provider.get_track_metadata(track)
333 except Exception as err:
334 self.logger.warning(
335 "Error fetching metadata for Track %s from provider %s: %s",
336 track.name,
337 provider.name,
338 err,
339 exc_info=err if self.logger.isEnabledFor(10) else None,
340 )
341 continue
342 if metadata:
343 if prefer_local_genres:
344 metadata = replace(metadata, genres=None)
345 track.metadata.update(metadata)
346 self.logger.debug(
347 "Fetched metadata for Track %s on provider %s",
348 track.name,
349 provider.name,
350 )
351 # set timestamp, used to determine when this function was last called
352 track.metadata.last_refresh = int(time())
353 # update final item in library database
354 await self.mass.music.tracks.update_item_in_library(track.item_id, track)
355
356 async def _update_playlist_metadata(
357 self, playlist: Playlist, force_refresh: bool = False
358 ) -> None:
359 """Get/update rich metadata for a playlist."""
360 # collect metadata + create collage images
361 # NOTE: we only do/allow this every REFRESH_INTERVAL
362 needs_refresh = (time() - (playlist.metadata.last_refresh or 0)) > REFRESH_INTERVAL
363 if not (force_refresh or needs_refresh):
364 return
365 self.logger.debug("Updating metadata for Playlist %s", playlist.name)
366 playlist.metadata.genres = set()
367 all_playlist_tracks_images: list[MediaItemImage] = []
368 playlist_genres: dict[str, int] = {}
369 # retrieve metadata for the playlist from the tracks (such as genres etc.)
370 # TODO: retrieve style/mood ?
371 async for track in self.mass.music.playlists.tracks(playlist.item_id, playlist.provider):
372 if (
373 track.image
374 and track.image not in all_playlist_tracks_images
375 and (
376 track.image.provider in ("url", "builtin", "http")
377 or self.mass.get_provider(track.image.provider)
378 )
379 ):
380 all_playlist_tracks_images.append(track.image)
381 if track.metadata.genres:
382 genres = track.metadata.genres
383 elif (
384 isinstance(track, Track)
385 and track.album
386 and isinstance(track.album, Album)
387 and track.album.metadata.genres
388 ):
389 genres = track.album.metadata.genres
390 else:
391 genres = set()
392 for genre in genres:
393 if genre not in playlist_genres:
394 playlist_genres[genre] = 0
395 playlist_genres[genre] += 1
396 await asyncio.sleep(0) # yield to eventloop
397
398 playlist_genres_filtered = {genre for genre, count in playlist_genres.items() if count > 5}
399 playlist_genres_filtered = set(list(playlist_genres_filtered)[:8])
400 playlist.metadata.genres.update(playlist_genres_filtered)
401
402 # Collect metadata from metadata providers (e.g. playlist_metadata)
403 for provider in self.providers:
404 if ProviderFeature.PLAYLIST_METADATA not in provider.supported_features:
405 continue
406 try:
407 if prov_metadata := await provider.get_playlist_metadata(playlist):
408 playlist.metadata.update(prov_metadata)
409 self.logger.debug(
410 "Retrieved playlist metadata from provider %s for %s",
411 provider.name,
412 playlist.name,
413 )
414 except MusicAssistantError as err:
415 self.logger.warning(
416 "Error retrieving playlist metadata from provider %s for %s: %s",
417 provider.name,
418 playlist.name,
419 err,
420 exc_info=err if self.logger.isEnabledFor(10) else None,
421 )
422 # set timestamp, used to determine when this function was last called
423 playlist.metadata.last_refresh = int(time())
424 # update final item in library database
425 await self.mass.music.playlists.update_item_in_library(
426 playlist.item_id, playlist, overwrite=True
427 )
428
429 async def _update_audiobook_metadata(
430 self, audiobook: Audiobook, force_refresh: bool = False
431 ) -> None:
432 """Get/update rich metadata for an audiobook."""
433 # collect metadata from all (online) music + metadata providers
434 # NOTE: we only do/allow this every REFRESH_INTERVAL
435 needs_refresh = (time() - (audiobook.metadata.last_refresh or 0)) > REFRESH_INTERVAL
436 if not (force_refresh or needs_refresh):
437 return
438
439 self.logger.debug("Updating metadata for Audiobook %s", audiobook.name)
440
441 # collect metadata from all [music] providers
442 # note that we sort the providers by priority so that we always
443 # prefer local providers over online providers
444 unique_keys: set[str] = set()
445 prov_images: UniqueList[MediaItemImage] | None = None
446 for prov_mapping in sorted(
447 audiobook.provider_mappings, key=lambda x: x.priority, reverse=True
448 ):
449 prov = self.mass.get_provider(
450 prov_mapping.provider_instance, provider_type=MusicProvider
451 )
452 if prov is None:
453 continue
454 # prefer domain for streaming providers as the catalog is the same across instances
455 prov_key = prov.domain if prov.is_streaming_provider else prov.instance_id
456 if prov_key in unique_keys:
457 continue
458 unique_keys.add(prov_key)
459 with suppress(MediaNotFoundError):
460 prov_item = await self.mass.music.audiobooks.get_provider_item(
461 prov_mapping.item_id, prov_mapping.provider_instance
462 )
463 if prov_images is None and prov_item.metadata.images:
464 prov_images = prov_item.metadata.images
465 audiobook.metadata.update(prov_item.metadata)
466 if audiobook.publisher is None and prov_item.publisher:
467 audiobook.publisher = prov_item.publisher
468 if not audiobook.authors and prov_item.authors:
469 audiobook.authors = prov_item.authors
470 if not audiobook.narrators and prov_item.narrators:
471 audiobook.narrators = prov_item.narrators
472 if not audiobook.duration and prov_item.duration:
473 audiobook.duration = prov_item.duration
474
475 # no way to select a cover for audiobooks, so replace rather than merge the
476 # images to keep it in sync with the provider; revisit if a picker is added
477 if prov_images is not None:
478 audiobook.metadata.images = prov_images
479
480 # update final item in library database
481 # set timestamp, used to determine when this function was last called
482 audiobook.metadata.last_refresh = int(time())
483 await self.mass.music.audiobooks.update_item_in_library(audiobook.item_id, audiobook)
484
485 async def _update_podcast_metadata(self, podcast: Podcast, force_refresh: bool = False) -> None:
486 """Get/update rich metadata for a podcast."""
487 # collect metadata from all (online) music + metadata providers
488 # NOTE: we only do/allow this every REFRESH_INTERVAL
489 needs_refresh = (time() - (podcast.metadata.last_refresh or 0)) > REFRESH_INTERVAL
490 if not (force_refresh or needs_refresh):
491 return
492
493 self.logger.debug("Updating metadata for Podcast %s", podcast.name)
494
495 # collect metadata from all [music] providers
496 # note that we sort the providers by priority so that we always
497 # prefer local providers over online providers
498 unique_keys: set[str] = set()
499 prov_images: UniqueList[MediaItemImage] | None = None
500 for prov_mapping in sorted(
501 podcast.provider_mappings, key=lambda x: x.priority, reverse=True
502 ):
503 prov = self.mass.get_provider(
504 prov_mapping.provider_instance, provider_type=MusicProvider
505 )
506 if prov is None:
507 continue
508 # prefer domain for streaming providers as the catalog is the same across instances
509 prov_key = prov.domain if prov.is_streaming_provider else prov.instance_id
510 if prov_key in unique_keys:
511 continue
512 unique_keys.add(prov_key)
513 with suppress(MediaNotFoundError):
514 prov_item = await self.mass.music.podcasts.get_provider_item(
515 prov_mapping.item_id, prov_mapping.provider_instance
516 )
517 if prov_images is None and prov_item.metadata.images:
518 prov_images = prov_item.metadata.images
519 podcast.metadata.update(prov_item.metadata)
520 if podcast.publisher is None and prov_item.publisher:
521 podcast.publisher = prov_item.publisher
522 if not podcast.total_episodes and prov_item.total_episodes:
523 podcast.total_episodes = prov_item.total_episodes
524
525 # no way to select a cover for podcasts, so replace rather than merge the
526 # images to keep it in sync with the provider; revisit if a picker is added
527 if prov_images is not None:
528 podcast.metadata.images = prov_images
529
530 # update final item in library database
531 # set timestamp, used to determine when this function was last called
532 podcast.metadata.last_refresh = int(time())
533 await self.mass.music.podcasts.update_item_in_library(podcast.item_id, podcast)
534
535 async def _get_artist_mbid(self, artist: Artist) -> str | None:
536 """Fetch musicbrainz id by performing search using the artist name, albums and tracks."""
537 if artist.mbid:
538 return artist.mbid
539 if compare_strings(artist.name, VARIOUS_ARTISTS_NAME):
540 return VARIOUS_ARTISTS_MBID
541
542 musicbrainz_provider = self.mass.get_provider("musicbrainz")
543 if not musicbrainz_provider:
544 return None
545 musicbrainz: MusicbrainzProvider = cast("MusicbrainzProvider", musicbrainz_provider)
546 if TYPE_CHECKING:
547 assert isinstance(musicbrainz, MusicbrainzProvider)
548 # first try with resource URL (e.g. streaming provider share URL)
549 for prov_mapping in artist.provider_mappings:
550 if prov_mapping.url and prov_mapping.url.startswith("http"):
551 if mb_artist := await musicbrainz.get_artist_details_by_resource_url(
552 prov_mapping.url
553 ):
554 return mb_artist.id
555
556 # start lookup of musicbrainz id using artist name, albums and tracks
557 ref_albums = await self.mass.music.artists.albums(artist.item_id, artist.provider)
558 # prefer the (widely supported) top tracks listing, falling back to all tracks
559 ref_tracks = await self.mass.music.artists.top_tracks(artist.item_id, artist.provider)
560 if not ref_tracks:
561 ref_tracks = await self.mass.music.artists.tracks(artist.item_id, artist.provider)
562 # try with (strict) ref track(s), using recording id
563 for ref_track in ref_tracks:
564 if mb_artist := await musicbrainz.get_artist_details_by_track(artist.name, ref_track):
565 return mb_artist.id
566 # try with (strict) ref album(s), using releasegroup id
567 for ref_album in ref_albums:
568 if mb_artist := await musicbrainz.get_artist_details_by_album(artist.name, ref_album):
569 return mb_artist.id
570 # last resort: track matching by name
571 for ref_track in ref_tracks:
572 if not ref_track.album:
573 continue
574 if result := await musicbrainz.search(
575 artistname=artist.name,
576 albumname=ref_track.album.name,
577 trackname=ref_track.name,
578 trackversion=ref_track.version,
579 ):
580 return result[0].id
581
582 # lookup failed
583 ref_albums_str = "/".join(x.name for x in ref_albums) or "none"
584 ref_tracks_str = "/".join(x.name for x in ref_tracks) or "none"
585 self.logger.debug(
586 "Unable to get musicbrainz ID for artist %s (albums: %s, tracks: %s)",
587 artist.name,
588 ref_albums_str,
589 ref_tracks_str,
590 )
591 return None
592