music-assistant-server

1"""Base (ABC) MediaType specific controller."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from abc import ABCMeta, abstractmethod
8from collections.abc import Iterable
9from contextlib import suppress
10from contextvars import ContextVar
11from dataclasses import dataclass
12from datetime import UTC, datetime
13from typing import TYPE_CHECKING, Any, Literal, TypeVar, cast, final, overload
14
15from music_assistant_models.auth import Scope
16from music_assistant_models.enums import (
17    EventType,
18    ExternalID,
19    ImageType,
20    MediaType,
21    ProviderFeature,
22    ProviderType,
23)
24from music_assistant_models.errors import (
25    InsufficientPermissions,
26    MediaNotFoundError,
27    ProviderUnavailableError,
28)
29from music_assistant_models.helpers import create_safe_string, get_global_cache_value
30from music_assistant_models.media_items import (
31    AudioFormat,
32    ItemMapping,
33    ItemMappingSummary,
34    MediaCollection,
35    MediaItemImage,
36    MediaItemMetadata,
37    MediaItemMetadataSummary,
38    MediaItemSummaryType,
39    MediaItemType,
40    ProviderMapping,
41    UniqueList,
42)
43
44from music_assistant.constants import (
45    DB_TABLE_AUDIO_ANALYSIS,
46    DB_TABLE_EXTERNAL_ID_LOOKUP,
47    DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
48    DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
49    DB_TABLE_PLAYLOG,
50    DB_TABLE_PROVIDER_MAPPINGS,
51    MASS_LOGGER_NAME,
52)
53from music_assistant.controllers.music.helpers import search_name_match_clause
54from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
55from music_assistant.helpers.collections import (
56    get_collection_item_id,
57    get_collection_name_from_item_id,
58)
59from music_assistant.helpers.compare import compare_media_item
60from music_assistant.helpers.database import UNSET
61from music_assistant.helpers.json import json_loads, serialize_to_json
62from music_assistant.helpers.util import guard_single_request, parse_optional_bool
63
64if TYPE_CHECKING:
65    from collections.abc import AsyncGenerator, Mapping
66
67    from music_assistant import MusicAssistant
68    from music_assistant.models.music_provider import MusicProvider
69    from music_assistant.models.plugin import PluginProvider
70
71
72ItemCls = TypeVar("ItemCls", bound="MediaItemType")
73
74
75JSON_KEYS = (
76    "artists",
77    "track_album",
78    "metadata",
79    "provider_mappings",
80    "external_ids",
81    "narrators",
82    "authors",
83    "genre_aliases",
84    "supported_mediatypes",
85    "translation_params",
86    "audiobook_artists",
87)
88
89# When set (task-local), per-item MEDIA_ITEM_ADDED/UPDATED events and the on_item_updated
90# provider write-back are suppressed, so bulk operations (provider sync, provider cleanup)
91# don't flood subscribers with one event per touched item.
92SUPPRESS_MEDIA_ITEM_UPDATES: ContextVar[bool] = ContextVar(
93    "SUPPRESS_MEDIA_ITEM_UPDATES", default=False
94)
95
96SORT_KEYS = {
97    # sqlite has no builtin support for natural sorting
98    # so we have use an additional column for this
99    # this also improves searching and sorting performance
100    "name": "search_name ASC",
101    "name_desc": "search_name DESC",
102    "duration": "duration ASC",
103    "duration_desc": "duration DESC",
104    "sort_name": "search_sort_name ASC",
105    "sort_name_desc": "search_sort_name DESC",
106    "timestamp_added": "timestamp_added ASC",
107    "timestamp_added_desc": "timestamp_added DESC",
108    "timestamp_modified": "timestamp_modified ASC",
109    "timestamp_modified_desc": "timestamp_modified DESC",
110    "last_played": "last_played ASC",
111    "last_played_desc": "last_played DESC",
112    "play_count": "play_count ASC",
113    "play_count_desc": "play_count DESC",
114    "year": "year ASC",
115    "year_desc": "year DESC",
116    "position": "position ASC",
117    "position_desc": "position DESC",
118    "artist_name": "artists.search_name ASC, year DESC",
119    "artist_name_desc": "artists.search_name DESC, year DESC",
120    "random": "RANDOM()",
121    "random_play_count": "RANDOM(), play_count ASC",
122}
123
124
125@dataclass(slots=True)
126class LibraryItemSyncDetails:
127    """
128    Lightweight snapshot of a library item with just the fields the library sync needs.
129
130    Used by the provider sync loops to detect (un)changed items without hydrating
131    full MediaItem objects from the database.
132    """
133
134    item_id: int
135    favorite: bool
136    date_added: datetime
137    provider_mappings: set[ProviderMapping]
138
139
140@dataclass(slots=True)
141class TrackSyncDetails(LibraryItemSyncDetails):
142    """Lightweight sync snapshot of a library track."""
143
144    has_album: bool
145
146
147@dataclass(slots=True)
148class AudiobookSyncDetails(LibraryItemSyncDetails):
149    """Lightweight sync snapshot of a library audiobook."""
150
151    author_is_str: bool
152    narrator_is_str: bool
153    fully_played: bool | None
154    resume_position_ms: int | None
155
156
157class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta):
158    """Base model for controller managing a MediaType."""
159
160    media_type: MediaType
161    item_cls: type[MediaItemType]
162    summary_item_cls: type[MediaItemSummaryType]
163    db_table: str
164
165    def __init__(self, mass: MusicAssistant) -> None:
166        """Initialize class."""
167        self.mass = mass
168        self.logger = logging.getLogger(f"{MASS_LOGGER_NAME}.music.{self.media_type.value}")
169        # register (base) api handlers
170        self.api_base = api_base = f"{self.media_type}s"
171        self.mass.register_api_command(
172            f"music/{api_base}/count", self.library_count, required_scope=Scope.LIBRARY_READ
173        )
174        self.mass.register_api_command(
175            f"music/{api_base}/library_items",
176            self.library_items,
177            required_scope=Scope.LIBRARY_READ,
178            allow_impersonation=True,
179        )
180        self.mass.register_api_command(
181            f"music/{api_base}/get", self.get, required_scope=Scope.LIBRARY_READ
182        )
183        self.mass.register_api_command(
184            f"music/{api_base}/get_by_external_id",
185            self.get_library_item_by_external_id,
186            required_scope=Scope.LIBRARY_READ,
187        )
188        self.mass.register_api_command(
189            f"music/{api_base}/get_collection",
190            self.get_collection,
191            required_scope=Scope.LIBRARY_READ,
192            allow_impersonation=True,
193        )
194        # Backward compatibility alias - prefer the generic "get" endpoint
195        self.mass.register_api_command(
196            f"music/{api_base}/get_{self.media_type}",
197            self.get,
198            required_scope=Scope.LIBRARY_READ,
199            alias=True,
200        )
201        self.mass.register_api_command(
202            f"music/{api_base}/update",
203            self.update_item_in_library,
204            required_scope=Scope.LIBRARY_MANAGE,
205        )
206        self.mass.register_api_command(
207            f"music/{api_base}/remove",
208            self.remove_item_from_library,
209            required_scope=Scope.LIBRARY_MANAGE,
210        )
211        self._db_add_lock = asyncio.Lock()
212
213    @property
214    def translation_owner(self) -> str:
215        """Return the "core.music" namespace these media controllers' translation strings live under."""
216        return "core.music"
217
218    @property
219    def base_query(self) -> tuple[str, dict[str, Any]]:
220        """
221        Return the base SELECT query for this media type and its bound query params.
222
223        Override in a subclass to customize the query (extra joins/columns) and/or to
224        inject dynamic, parameterized filters.
225        """
226        query = f"""
227        SELECT
228            {self.db_table}.*,
229            {self._external_ids_query()} AS external_ids,
230            {self._provider_mappings_query()} AS provider_mappings
231            FROM {self.db_table} """
232        return query, {}
233
234    @property
235    def summary_query(self) -> tuple[str, dict[str, Any]]:
236        """
237        Return the slim SELECT query used for summary listings and its bound query params.
238
239        Selects only the columns needed to build summary items. Override in a subclass
240        to select additional per-type columns.
241        """
242        query = f"""
243        SELECT
244            {self._summary_base_columns()},
245            {self._provider_mappings_query()} AS provider_mappings
246            FROM {self.db_table} """
247        return query, {}
248
249    @final
250    async def add_item_to_library(
251        self,
252        item: ItemCls,
253        overwrite_existing: bool = False,
254    ) -> ItemCls:
255        """Add item to library and return the new (or updated) database item."""
256        new_item = False
257        # batch the many writes of an item add/update into a single commit
258        async with self.mass.music.database.deferred_commit():
259            # check for existing item first
260            if library_id := await self._get_library_item_by_match(item):
261                # update existing item
262                await self._update_library_item(library_id, item, overwrite=overwrite_existing)
263            else:
264                # actually add a new item in the library db
265                self.mass.music.match_provider_instances(item)
266                async with self._db_add_lock:
267                    library_id = await self._add_library_item(item)
268                    new_item = True
269        # return final library_item
270        library_item = await self.get_library_item(library_id)
271        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
272            self.mass.signal_event(
273                EventType.MEDIA_ITEM_ADDED if new_item else EventType.MEDIA_ITEM_UPDATED,
274                library_item.uri,
275                library_item,
276            )
277        return library_item
278
279    @final
280    async def update_item_in_library(
281        self, item_id: str | int, update: ItemCls, overwrite: bool = False
282    ) -> ItemCls:
283        """Update existing library record in the library database."""
284        self.mass.music.match_provider_instances(update)
285        # batch the many writes of an item update into a single commit
286        async with self.mass.music.database.deferred_commit():
287            await self._update_library_item(item_id, update, overwrite=overwrite)
288        # return the updated object
289        library_item = await self.get_library_item(item_id)
290        if SUPPRESS_MEDIA_ITEM_UPDATES.get():
291            # during a sync the update originates from the provider itself,
292            # so skip both the event and the write-back to that provider
293            return library_item
294        # drop cached artwork for the updated item so replaced art is served fresh
295        for img in library_item.metadata.images or []:
296            await self.mass.metadata.invalidate_image_cache(img.provider, img.path)
297        self.mass.signal_event(
298            EventType.MEDIA_ITEM_UPDATED,
299            library_item.uri,
300            library_item,
301        )
302        # notify music providers of the update so they can sync their own storage
303        for prov_mapping in library_item.provider_mappings:
304            if provider := self.mass.get_provider(prov_mapping.provider_instance):
305                if provider.type != ProviderType.MUSIC:
306                    continue
307                provider = cast("MusicProvider", provider)
308                await provider.on_item_updated(library_item)
309        return library_item
310
311    async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
312        """Delete library record from the database."""
313        db_id = int(item_id)  # ensure integer
314        library_item = await self.get_library_item(db_id)
315        assert library_item, f"Item does not exist: {db_id}"
316        # delete item
317        await self.mass.music.database.delete(
318            self.db_table,
319            {"item_id": db_id},
320        )
321        # update provider_mappings table
322        await self.mass.music.database.delete(
323            DB_TABLE_PROVIDER_MAPPINGS,
324            {"media_type": self.media_type.value, "item_id": db_id},
325        )
326        # cleanup external_id_lookup table
327        await self.mass.music.database.delete(
328            DB_TABLE_EXTERNAL_ID_LOOKUP,
329            {"media_type": self.media_type.value, "item_id": db_id},
330        )
331        # cleanup playlog table
332        await self.mass.music.database.delete(
333            DB_TABLE_PLAYLOG,
334            {
335                "media_type": self.media_type.value,
336                "item_id": db_id,
337                "provider": "library",
338            },
339        )
340        for prov_mapping in library_item.provider_mappings:
341            await self.mass.music.database.delete(
342                DB_TABLE_PLAYLOG,
343                {
344                    "media_type": self.media_type.value,
345                    "item_id": prov_mapping.item_id,
346                    "provider": prov_mapping.provider_instance,
347                },
348            )
349            # cleanup audio analysis rows for this provider mapping
350            for prov_key in (prov_mapping.provider_domain, prov_mapping.provider_instance):
351                await self.mass.music.database.delete(
352                    DB_TABLE_AUDIO_ANALYSIS,
353                    {
354                        "media_type": self.media_type.value,
355                        "item_id": prov_mapping.item_id,
356                        "provider": prov_key,
357                    },
358                )
359        # delete genre exclusions for this media item
360        await self.mass.music.database.delete(
361            DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
362            {"media_type": self.media_type.value, "media_id": db_id},
363        )
364        # NOTE: this does not delete any references to this item in other records,
365        # this is handled/overridden in the mediatype specific controllers
366        # drop cached artwork for the removed item
367        for img in library_item.metadata.images or []:
368            await self.mass.metadata.invalidate_image_cache(img.provider, img.path)
369        self.mass.signal_event(EventType.MEDIA_ITEM_DELETED, library_item.uri, library_item)
370        self.logger.debug("deleted item with id %s from database", db_id)
371
372    async def library_count(self, favorite_only: bool = False) -> int:
373        """
374        Return the number of items in the library.
375
376        Restricted to the providers the current user is allowed to see when that user
377        has a provider filter set.
378
379        :param favorite_only: Only count items marked as favorite.
380        """
381        query_parts: list[str] = []
382        query_params: dict[str, Any] = {}
383        if favorite_only:
384            query_parts.append("favorite = 1")
385        if provider_filter := self._ensure_provider_filter(None):
386            query_parts.append(
387                self._provider_filter_clause(query_params, provider_filter, in_library_only=True)
388            )
389        if not query_parts:
390            return await self.mass.music.database.get_count(self.db_table)
391        sql_query = f"SELECT item_id FROM {self.db_table} WHERE {' AND '.join(query_parts)}"
392        return await self.mass.music.database.get_count_from_query(sql_query, query_params)
393
394    if TYPE_CHECKING:
395
396        @overload
397        async def library_items(
398            self,
399            favorite: bool | None = None,
400            search: str | None = None,
401            limit: int = 500,
402            offset: int = 0,
403            order_by: str = "sort_name",
404            provider: str | list[str] | None = None,
405            genre: int | list[int] | None = None,
406            played_only: bool = False,
407            *,
408            summary: bool = True,
409            collapse_collections: Literal[False] = False,
410            **kwargs: Any,
411        ) -> list[ItemCls]: ...
412
413        @overload
414        async def library_items(
415            self,
416            favorite: bool | None = None,
417            search: str | None = None,
418            limit: int = 500,
419            offset: int = 0,
420            order_by: str = "sort_name",
421            provider: str | list[str] | None = None,
422            genre: int | list[int] | None = None,
423            played_only: bool = False,
424            *,
425            summary: bool = True,
426            collapse_collections: Literal[True],
427            **kwargs: Any,
428        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
429
430        @overload
431        async def library_items(
432            self,
433            favorite: bool | None = None,
434            search: str | None = None,
435            limit: int = 500,
436            offset: int = 0,
437            order_by: str = "sort_name",
438            provider: str | list[str] | None = None,
439            genre: int | list[int] | None = None,
440            played_only: bool = False,
441            *,
442            summary: bool = True,
443            collapse_collections: bool,
444            **kwargs: Any,
445        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
446
447    async def library_items(
448        self,
449        favorite: bool | None = None,
450        search: str | None = None,
451        limit: int = 500,
452        offset: int = 0,
453        order_by: str = "sort_name",
454        provider: str | list[str] | None = None,
455        genre: int | list[int] | None = None,
456        played_only: bool = False,
457        *,
458        summary: bool = True,
459        collapse_collections: bool = False,
460        **kwargs: Any,
461    ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]:
462        """
463        Get the library items for this mediatype.
464
465        :param favorite: Filter by favorite status.
466        :param search: Filter by search query.
467        :param limit: Maximum number of items to return.
468        :param offset: Number of items to skip.
469        :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
470        :param provider: Filter by provider instance ID (single string or list).
471        :param genre: Filter by genre id(s).
472        :param played_only: Only include items that have been played (last_played > 0).
473        :param summary: When True (default), return slim summary items containing only the
474            fields needed for a list view. Set to False to get fully hydrated items.
475        :param collapse_collections: Collapse available collections. Items in a collection won't
476            be returned individually.
477        """
478        items = await self.get_library_items_by_query(
479            favorite=favorite,
480            search=search,
481            limit=limit,
482            offset=offset,
483            order_by=order_by,
484            provider_filter=self._ensure_provider_filter(provider),
485            genre_ids=genre,
486            played_only=played_only,
487            in_library_only=True,
488            summary=summary,
489            collapse_collections=collapse_collections,
490        )
491        if (
492            kwargs.get("_localized_fallback", True)
493            and search
494            and not items
495            and self.media_type in (MediaType.GENRE, MediaType.PLAYLIST)
496        ):
497            return await self._localized_search_fallback(
498                search,
499                limit=limit,
500                offset=offset,
501                favorite=favorite,
502                order_by=order_by,
503                provider=provider,
504                genre=genre,
505                summary=summary,
506            )
507        return items
508
509    async def iter_library_items(
510        self,
511        favorite: bool | None = None,
512        search: str | None = None,
513        order_by: str = "sort_name",
514        provider: str | list[str] | None = None,
515        genre: int | list[int] | None = None,
516        library_items_only: bool = True,
517    ) -> AsyncGenerator[ItemCls]:
518        """Iterate all in-database items."""
519        limit: int = 500
520        offset: int = 0
521        if provider is not None:
522            provider_filter = provider if isinstance(provider, list) else [provider]
523        else:
524            provider_filter = None
525        while True:
526            next_items = await self.get_library_items_by_query(
527                favorite=favorite,
528                search=search,
529                genre_ids=genre,
530                limit=limit,
531                offset=offset,
532                order_by=order_by,
533                provider_filter=provider_filter,
534                in_library_only=library_items_only,
535            )
536            for item in next_items:
537                yield item
538            if len(next_items) < limit:
539                break
540            offset += limit
541
542    async def get(
543        self,
544        item_id: str,
545        provider_instance_id_or_domain: str,
546        allow_update_metadata: bool = True,
547    ) -> ItemCls:
548        """
549        Return (full) details for a single media item.
550
551        Tries to find the item in the library first, falling back to
552        fetching directly from the provider if not found.
553
554        :param item_id: The provider item id to fetch.
555        :param provider_instance_id_or_domain: The provider instance id or
556            domain to fetch the item from.
557        :param allow_update_metadata: Schedule a metadata refresh on access.
558            Set to False when fetching items in bulk (e.g. provider sync).
559        """
560        # always prefer the full library item if we have it
561        if library_item := await self.get_library_item_by_prov_id(
562            item_id,
563            provider_instance_id_or_domain,
564        ):
565            # schedule a refresh of the metadata on access of the item
566            # e.g. the item is being played or opened in the UI
567            if allow_update_metadata:
568                assert library_item.uri is not None
569                self.mass.metadata.schedule_update_metadata(library_item)
570            return library_item
571        # grab full details from the provider
572        return await self.get_provider_item(
573            item_id,
574            provider_instance_id_or_domain,
575        )
576
577    async def search(
578        self,
579        search_query: str,
580        provider_instance_id_or_domain: str,
581        limit: int = 25,
582    ) -> list[ItemCls]:
583        """Search database or provider with given query."""
584        # create safe search string
585        search_query = search_query.replace("/", " ").replace("'", "")
586        if provider_instance_id_or_domain == "library":
587            return await self.library_items(
588                search=search_query, limit=limit, summary=False, collapse_collections=False
589            )
590        if not (prov := self.mass.get_provider(provider_instance_id_or_domain)):
591            return []
592        prov = cast("MusicProvider", prov)
593        if ProviderFeature.SEARCH not in prov.supported_features:
594            return []
595        if not self.mass.music.library_supported(prov, self.media_type):
596            # assume library supported also means that this mediatype is supported
597            return []
598        searchresult = await prov.search(
599            search_query,
600            [self.media_type],
601            limit,
602        )
603        match self.media_type:
604            case MediaType.ARTIST:
605                return cast("list[ItemCls]", searchresult.artists)
606            case MediaType.ALBUM:
607                return cast("list[ItemCls]", searchresult.albums)
608            case MediaType.TRACK:
609                return cast("list[ItemCls]", searchresult.tracks)
610            case MediaType.PLAYLIST:
611                return cast("list[ItemCls]", searchresult.playlists)
612            case MediaType.AUDIOBOOK:
613                return cast("list[ItemCls]", searchresult.audiobooks)
614            case MediaType.PODCAST:
615                return cast("list[ItemCls]", searchresult.podcasts)
616            case MediaType.RADIO:
617                return cast("list[ItemCls]", searchresult.radio)
618            case _:
619                return []
620
621    async def get_collection(self, item_id: str) -> MediaCollection[ItemCls]:
622        """Get a single collection."""
623        name = get_collection_name_from_item_id(item_id)
624        query_params: dict[str, Any] = {"collection_name": name}
625        sql_query, base_query_params = self._build_final_query([], [], None, summary=False)
626        for key, value in base_query_params.items():
627            query_params.setdefault(key, value)
628        sql_query = await self._adapt_query_for_collections(
629            sql_query, query_params, summary=False, order_by=None, collection_name=name
630        )
631        db_rows = await self.mass.music.database.get_rows_from_query(
632            sql_query, query_params, limit=1, offset=0
633        )
634        if len(db_rows) != 1:
635            raise MediaNotFoundError(f"Collection {name} not found.")
636
637        return cast(
638            "MediaCollection[ItemCls]",
639            MediaCollection(
640                item_id=get_collection_item_id(db_rows[0]["name"], item_media_type=self.media_type),
641                name=db_rows[0]["name"],
642                provider="library",
643                provider_mappings=set(),
644                items=UniqueList(
645                    [
646                        self.item_cls.from_dict(self._parse_db_row(json_loads(x)))
647                        for x in json_loads(db_rows[0]["media_data"])
648                    ]
649                ),
650            ),
651        )
652
653    async def get_library_item(self, item_id: int | str) -> ItemCls:
654        """Get single library item by id."""
655        db_id = int(item_id)  # ensure integer
656        extra_query = f"WHERE {self.db_table}.item_id = :item_id"
657        for db_item in await self.get_library_items_by_query(
658            extra_query_parts=[extra_query],
659            extra_query_params={"item_id": db_id},
660            in_library_only=False,
661        ):
662            return db_item
663        msg = f"{self.media_type.value} not found in library: {db_id}"
664        raise MediaNotFoundError(msg)
665
666    async def get_library_item_by_prov_id(
667        self,
668        item_id: str,
669        provider_instance_id_or_domain: str,
670    ) -> ItemCls | None:
671        """Get the library item for the given provider item, if present."""
672        assert item_id
673        assert provider_instance_id_or_domain
674        if provider_instance_id_or_domain == "library":
675            try:
676                return await self.get_library_item(item_id)
677            except MediaNotFoundError:
678                return None
679        for item in await self.get_library_items_by_prov_id(
680            provider_instance_id_or_domain=provider_instance_id_or_domain,
681            provider_item_id=item_id,
682        ):
683            return item
684        return None
685
686    @final
687    async def get_library_item_by_prov_mappings(
688        self,
689        provider_mappings: Iterable[ProviderMapping],
690    ) -> ItemCls | None:
691        """Get the library item for the given provider_instance."""
692        # always prefer provider instance first
693        for mapping in provider_mappings:
694            for item in await self.get_library_items_by_prov_id(
695                provider_instance=mapping.provider_instance,
696                provider_item_id=mapping.item_id,
697            ):
698                return item
699        # check by domain too
700        for mapping in provider_mappings:
701            for item in await self.get_library_items_by_prov_id(
702                provider_domain=mapping.provider_domain,
703                provider_item_id=mapping.item_id,
704            ):
705                return item
706        return None
707
708    @final
709    async def get_library_item_sync_details(
710        self,
711        provider_mappings: Iterable[ProviderMapping],
712    ) -> LibraryItemSyncDetails | None:
713        """
714        Get a lightweight sync snapshot of the library item for the given provider mappings.
715
716        Returns only the scalar columns and raw provider mapping rows the library sync
717        needs for its change detection, without hydrating a full MediaItem object.
718        Resolution order matches get_library_item_by_prov_mappings (instance first,
719        then domain).
720        """
721        extra_columns, extra_joins, extra_params = self._sync_details_query_parts()
722        base_sql = f"""
723            SELECT
724                {self.db_table}.item_id,
725                {self.db_table}.favorite,
726                {self.db_table}.timestamp_added,
727                (SELECT JSON_GROUP_ARRAY(
728                    json_object(
729                        'item_id', pm.provider_item_id,
730                        'provider_domain', pm.provider_domain,
731                        'provider_instance', pm.provider_instance,
732                        'available', pm.available,
733                        'in_library', pm.in_library,
734                        'is_unique', pm.is_unique
735                    )) FROM provider_mappings pm WHERE pm.item_id = {self.db_table}.item_id
736                        AND pm.media_type = '{self.media_type.value}') AS provider_mappings
737                {extra_columns}
738            FROM {self.db_table}
739            {extra_joins}
740            WHERE {self.db_table}.item_id IN (
741                SELECT item_id FROM provider_mappings
742                WHERE provider_mappings.media_type = '{self.media_type.value}'
743                AND provider_mappings.{{prov_column}} = :prov_id
744                AND provider_mappings.provider_item_id = :prov_item_id
745            )
746        """
747        # always prefer provider instance first, then domain
748        # (same resolution order as get_library_item_by_prov_mappings)
749        for prov_column in ("provider_instance", "provider_domain"):
750            for mapping in provider_mappings:
751                for db_row in await self.mass.music.database.get_rows_from_query(
752                    base_sql.format(prov_column=prov_column),
753                    {
754                        **extra_params,
755                        "prov_id": getattr(mapping, prov_column),
756                        "prov_item_id": mapping.item_id,
757                    },
758                    limit=1,
759                ):
760                    return self._parse_sync_details_row(db_row)
761        return None
762
763    @final
764    async def get_library_item_by_external_id(
765        self, external_id: str, external_id_type: ExternalID | None = None
766    ) -> ItemCls | None:
767        """Get the library item for the given external id."""
768        subquery_parts = [
769            "media_type = :ext_id_media_type",
770            "external_id = :external_id",
771        ]
772        query_params: dict[str, Any] = {
773            "ext_id_media_type": self.media_type.value,
774            "external_id": external_id,
775        }
776        if external_id_type:
777            subquery_parts.append("external_id_type = :external_id_type")
778            query_params["external_id_type"] = str(external_id_type)
779        subquery = (
780            f"SELECT item_id FROM {DB_TABLE_EXTERNAL_ID_LOOKUP} "
781            f"WHERE {' AND '.join(subquery_parts)}"
782        )
783        query = f"{self.db_table}.item_id IN ({subquery})"
784        for item in await self.get_library_items_by_query(
785            limit=1,
786            extra_query_parts=[query],
787            extra_query_params=query_params,
788        ):
789            return item
790        return None
791
792    @final
793    async def get_library_item_by_external_ids(
794        self, external_ids: set[tuple[ExternalID, str]]
795    ) -> ItemCls | None:
796        """Get the library item for (one of) the given external ids."""
797        for external_id_type, external_id in external_ids:
798            if match := await self.get_library_item_by_external_id(external_id, external_id_type):
799                return match
800        return None
801
802    @final
803    async def get_library_items_by_prov_id(
804        self,
805        provider_domain: str | None = None,
806        provider_instance: str | None = None,
807        provider_instance_id_or_domain: str | None = None,
808        provider_item_id: str | None = None,
809        provider_item_ids: list[str] | None = None,
810        limit: int = 500,
811        offset: int = 0,
812    ) -> list[ItemCls]:
813        """
814        Fetch all records from library for given provider.
815
816        :param provider_item_ids: When given, batch-match this list of provider
817            item ids in a single query (the plural form of provider_item_id);
818            takes precedence over provider_item_id when both are passed. An
819            empty list matches nothing (distinct from None, which applies no
820            item-id filter).
821        """
822        assert provider_instance_id_or_domain != "library"
823        assert provider_domain != "library"
824        assert provider_instance != "library"
825        if provider_item_ids is not None and not provider_item_ids:
826            return []
827        subquery_parts: list[str] = []
828        query_params: dict[str, Any] = {}
829        if provider_instance:
830            query_params = {"prov_id": provider_instance}
831            subquery_parts.append("provider_mappings.provider_instance = :prov_id")
832        elif provider_domain:
833            query_params = {"prov_id": provider_domain}
834            subquery_parts.append("provider_mappings.provider_domain = :prov_id")
835        else:
836            query_params = {"prov_id": provider_instance_id_or_domain}
837            subquery_parts.append(
838                "(provider_mappings.provider_instance = :prov_id "
839                "OR provider_mappings.provider_domain = :prov_id)"
840            )
841        if provider_item_ids:
842            placeholders = ", ".join(f":item_id_{i}" for i in range(len(provider_item_ids)))
843            subquery_parts.append(f"provider_mappings.provider_item_id IN ({placeholders})")
844            for i, item_id in enumerate(provider_item_ids):
845                query_params[f"item_id_{i}"] = item_id
846        elif provider_item_id:
847            subquery_parts.append("provider_mappings.provider_item_id = :item_id")
848            query_params["item_id"] = provider_item_id
849        subquery = f"SELECT item_id FROM provider_mappings WHERE {' AND '.join(subquery_parts)}"
850        query = f"WHERE {self.db_table}.item_id IN ({subquery})"
851        return await self.get_library_items_by_query(
852            limit=limit,
853            offset=offset,
854            extra_query_parts=[query],
855            extra_query_params=query_params,
856            in_library_only=False,
857        )
858
859    @final
860    async def iter_library_items_by_prov_id(
861        self,
862        provider_instance_id_or_domain: str,
863        provider_item_id: str | None = None,
864    ) -> AsyncGenerator[ItemCls]:
865        """Iterate all records from database for given provider."""
866        limit: int = 500
867        offset: int = 0
868        while True:
869            next_items = await self.get_library_items_by_prov_id(
870                provider_instance_id_or_domain=provider_instance_id_or_domain,
871                provider_item_id=provider_item_id,
872                limit=limit,
873                offset=offset,
874            )
875            for item in next_items:
876                yield item
877            if len(next_items) < limit:
878                break
879            offset += limit
880
881    @final
882    async def set_favorite(self, item_id: str | int, favorite: bool) -> None:
883        """Set the favorite bool on a database item."""
884        db_id = int(item_id)  # ensure integer
885        library_item = await self.get_library_item(db_id)
886        if library_item.favorite == favorite:
887            return
888        match = {"item_id": db_id}
889        await self.mass.music.database.update(self.db_table, match, {"favorite": favorite})
890        library_item = await self.get_library_item(db_id)
891        self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, library_item.uri, library_item)
892
893    @guard_single_request
894    @final
895    async def get_provider_item(
896        self,
897        item_id: str,
898        provider_instance_id_or_domain: str,
899        force_refresh: bool = False,
900        fallback: ItemMapping | ItemCls | None = None,
901    ) -> ItemCls:
902        """Return item details for the given provider item id."""
903        if provider_instance_id_or_domain == "library":
904            return await self.get_library_item(item_id)
905        if not (provider := self.mass.get_provider(provider_instance_id_or_domain)):
906            raise ProviderUnavailableError(f"{provider_instance_id_or_domain} is not available")
907        if provider := self.mass.get_provider(provider_instance_id_or_domain):
908            provider = cast("MusicProvider | PluginProvider", provider)
909            with suppress(MediaNotFoundError):
910                async with self.mass.cache.handle_refresh(force_refresh):
911                    if self.media_type == MediaType.PLAYLIST:
912                        return cast("ItemCls", await provider.get_playlist(item_id))
913                    music_prov = cast("MusicProvider", provider)
914                    if self.media_type == MediaType.ARTIST:
915                        return cast("ItemCls", await music_prov.get_artist(item_id))
916                    if self.media_type == MediaType.ALBUM:
917                        return cast("ItemCls", await music_prov.get_album(item_id))
918                    if self.media_type == MediaType.TRACK:
919                        return cast("ItemCls", await music_prov.get_track(item_id))
920                    if self.media_type == MediaType.RADIO:
921                        return cast("ItemCls", await music_prov.get_radio(item_id))
922                    if self.media_type == MediaType.AUDIOBOOK:
923                        return cast("ItemCls", await music_prov.get_audiobook(item_id))
924                    if self.media_type == MediaType.PODCAST:
925                        return cast("ItemCls", await music_prov.get_podcast(item_id))
926        # if we reach this point all possibilities failed and the item could not be found.
927        # There is a possibility that the (streaming) provider changed the id of the item
928        # so we return the previous details (if we have any) marked as unavailable, so
929        # at least we have the possibility to sort out the new id through matching logic.
930        fallback = fallback or await self.get_library_item_by_prov_id(
931            item_id, provider_instance_id_or_domain
932        )
933        if (
934            fallback
935            and isinstance(fallback, ItemMapping)
936            and (fallback_provider := self.mass.get_provider(fallback.provider))
937        ):
938            # fallback is a ItemMapping, try to convert to full item
939            with suppress(LookupError, TypeError, ValueError):
940                return cast(
941                    "ItemCls",
942                    self.item_cls.from_dict(
943                        {
944                            **fallback.to_dict(),
945                            "provider_mappings": [
946                                {
947                                    "item_id": fallback.item_id,
948                                    "provider_domain": fallback_provider.domain,
949                                    "provider_instance": fallback_provider.instance_id,
950                                    "available": fallback.available,
951                                }
952                            ],
953                        }
954                    ),
955                )
956        if fallback:
957            # simply return the fallback item
958            return cast("ItemCls", fallback)
959        # all options exhausted, we really can not find this item
960        msg = (
961            f"{self.media_type.value}://{item_id} not "
962            f"found on provider {provider_instance_id_or_domain}"
963        )
964        raise MediaNotFoundError(msg)
965
966    @final
967    async def add_provider_mapping(
968        self, item_id: str | int, provider_mapping: ProviderMapping
969    ) -> None:
970        """Add provider mapping to existing library item."""
971        await self.add_provider_mappings(item_id, [provider_mapping])
972
973    @final
974    async def add_provider_mappings(
975        self, item_id: str | int, provider_mappings: Iterable[ProviderMapping]
976    ) -> None:
977        """
978        Add provider mappings to existing library item.
979
980        :param item_id: The library item ID to add mappings to.
981        :param provider_mappings: The provider mappings to add.
982        """
983        db_id = int(item_id)  # ensure integer
984        library_item = await self.get_library_item(db_id)
985        new_mappings: set[ProviderMapping] = set()
986        for provider_mapping in provider_mappings:
987            # ignore if the mapping is already present
988            if provider_mapping not in library_item.provider_mappings:
989                new_mappings.add(provider_mapping)
990        if not new_mappings:
991            return
992        # handle special case where the user wants to merge 2 library items
993        for mapping in new_mappings:
994            if _library_item := await self.get_library_item_by_prov_id(
995                mapping.item_id, mapping.provider_instance
996            ):
997                if _library_item.item_id != library_item.item_id:
998                    # merging items
999                    self.logger.debug(
1000                        "merging item id %s into item id %s based on provider mapping %s/%s",
1001                        _library_item.item_id,
1002                        library_item.item_id,
1003                        mapping.provider_instance,
1004                        mapping.item_id,
1005                    )
1006                    await self.remove_item_from_library(_library_item.item_id, recursive=True)
1007                    break
1008        library_item.provider_mappings.update(new_mappings)
1009        self.mass.music.match_provider_instances(library_item)
1010        await self.set_provider_mappings(db_id, library_item.provider_mappings)
1011        self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, library_item.uri, library_item)
1012
1013    @final
1014    async def update_provider_mapping(
1015        self,
1016        item_id: str | int,
1017        provider_instance_id: str,
1018        provider_item_id: str,
1019        *,
1020        available: bool | Any = UNSET,
1021        in_library: bool | Any = UNSET,
1022        is_unique: bool | None | Any = UNSET,
1023        url: str | None | Any = UNSET,
1024        details: str | None | Any = UNSET,
1025        audio_format: AudioFormat | Any = UNSET,
1026    ) -> None:
1027        """Update an existing provider mapping for a library item."""
1028        db_id = int(item_id)  # ensure integer
1029        library_item = await self.get_library_item(db_id)
1030
1031        # find the current mapping (strictly by provider instance + provider item id)
1032        cur_mapping: ProviderMapping | None = None
1033        for mapping in library_item.provider_mappings:
1034            if (
1035                mapping.provider_instance == provider_instance_id
1036                and mapping.item_id == provider_item_id
1037            ):
1038                cur_mapping = mapping
1039                break
1040        if cur_mapping is None:
1041            msg = (
1042                f"Provider mapping {provider_instance_id}/{provider_item_id} "
1043                f"not found for item {db_id}"
1044            )
1045            raise MediaNotFoundError(msg)
1046
1047        # guard against nulls for NOT NULL columns
1048        if available is None:
1049            available = UNSET
1050        if in_library is None:
1051            in_library = UNSET
1052
1053        updates: dict[str, Any] = {}
1054        if available is not UNSET:
1055            updates["available"] = bool(available)
1056        if in_library is not UNSET:
1057            updates["in_library"] = bool(in_library)
1058        if is_unique is not UNSET:
1059            updates["is_unique"] = is_unique
1060        if url is not UNSET:
1061            updates["url"] = url
1062        if details is not UNSET:
1063            updates["details"] = details
1064        if audio_format is not UNSET:
1065            updates["audio_format"] = serialize_to_json(audio_format)
1066
1067        if not updates:
1068            return
1069
1070        match = {
1071            "media_type": self.media_type.value,
1072            "item_id": db_id,
1073            "provider_instance": provider_instance_id,
1074            "provider_item_id": provider_item_id,
1075        }
1076        await self.mass.music.database.update(DB_TABLE_PROVIDER_MAPPINGS, match, updates)
1077
1078        # Re-fetch the updated item so the event payload reflects persisted DB state.
1079        updated_item = await self.get_library_item(db_id)
1080        self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, updated_item.uri, updated_item)
1081
1082    @final
1083    async def remove_provider_mapping(
1084        self, item_id: str | int, provider_instance_id: str, provider_item_id: str
1085    ) -> None:
1086        """Remove provider mapping(s) from item."""
1087        db_id = int(item_id)  # ensure integer
1088        try:
1089            library_item = await self.get_library_item(db_id)
1090        except MediaNotFoundError:
1091            # edge case: already deleted / race condition
1092            return
1093
1094        # update provider_mappings table
1095        await self.mass.music.database.delete(
1096            DB_TABLE_PROVIDER_MAPPINGS,
1097            {
1098                "media_type": self.media_type.value,
1099                "item_id": db_id,
1100                "provider_instance": provider_instance_id,
1101                "provider_item_id": provider_item_id,
1102            },
1103        )
1104        # cleanup playlog table
1105        await self.mass.music.database.delete(
1106            DB_TABLE_PLAYLOG,
1107            {
1108                "media_type": self.media_type.value,
1109                "item_id": provider_item_id,
1110                "provider": provider_instance_id,
1111            },
1112        )
1113        library_item.provider_mappings = {
1114            x
1115            for x in library_item.provider_mappings
1116            if not (x.provider_instance == provider_instance_id and x.item_id == provider_item_id)
1117        }
1118        if library_item.provider_mappings:
1119            # if this was the last mapping for the provider instance, strip any artwork
1120            # that belonged to it (e.g. local file paths that are no longer resolvable)
1121            images_changed = not any(
1122                x.provider_instance == provider_instance_id for x in library_item.provider_mappings
1123            ) and await self._remove_provider_images(db_id, provider_instance_id)
1124            self.logger.debug(
1125                "removed provider_mapping %s/%s from item id %s",
1126                provider_instance_id,
1127                provider_item_id,
1128                db_id,
1129            )
1130            # the removed provider mapping is itself a change to the item, so always notify
1131            # (unless suppressed during a bulk cleanup); re-fetch first when images were
1132            # stripped so the event payload stays accurate
1133            if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
1134                event_item = await self.get_library_item(db_id) if images_changed else library_item
1135                self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, event_item.uri, event_item)
1136        else:
1137            # remove item if it has no more providers
1138            with suppress(AssertionError):
1139                await self.remove_item_from_library(db_id)
1140
1141    @final
1142    async def remove_provider_mappings(self, item_id: str | int, provider_instance_id: str) -> None:
1143        """Remove all provider mappings from an item."""
1144        db_id = int(item_id)  # ensure integer
1145        try:
1146            library_item = await self.get_library_item(db_id)
1147        except MediaNotFoundError:
1148            # edge case: already deleted / race condition
1149            library_item = None
1150        # update provider_mappings table
1151        await self.mass.music.database.delete(
1152            DB_TABLE_PROVIDER_MAPPINGS,
1153            {
1154                "media_type": self.media_type.value,
1155                "item_id": db_id,
1156                "provider_instance": provider_instance_id,
1157            },
1158        )
1159        if library_item is None:
1160            return
1161        # update the item's provider mappings (and check if we still have any)
1162        library_item.provider_mappings = {
1163            x for x in library_item.provider_mappings if x.provider_instance != provider_instance_id
1164        }
1165        if library_item.provider_mappings:
1166            # the item is kept (it still has other providers), but it may carry artwork
1167            # that belonged to the removed provider (e.g. local file paths that are no
1168            # longer resolvable), so strip those images from the stored metadata
1169            images_changed = await self._remove_provider_images(db_id, provider_instance_id)
1170            self.logger.debug(
1171                "removed all provider mappings for provider %s from item id %s",
1172                provider_instance_id,
1173                db_id,
1174            )
1175            # the removed provider mapping(s) are themselves a change to the item, so
1176            # always notify (unless suppressed during a bulk cleanup); re-fetch first when
1177            # images were stripped so the event payload stays accurate
1178            if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
1179                event_item = await self.get_library_item(db_id) if images_changed else library_item
1180                self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, event_item.uri, event_item)
1181        else:
1182            # remove item if it has no more providers
1183            with suppress(AssertionError):
1184                await self.remove_item_from_library(db_id)
1185
1186    @final
1187    async def set_provider_mappings(
1188        self,
1189        item_id: str | int,
1190        provider_mappings: Iterable[ProviderMapping],
1191        overwrite: bool = False,
1192    ) -> None:
1193        """Update the provider_items table for the media item."""
1194        db_id = int(item_id)  # ensure integer
1195        if overwrite:
1196            # on overwrite, clear the provider_mappings table first
1197            # this is done for filesystem provider changing the path (and thus item_id)
1198            await self.mass.music.database.delete(
1199                DB_TABLE_PROVIDER_MAPPINGS,
1200                {"media_type": self.media_type.value, "item_id": db_id},
1201            )
1202        prov_map_objs: list[dict[str, Any]] = []
1203        for provider_mapping in provider_mappings:
1204            prov_map_obj = {
1205                "media_type": self.media_type.value,
1206                "item_id": db_id,
1207                "provider_domain": provider_mapping.provider_domain,
1208                "provider_instance": provider_mapping.provider_instance,
1209                "provider_item_id": provider_mapping.item_id,
1210                "available": provider_mapping.available,
1211                "audio_format": serialize_to_json(provider_mapping.audio_format),
1212            }
1213            for key in ("url", "details", "in_library", "is_unique"):
1214                if (value := getattr(provider_mapping, key, None)) is not None:
1215                    prov_map_obj[key] = value
1216            prov_map_objs.append(prov_map_obj)
1217        await self.mass.music.database.upsert_many(
1218            DB_TABLE_PROVIDER_MAPPINGS,
1219            prov_map_objs,
1220        )
1221
1222    @final
1223    async def set_external_ids(
1224        self,
1225        item_id: str | int,
1226        external_ids: Iterable[tuple[ExternalID, str]],
1227    ) -> None:
1228        """Update the external_id_lookup table rows for the media item."""
1229        db_id = int(item_id)  # ensure integer
1230        await self.mass.music.database.delete(
1231            DB_TABLE_EXTERNAL_ID_LOOKUP,
1232            {"media_type": self.media_type.value, "item_id": db_id},
1233        )
1234        if lookup_rows := [
1235            {
1236                "media_type": self.media_type.value,
1237                "external_id_type": external_id_type,
1238                "external_id": external_id,
1239                "item_id": db_id,
1240            }
1241            for external_id_type, external_id in external_ids
1242        ]:
1243            await self.mass.music.database.upsert_many(DB_TABLE_EXTERNAL_ID_LOOKUP, lookup_rows)
1244
1245    @abstractmethod
1246    async def match_providers(self, db_item: ItemCls) -> None:
1247        """
1248        Try to find match on all (streaming) providers for the provided (database) item.
1249
1250        This is used to link objects of different providers/qualities together.
1251        """
1252
1253    if TYPE_CHECKING:
1254
1255        @overload
1256        async def get_library_items_by_query(
1257            self,
1258            favorite: bool | None = None,
1259            search: str | None = None,
1260            limit: int = 500,
1261            offset: int = 0,
1262            order_by: str | None = None,
1263            provider_filter: list[str] | None = None,
1264            extra_query_parts: list[str] | None = None,
1265            extra_query_params: dict[str, Any] | None = None,
1266            extra_join_parts: list[str] | None = None,
1267            genre_ids: int | list[int] | None = None,
1268            played_only: bool = False,
1269            in_library_only: bool = False,
1270            summary: bool = False,
1271            *,
1272            collapse_collections: Literal[True],
1273        ) -> list[ItemCls | MediaCollection[ItemCls]]: ...
1274
1275        @overload
1276        async def get_library_items_by_query(
1277            self,
1278            favorite: bool | None = None,
1279            search: str | None = None,
1280            limit: int = 500,
1281            offset: int = 0,
1282            order_by: str | None = None,
1283            provider_filter: list[str] | None = None,
1284            extra_query_parts: list[str] | None = None,
1285            extra_query_params: dict[str, Any] | None = None,
1286            extra_join_parts: list[str] | None = None,
1287            genre_ids: int | list[int] | None = None,
1288            played_only: bool = False,
1289            in_library_only: bool = False,
1290            summary: bool = False,
1291            *,
1292            collapse_collections: Literal[False] = False,
1293        ) -> list[ItemCls]: ...
1294
1295        @overload
1296        async def get_library_items_by_query(
1297            self,
1298            favorite: bool | None = None,
1299            search: str | None = None,
1300            limit: int = 500,
1301            offset: int = 0,
1302            order_by: str | None = None,
1303            provider_filter: list[str] | None = None,
1304            extra_query_parts: list[str] | None = None,
1305            extra_query_params: dict[str, Any] | None = None,
1306            extra_join_parts: list[str] | None = None,
1307            genre_ids: int | list[int] | None = None,
1308            played_only: bool = False,
1309            in_library_only: bool = False,
1310            summary: bool = False,
1311            *,
1312            collapse_collections: bool,
1313        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
1314
1315    @final
1316    async def get_library_items_by_query(  # noqa: PLR0913
1317        self,
1318        favorite: bool | None = None,
1319        search: str | None = None,
1320        limit: int = 500,
1321        offset: int = 0,
1322        order_by: str | None = None,
1323        provider_filter: list[str] | None = None,
1324        extra_query_parts: list[str] | None = None,
1325        extra_query_params: dict[str, Any] | None = None,
1326        extra_join_parts: list[str] | None = None,
1327        genre_ids: int | list[int] | None = None,
1328        played_only: bool = False,
1329        in_library_only: bool = False,
1330        summary: bool = False,
1331        *,
1332        collapse_collections: bool = False,
1333    ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]:
1334        """Fetch MediaItem records from database by building the query."""
1335        query_params = dict(extra_query_params) if extra_query_params else {}
1336        query_parts: list[str] = list(extra_query_parts) if extra_query_parts else []
1337        join_parts: list[str] = list(extra_join_parts) if extra_join_parts else []
1338        search = self._preprocess_search(search)
1339        genre_ids = self._preprocess_genre_ids(genre_ids)
1340        # create special performant random query
1341        if order_by and order_by.startswith("random"):
1342            self._apply_random_subquery(
1343                query_parts=query_parts,
1344                query_params=query_params,
1345                join_parts=join_parts,
1346                favorite=favorite,
1347                search=search if not collapse_collections else None,
1348                genre_ids=genre_ids,
1349                provider_filter=provider_filter,
1350                played_only=played_only,
1351                limit=limit,
1352                in_library_only=in_library_only,
1353            )
1354        else:
1355            # apply filters
1356            self._apply_filters(
1357                query_parts=query_parts,
1358                query_params=query_params,
1359                favorite=favorite,
1360                search=search if not collapse_collections else None,
1361                genre_ids=genre_ids,
1362                provider_filter=provider_filter,
1363                played_only=played_only,
1364                in_library_only=in_library_only,
1365            )
1366        # build and execute final query
1367        sql_query, base_query_params = self._build_final_query(
1368            query_parts, join_parts, order_by, summary=summary
1369        )
1370        # base query params act as defaults: callers may override them via extra_query_params
1371        for key, value in base_query_params.items():
1372            query_params.setdefault(key, value)
1373
1374        if collapse_collections:
1375            if search:
1376                query_params["search"] = f"%{search}%"
1377            sql_query = await self._adapt_query_for_collections(
1378                sql_query, query_params, summary=summary, order_by=order_by, search=search
1379            )
1380
1381        db_rows = await self.mass.music.database.get_rows_from_query(
1382            sql_query, query_params, limit=limit, offset=offset
1383        )
1384        if collapse_collections:
1385            items: list[ItemCls | MediaCollection[ItemCls]] = []
1386
1387            def _parse_method(x: str) -> ItemCls:
1388                if summary:
1389                    return cast("ItemCls", self._parse_summary_row(json_loads(x)))
1390                return cast(
1391                    "ItemCls",
1392                    self.item_cls.from_dict(self._parse_db_row(json_loads(x))),
1393                )
1394
1395            for db_row in db_rows:
1396                if db_row["type"] == "single":
1397                    items.append(_parse_method(db_row["media_data"]))
1398                elif db_row["type"] == "collection":
1399                    items.append(
1400                        MediaCollection[ItemCls](
1401                            item_id=get_collection_item_id(
1402                                db_row["name"], item_media_type=self.media_type
1403                            ),
1404                            name=db_row["name"],
1405                            provider="library",
1406                            provider_mappings=set(),
1407                            items=UniqueList(
1408                                [_parse_method(x) for x in json_loads(db_row["media_data"])]
1409                            ),
1410                        )
1411                    )
1412            return items
1413        if summary:
1414            return [cast("ItemCls", self._parse_summary_row(db_row)) for db_row in db_rows]
1415        return [
1416            cast("ItemCls", self.item_cls.from_dict(self._parse_db_row(db_row)))
1417            for db_row in db_rows
1418        ]
1419
1420    @final
1421    async def _get_library_item_by_match(self, item: ItemCls | ItemMapping) -> int | None:
1422        if item.provider == "library":
1423            return int(item.item_id)
1424        # search by provider mappings if item is ItemMapping
1425        if isinstance(item, ItemMapping):
1426            if cur_item := await self.get_library_item_by_prov_id(item.item_id, item.provider):
1427                return int(cur_item.item_id)
1428
1429        # for all other items that are MediaItemType, check provider_mappings if it exists
1430        provider_mappings = getattr(item, "provider_mappings", None)
1431        if provider_mappings:
1432            if cur_item := await self.get_library_item_by_prov_mappings(provider_mappings):
1433                return int(cur_item.item_id)
1434        if cur_item := await self.get_library_item_by_external_ids(item.external_ids):
1435            # existing item match by external id
1436            # Double check external IDs - if MBID exists, regards that as overriding
1437            if compare_media_item(item, cur_item):
1438                return int(cur_item.item_id)
1439        # search by (exact) name match
1440        query = f"{self.db_table}.name = :name OR {self.db_table}.sort_name = :sort_name"
1441        query_params = {"name": item.name, "sort_name": item.sort_name}
1442        for db_item in await self.get_library_items_by_query(
1443            extra_query_parts=[query], extra_query_params=query_params
1444        ):
1445            if compare_media_item(db_item, item, True):
1446                return int(db_item.item_id)
1447        return None
1448
1449    def _external_ids_query(
1450        self, media_type: MediaType | None = None, table_alias: str | None = None
1451    ) -> str:
1452        """
1453        Return a subquery that selects the external ids of a media item as a JSON array.
1454
1455        :param media_type: Media type to select the external ids for, defaults to
1456            this controller's media type.
1457        :param table_alias: (Aliased) table name the subquery correlates against,
1458            defaults to this controller's table.
1459        """
1460        media_type = media_type or self.media_type
1461        table_alias = table_alias or self.db_table
1462        return (
1463            f"(SELECT JSON_GROUP_ARRAY(json_array("
1464            f"{DB_TABLE_EXTERNAL_ID_LOOKUP}.external_id_type, "
1465            f"{DB_TABLE_EXTERNAL_ID_LOOKUP}.external_id)) "
1466            f"FROM {DB_TABLE_EXTERNAL_ID_LOOKUP} "
1467            f"WHERE {DB_TABLE_EXTERNAL_ID_LOOKUP}.media_type = '{media_type.value}' "
1468            f"AND {DB_TABLE_EXTERNAL_ID_LOOKUP}.item_id = {table_alias}.item_id)"
1469        )
1470
1471    def _provider_mappings_query(self) -> str:
1472        """Return a subquery that selects the provider mappings of a media item as a JSON array."""
1473        return f"""(SELECT JSON_GROUP_ARRAY(
1474            json_object(
1475                'item_id', pm.provider_item_id,
1476                'provider_domain', pm.provider_domain,
1477                'provider_instance', pm.provider_instance,
1478                'available', pm.available,
1479                'audio_format', json(pm.audio_format),
1480                'url', pm.url,
1481                'details', pm.details,
1482                'in_library', pm.in_library,
1483                'is_unique', pm.is_unique
1484            )) FROM {DB_TABLE_PROVIDER_MAPPINGS} pm
1485            WHERE pm.item_id = {self.db_table}.item_id
1486            AND pm.media_type = '{self.media_type.value}')"""
1487
1488    def _artist_mappings_summary_query(
1489        self, m2m_table: str, m2m_key: str, include_artist_type: bool = False
1490    ) -> str:
1491        """
1492        Return a subquery selecting the slim artist mappings JSON of a summary row.
1493
1494        :param m2m_table: The many-to-many table linking artists to this media type.
1495        :param m2m_key: The column in the m2m table referencing this media type's item id.
1496        :param include_artist_type: Also select the artist_type of each artist.
1497        """
1498        artist_type_part = ",\n                'artist_type', artists.artist_type"
1499        return f"""(SELECT JSON_GROUP_ARRAY(
1500            json_object(
1501                'item_id', artists.item_id,
1502                'name', artists.name,
1503                'sort_name', artists.sort_name{artist_type_part if include_artist_type else ""}
1504            )) FROM artists
1505            JOIN {m2m_table} ON artists.item_id = {m2m_table}.artist_id
1506            WHERE {m2m_table}.{m2m_key} = {self.db_table}.item_id)"""
1507
1508    def _summary_base_columns(self) -> str:
1509        """Return the SELECT columns shared by every summary query."""
1510        # the search/sort/statistics columns are selected so ORDER BY (see SORT_KEYS)
1511        # resolves them from the result set, like the full query's SELECT * does
1512        return f"""
1513            {self.db_table}.item_id,
1514            {self.db_table}.name,
1515            {self.db_table}.sort_name,
1516            {self.db_table}.favorite,
1517            {self.db_table}.search_name AS search_name,
1518            {self.db_table}.search_sort_name AS search_sort_name,
1519            {self.db_table}.play_count AS play_count,
1520            {self.db_table}.last_played AS last_played,
1521            {self.db_table}.timestamp_added AS timestamp_added,
1522            {self.db_table}.timestamp_modified AS timestamp_modified,
1523            json_extract({self.db_table}.metadata, '$.images') AS images,
1524            json_extract({self.db_table}.metadata, '$.collections') AS collections"""
1525
1526    async def _localized_search_fallback(
1527        self, search_query: str, limit: int, offset: int = 0, **call_kwargs: Any
1528    ) -> list[ItemCls]:
1529        """
1530        Retry a library search using the canonical names behind a localized query.
1531
1532        For genre/playlist searches that return nothing literally, reverse-resolve the query to the
1533        canonical (English) names of matching localized items and search those, so an item is
1534        findable by the localized name the user sees. The caller's other filters (favorite,
1535        order_by, provider and any controller-specific kwargs) are forwarded unchanged so the retry
1536        behaves like the literal search; results are merged, de-duplicated and paginated here. See
1537        ``TranslationController.reverse_lookup_media_names``.
1538        """
1539        seen: set[Any] = set()
1540        merged: list[ItemCls] = []
1541        # iterate the canonical names in a stable order, and fetch each from the start so the
1542        # offset/limit window can be applied to the merged, de-duplicated result set
1543        for name in sorted(await self.mass.translations.reverse_lookup_media_names(search_query)):
1544            for item in await self.library_items(
1545                search=name,
1546                limit=limit + offset,
1547                offset=0,
1548                _localized_fallback=False,
1549                **call_kwargs,
1550            ):
1551                if item.item_id not in seen:
1552                    seen.add(item.item_id)
1553                    merged.append(item)
1554        return merged[offset : offset + limit]
1555
1556    @abstractmethod
1557    async def _add_library_item(
1558        self,
1559        item: ItemCls,
1560        overwrite_existing: bool = False,
1561    ) -> int:
1562        """Add item to library and return the database id."""
1563
1564    @abstractmethod
1565    async def _update_library_item(
1566        self, item_id: str | int, update: ItemCls, overwrite: bool = False
1567    ) -> None:
1568        """Update existing library record in the database."""
1569
1570    def _search_filter_clause(self, search: str, query_params: dict[str, Any]) -> str:
1571        """Return the SQL WHERE clause fragment used for search filtering."""
1572        return search_name_match_clause(self.db_table, search, "search", query_params)
1573
1574    @final
1575    def _preprocess_search(self, search: str | None) -> str | None:
1576        """Normalize the search string for use in the search filter clauses."""
1577        return create_safe_string(search, True, True) if search else search
1578
1579    @final
1580    @staticmethod
1581    def _preprocess_genre_ids(genre_ids: int | list[int] | None) -> list[int] | None:
1582        if genre_ids is None:
1583            return None
1584        if isinstance(genre_ids, list):
1585            normalized = [int(x) for x in genre_ids]
1586        else:
1587            normalized = [int(genre_ids)]
1588        return normalized or None
1589
1590    @final
1591    @staticmethod
1592    def _clean_query_parts(query_parts: list[str]) -> list[str]:
1593        """Clean the query parts list by removing duplicate where statements."""
1594        return [x[5:] if x.lower().startswith("where ") else x for x in query_parts]
1595
1596    @final
1597    def _apply_random_subquery(
1598        self,
1599        query_parts: list[str],
1600        query_params: dict[str, Any],
1601        join_parts: list[str],
1602        favorite: bool | None,
1603        search: str | None,
1604        genre_ids: list[int] | None,
1605        provider_filter: list[str] | None,
1606        played_only: bool = False,
1607        limit: int = 500,
1608        in_library_only: bool = False,
1609    ) -> None:
1610        """Build a fast random subquery with all filters applied."""
1611        sub_query_parts = query_parts.copy()
1612        sub_join_parts = join_parts.copy()
1613
1614        # Apply all filters to the subquery
1615        self._apply_filters(
1616            query_parts=sub_query_parts,
1617            query_params=query_params,
1618            favorite=favorite,
1619            search=search,
1620            genre_ids=genre_ids,
1621            provider_filter=provider_filter,
1622            played_only=played_only,
1623            in_library_only=in_library_only,
1624        )
1625
1626        # Build the subquery
1627        sub_query = f"SELECT {self.db_table}.item_id FROM {self.db_table}"
1628
1629        if sub_join_parts:
1630            sub_query += f" {' '.join(sub_join_parts)}"
1631
1632        if sub_query_parts:
1633            sub_query += " WHERE " + " AND ".join(self._clean_query_parts(sub_query_parts))
1634
1635        sub_query += f" ORDER BY RANDOM() LIMIT {limit}"
1636
1637        # The query now only consists of the random subquery, which applies all filters
1638        # within itself
1639        query_parts.clear()
1640        query_parts.append(f"{self.db_table}.item_id in ({sub_query})")
1641        join_parts.clear()
1642
1643    @final
1644    def _apply_filters(
1645        self,
1646        query_parts: list[str],
1647        query_params: dict[str, Any],
1648        favorite: bool | None,
1649        search: str | None,
1650        genre_ids: list[int] | None,
1651        provider_filter: list[str] | None,
1652        played_only: bool = False,
1653        in_library_only: bool = False,
1654    ) -> None:
1655        """Apply search, favorite, and provider filters."""
1656        # handle search
1657        if search:
1658            query_parts.append(self._search_filter_clause(search, query_params))
1659        # handle favorite filter
1660        if favorite is not None:
1661            query_parts.append(f"{self.db_table}.favorite = :favorite")
1662            query_params["favorite"] = favorite
1663        # handle played_only filter
1664        if played_only:
1665            query_parts.append(f"{self.db_table}.last_played > 0")
1666        # handle genre filter
1667        if genre_ids:
1668            query_params["genre_ids"] = genre_ids
1669            query_params["genre_media_type"] = self.media_type.value
1670            query_parts.append(
1671                f"EXISTS("
1672                f"SELECT 1 FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING} gm "
1673                f"WHERE gm.media_id = {self.db_table}.item_id "
1674                "AND gm.media_type = :genre_media_type "
1675                "AND gm.genre_id IN :genre_ids)"
1676            )
1677        # Apply the provider filter
1678        if provider_filter or in_library_only:
1679            query_parts.append(
1680                self._provider_filter_clause(query_params, provider_filter, in_library_only)
1681            )
1682
1683    @final
1684    def _provider_filter_clause(
1685        self,
1686        query_params: dict[str, Any],
1687        provider_filter: list[str] | None,
1688        in_library_only: bool = False,
1689    ) -> str:
1690        """
1691        Return the SQL clause that restricts items by their provider mappings.
1692
1693        At least one of provider_filter/in_library_only must be set, otherwise the
1694        returned clause only asserts that the item has any mapping at all.
1695
1696        :param query_params: Query params dict; the clause's bound params are added to it.
1697        :param provider_filter: Only match items mapped to one of these provider instances.
1698        :param in_library_only: Only match provider mappings that are in the provider's library.
1699        """
1700        # NOTE: provider mapping filters are applied as a correlated EXISTS subquery
1701        # instead of a JOIN + GROUP BY, so SQLite can stream results straight from the
1702        # sort index instead of materializing/sorting the whole (deduped) result set.
1703        query_params["provider_media_type"] = self.media_type.value
1704        conditions = [
1705            f"provider_mappings.item_id = {self.db_table}.item_id",
1706            "provider_mappings.media_type = :provider_media_type",
1707        ]
1708        if in_library_only:
1709            conditions.append("provider_mappings.in_library = 1")
1710        if provider_filter:
1711            provider_conditions = []
1712            for idx, prov in enumerate(provider_filter):
1713                param_name = f"provider_filter_{idx}"
1714                provider_conditions.append(f"provider_mappings.provider_instance = :{param_name}")
1715                query_params[param_name] = prov
1716            conditions.append(f"({' OR '.join(provider_conditions)})")
1717        return f"EXISTS(SELECT 1 FROM provider_mappings WHERE {' AND '.join(conditions)})"
1718
1719    @final
1720    def _build_final_query(
1721        self,
1722        query_parts: list[str],
1723        join_parts: list[str],
1724        order_by: str | None,
1725        summary: bool = False,
1726    ) -> tuple[str, dict[str, Any]]:
1727        """Build the final SQL query string and its (base) bound query params."""
1728        sql_query, base_query_params = self.summary_query if summary else self.base_query
1729
1730        # Add joins
1731        if join_parts:
1732            sql_query += f" {' '.join(join_parts)} "
1733
1734        # Add where clauses
1735        if query_parts:
1736            # prevent duplicate where statement
1737            sql_query += " WHERE " + " AND ".join(self._clean_query_parts(query_parts))
1738
1739        # Add grouping (only needed when caller-provided joins can fan out rows)
1740        # and ordering. Without a GROUP BY, SQLite can stream results directly
1741        # from the sort index instead of sorting the whole result set.
1742        if join_parts:
1743            sql_query += f" GROUP BY {self.db_table}.item_id"
1744
1745        if order_by:
1746            if sort_key := SORT_KEYS.get(order_by):
1747                sql_query += f" ORDER BY {sort_key}"
1748
1749        return sql_query, base_query_params
1750
1751    @final
1752    @staticmethod
1753    def _parse_db_row(db_row: Mapping[str, Any]) -> dict[str, Any]:
1754        """Parse raw db Mapping into a dict."""
1755        db_row_dict = dict(db_row)
1756        db_row_dict["provider"] = "library"
1757        db_row_dict["favorite"] = bool(db_row_dict["favorite"])
1758        db_row_dict["item_id"] = str(db_row_dict["item_id"])
1759        db_row_dict["date_added"] = datetime.fromtimestamp(
1760            db_row_dict["timestamp_added"], tz=UTC
1761        ).isoformat()
1762
1763        for key in JSON_KEYS:
1764            if key not in db_row_dict:
1765                continue
1766            if not (raw_value := db_row_dict[key]):
1767                continue
1768            db_row_dict[key] = json_loads(raw_value)
1769
1770        # parse "fully_played" as bool if present in the row
1771        if "fully_played" in db_row_dict:
1772            db_row_dict["fully_played"] = parse_optional_bool(db_row_dict["fully_played"])
1773
1774        # copy track_album --> album
1775        if track_album := db_row_dict.get("track_album"):
1776            db_row_dict["album"] = track_album
1777            db_row_dict["disc_number"] = track_album["disc_number"]
1778            db_row_dict["track_number"] = track_album["track_number"]
1779            # always prefer album image over track image
1780            if (album_images := track_album.get("images")) and (
1781                album_thumb := next((x for x in album_images if x["type"] == "thumb"), None)
1782            ):
1783                # copy album image to itemmapping single image (on the track)
1784                db_row_dict["image"] = album_thumb
1785                # also set image on the album dict for ItemMapping compatibility
1786                track_album["image"] = album_thumb
1787                if db_row_dict["metadata"].get("images"):
1788                    # merge album image with existing images
1789                    db_row_dict["metadata"]["images"] = [
1790                        album_thumb,
1791                        *db_row_dict["metadata"]["images"],
1792                    ]
1793                else:
1794                    db_row_dict["metadata"]["images"] = [album_thumb]
1795
1796        if audiobook_artists := db_row_dict.get("audiobook_artists"):
1797            _narrators = []
1798            _authors = []
1799            for artist in audiobook_artists:
1800                artist_type = artist.get("artist_type")
1801                if artist_type == "author":
1802                    _authors.append(artist)
1803                elif artist_type == "narrator":
1804                    _narrators.append(artist)
1805            if _authors:
1806                # prevent overwriting string values
1807                db_row_dict["authors"] = _authors
1808            if _narrators:
1809                # prevent overwriting string values
1810                db_row_dict["narrators"] = _narrators
1811
1812        return db_row_dict
1813
1814    @final
1815    def _ensure_provider_filter(
1816        self,
1817        provider: str | list[str] | None,
1818    ) -> list[str] | None:
1819        """Ensure the provider filter respects the current user's provider filter."""
1820        # Apply user provider filter if needed
1821        user = get_current_user()
1822        user_provider_filter = user.provider_filter if user and user.provider_filter else None
1823        final_provider_filter: list[str] | None = None
1824        if user_provider_filter:
1825            plugin_provider_instances = {
1826                prov.instance_id for prov in self.mass.providers if prov.type == ProviderType.PLUGIN
1827            }
1828            # User has a provider filter set
1829            if provider:
1830                # Explicit provider filter provided - validate against user's allowed providers
1831                requested_providers = [provider] if isinstance(provider, str) else provider
1832                # Only restrict access to music providers.
1833                final_provider_filter = [
1834                    p
1835                    for p in requested_providers
1836                    if p in user_provider_filter or p in plugin_provider_instances
1837                ]
1838                if not final_provider_filter:
1839                    # No overlap - user requested providers they don't have access to
1840                    raise InsufficientPermissions(
1841                        "User does not have permission to access the requested provider(s)."
1842                    )
1843            else:
1844                # No explicit filter - apply user music provider filter but keep plugin providers.
1845                final_provider_filter = list(
1846                    dict.fromkeys([*user_provider_filter, *plugin_provider_instances])
1847                )
1848        elif provider is not None:
1849            # No user filter - use the provided filter as is
1850            final_provider_filter = [provider] if isinstance(provider, str) else provider
1851        return final_provider_filter
1852
1853    @final
1854    def _select_provider_id(self, library_item: ItemCls) -> tuple[str, str]:
1855        """Select the correct provider id to use for fetching the item."""
1856        user = get_current_user()
1857        user_provider_filter = user.provider_filter if user and user.provider_filter else None
1858        if not user_provider_filter:
1859            mapping = next(iter(library_item.provider_mappings))
1860            return (mapping.provider_instance, mapping.item_id)
1861
1862        # First prefer music provider mappings that are explicitly allowed for this user.
1863        # prefer user provider filter if available
1864        for mapping in library_item.provider_mappings:
1865            provider = self.mass.get_provider(mapping.provider_instance)
1866            if provider and provider.type == ProviderType.MUSIC:
1867                if mapping.provider_instance in user_provider_filter:
1868                    return (mapping.provider_instance, mapping.item_id)
1869
1870        # If no allowed music mapping exists, fall back to plugin mappings.
1871        for mapping in library_item.provider_mappings:
1872            provider = self.mass.get_provider(mapping.provider_instance)
1873            if provider and provider.type == ProviderType.PLUGIN:
1874                return (mapping.provider_instance, mapping.item_id)
1875
1876        # As a final fallback, preserve previous behavior.
1877        for mapping in library_item.provider_mappings:
1878            if mapping.provider_instance in user_provider_filter:
1879                return (mapping.provider_instance, mapping.item_id)
1880
1881        # fallback to first mapping
1882        mapping = next(iter(library_item.provider_mappings))
1883        return (mapping.provider_instance, mapping.item_id)
1884
1885    async def _remove_provider_images(self, db_id: int, provider_instance_id: str) -> bool:
1886        """
1887        Remove images belonging to a provider from a library item's stored metadata.
1888
1889        :param db_id: The library (database) id of the item.
1890        :param provider_instance_id: The provider instance whose images should be removed.
1891        :return: True if any images were removed and the db record was updated.
1892        """
1893        # read the raw metadata straight from the db (instead of via get_library_item)
1894        # to avoid persisting any images that are only injected at read time (such as
1895        # the album thumb that gets merged into a track's images)
1896        db_row = await self.mass.music.database.get_row(self.db_table, {"item_id": db_id})
1897        if not db_row or not (raw_metadata := db_row["metadata"]):
1898            return False
1899        metadata = MediaItemMetadata.from_dict(json_loads(raw_metadata))
1900        if not metadata.images:
1901            return False
1902        remaining = UniqueList(
1903            img for img in metadata.images if img.provider != provider_instance_id
1904        )
1905        if len(remaining) == len(metadata.images):
1906            # nothing belonged to this provider
1907            return False
1908        metadata.images = remaining or None
1909        await self.mass.music.database.update(
1910            self.db_table,
1911            {"item_id": db_id},
1912            {"metadata": serialize_to_json(metadata)},
1913        )
1914        return True
1915
1916    def _sync_details_query_parts(self) -> tuple[str, str, dict[str, Any]]:
1917        """
1918        Return extra (columns, joins, params) for this media type's sync-details query.
1919
1920        Override in a subclass to select additional lightweight columns needed by the
1921        library sync change detection for this media type.
1922        """
1923        return "", "", {}
1924
1925    def _parse_sync_details_row(self, db_row: Mapping[str, Any]) -> LibraryItemSyncDetails:
1926        """Parse a raw sync-details db row into a LibraryItemSyncDetails object."""
1927        return LibraryItemSyncDetails(
1928            item_id=db_row["item_id"],
1929            favorite=bool(db_row["favorite"]),
1930            date_added=datetime.fromtimestamp(db_row["timestamp_added"], tz=UTC),
1931            provider_mappings=self._parse_sync_details_mappings(db_row),
1932        )
1933
1934    @final
1935    def _parse_sync_details_mappings(self, db_row: Mapping[str, Any]) -> set[ProviderMapping]:
1936        """Parse the aggregated raw provider mapping rows of a sync-details db row."""
1937        return {
1938            ProviderMapping(
1939                item_id=raw_mapping["item_id"],
1940                provider_domain=raw_mapping["provider_domain"],
1941                provider_instance=raw_mapping["provider_instance"],
1942                available=bool(raw_mapping["available"]),
1943                in_library=parse_optional_bool(raw_mapping["in_library"]),
1944                is_unique=parse_optional_bool(raw_mapping["is_unique"]),
1945            )
1946            for raw_mapping in json_loads(db_row["provider_mappings"])
1947        }
1948
1949    def _parse_summary_row(self, db_row: Mapping[str, Any]) -> MediaItemSummaryType:
1950        """
1951        Parse a raw summary db row into a summary item of this controller's media type.
1952
1953        Override in a subclass to fill additional per-type fields (selected by the
1954        subclass's summary_query).
1955        """
1956        provider_mappings = self._parse_summary_provider_mappings(db_row)
1957        return self.summary_item_cls(
1958            item_id=str(db_row["item_id"]),
1959            provider="library",
1960            name=db_row["name"],
1961            sort_name=db_row["sort_name"],
1962            favorite=bool(db_row["favorite"]),
1963            provider_mappings=provider_mappings,
1964            available=self._summary_available(provider_mappings),
1965            metadata=self._parse_summary_metadata(db_row),
1966        )
1967
1968    @final
1969    @staticmethod
1970    def _parse_summary_provider_mappings(db_row: Mapping[str, Any]) -> set[ProviderMapping]:
1971        """Hydrate the provider mappings of a summary row into ProviderMapping objects."""
1972        if not (raw_mappings := db_row["provider_mappings"]):
1973            return set()
1974        return {ProviderMapping.from_dict(x) for x in json_loads(raw_mappings)}
1975
1976    @final
1977    @staticmethod
1978    def _summary_available(provider_mappings: set[ProviderMapping]) -> bool:
1979        """Compute the availability flag from a summary item's provider mappings."""
1980        # same semantics as the MediaItem.available property
1981        if not (available_providers := get_global_cache_value("available_providers")):
1982            return any(x.available for x in provider_mappings)
1983        if TYPE_CHECKING:
1984            available_providers = cast("set[str]", available_providers)
1985        return any(
1986            x.available and x.provider_instance in available_providers for x in provider_mappings
1987        )
1988
1989    @final
1990    @staticmethod
1991    def _parse_summary_metadata(db_row: Mapping[str, Any]) -> MediaItemMetadataSummary:
1992        """Build the slim metadata of a summary row, carrying only the (first) thumb image."""
1993        thumb: MediaItemImage | None = None
1994        if raw_images := db_row["images"]:
1995            for image in json_loads(raw_images):
1996                if image["type"] != ImageType.THUMB.value:
1997                    continue
1998                thumb = MediaItemImage(
1999                    type=ImageType.THUMB,
2000                    path=image["path"],
2001                    provider=image["provider"],
2002                    remotely_accessible=image.get("remotely_accessible", False),
2003                )
2004                break
2005        return MediaItemMetadataSummary(images=UniqueList([thumb]) if thumb else None)
2006
2007    @final
2008    def _parse_summary_artist_mappings(
2009        self, db_row: Mapping[str, Any]
2010    ) -> UniqueList[ItemMappingSummary]:
2011        """Parse the aggregated slim artist mapping rows of a summary db row."""
2012        return UniqueList(
2013            ItemMappingSummary(
2014                media_type=MediaType.ARTIST,
2015                item_id=str(raw_mapping["item_id"]),
2016                provider="library",
2017                name=raw_mapping["name"],
2018                sort_name=raw_mapping["sort_name"],
2019            )
2020            for raw_mapping in json_loads(db_row["artists"])
2021        )
2022
2023    async def _adapt_query_for_collections(
2024        self,
2025        sql_query: str,
2026        query_params: dict[str, Any],
2027        summary: bool,
2028        order_by: str | None,
2029        collection_name: str | None = None,
2030        search: str | None = None,
2031    ) -> str:
2032        cache_key_json_object = f"collection_{self.api_base}"
2033        json_object = await self.mass.cache.get(key=cache_key_json_object, category=int(summary))
2034        if json_object is None:
2035            # get column names of base query
2036            db_rows = await self.mass.music.database.get_rows_from_query(
2037                sql_query, query_params, limit=1, offset=0
2038            )
2039            # create a sql json_object which queries all these columns
2040            if db_rows:
2041                json_object = (
2042                    "json_object(" + ",".join([f"'{x}',{x}" for x in db_rows[0].keys()]) + ")"  # noqa: SIM118
2043                )
2044                await self.mass.cache.set(
2045                    key=cache_key_json_object, category=int(summary), data=json_object
2046                )
2047            else:
2048                json_object = "json_object()"
2049
2050        collections_column = "collections" if summary else "json_extract(metadata, '$.collections')"
2051
2052        supported_order_keys = [
2053            "name",
2054            "name_desc",
2055            "sort_name",
2056            "sort_name_desc",
2057            "timestamp_added",
2058            "timestamp_added_desc",
2059            "timestamp_modified",
2060            "timestamp_modified_desc",
2061            "last_played",
2062            "last_played_desc",
2063            "play_count",
2064            "play_count_desc",
2065        ]
2066
2067        # additional order options subject to media type
2068        # single is targeting a single media item, collection the aggregated ones
2069        single_extra_order_keys = ""
2070        collection_extra_order_keys = ""
2071        if MediaType.AUDIOBOOK.value in self.api_base:
2072            single_extra_order_keys = "duration,"
2073            collection_extra_order_keys = "SUM(duration) as duration,"
2074            supported_order_keys += ["duration", "duration_desc"]
2075
2076        sql_query = f"""
2077        SELECT * FROM (
2078
2079            WITH
2080                joined_table as ({sql_query}),
2081                collection_extract as (
2082                    SELECT
2083                        name as media_name,
2084                        timestamp_added,
2085                        timestamp_modified,
2086                        last_played,
2087                        play_count,
2088                        {single_extra_order_keys}
2089                        json_extract(iter_coll.value, '$.title') as collection_title,
2090                        json_extract(iter_coll.value, '$.sequence') as collection_sequence,
2091                        json_extract(iter_coll.value, '$.search_title') as collection_search_title,
2092                        json_extract(iter_coll.value, '$.search_sort_title') as collection_search_sort_title,
2093                        CASE
2094                            WHEN json_type(iter_coll.value, '$.sequence') IN ('integer', 'real')
2095                            THEN 1
2096                            WHEN json_type(iter_coll.value, '$.sequence') = 'text'
2097                                AND json_valid(json_extract(iter_coll.value, '$.sequence'))
2098                            THEN CASE
2099                                WHEN json_type(json_extract(iter_coll.value, '$.sequence'))
2100                                    IN ('integer', 'real')
2101                                THEN 1
2102                                ELSE 0
2103                            END
2104                            ELSE 0
2105                        END as collection_sequence_is_numeric,
2106                        {json_object} as media_data
2107                    FROM (
2108                        SELECT * FROM joined_table
2109                    ), json_each({collections_column}) as iter_coll
2110                )
2111            SELECT
2112                'collection' as type,
2113                collection_title as name,
2114                COALESCE(MAX(collection_search_title), replace(lower(collection_title),' ','')) AS search_name,
2115                COALESCE(MAX(collection_search_sort_title), replace(lower(collection_title),' ','')) AS search_sort_name,
2116                MAX(timestamp_added) as timestamp_added,
2117                MAX(timestamp_modified) as timestamp_modified,
2118                MAX(last_played) as last_played,
2119                SUM(play_count) as play_count,
2120                {collection_extra_order_keys}
2121                json_group_array(media_data) as media_data
2122            FROM (
2123                SELECT * FROM collection_extract
2124                -- NOTE: The following ORDER_BY to control the aggregation order of json_group_array is undocumented sqlite behavior
2125                -- Confirmed working with sqlite 3.40.1 & 3.53
2126                -- Once our image moves to sqlite 3.44 we can and should make use of ORDER_BY in the aggregate itself
2127                ORDER BY collection_title,
2128                -- null case
2129                CASE WHEN collection_sequence IS NULL THEN 1 ELSE 0 END,
2130                -- numeric before text
2131                CASE WHEN collection_sequence_is_numeric THEN 0 ELSE 1 END,
2132                -- order NUMERIC
2133                CASE WHEN collection_sequence_is_numeric
2134                    THEN CAST(collection_sequence AS REAL)
2135                END,
2136                -- order TEXT
2137                CASE WHEN NOT collection_sequence_is_numeric
2138                    THEN collection_sequence
2139                END COLLATE NOCASE,
2140                -- order by media name if no sequence given
2141                CASE
2142                    WHEN collection_sequence IS NULL
2143                    THEN media_name
2144                END COLLATE NOCASE
2145            )
2146            GROUP BY collection_title
2147
2148            UNION ALL
2149
2150            SELECT 'single', name, search_name, search_sort_name,
2151                timestamp_added, timestamp_modified, last_played, play_count,
2152                {single_extra_order_keys}
2153                {json_object} FROM joined_table
2154                WHERE {collections_column} IS NULL
2155                    OR {collections_column} = '[]'
2156        )
2157        """
2158
2159        if collection_name:
2160            sql_query += " WHERE type = 'collection' AND name = :collection_name"
2161            return sql_query
2162
2163        if search:
2164            sql_query += " WHERE search_name LIKE :search"
2165
2166        if order_by:
2167            if order_by not in supported_order_keys:
2168                self.logger.warning("%s is not supported for order_by key in collections", order_by)
2169                order_by = "name"  # fallback
2170            if sort_key := SORT_KEYS.get(order_by):
2171                sql_query += f" ORDER BY {sort_key}"
2172
2173        return sql_query
2174