music-assistant-server

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                if album.year is None and prov_item.year:
240                    album.year = prov_item.year
241                if album.album_type == AlbumType.UNKNOWN:
242                    album.album_type = prov_item.album_type
243
244        # don't merge online genres on top of source-supplied ones; propagation-derived
245        # genres also count as a local source so they survive metadata refreshes
246        prefer_local_genres = self.config.get_value(CONF_PREFER_LOCAL_GENRES) and (
247            bool(album.metadata.genres)
248            or await self.mass.music.genres.has_derived_genre_mappings(
249                MediaType.ALBUM, album.item_id
250            )
251        )
252
253        # collect metadata from all (online) [metadata] providers
254        # TODO: Utilize a global (cloud) cache for metadata lookups to save on API calls
255        if self.config.get_value(CONF_ENABLE_ONLINE_METADATA):
256            for provider in self.providers:
257                if ProviderFeature.ALBUM_METADATA not in provider.supported_features:
258                    continue
259                try:
260                    metadata = await provider.get_album_metadata(album)
261                except Exception as err:
262                    self.logger.warning(
263                        "Error fetching metadata for Album %s from provider %s: %s",
264                        album.name,
265                        provider.name,
266                        err,
267                        exc_info=err if self.logger.isEnabledFor(10) else None,
268                    )
269                    continue
270                if metadata:
271                    if prefer_local_genres:
272                        metadata = replace(metadata, genres=None)
273                    album.metadata.update(metadata)
274                    self.logger.debug(
275                        "Fetched metadata for Album %s on provider %s",
276                        album.name,
277                        provider.name,
278                    )
279        # update final item in library database
280        # set timestamp, used to determine when this function was last called
281        album.metadata.last_refresh = int(time())
282        await self.mass.music.albums.update_item_in_library(album.item_id, album)
283
284    async def _update_track_metadata(self, track: Track, force_refresh: bool = False) -> None:
285        """Get/update rich metadata for a track."""
286        # collect metadata from all (online) music + metadata providers
287        # NOTE: we only do/allow this every REFRESH_INTERVAL
288        needs_refresh = (time() - (track.metadata.last_refresh or 0)) > REFRESH_INTERVAL
289        if not (force_refresh or needs_refresh):
290            return
291
292        self.logger.debug("Updating metadata for Track %s", track.name)
293
294        # collect metadata from all [music] providers
295        # note that we sort the providers by priority so that we always
296        # prefer local providers over online providers
297        unique_keys: set[str] = set()
298        for prov_mapping in sorted(track.provider_mappings, key=lambda x: x.priority, reverse=True):
299            prov = self.mass.get_provider(
300                prov_mapping.provider_instance, provider_type=MusicProvider
301            )
302            if prov is None:
303                continue
304            # prefer domain for streaming providers as the catalog is the same across instances
305            prov_key = prov.domain if prov.is_streaming_provider else prov.instance_id
306            if prov_key in unique_keys:
307                continue
308            unique_keys.add(prov_key)
309            with suppress(MediaNotFoundError):
310                prov_item = await self.mass.music.tracks.get_provider_item(
311                    prov_mapping.item_id, prov_mapping.provider_instance
312                )
313                track.metadata.update(prov_item.metadata)
314
315        # don't merge online genres on top of source-supplied ones
316        prefer_local_genres = self.config.get_value(CONF_PREFER_LOCAL_GENRES) and bool(
317            track.metadata.genres
318        )
319
320        # collect metadata from all [metadata] providers
321        # Only fetch metadata from these sources if force_refresh is set OR
322        # if the track needs a refresh (based on REFRESH_INTERVAL) AND
323        # online metadata is enabled.
324        if (force_refresh or needs_refresh) and self.config.get_value(CONF_ENABLE_ONLINE_METADATA):
325            for provider in self.providers:
326                if ProviderFeature.TRACK_METADATA not in provider.supported_features:
327                    continue
328
329                try:
330                    metadata = await provider.get_track_metadata(track)
331                except Exception as err:
332                    self.logger.warning(
333                        "Error fetching metadata for Track %s from provider %s: %s",
334                        track.name,
335                        provider.name,
336                        err,
337                        exc_info=err if self.logger.isEnabledFor(10) else None,
338                    )
339                    continue
340                if metadata:
341                    if prefer_local_genres:
342                        metadata = replace(metadata, genres=None)
343                    track.metadata.update(metadata)
344                    self.logger.debug(
345                        "Fetched metadata for Track %s on provider %s",
346                        track.name,
347                        provider.name,
348                    )
349        # set timestamp, used to determine when this function was last called
350        track.metadata.last_refresh = int(time())
351        # update final item in library database
352        await self.mass.music.tracks.update_item_in_library(track.item_id, track)
353
354    async def _update_playlist_metadata(
355        self, playlist: Playlist, force_refresh: bool = False
356    ) -> None:
357        """Get/update rich metadata for a playlist."""
358        # collect metadata + create collage images
359        # NOTE: we only do/allow this every REFRESH_INTERVAL
360        needs_refresh = (time() - (playlist.metadata.last_refresh or 0)) > REFRESH_INTERVAL
361        if not (force_refresh or needs_refresh):
362            return
363        self.logger.debug("Updating metadata for Playlist %s", playlist.name)
364        playlist.metadata.genres = set()
365        all_playlist_tracks_images: list[MediaItemImage] = []
366        playlist_genres: dict[str, int] = {}
367        # retrieve metadata for the playlist from the tracks (such as genres etc.)
368        # TODO: retrieve style/mood ?
369        async for track in self.mass.music.playlists.tracks(playlist.item_id, playlist.provider):
370            if (
371                track.image
372                and track.image not in all_playlist_tracks_images
373                and (
374                    track.image.provider in ("url", "builtin", "http")
375                    or self.mass.get_provider(track.image.provider)
376                )
377            ):
378                all_playlist_tracks_images.append(track.image)
379            if track.metadata.genres:
380                genres = track.metadata.genres
381            elif (
382                isinstance(track, Track)
383                and track.album
384                and isinstance(track.album, Album)
385                and track.album.metadata.genres
386            ):
387                genres = track.album.metadata.genres
388            else:
389                genres = set()
390            for genre in genres:
391                if genre not in playlist_genres:
392                    playlist_genres[genre] = 0
393                playlist_genres[genre] += 1
394            await asyncio.sleep(0)  # yield to eventloop
395
396        playlist_genres_filtered = {genre for genre, count in playlist_genres.items() if count > 5}
397        playlist_genres_filtered = set(list(playlist_genres_filtered)[:8])
398        playlist.metadata.genres.update(playlist_genres_filtered)
399
400        # Collect metadata from metadata providers (e.g. playlist_metadata)
401        for provider in self.providers:
402            if ProviderFeature.PLAYLIST_METADATA not in provider.supported_features:
403                continue
404            try:
405                if prov_metadata := await provider.get_playlist_metadata(playlist):
406                    playlist.metadata.update(prov_metadata)
407                    self.logger.debug(
408                        "Retrieved playlist metadata from provider %s for %s",
409                        provider.name,
410                        playlist.name,
411                    )
412            except MusicAssistantError as err:
413                self.logger.warning(
414                    "Error retrieving playlist metadata from provider %s for %s: %s",
415                    provider.name,
416                    playlist.name,
417                    err,
418                    exc_info=err if self.logger.isEnabledFor(10) else None,
419                )
420        # set timestamp, used to determine when this function was last called
421        playlist.metadata.last_refresh = int(time())
422        # update final item in library database
423        await self.mass.music.playlists.update_item_in_library(
424            playlist.item_id, playlist, overwrite=True
425        )
426
427    async def _update_audiobook_metadata(
428        self, audiobook: Audiobook, force_refresh: bool = False
429    ) -> None:
430        """Get/update rich metadata for an audiobook."""
431        # collect metadata from all (online) music + metadata providers
432        # NOTE: we only do/allow this every REFRESH_INTERVAL
433        needs_refresh = (time() - (audiobook.metadata.last_refresh or 0)) > REFRESH_INTERVAL
434        if not (force_refresh or needs_refresh):
435            return
436
437        self.logger.debug("Updating metadata for Audiobook %s", audiobook.name)
438
439        # collect metadata from all [music] providers
440        # note that we sort the providers by priority so that we always
441        # prefer local providers over online providers
442        unique_keys: set[str] = set()
443        prov_images: UniqueList[MediaItemImage] | None = None
444        for prov_mapping in sorted(
445            audiobook.provider_mappings, key=lambda x: x.priority, reverse=True
446        ):
447            prov = self.mass.get_provider(
448                prov_mapping.provider_instance, provider_type=MusicProvider
449            )
450            if prov is None:
451                continue
452            # prefer domain for streaming providers as the catalog is the same across instances
453            prov_key = prov.domain if prov.is_streaming_provider else prov.instance_id
454            if prov_key in unique_keys:
455                continue
456            unique_keys.add(prov_key)
457            with suppress(MediaNotFoundError):
458                prov_item = await self.mass.music.audiobooks.get_provider_item(
459                    prov_mapping.item_id, prov_mapping.provider_instance
460                )
461                if prov_images is None and prov_item.metadata.images:
462                    prov_images = prov_item.metadata.images
463                audiobook.metadata.update(prov_item.metadata)
464                if audiobook.publisher is None and prov_item.publisher:
465                    audiobook.publisher = prov_item.publisher
466                if not audiobook.authors and prov_item.authors:
467                    audiobook.authors = prov_item.authors
468                if not audiobook.narrators and prov_item.narrators:
469                    audiobook.narrators = prov_item.narrators
470                if not audiobook.duration and prov_item.duration:
471                    audiobook.duration = prov_item.duration
472
473        # no way to select a cover for audiobooks, so replace rather than merge the
474        # images to keep it in sync with the provider; revisit if a picker is added
475        if prov_images is not None:
476            audiobook.metadata.images = prov_images
477
478        # update final item in library database
479        # set timestamp, used to determine when this function was last called
480        audiobook.metadata.last_refresh = int(time())
481        await self.mass.music.audiobooks.update_item_in_library(audiobook.item_id, audiobook)
482
483    async def _update_podcast_metadata(self, podcast: Podcast, force_refresh: bool = False) -> None:
484        """Get/update rich metadata for a podcast."""
485        # collect metadata from all (online) music + metadata providers
486        # NOTE: we only do/allow this every REFRESH_INTERVAL
487        needs_refresh = (time() - (podcast.metadata.last_refresh or 0)) > REFRESH_INTERVAL
488        if not (force_refresh or needs_refresh):
489            return
490
491        self.logger.debug("Updating metadata for Podcast %s", podcast.name)
492
493        # collect metadata from all [music] providers
494        # note that we sort the providers by priority so that we always
495        # prefer local providers over online providers
496        unique_keys: set[str] = set()
497        prov_images: UniqueList[MediaItemImage] | None = None
498        for prov_mapping in sorted(
499            podcast.provider_mappings, key=lambda x: x.priority, reverse=True
500        ):
501            prov = self.mass.get_provider(
502                prov_mapping.provider_instance, provider_type=MusicProvider
503            )
504            if prov is None:
505                continue
506            # prefer domain for streaming providers as the catalog is the same across instances
507            prov_key = prov.domain if prov.is_streaming_provider else prov.instance_id
508            if prov_key in unique_keys:
509                continue
510            unique_keys.add(prov_key)
511            with suppress(MediaNotFoundError):
512                prov_item = await self.mass.music.podcasts.get_provider_item(
513                    prov_mapping.item_id, prov_mapping.provider_instance
514                )
515                if prov_images is None and prov_item.metadata.images:
516                    prov_images = prov_item.metadata.images
517                podcast.metadata.update(prov_item.metadata)
518                if podcast.publisher is None and prov_item.publisher:
519                    podcast.publisher = prov_item.publisher
520                if not podcast.total_episodes and prov_item.total_episodes:
521                    podcast.total_episodes = prov_item.total_episodes
522
523        # no way to select a cover for podcasts, so replace rather than merge the
524        # images to keep it in sync with the provider; revisit if a picker is added
525        if prov_images is not None:
526            podcast.metadata.images = prov_images
527
528        # update final item in library database
529        # set timestamp, used to determine when this function was last called
530        podcast.metadata.last_refresh = int(time())
531        await self.mass.music.podcasts.update_item_in_library(podcast.item_id, podcast)
532
533    async def _get_artist_mbid(self, artist: Artist) -> str | None:
534        """Fetch musicbrainz id by performing search using the artist name, albums and tracks."""
535        if artist.mbid:
536            return artist.mbid
537        if compare_strings(artist.name, VARIOUS_ARTISTS_NAME):
538            return VARIOUS_ARTISTS_MBID
539
540        musicbrainz_provider = self.mass.get_provider("musicbrainz")
541        if not musicbrainz_provider:
542            return None
543        musicbrainz: MusicbrainzProvider = cast("MusicbrainzProvider", musicbrainz_provider)
544        if TYPE_CHECKING:
545            assert isinstance(musicbrainz, MusicbrainzProvider)
546        # first try with resource URL (e.g. streaming provider share URL)
547        for prov_mapping in artist.provider_mappings:
548            if prov_mapping.url and prov_mapping.url.startswith("http"):
549                if mb_artist := await musicbrainz.get_artist_details_by_resource_url(
550                    prov_mapping.url
551                ):
552                    return mb_artist.id
553
554        # start lookup of musicbrainz id using artist name, albums and tracks
555        ref_albums = await self.mass.music.artists.albums(artist.item_id, artist.provider)
556        # prefer the (widely supported) top tracks listing, falling back to all tracks
557        ref_tracks = await self.mass.music.artists.top_tracks(artist.item_id, artist.provider)
558        if not ref_tracks:
559            ref_tracks = await self.mass.music.artists.tracks(artist.item_id, artist.provider)
560        # try with (strict) ref track(s), using recording id
561        for ref_track in ref_tracks:
562            if mb_artist := await musicbrainz.get_artist_details_by_track(artist.name, ref_track):
563                return mb_artist.id
564        # try with (strict) ref album(s), using releasegroup id
565        for ref_album in ref_albums:
566            if mb_artist := await musicbrainz.get_artist_details_by_album(artist.name, ref_album):
567                return mb_artist.id
568        # last resort: track matching by name
569        for ref_track in ref_tracks:
570            if not ref_track.album:
571                continue
572            if result := await musicbrainz.search(
573                artistname=artist.name,
574                albumname=ref_track.album.name,
575                trackname=ref_track.name,
576                trackversion=ref_track.version,
577            ):
578                return result[0].id
579
580        # lookup failed
581        ref_albums_str = "/".join(x.name for x in ref_albums) or "none"
582        ref_tracks_str = "/".join(x.name for x in ref_tracks) or "none"
583        self.logger.debug(
584            "Unable to get musicbrainz ID for artist %s (albums: %s, tracks: %s)",
585            artist.name,
586            ref_albums_str,
587            ref_tracks_str,
588        )
589        return None
590