Adapt artist / audiobook controller for authors and narrators

music_assistant/constants.py

@@ -160,6 +160,7 @@
 DB_TABLE_ALBUM_TRACKS: Final[str] = "album_tracks"
 DB_TABLE_TRACK_ARTISTS: Final[str] = "track_artists"
 DB_TABLE_ALBUM_ARTISTS: Final[str] = "album_artists"
+DB_TABLE_AUDIOBOOK_ARTISTS: Final[str] = "audiobook_artists"
 DB_TABLE_LOUDNESS_MEASUREMENTS: Final[str] = "loudness_measurements"
 DB_TABLE_AUDIO_ANALYSIS: Final[str] = "audio_analysis"
 DB_TABLE_GENRES: Final[str] = "genres"

music_assistant/controllers/media/artists.py

@@ -6,17 +6,25 @@
 import contextlib
 from typing import TYPE_CHECKING, Any, cast
 
-from music_assistant_models.enums import AlbumType, MediaType, ProviderFeature
+from music_assistant_models.enums import AlbumType, ArtistType, MediaType, ProviderFeature
 from music_assistant_models.errors import (
     MediaNotFoundError,
     MusicAssistantError,
     ProviderUnavailableError,
 )
-from music_assistant_models.media_items import Album, Artist, ItemMapping, ProviderMapping, Track
+from music_assistant_models.media_items import (
+    Album,
+    Artist,
+    Audiobook,
+    ItemMapping,
+    ProviderMapping,
+    Track,
+)
 
 from music_assistant.constants import (
     DB_TABLE_ALBUM_ARTISTS,
     DB_TABLE_ARTISTS,
+    DB_TABLE_AUDIOBOOK_ARTISTS,
     DB_TABLE_TRACK_ARTISTS,
     VARIOUS_ARTISTS_MBID,
     VARIOUS_ARTISTS_NAME,
@@ -46,13 +54,17 @@ def __init__(self, mass: MusicAssistant) -> None:
         api_base = self.api_base
         self.mass.register_api_command(f"music/{api_base}/artist_albums", self.albums)
         self.mass.register_api_command(f"music/{api_base}/artist_tracks", self.tracks)
+        self.mass.register_api_command(f"music/{api_base}/artist_audiobooks", self.audiobooks)
 
     async def library_count(
-        self, favorite_only: bool = False, album_artists_only: bool = False
+        self,
+        favorite_only: bool = False,
+        album_artists_only: bool = False,
+        artist_type: ArtistType = ArtistType.ARTIST,
     ) -> int:
         """Return the total number of items in the library."""
         sql_query = f"SELECT item_id FROM {self.db_table}"
-        query_parts: list[str] = []
+        query_parts = [f"artist_type = '{artist_type}'"]
         if favorite_only:
             query_parts.append("favorite = 1")
         if album_artists_only:
@@ -74,6 +86,7 @@ async def library_items(
         provider: str | list[str] | None = None,
         genre: int | list[int] | None = None,
         album_artists_only: bool = False,
+        artist_type: ArtistType = ArtistType.ARTIST,
         **kwargs: Any,
     ) -> list[Artist]:
         """Get in-database (album) artists.
@@ -86,10 +99,11 @@ async def library_items(
         :param provider: Filter by provider instance ID (single string or list).
         :param album_artists_only: Only return artists that have albums.
         :param genre: Filter by genre id(s).
+        :param artist_type: The artist's type
         """
         extra_query_params: dict[str, Any] = {}
-        extra_query_parts: list[str] = []
-        if album_artists_only:
+        extra_query_parts = [f"artist_type = '{artist_type}'"]
+        if album_artists_only and artist_type == ArtistType.ARTIST:
             extra_query_parts.append(
                 f"artists.item_id in (select {DB_TABLE_ALBUM_ARTISTS}.artist_id "
                 f"from {DB_TABLE_ALBUM_ARTISTS})"
@@ -114,7 +128,7 @@ async def tracks(
         in_library_only: bool = False,
         provider_filter: str | list[str] | None = None,
     ) -> list[Track]:
-        """Return all/top tracks for an artist."""
+        """Return all/top tracks for a music artist."""
         if provider_filter and provider_instance_id_or_domain != "library":
             raise MusicAssistantError("Cannot use provider_filter with specific provider request")
         if isinstance(provider_filter, str):
@@ -123,6 +137,11 @@ async def tracks(
         library_artist = await self.get_library_item_by_prov_id(
             item_id, provider_instance_id_or_domain
         )
+        if library_artist and library_artist.artist_type != ArtistType.ARTIST:
+            self.logger.debug(
+                "Ignoring tracks request for artist of type %s", library_artist.artist_type
+            )
+            return []
         if not library_artist:
             return await self.get_provider_artist_toptracks(item_id, provider_instance_id_or_domain)
         db_items = await self.get_library_artist_tracks(library_artist.item_id)
@@ -167,6 +186,11 @@ async def albums(
         library_artist = await self.get_library_item_by_prov_id(
             item_id, provider_instance_id_or_domain
         )
+        if library_artist and library_artist.artist_type != ArtistType.ARTIST:
+            self.logger.debug(
+                "Ignoring album request for artist of type %s", library_artist.artist_type
+            )
+            return []
         if not library_artist:
             return await self.get_provider_artist_albums(item_id, provider_instance_id_or_domain)
         db_items = await self.get_library_artist_albums(library_artist.item_id)
@@ -201,7 +225,20 @@ async def albums(
     async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
         """Delete record from the database."""
         db_id = int(item_id)  # ensure integer
+        library_item = await self.get_library_item(db_id)
 
+        if library_item.artist_type == ArtistType.ARTIST:
+            await self._remove_music_artist_from_library(db_id=db_id, recursive=recursive)
+        elif library_item.artist_type in (ArtistType.AUTHOR, ArtistType.NARRATOR):
+            await self._remove_author_narrator_from_library(db_id=db_id, recursive=recursive)
+        else:
+            raise MusicAssistantError(f"Unknown artist_type {library_item.artist_type}.")
+
+        # delete the artist itself from db
+        # this will raise if the item still has references and recursive is false
+        await super().remove_item_from_library(db_id)
+
+    async def _remove_music_artist_from_library(self, db_id: int, recursive: bool) -> None:
         # recursively also remove artist albums
         for db_row in await self.mass.music.database.get_rows_from_query(
             f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id",
@@ -223,9 +260,17 @@ async def remove_item_from_library(self, item_id: str | int, recursive: bool = T
             with contextlib.suppress(MediaNotFoundError):
                 await self.mass.music.tracks.remove_item_from_library(db_row["track_id"])
 
-        # delete the artist itself from db
-        # this will raise if the item still has references and recursive is false
-        await super().remove_item_from_library(db_id)
+    async def _remove_author_narrator_from_library(self, db_id: int, recursive: bool) -> None:
+        # recursively also remove author/ narrator audiobooks
+        for db_row in await self.mass.music.database.get_rows_from_query(
+            f"SELECT album_id FROM {DB_TABLE_AUDIOBOOK_ARTISTS} WHERE artist_id = :artist_id",
+            {"artist_id": db_id},
+            limit=5000,
+        ):
+            if not recursive:
+                raise MusicAssistantError("Artist still has albums linked")
+            with contextlib.suppress(MediaNotFoundError):
+                await self.mass.music.audiobooks.remove_item_from_library(db_row["album_id"])
 
     async def get_provider_artist_toptracks(
         self,
@@ -244,6 +289,9 @@ async def get_provider_artist_toptracks(
             item_id,
             provider_instance_id_or_domain,
         ):
+            if db_artist.artist_type != ArtistType.ARTIST:
+                self.logger.debug("Top tracks only available for artists of type ARTIST")
+                return []
             db_artist_id = int(db_artist.item_id)  # ensure integer
             subquery = f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id"
             query = f"tracks.item_id in ({subquery})"
@@ -260,6 +308,10 @@ async def get_library_artist_tracks(
     ) -> list[Track]:
         """Return all tracks for an artist in the library/db."""
         db_id = int(item_id)  # ensure integer
+        library_item = await self.get_library_item(db_id)
+        if library_item.artist_type != ArtistType.ARTIST:
+            self.logger.debug("Tracks only available for artists of type ARTIST")
+            return []
         subquery = f"SELECT track_id FROM {DB_TABLE_TRACK_ARTISTS} WHERE artist_id = :artist_id"
         query = f"tracks.item_id in ({subquery})"
         return await self.mass.music.tracks.get_library_items_by_query(
@@ -284,6 +336,9 @@ async def get_provider_artist_albums(
             item_id,
             provider_instance_id_or_domain,
         ):
+            if db_artist.artist_type != ArtistType.ARTIST:
+                self.logger.debug("Albums only available for artists of type ARTIST")
+                return []
             db_artist_id = int(db_artist.item_id)  # ensure integer
             subquery = f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id"
             query = f"albums.item_id in ({subquery})"
@@ -300,6 +355,10 @@ async def get_library_artist_albums(
     ) -> list[Album]:
         """Return all in-library albums for an artist."""
         db_id = int(item_id)  # ensure integer
+        library_item = await self.get_library_item(db_id)
+        if library_item.artist_type != ArtistType.ARTIST:
+            self.logger.debug("Albums only available for artists of type ARTIST")
+            return []
         subquery = f"SELECT album_id FROM {DB_TABLE_ALBUM_ARTISTS} WHERE artist_id = :artist_id"
         query = f"albums.item_id in ({subquery})"
         return await self.mass.music.albums.get_library_items_by_query(
@@ -331,6 +390,7 @@ async def _add_library_item(
                 "search_name": create_safe_string(item.name, True, True),
                 "search_sort_name": create_safe_string(item.sort_name or "", True, True),
                 "timestamp_added": int(item.date_added.timestamp()) if item.date_added else UNSET,
+                "artist_type": item.artist_type,
             },
         )
         # update/set provider_mappings table
@@ -377,6 +437,7 @@ async def _update_library_item(
                 "timestamp_added": int(update.date_added.timestamp())
                 if update.date_added
                 else UNSET,
+                "artist_type": update.artist_type,
             },
         )
         self.logger.debug("updated %s in database: %s", update.name, db_id)
@@ -400,6 +461,8 @@ async def radio_mode_base_tracks(
         :param item: The Artist to get base tracks for.
         :param preferred_provider_instances: List of preferred provider instance IDs to use.
         """
+        if item.artist_type != ArtistType.ARTIST:
+            raise MusicAssistantError("Radio mode tracks only exists for artists of type ARTIST.")
         return await self.tracks(
             item.item_id,
             item.provider,
@@ -535,3 +598,146 @@ def artist_from_item_mapping(self, item: ItemMapping) -> Artist:
                 ],
             }
         )
+
+    async def audiobooks(
+        self,
+        item_id: str,
+        provider_instance_id_or_domain: str,
+        artist_type: ArtistType = ArtistType.AUTHOR,
+        in_library_only: bool = False,
+    ) -> list[Audiobook]:
+        """Return audiobooks for an artist.
+
+        Artist_type can be omitted for in-library artists.
+        """
+        if artist_type == ArtistType.ARTIST:
+            self.logger.warning("Audiobooks not supported for artist_type ARTIST.")
+            return []
+        # always check if we have a library item for this artist
+        library_artist = await self.get_library_item_by_prov_id(
+            item_id, provider_instance_id_or_domain
+        )
+        if library_artist and library_artist.artist_type == ArtistType.ARTIST:
+            self.logger.debug(
+                "Ignoring audiobook request for artist of type %s", library_artist.artist_type
+            )
+            return []
+        if not library_artist:
+            if artist_type == ArtistType.AUTHOR:
+                return await self.get_provider_author_audiobooks(
+                    item_id, provider_instance_id_or_domain
+                )
+            if artist_type == ArtistType.NARRATOR:
+                return await self.get_provider_narrator_audiobooks(
+                    item_id, provider_instance_id_or_domain
+                )
+            return []
+
+        db_items = await self.get_library_author_narrator_audiobooks(
+            library_artist.item_id, artist_type=library_artist.artist_type
+        )
+        result: list[Audiobook] = db_items
+        if in_library_only:
+            # return in-library items only
+            return result
+        # return all (unique) items from all providers
+        # initialize unique_ids with db_items to prevent duplicates
+        unique_ids: set[str] = {f"{item.name}.{item.version}" for item in db_items}
+        unique_providers = self.mass.music.get_unique_providers()
+        for provider_mapping in library_artist.provider_mappings:
+            if provider_mapping.provider_instance not in unique_providers:
+                continue
+            provider_audiobooks = await self.get_provider_author_audiobooks(
+                provider_mapping.item_id, provider_mapping.provider_instance
+            )
+            for provider_audiobook in provider_audiobooks:
+                unique_id = f"{provider_audiobook.name}.{provider_audiobook.version}"
+                if unique_id in unique_ids:
+                    continue
+                unique_ids.add(unique_id)
+                # prefer db item
+                if db_item := await self.mass.music.audiobooks.get_library_item_by_prov_id(
+                    provider_audiobook.item_id, provider_audiobook.provider
+                ):
+                    result.append(db_item)
+                elif not in_library_only:
+                    result.append(provider_audiobook)
+        return result
+
+    async def get_provider_author_audiobooks(
+        self,
+        item_id: str,
+        provider_instance_id_or_domain: str,
+    ) -> list[Audiobook]:
+        """Return audiobooks for an author on given provider."""
+        assert provider_instance_id_or_domain != "library"
+        if not (prov := self.mass.get_provider(provider_instance_id_or_domain)):
+            return []
+        prov = cast("MusicProvider", prov)
+        if ProviderFeature.AUTHOR_AUDIOBOOKS in prov.supported_features:
+            return await prov.get_author_audiobooks(item_id)
+        # fallback implementation using the db
+        return await self._get_db_author_narrator_audiobooks(
+            item_id=item_id,
+            provider_instance_id_or_domain=provider_instance_id_or_domain,
+            artist_type=ArtistType.AUTHOR,
+        )
+
+    async def get_provider_narrator_audiobooks(
+        self,
+        item_id: str,
+        provider_instance_id_or_domain: str,
+    ) -> list[Audiobook]:
+        """Return audiobooks for an author on given provider."""
+        assert provider_instance_id_or_domain != "library"
+        if not (prov := self.mass.get_provider(provider_instance_id_or_domain)):
+            return []
+        prov = cast("MusicProvider", prov)
+        if ProviderFeature.NARRATOR_AUDIOBOOKS in prov.supported_features:
+            return await prov.get_author_audiobooks(item_id)
+        # fallback implementation using the db
+        return await self._get_db_author_narrator_audiobooks(
+            item_id=item_id,
+            provider_instance_id_or_domain=provider_instance_id_or_domain,
+            artist_type=ArtistType.NARRATOR,
+        )
+
+    async def _get_db_author_narrator_audiobooks(
+        self, item_id: str, provider_instance_id_or_domain: str, artist_type: ArtistType
+    ) -> list[Audiobook]:
+        if db_author_narrator := await self.mass.music.artists.get_library_item_by_prov_id(
+            item_id,
+            provider_instance_id_or_domain,
+        ):
+            if db_author_narrator.artist_type != artist_type:
+                self.logger.debug("Artist type must be %s.", artist_type)
+                return []
+            db_artist_id = int(db_author_narrator.item_id)  # ensure integer
+            subquery = f"SELECT audiobook_id FROM {DB_TABLE_AUDIOBOOK_ARTISTS} WHERE artist_id = :artist_id"
+            query = f"audiobooks.item_id in ({subquery})"
+            return await self.mass.music.audiobooks.get_library_items_by_query(
+                extra_query_parts=[query],
+                extra_query_params={"artist_id": db_artist_id},
+                provider_filter=[provider_instance_id_or_domain],
+            )
+        return []
+
+    async def get_library_author_narrator_audiobooks(
+        self,
+        item_id: str | int,
+        artist_type: ArtistType,
+    ) -> list[Audiobook]:
+        """Return all in-library audiobooks for an author/ narrator."""
+        db_id = int(item_id)  # ensure integer
+        library_item = await self.get_library_item(db_id)
+        if library_item.artist_type != artist_type:
+            self.logger.debug("Audiobooks only available for artists of type %s", artist_type)
+            return []
+        subquery = (
+            f"SELECT audiobook_id FROM {DB_TABLE_AUDIOBOOK_ARTISTS} WHERE artist_id = :artist_id"
+        )
+        query = f"audiobooks.item_id in ({subquery})"
+        return await self.mass.music.audiobooks.get_library_items_by_query(
+            extra_query_parts=[query],
+            extra_query_params={"artist_id": db_id},
+        )