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