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