music-assistant-server

1"""Manage MediaItems of type Artist."""
2
3from __future__ import annotations
4
5import asyncio
6import contextlib
7from itertools import zip_longest
8from typing import TYPE_CHECKING, Any, cast
9
10from music_assistant_models.auth import Scope
11from music_assistant_models.enums import (
12    AlbumType,
13    ArtistType,
14    MediaType,
15    ProviderFeature,
16    ProviderType,
17)
18from music_assistant_models.errors import (
19    MediaNotFoundError,
20    MusicAssistantError,
21    ProviderUnavailableError,
22)
23from music_assistant_models.helpers import create_safe_string
24from music_assistant_models.media_items import (
25    Album,
26    Artist,
27    ArtistSummary,
28    Audiobook,
29    ItemMapping,
30    ProviderMapping,
31    Track,
32)
33
34from music_assistant.constants import (
35    DB_TABLE_ALBUM_ARTISTS,
36    DB_TABLE_ARTISTS,
37    DB_TABLE_AUDIOBOOK_ARTISTS,
38    DB_TABLE_TRACK_ARTISTS,
39    VARIOUS_ARTISTS_MBID,
40    VARIOUS_ARTISTS_NAME,
41)
42from music_assistant.helpers.compare import (
43    compare_album,
44    compare_artist,
45    compare_strings,
46    compare_track,
47)
48from music_assistant.helpers.database import UNSET
49from music_assistant.helpers.json import serialize_to_json
50from music_assistant.models.music_provider import MusicProvider
51
52from .base import MediaControllerBase
53
54if TYPE_CHECKING:
55    from collections.abc import Mapping
56
57    from music_assistant import MusicAssistant
58    from music_assistant.models.metadata_provider import MetadataProvider
59
60
61class ArtistsController(MediaControllerBase[Artist]):
62    """Controller managing MediaItems of type Artist."""
63
64    db_table = DB_TABLE_ARTISTS
65    media_type = MediaType.ARTIST
66    item_cls = Artist
67    summary_item_cls = ArtistSummary
68
69    def __init__(self, mass: MusicAssistant) -> None:
70        """Initialize class."""
71        super().__init__(mass)
72        self._db_add_lock = asyncio.Lock()
73        # register (extra) api handlers
74        api_base = self.api_base
75        self.mass.register_api_command(
76            f"music/{api_base}/artist_albums", self.albums, required_scope=Scope.LIBRARY_READ
77        )
78        self.mass.register_api_command(
79            f"music/{api_base}/artist_tracks", self.tracks, required_scope=Scope.LIBRARY_READ
80        )
81        self.mass.register_api_command(
82            f"music/{api_base}/top_tracks", self.top_tracks, required_scope=Scope.LIBRARY_READ
83        )
84        self.mass.register_api_command(
85            f"music/{api_base}/top_albums", self.top_albums, required_scope=Scope.LIBRARY_READ
86        )
87        self.mass.register_api_command(
88            f"music/{api_base}/artist_audiobooks",
89            self.audiobooks,
90            required_scope=Scope.LIBRARY_READ,
91        )
92        self.mass.register_api_command(
93            f"music/{api_base}/similar_artists",
94            self.similar_artists,
95            required_scope=Scope.LIBRARY_READ,
96        )
97        self.mass.register_api_command(
98            f"music/{api_base}/library_artist_types",
99            self.get_library_artist_types,
100            required_scope=Scope.LIBRARY_READ,
101        )
102
103    @property
104    def summary_query(self) -> tuple[str, dict[str, Any]]:
105        """Return the slim SELECT query used for artist summary listings."""
106        query = f"""
107        SELECT
108            {self._summary_base_columns()},
109            artists.artist_type,
110            {self._provider_mappings_query()} AS provider_mappings
111            FROM artists"""
112        return query, {}
113
114    async def library_count(
115        self,
116        favorite_only: bool = False,
117        album_artists_only: bool = False,
118        artist_type: ArtistType | None = None,
119    ) -> int:
120        """
121        Return the number of artists in the library.
122
123        Restricted to the providers the current user is allowed to see when that user
124        has a provider filter set.
125
126        :param favorite_only: Only count artists marked as favorite.
127        :param album_artists_only: Only count artists that have albums.
128        :param artist_type: Only count artists of this type.
129        """
130        sql_query = f"SELECT item_id FROM {self.db_table}"
131        query_parts = []
132        query_params: dict[str, Any] = {}
133        if artist_type:
134            query_parts.append(f"artist_type = '{artist_type}'")
135        if favorite_only:
136            query_parts.append("favorite = 1")
137        if album_artists_only:
138            query_parts.append(
139                f"item_id in (select {DB_TABLE_ALBUM_ARTISTS}.artist_id "
140                f"FROM {DB_TABLE_ALBUM_ARTISTS})"
141            )
142        if provider_filter := self._ensure_provider_filter(None):
143            query_parts.append(
144                self._provider_filter_clause(query_params, provider_filter, in_library_only=True)
145            )
146        if query_parts:
147            sql_query += f" WHERE {' AND '.join(query_parts)}"
148        return await self.mass.music.database.get_count_from_query(sql_query, query_params)
149
150    async def library_items(  # noqa: PLR0913
151        self,
152        favorite: bool | None = None,
153        search: str | None = None,
154        limit: int = 500,
155        offset: int = 0,
156        order_by: str = "sort_name",
157        provider: str | list[str] | None = None,
158        genre: int | list[int] | None = None,
159        played_only: bool = False,
160        album_artists_only: bool = False,
161        artist_type: ArtistType | None = None,
162        *,
163        summary: bool = True,
164        **kwargs: Any,
165    ) -> list[Artist]:
166        """
167        Get in-database (album) artists.
168
169        :param favorite: Filter by favorite status.
170        :param search: Filter by search query.
171        :param limit: Maximum number of items to return.
172        :param offset: Number of items to skip.
173        :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
174        :param provider: Filter by provider instance ID (single string or list).
175        :param album_artists_only: Only return artists that have albums.
176        :param genre: Filter by genre id(s).
177        :param artist_type: The artist's type
178        :param summary: When True (default), return slim summary items containing only the
179            fields needed for a list view. Set to False to get fully hydrated items.
180        """
181        extra_query_params: dict[str, Any] = {}
182        extra_query_parts: list[str] = []
183        if artist_type:
184            extra_query_parts = [f"artist_type = '{artist_type}'"]
185        if album_artists_only and artist_type in (None, ArtistType.SINGER):
186            extra_query_parts.append(
187                f"artists.item_id in (select {DB_TABLE_ALBUM_ARTISTS}.artist_id "
188                f"from {DB_TABLE_ALBUM_ARTISTS})"
189            )
190        return await self.get_library_items_by_query(
191            favorite=favorite,
192            search=search,
193            genre_ids=genre,
194            limit=limit,
195            offset=offset,
196            order_by=order_by,
197            provider_filter=self._ensure_provider_filter(provider),
198            extra_query_parts=extra_query_parts,
199            extra_query_params=extra_query_params,
200            played_only=played_only,
201            in_library_only=True,
202            summary=summary,
203        )
204
205    async def tracks(
206        self,
207        item_id: str,
208        provider_instance_id_or_domain: str,
209        provider_filter: str | None = None,
210    ) -> list[Track]:
211        """
212        Return the tracks for a artist.
213
214        For a library item, the in-library tracks are returned, optionally limited to a single
215        provider instance with the provider_filter. For a provider item, that provider's
216        tracks listing is returned (which may be empty if it is not supported).
217
218        :param item_id: The item ID of the artist.
219        :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
220        :param provider_filter: Optional provider instance ID to limit the (library) result to.
221        """
222        if provider_instance_id_or_domain == "library":
223            return await self.get_library_artist_tracks(item_id, provider_filter=provider_filter)
224        self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
225        return await self.get_provider_artist_tracks(item_id, provider_instance_id_or_domain)
226
227    async def albums(
228        self,
229        item_id: str,
230        provider_instance_id_or_domain: str,
231        provider_filter: str | None = None,
232    ) -> list[Album]:
233        """
234        Return the albums for an artist.
235
236        For a library item, the in-library albums are returned, optionally limited to a single
237        provider instance with the provider_filter. For a provider item, that provider's
238        albums listing is returned (which may be empty if it is not supported).
239
240        :param item_id: The item ID of the artist.
241        :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
242        :param provider_filter: Optional provider instance ID to limit the (library) result to.
243        """
244        if provider_instance_id_or_domain == "library":
245            return await self.get_library_artist_albums(item_id, provider_filter=provider_filter)
246        self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
247        return await self.get_provider_artist_albums(item_id, provider_instance_id_or_domain)
248
249    async def top_tracks(
250        self,
251        item_id: str,
252        provider_instance_id_or_domain: str,
253        provider_filter: str | None = None,
254    ) -> list[Track]:
255        """
256        Return the top/featured tracks for an artist.
257
258        For a library item, the top tracks of all the artist's providers are aggregated (and
259        deduplicated), optionally limited to a single provider instance. For a provider
260        item, that provider's top tracks listing is returned (may be empty if not supported).
261
262        :param item_id: The item ID of the artist.
263        :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
264        :param provider_filter: Optional provider instance ID to limit the result to.
265        """
266        if provider_instance_id_or_domain == "library":
267            return await self.get_library_artist_toptracks(item_id, provider_filter=provider_filter)
268        self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
269        return await self.get_provider_artist_toptracks(item_id, provider_instance_id_or_domain)
270
271    async def top_albums(
272        self,
273        item_id: str,
274        provider_instance_id_or_domain: str,
275        provider_filter: str | None = None,
276    ) -> list[Album]:
277        """
278        Return the top/featured albums for an artist.
279
280        For a library item, the top albums of all the artist's providers are aggregated (and
281        deduplicated), optionally limited to a single provider instance. For a provider
282        item, that provider's top albums listing is returned (may be empty if not supported).
283
284        :param item_id: The item ID of the artist.
285        :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
286        :param provider_filter: Optional provider instance ID to limit the result to.
287        """
288        if provider_instance_id_or_domain == "library":
289            return await self.get_library_artist_topalbums(item_id, provider_filter=provider_filter)
290        self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
291        return await self.get_provider_artist_topalbums(item_id, provider_instance_id_or_domain)
292
293    async def similar_artists(
294        self,
295        item_id: str,
296        provider_instance_id_or_domain: str,
297        provider_filter: str | None = None,
298        limit: int = 25,
299    ) -> list[Artist]:
300        """
301        Return similar artists for an artist.
302
303        For a library item, the similar artists of all the artist's providers are aggregated
304        (and deduplicated), optionally limited to a single provider instance. For a provider
305        item, that provider's similar artists listing is returned (may be empty if not
306        supported).
307
308        :param item_id: The item ID of the artist.
309        :param provider_instance_id_or_domain: The provider instance ID or domain of the artist.
310        :param provider_filter: Optional provider instance ID to limit the result to.
311        :param limit: Maximum number of similar artists to return.
312        """
313        if provider_instance_id_or_domain == "library":
314            return await self.get_library_artist_similar_artists(
315                item_id, provider_filter=provider_filter, limit=limit
316            )
317        self._validate_provider_filter(provider_instance_id_or_domain, provider_filter)
318        return await self.get_provider_artist_similar_artists(
319            item_id, provider_instance_id_or_domain, limit=limit
320        )
321
322    async def audiobooks(
323        self,
324        item_id: str,
325        provider_instance_id_or_domain: str,
326        artist_type: ArtistType = ArtistType.AUTHOR,
327        in_library_only: bool = False,
328    ) -> list[Audiobook]:
329        """
330        Return audiobooks for an artist.
331
332        Artist_type can be omitted for in-library artists.
333        """
334        if artist_type == ArtistType.SINGER:
335            self.logger.warning("Audiobooks not supported for artist_type SINGER.")
336            return []
337        # always check if we have a library item for this artist
338        library_artist = await self.get_library_item_by_prov_id(
339            item_id, provider_instance_id_or_domain
340        )
341        if library_artist and library_artist.artist_type == ArtistType.SINGER:
342            self.logger.debug(
343                "Ignoring audiobook request for artist of type %s", library_artist.artist_type
344            )
345            return []
346        if not library_artist:
347            if artist_type == ArtistType.AUTHOR:
348                return await self.get_provider_author_audiobooks(
349                    item_id, provider_instance_id_or_domain
350                )
351            if artist_type == ArtistType.NARRATOR:
352                return await self.get_provider_narrator_audiobooks(
353                    item_id, provider_instance_id_or_domain
354                )
355            return []
356
357        db_items = await self.get_library_author_narrator_audiobooks(
358            library_artist.item_id, artist_type=library_artist.artist_type
359        )
360        result: list[Audiobook] = db_items
361        if in_library_only:
362            # return in-library items only
363            return result
364        # return all (unique) items from all providers
365        # initialize unique_ids with db_items to prevent duplicates
366        unique_ids: set[str] = {f"{item.name}.{item.version}" for item in db_items}
367        unique_providers = self.mass.music.get_unique_providers()
368        audiobook_method = (
369            self.get_provider_author_audiobooks
370            if artist_type == ArtistType.AUTHOR
371            else self.get_provider_narrator_audiobooks
372        )
373        for provider_mapping in library_artist.provider_mappings:
374            if provider_mapping.provider_instance not in unique_providers:
375                continue
376            provider_audiobooks = await audiobook_method(
377                provider_mapping.item_id, provider_mapping.provider_instance
378            )
379            for provider_audiobook in provider_audiobooks:
380                unique_id = f"{provider_audiobook.name}.{provider_audiobook.version}"
381                if unique_id in unique_ids:
382                    continue
383                unique_ids.add(unique_id)
384                # prefer db item
385                if db_item := await self.mass.music.audiobooks.get_library_item_by_prov_id(
386                    provider_audiobook.item_id, provider_audiobook.provider
387                ):
388                    result.append(db_item)
389                elif not in_library_only:
390                    result.append(provider_audiobook)
391        return result
392
393    async def get_library_author_narrator_audiobooks(
394        self,
395        item_id: str | int,
396        artist_type: ArtistType,
397    ) -> list[Audiobook]:
398        """Return all in-library audiobooks for an author/ narrator."""
399        db_id = int(item_id)  # ensure integer
400        library_item = await self.get_library_item(db_id)
401        if library_item.artist_type != artist_type:
402            self.logger.debug("Audiobooks only available for artists of type %s", artist_type)
403            return []
404        subquery = (
405            f"SELECT audiobook_id FROM {DB_TABLE_AUDIOBOOK_ARTISTS} WHERE artist_id = :artist_id"
406        )
407        query = f"audiobooks.item_id in ({subquery})"
408        return await self.mass.music.audiobooks.get_library_items_by_query(
409            extra_query_parts=[query],
410            extra_query_params={"artist_id": db_id},
411        )
412
413    async def get_provider_author_audiobooks(
414        self,
415        item_id: str,
416        provider_instance_id_or_domain: str,
417    ) -> list[Audiobook]:
418        """Return audiobooks for an author on given provider."""
419        assert provider_instance_id_or_domain != "library"
420        if not (prov := self.mass.get_provider(provider_instance_id_or_domain)):
421            return []
422        prov = cast("MusicProvider", prov)
423        if ProviderFeature.AUTHOR_AUDIOBOOKS in prov.supported_features:
424            return await prov.get_author_audiobooks(item_id)
425        # fallback implementation using the db
426        return await self._get_db_author_narrator_audiobooks(
427            item_id=item_id,
428            provider_instance_id_or_domain=provider_instance_id_or_domain,
429            artist_type=ArtistType.AUTHOR,
430        )
431
432    async def get_provider_narrator_audiobooks(
433        self,
434        item_id: str,
435        provider_instance_id_or_domain: str,
436    ) -> list[Audiobook]:
437        """Return audiobooks for an author on given provider."""
438        assert provider_instance_id_or_domain != "library"
439        if not (prov := self.mass.get_provider(provider_instance_id_or_domain)):
440            return []
441        prov = cast("MusicProvider", prov)
442        if ProviderFeature.NARRATOR_AUDIOBOOKS in prov.supported_features:
443            return await prov.get_narrator_audiobooks(item_id)
444        # fallback implementation using the db
445        return await self._get_db_author_narrator_audiobooks(
446            item_id=item_id,
447            provider_instance_id_or_domain=provider_instance_id_or_domain,
448            artist_type=ArtistType.NARRATOR,
449        )
450
451    async def get_provider_artist_toptracks(
452        self,
453        item_id: str,
454        provider_instance_id_or_domain: str,
455    ) -> list[Track]:
456        """
457        Return the top tracks for an artist on the given provider.
458
459        Each track is resolved to its in-library equivalent where available.
460        """
461        provider = self.mass.get_provider(
462            provider_instance_id_or_domain, provider_type=MusicProvider
463        )
464        if provider is None or not provider.available:
465            return []  # guard against unavailable provider
466        if not provider.supports_feature(ProviderFeature.ARTIST_TOPTRACKS):
467            self.logger.warning(
468                "Provider %s does not support fetching artist top tracks.",
469                provider.name,
470            )
471            return []  # guard against unsupported feature
472        tracks = await provider.get_artist_toptracks(item_id)
473        # resolve to in-library equivalents (in parallel) where available
474        resolved = await asyncio.gather(
475            *(
476                self.mass.music.tracks.get_library_item_by_prov_id(track.item_id, track.provider)
477                for track in tracks
478            )
479        )
480        return [
481            library_track or track for library_track, track in zip(resolved, tracks, strict=True)
482        ]
483
484    async def get_library_artist_toptracks(
485        self,
486        item_id: str | int,
487        provider_filter: str | None = None,
488    ) -> list[Track]:
489        """
490        Return the top tracks for an in-library artist, aggregated across all its providers.
491
492        The result combines (and deduplicates, preserving order) the top tracks from every
493        provider attached to the artist and any metadata/plugin provider implementing the
494        feature. Empty when no provider yields a result.
495
496        :param item_id: The library item ID of the artist.
497        :param provider_filter: Optional provider instance ID to limit the result to.
498        """
499        ref_item = await self.get_library_item(item_id)
500        allowed = self._ensure_provider_filter(provider_filter)
501        # fetch each provider's ranked top tracks in parallel
502        fetches = []
503        # streaming providers attached to the artist (results resolved to library items)
504        for provider_mapping in ref_item.provider_mappings:
505            if allowed is not None and provider_mapping.provider_instance not in allowed:
506                continue
507            music_prov = self.mass.get_provider(
508                provider_mapping.provider_instance, provider_type=MusicProvider
509            )
510            if (
511                music_prov is None
512                or ProviderFeature.ARTIST_TOPTRACKS not in music_prov.supported_features
513            ):
514                continue
515            fetches.append(
516                self.get_provider_artist_toptracks(
517                    provider_mapping.item_id, provider_mapping.provider_instance
518                )
519            )
520        # metadata/plugin providers implementing the feature
521        for prov in self.mass.get_providers_supporting_feature(
522            ProviderFeature.ARTIST_TOPTRACKS,
523            priority=(ProviderType.METADATA, ProviderType.PLUGIN),
524        ):
525            if allowed is not None and prov.instance_id not in allowed:
526                continue
527            fetches.append(cast("MetadataProvider", prov).get_artist_toptracks(ref_item))
528        per_provider = await asyncio.gather(*fetches, return_exceptions=True)
529        # drop (and log) any provider that failed so one bad provider can't sink the listing
530        listings: list[list[Track]] = []
531        for listing in per_provider:
532            if isinstance(listing, BaseException):
533                self.logger.warning(
534                    "Error fetching top tracks for artist %s from a provider",
535                    ref_item.name,
536                    exc_info=listing,
537                )
538                continue
539            listings.append(listing)
540        # interleave the providers' rankings by position (zip), deduplicating with the compare
541        # helper (which also matches on version/duration)
542        result: list[Track] = []
543        for row in zip_longest(*listings):
544            for candidate in row:
545                if candidate is None or any(
546                    compare_track(existing, candidate) for existing in result
547                ):
548                    continue
549                result.append(candidate)
550        return result
551
552    async def get_provider_artist_topalbums(
553        self,
554        item_id: str,
555        provider_instance_id_or_domain: str,
556    ) -> list[Album]:
557        """
558        Return the top/featured albums for an artist on the given provider.
559
560        Each album is resolved to its in-library equivalent where available.
561        """
562        provider = self.mass.get_provider(
563            provider_instance_id_or_domain, provider_type=MusicProvider
564        )
565        if provider is None or not provider.available:
566            return []  # guard against unavailable provider
567        if not provider.supports_feature(ProviderFeature.ARTIST_TOPALBUMS):
568            self.logger.warning(
569                "Provider %s does not support fetching artist top albums.",
570                provider.name,
571            )
572            return []  # guard against unsupported feature
573        albums = await provider.get_artist_topalbums(item_id)
574        # resolve to in-library equivalents (in parallel) where available
575        resolved = await asyncio.gather(
576            *(
577                self.mass.music.albums.get_library_item_by_prov_id(album.item_id, album.provider)
578                for album in albums
579            )
580        )
581        return [
582            library_album or album for library_album, album in zip(resolved, albums, strict=True)
583        ]
584
585    async def get_library_artist_topalbums(
586        self,
587        item_id: str | int,
588        provider_filter: str | None = None,
589    ) -> list[Album]:
590        """
591        Return the top albums for an in-library artist, aggregated across all its providers.
592
593        The result combines (and deduplicates, preserving order) the top albums from every
594        provider attached to the artist and any metadata/plugin provider implementing the
595        feature. Empty when no provider yields a result.
596
597        :param item_id: The library item ID of the artist.
598        :param provider_filter: Optional provider instance ID to limit the result to.
599        """
600        ref_item = await self.get_library_item(item_id)
601        allowed = self._ensure_provider_filter(provider_filter)
602        # fetch each provider's ranked top albums in parallel
603        fetches = []
604        # streaming providers attached to the artist (results resolved to library items)
605        for provider_mapping in ref_item.provider_mappings:
606            if allowed is not None and provider_mapping.provider_instance not in allowed:
607                continue
608            music_prov = self.mass.get_provider(
609                provider_mapping.provider_instance, provider_type=MusicProvider
610            )
611            if (
612                music_prov is None
613                or ProviderFeature.ARTIST_TOPALBUMS not in music_prov.supported_features
614            ):
615                continue
616            fetches.append(
617                self.get_provider_artist_topalbums(
618                    provider_mapping.item_id, provider_mapping.provider_instance
619                )
620            )
621        # metadata/plugin providers implementing the feature
622        for prov in self.mass.get_providers_supporting_feature(
623            ProviderFeature.ARTIST_TOPALBUMS,
624            priority=(ProviderType.METADATA, ProviderType.PLUGIN),
625        ):
626            if allowed is not None and prov.instance_id not in allowed:
627                continue
628            fetches.append(cast("MetadataProvider", prov).get_artist_topalbums(ref_item))
629        per_provider = await asyncio.gather(*fetches, return_exceptions=True)
630        # drop (and log) any provider that failed so one bad provider can't sink the listing
631        listings: list[list[Album]] = []
632        for listing in per_provider:
633            if isinstance(listing, BaseException):
634                self.logger.warning(
635                    "Error fetching top albums for artist %s from a provider",
636                    ref_item.name,
637                    exc_info=listing,
638                )
639                continue
640            listings.append(listing)
641        # interleave the providers' rankings by position (zip), deduplicating with the compare
642        # helper (which also matches on version/duration)
643        result: list[Album] = []
644        for row in zip_longest(*listings):
645            for candidate in row:
646                if candidate is None or any(
647                    compare_album(existing, candidate) for existing in result
648                ):
649                    continue
650                result.append(candidate)
651        return result
652
653    async def get_provider_artist_tracks(
654        self,
655        item_id: str,
656        provider_instance_id_or_domain: str,
657    ) -> list[Track]:
658        """Return all tracks for an artist on given provider."""
659        provider = self.mass.get_provider(
660            provider_instance_id_or_domain, provider_type=MusicProvider
661        )
662        if provider is None or not provider.available:
663            return []  # guard against unavailable provider
664        if provider.supports_feature(ProviderFeature.ARTIST_TRACKS):
665            return await provider.get_artist_tracks(item_id)
666        # fallback: enumerate (and dedupe) the tracks of all the artist's albums on the provider
667        result: list[Track] = []
668        unique_ids: set[str] = set()
669        for album in await self.get_provider_artist_albums(item_id, provider_instance_id_or_domain):
670            for track in await self.mass.music.albums.tracks(album.item_id, album.provider):
671                unique_id = f"{track.name}.{track.version}"
672                if unique_id in unique_ids:
673                    continue
674                unique_ids.add(unique_id)
675                result.append(track)
676        return result
677
678    async def get_library_artist_tracks(
679        self,
680        item_id: str | int,
681        provider_filter: str | None = None,
682    ) -> list[Track]:
683        """Return all in-library tracks for an artist, optionally limited to a single provider."""
684        db_id = int(item_id)  # ensure integer
685        library_item = await self.get_library_item(db_id)
686        if library_item.artist_type != ArtistType.SINGER:
687            self.logger.debug("Tracks only available for artists of type ARTIST")
688            return []
689        subquery = f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id"
690        query = f"tracks.item_id in ({subquery})"
691        return await self.mass.music.tracks.get_library_items_by_query(
692            extra_query_parts=[query],
693            extra_query_params={"artist_id": db_id},
694            provider_filter=self._ensure_provider_filter(provider_filter),
695            in_library_only=True,
696        )
697
698    async def get_provider_artist_albums(
699        self,
700        item_id: str,
701        provider_instance_id_or_domain: str,
702    ) -> list[Album]:
703        """Return albums for an artist on given provider."""
704        provider = self.mass.get_provider(
705            provider_instance_id_or_domain, provider_type=MusicProvider
706        )
707        if provider is None or not provider.available:
708            return []  # guard against unavailable provider
709        if not provider.supports_feature(ProviderFeature.ARTIST_ALBUMS):
710            self.logger.warning(
711                "Provider %s does not support fetching all artist albums.",
712                provider.name,
713            )
714            return []  # guard against unsupported feature
715        return await provider.get_artist_albums(item_id)
716
717    async def get_library_artist_albums(
718        self,
719        item_id: str | int,
720        provider_filter: str | None = None,
721    ) -> list[Album]:
722        """Return all in-library albums for an artist, optionally limited to a single provider."""
723        db_id = int(item_id)  # ensure integer
724        library_item = await self.get_library_item(db_id)
725        if library_item.artist_type != ArtistType.SINGER:
726            self.logger.debug("Albums only available for artists of type ARTIST")
727            return []
728        subquery = f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id"
729        query = f"albums.item_id in ({subquery})"
730        return await self.mass.music.albums.get_library_items_by_query(
731            extra_query_parts=[query],
732            extra_query_params={"artist_id": db_id},
733            provider_filter=self._ensure_provider_filter(provider_filter),
734            in_library_only=True,
735        )
736
737    async def get_provider_artist_similar_artists(
738        self,
739        item_id: str,
740        provider_instance_id_or_domain: str,
741        limit: int = 25,
742    ) -> list[Artist]:
743        """
744        Return similar artists for an artist on the given provider.
745
746        Each artist is resolved to its in-library equivalent where available.
747        """
748        provider = self.mass.get_provider(
749            provider_instance_id_or_domain, provider_type=MusicProvider
750        )
751        if provider is None or not provider.available:
752            return []  # guard against unavailable provider
753        if not provider.supports_feature(ProviderFeature.SIMILAR_ARTISTS):
754            self.logger.warning(
755                "Provider %s does not support fetching similar artists.",
756                provider.name,
757            )
758            return []  # guard against unsupported feature
759        artists = await provider.get_similar_artists(item_id, limit=limit)
760        # resolve to in-library equivalents (in parallel) where available
761        resolved = await asyncio.gather(
762            *(
763                self.get_library_item_by_prov_id(artist.item_id, artist.provider)
764                for artist in artists
765            )
766        )
767        return [
768            library_artist or artist
769            for library_artist, artist in zip(resolved, artists, strict=True)
770        ]
771
772    async def get_library_artist_similar_artists(
773        self,
774        item_id: str | int,
775        provider_filter: str | None = None,
776        limit: int = 25,
777    ) -> list[Artist]:
778        """
779        Return similar artists for an in-library artist, aggregated across all its providers.
780
781        The result combines (and deduplicates, preserving order) the similar artists from
782        every provider attached to the artist and any metadata/plugin provider implementing
783        the feature. Empty when no provider yields a result.
784
785        :param item_id: The library item ID of the artist.
786        :param provider_filter: Optional provider instance ID to limit the result to.
787        :param limit: Maximum number of similar artists to return.
788        """
789        ref_item = await self.get_library_item(item_id)
790        allowed = self._ensure_provider_filter(provider_filter)
791        # fetch each provider's similar artists in parallel
792        fetches = []
793        # streaming providers attached to the artist (results resolved to library items)
794        for provider_mapping in ref_item.provider_mappings:
795            if allowed is not None and provider_mapping.provider_instance not in allowed:
796                continue
797            music_prov = self.mass.get_provider(
798                provider_mapping.provider_instance, provider_type=MusicProvider
799            )
800            if (
801                music_prov is None
802                or ProviderFeature.SIMILAR_ARTISTS not in music_prov.supported_features
803            ):
804                continue
805            fetches.append(
806                self.get_provider_artist_similar_artists(
807                    provider_mapping.item_id, provider_mapping.provider_instance, limit=limit
808                )
809            )
810        # metadata/plugin providers implementing the feature
811        for prov in self.mass.get_providers_supporting_feature(
812            ProviderFeature.SIMILAR_ARTISTS,
813            priority=(ProviderType.METADATA, ProviderType.PLUGIN),
814        ):
815            if allowed is not None and prov.instance_id not in allowed:
816                continue
817            fetches.append(
818                cast("MetadataProvider", prov).get_similar_artists(ref_item, limit=limit)
819            )
820        per_provider = await asyncio.gather(*fetches, return_exceptions=True)
821        # drop (and log) any provider that failed so one bad provider can't sink the listing
822        listings: list[list[Artist]] = []
823        for listing in per_provider:
824            if isinstance(listing, BaseException):
825                self.logger.warning(
826                    "Error fetching similar artists for %s from a provider",
827                    ref_item.name,
828                    exc_info=listing,
829                )
830                continue
831            listings.append(listing)
832        # interleave the providers' results by position (zip), deduplicating with the compare
833        # helper, and cap to the requested limit
834        result: list[Artist] = []
835        for row in zip_longest(*listings):
836            for candidate in row:
837                if candidate is None or any(
838                    compare_artist(existing, candidate) for existing in result
839                ):
840                    continue
841                result.append(candidate)
842        return result[:limit]
843
844    async def get_library_artist_types(self) -> list[ArtistType]:
845        """Get all supported in-library artist types."""
846        artist_types: list[ArtistType] = []
847        query = f"SELECT DISTINCT artist_type FROM {DB_TABLE_ARTISTS}"
848        rows = await self.mass.music.database.get_rows_from_query(query)
849        for row in rows:
850            artist_types.append(ArtistType(row["artist_type"]))
851        return artist_types
852
853    async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
854        """Delete record from the database."""
855        db_id = int(item_id)  # ensure integer
856        library_item = await self.get_library_item(db_id)
857
858        if library_item.artist_type == ArtistType.SINGER:
859            await self._remove_music_artist_from_library(db_id=db_id, recursive=recursive)
860        elif library_item.artist_type in (ArtistType.AUTHOR, ArtistType.NARRATOR):
861            await self._remove_author_narrator_from_library(db_id=db_id, recursive=recursive)
862        else:
863            raise MusicAssistantError(f"Unknown artist_type {library_item.artist_type}.")
864
865        # delete the artist itself from db
866        # this will raise if the item still has references and recursive is false
867        await super().remove_item_from_library(db_id)
868
869    async def match_provider(
870        self, db_artist: Artist, provider: MusicProvider, strict: bool = True
871    ) -> list[ProviderMapping]:
872        """
873        Try to find match on (streaming) provider for the provided (database) artist.
874
875        This is used to link objects of different providers/qualities together.
876        """
877        self.logger.debug("Trying to match artist %s on provider %s", db_artist.name, provider.name)
878        matches: list[ProviderMapping] = []
879        # try to get a match with some reference tracks of this artist
880        ref_tracks = await self.mass.music.artists.tracks(db_artist.item_id, db_artist.provider)
881        if len(ref_tracks) < 10:
882            # fetch reference tracks from provider(s) attached to the artist
883            for provider_mapping in db_artist.provider_mappings:
884                with contextlib.suppress(ProviderUnavailableError, MediaNotFoundError):
885                    ref_tracks += await self.mass.music.artists.tracks(
886                        provider_mapping.item_id, provider_mapping.provider_instance
887                    )
888        for ref_track in ref_tracks:
889            search_str = f"{db_artist.name} - {ref_track.name}"
890            search_results = await self.mass.music.tracks.search(search_str, provider.domain)
891            for search_result_item in search_results:
892                if not compare_strings(search_result_item.name, ref_track.name, strict=strict):
893                    continue
894                # get matching artist from track
895                for search_item_artist in search_result_item.artists:
896                    if not compare_strings(search_item_artist.name, db_artist.name, strict=strict):
897                        continue
898                    # 100% track match
899                    # get full artist details so we have all metadata
900                    prov_artist = await self.get_provider_item(
901                        search_item_artist.item_id,
902                        search_item_artist.provider,
903                        fallback=search_item_artist,
904                    )
905                    # 100% match
906                    matches.extend(prov_artist.provider_mappings)
907                    if matches:
908                        return matches
909        # try to get a match with some reference albums of this artist
910        ref_albums = await self.mass.music.artists.albums(db_artist.item_id, db_artist.provider)
911        if len(ref_albums) < 10:
912            # fetch reference albums from provider(s) attached to the artist
913            for provider_mapping in db_artist.provider_mappings:
914                with contextlib.suppress(ProviderUnavailableError, MediaNotFoundError):
915                    ref_albums += await self.mass.music.artists.albums(
916                        provider_mapping.item_id, provider_mapping.provider_instance
917                    )
918        for ref_album in ref_albums:
919            if ref_album.album_type == AlbumType.COMPILATION:
920                continue
921            if not ref_album.artists:
922                continue
923            search_str = f"{db_artist.name} - {ref_album.name}"
924            search_result_albums = await self.mass.music.albums.search(search_str, provider.domain)
925            for search_result_album in search_result_albums:
926                if not search_result_album.artists:
927                    continue
928                if not compare_strings(search_result_album.name, ref_album.name, strict=strict):
929                    continue
930                # artist must match 100%
931                if not compare_artist(db_artist, search_result_album.artists[0], strict=strict):
932                    continue
933                # 100% match
934                # get full artist details so we have all metadata
935                prov_artist = await self.get_provider_item(
936                    search_result_album.artists[0].item_id,
937                    search_result_album.artists[0].provider,
938                    fallback=search_result_album.artists[0],
939                )
940                matches.extend(prov_artist.provider_mappings)
941                if matches:
942                    return matches
943        if not matches:
944            self.logger.debug(
945                "Could not find match for Artist %s on provider %s",
946                db_artist.name,
947                provider.name,
948            )
949        return matches
950
951    async def match_providers(self, db_artist: Artist) -> None:
952        """
953        Try to find matching artists on all providers for the provided (database) item_id.
954
955        This is used to link objects of different providers together.
956        """
957        if db_artist.provider != "library":
958            return  # Matching only supported for database items
959
960        # try to find match on all providers
961
962        cur_provider_domains = {
963            x.provider_domain for x in db_artist.provider_mappings if x.available
964        }
965        for provider in self.mass.music.providers:
966            if provider.domain in cur_provider_domains:
967                continue
968            if ProviderFeature.SEARCH not in provider.supported_features:
969                continue
970            if not self.mass.music.library_supported(provider, MediaType.ARTIST):
971                continue
972            if not provider.is_streaming_provider:
973                # matching on unique providers is pointless as they push (all) their content to MA
974                continue
975            if match := await self.match_provider(db_artist, provider):
976                # 100% match, we update the db with the additional provider mapping(s)
977                await self.add_provider_mappings(db_artist.item_id, match)
978                cur_provider_domains.add(provider.domain)
979
980    def artist_from_item_mapping(self, item: ItemMapping) -> Artist:
981        """Create an Artist object from an ItemMapping object."""
982        domain, instance_id = None, None
983        if prov := self.mass.get_provider(item.provider):
984            domain = prov.domain
985            instance_id = prov.instance_id
986        return Artist.from_dict(
987            {
988                **item.to_dict(),
989                "provider_mappings": [
990                    {
991                        "item_id": item.item_id,
992                        "provider_domain": domain,
993                        "provider_instance": instance_id,
994                        "available": item.available,
995                    }
996                ],
997            }
998        )
999
1000    def _validate_provider_filter(
1001        self, provider_instance_id_or_domain: str, provider_filter: str | None
1002    ) -> None:
1003        """Raise when a provider filter is set that does not match the requested provider."""
1004        if provider_filter is not None and provider_filter != provider_instance_id_or_domain:
1005            raise MusicAssistantError(
1006                f"provider_filter '{provider_filter}' does not match the requested "
1007                f"provider '{provider_instance_id_or_domain}'"
1008            )
1009
1010    async def _add_library_item(
1011        self, item: Artist | ItemMapping, overwrite_existing: bool = False
1012    ) -> int:
1013        """Add a new item record to the database."""
1014        # If item is an ItemMapping, convert it
1015        if isinstance(item, ItemMapping):
1016            item = self.artist_from_item_mapping(item)
1017        # enforce various artists name + id
1018        if compare_strings(item.name, VARIOUS_ARTISTS_NAME):
1019            item.mbid = VARIOUS_ARTISTS_MBID
1020        if item.mbid == VARIOUS_ARTISTS_MBID:
1021            item.name = VARIOUS_ARTISTS_NAME
1022        # no existing item matched: insert item
1023        db_id = await self.mass.music.database.insert(
1024            self.db_table,
1025            {
1026                "name": item.name,
1027                "sort_name": item.sort_name,
1028                "favorite": item.favorite,
1029                "metadata": serialize_to_json(item.metadata),
1030                "search_name": create_safe_string(item.name, True, True),
1031                "search_sort_name": create_safe_string(item.sort_name or "", True, True),
1032                "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
1033                "artist_type": item.artist_type,
1034            },
1035        )
1036        # update/set external id lookup table
1037        await self.set_external_ids(db_id, item.external_ids)
1038        # update/set provider_mappings table
1039        await self.set_provider_mappings(db_id, item.provider_mappings)
1040        self.logger.debug("added %s to database (id: %s)", item.name, db_id)
1041        return db_id
1042
1043    async def _update_library_item(
1044        self, item_id: str | int, update: Artist | ItemMapping, overwrite: bool = False
1045    ) -> None:
1046        """Update existing record in the database."""
1047        db_id = int(item_id)  # ensure integer
1048        cur_item = await self.get_library_item(db_id)
1049        if isinstance(update, ItemMapping):
1050            # NOTE that artist is the only mediatype where its accepted we
1051            # receive an itemmapping from streaming providers
1052            update = self.artist_from_item_mapping(update)
1053            metadata = cur_item.metadata
1054        else:
1055            metadata = update.metadata if overwrite else cur_item.metadata.update(update.metadata)
1056        cur_item.external_ids.update(update.external_ids)
1057        # enforce various artists name + id
1058        mbid = cur_item.mbid
1059        if (not mbid or overwrite) and getattr(update, "mbid", None):
1060            if compare_strings(update.name, VARIOUS_ARTISTS_NAME):
1061                update.mbid = VARIOUS_ARTISTS_MBID
1062            if update.mbid == VARIOUS_ARTISTS_MBID:
1063                update.name = VARIOUS_ARTISTS_NAME
1064
1065        name = update.name if overwrite else cur_item.name
1066        sort_name = update.sort_name if overwrite else cur_item.sort_name or update.sort_name
1067        await self.mass.music.database.update(
1068            self.db_table,
1069            {"item_id": db_id},
1070            {
1071                "name": name,
1072                "sort_name": sort_name,
1073                "metadata": serialize_to_json(metadata),
1074                "search_name": create_safe_string(name, True, True),
1075                "search_sort_name": create_safe_string(sort_name or "", True, True),
1076                "timestamp_added": int(update.date_added.timestamp())
1077                if update.date_added
1078                else UNSET,
1079                "artist_type": update.artist_type,
1080            },
1081        )
1082        self.logger.debug("updated %s in database: %s", update.name, db_id)
1083        # update/set external id lookup table
1084        await self.set_external_ids(
1085            db_id, update.external_ids if overwrite else cur_item.external_ids
1086        )
1087        # update/set provider_mappings table
1088        provider_mappings = (
1089            update.provider_mappings
1090            if overwrite
1091            else {*update.provider_mappings, *cur_item.provider_mappings}
1092        )
1093        await self.set_provider_mappings(db_id, provider_mappings, overwrite)
1094        self.logger.debug("updated %s in database: (id %s)", update.name, db_id)
1095
1096    async def _remove_music_artist_from_library(self, db_id: int, recursive: bool) -> None:
1097        # recursively also remove artist albums
1098        for db_row in await self.mass.music.database.get_rows_from_query(
1099            f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id",
1100            {"artist_id": db_id},
1101            limit=5000,
1102        ):
1103            if not recursive:
1104                raise MusicAssistantError("Artist still has albums linked")
1105            with contextlib.suppress(MediaNotFoundError):
1106                await self.mass.music.albums.remove_item_from_library(db_row["album_id"])
1107        # recursively also remove artist tracks
1108        for db_row in await self.mass.music.database.get_rows_from_query(
1109            f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id",
1110            {"artist_id": db_id},
1111            limit=5000,
1112        ):
1113            if not recursive:
1114                raise MusicAssistantError("Artist still has tracks linked")
1115            with contextlib.suppress(MediaNotFoundError):
1116                await self.mass.music.tracks.remove_item_from_library(db_row["track_id"])
1117
1118    async def _remove_author_narrator_from_library(self, db_id: int, recursive: bool) -> None:
1119        # recursively also remove author/ narrator audiobooks
1120        for db_row in await self.mass.music.database.get_rows_from_query(
1121            f"SELECT audiobook_id FROM {DB_TABLE_AUDIOBOOK_ARTISTS} WHERE artist_id = :artist_id",
1122            {"artist_id": db_id},
1123            limit=5000,
1124        ):
1125            if not recursive:
1126                raise MusicAssistantError("Artist still has audiobooks linked")
1127            with contextlib.suppress(MediaNotFoundError):
1128                await self.mass.music.audiobooks.remove_item_from_library(db_row["audiobook_id"])
1129
1130    async def _get_db_author_narrator_audiobooks(
1131        self, item_id: str, provider_instance_id_or_domain: str, artist_type: ArtistType
1132    ) -> list[Audiobook]:
1133        if db_author_narrator := await self.mass.music.artists.get_library_item_by_prov_id(
1134            item_id,
1135            provider_instance_id_or_domain,
1136        ):
1137            if db_author_narrator.artist_type != artist_type:
1138                self.logger.debug("Artist type must be %s.", artist_type)
1139                return []
1140            db_artist_id = int(db_author_narrator.item_id)  # ensure integer
1141            subquery = f"SELECT audiobook_id FROM {DB_TABLE_AUDIOBOOK_ARTISTS} WHERE artist_id = :artist_id"
1142            query = f"audiobooks.item_id in ({subquery})"
1143            return await self.mass.music.audiobooks.get_library_items_by_query(
1144                extra_query_parts=[query],
1145                extra_query_params={"artist_id": db_artist_id},
1146                provider_filter=[provider_instance_id_or_domain],
1147            )
1148        return []
1149
1150    def _parse_summary_row(self, db_row: Mapping[str, Any]) -> ArtistSummary:
1151        """Parse a raw summary db row into an ArtistSummary object."""
1152        item = cast("ArtistSummary", super()._parse_summary_row(db_row))
1153        item.artist_type = ArtistType(db_row["artist_type"])
1154        return item
1155