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    InvalidDataError,
27    MediaNotFoundError,
28    ProviderUnavailableError,
29)
30from music_assistant_models.helpers import create_safe_string, get_global_cache_value
31from music_assistant_models.media_items import (
32    AudioFormat,
33    ItemMapping,
34    ItemMappingSummary,
35    MediaCollection,
36    MediaItemImage,
37    MediaItemMetadata,
38    MediaItemMetadataSummary,
39    MediaItemSummaryType,
40    MediaItemType,
41    ProviderMapping,
42    UniqueList,
43)
44
45from music_assistant.constants import (
46    DB_TABLE_ALBUM_ARTISTS,
47    DB_TABLE_ALBUM_TRACKS,
48    DB_TABLE_AUDIO_ANALYSIS,
49    DB_TABLE_AUDIOBOOK_ARTISTS,
50    DB_TABLE_EXTERNAL_ID_LOOKUP,
51    DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
52    DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
53    DB_TABLE_PLAYLOG,
54    DB_TABLE_PROVIDER_MAPPINGS,
55    DB_TABLE_TRACK_ARTISTS,
56    MASS_LOGGER_NAME,
57)
58from music_assistant.controllers.music.helpers import search_name_match_clause
59from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
60from music_assistant.helpers.collections import (
61    get_collection_item_id,
62    get_collection_name_from_item_id,
63)
64from music_assistant.helpers.compare import compare_media_item
65from music_assistant.helpers.database import UNSET
66from music_assistant.helpers.external_ids import (
67    external_id_lookup_values,
68    external_id_lookup_values_untyped,
69    external_id_sort_key,
70    normalize_external_ids,
71)
72from music_assistant.helpers.json import json_loads, serialize_to_json
73from music_assistant.helpers.util import guard_single_request, parse_optional_bool
74
75if TYPE_CHECKING:
76    from collections.abc import AsyncGenerator, Mapping
77
78    from music_assistant import MusicAssistant
79    from music_assistant.models.music_provider import MusicProvider
80    from music_assistant.models.plugin import PluginProvider
81
82
83ItemCls = TypeVar("ItemCls", bound="MediaItemType")
84
85
86JSON_KEYS = (
87    "artists",
88    "track_album",
89    "metadata",
90    "provider_mappings",
91    "external_ids",
92    "narrators",
93    "authors",
94    "genre_aliases",
95    "supported_mediatypes",
96    "translation_params",
97    "audiobook_artists",
98)
99
100# The columns that make up a relation row, so a merge can copy it onto the target
101# without relying on SELECT *: album_tracks carries a surrogate autoincrement id that
102# must not be copied along.
103RELATION_TABLE_COLUMNS = {
104    DB_TABLE_ALBUM_ARTISTS: ("album_id", "artist_id"),
105    DB_TABLE_ALBUM_TRACKS: ("track_id", "album_id", "disc_number", "track_number"),
106    DB_TABLE_AUDIOBOOK_ARTISTS: ("audiobook_id", "artist_id"),
107    DB_TABLE_TRACK_ARTISTS: ("track_id", "artist_id"),
108}
109
110# When set (task-local), per-item MEDIA_ITEM_ADDED/UPDATED events and the on_item_updated
111# provider write-back are suppressed, so bulk operations (provider sync, provider cleanup)
112# don't flood subscribers with one event per touched item.
113SUPPRESS_MEDIA_ITEM_UPDATES: ContextVar[bool] = ContextVar(
114    "SUPPRESS_MEDIA_ITEM_UPDATES", default=False
115)
116
117SORT_KEYS = {
118    # sqlite has no builtin support for natural sorting
119    # so we have use an additional column for this
120    # this also improves searching and sorting performance
121    "name": "search_name ASC",
122    "name_desc": "search_name DESC",
123    "duration": "duration ASC",
124    "duration_desc": "duration DESC",
125    "sort_name": "search_sort_name ASC",
126    "sort_name_desc": "search_sort_name DESC",
127    "timestamp_added": "timestamp_added ASC",
128    "timestamp_added_desc": "timestamp_added DESC",
129    "timestamp_modified": "timestamp_modified ASC",
130    "timestamp_modified_desc": "timestamp_modified DESC",
131    "last_played": "last_played ASC",
132    "last_played_desc": "last_played DESC",
133    "play_count": "play_count ASC",
134    "play_count_desc": "play_count DESC",
135    "year": "year ASC",
136    "year_desc": "year DESC",
137    "position": "position ASC",
138    "position_desc": "position DESC",
139    "album_artist_name": "artists.search_name ASC, year DESC",
140    "album_artist_name_desc": "artists.search_name DESC, year DESC",
141    "track_artist_name": "artists.search_name ASC, search_name ASC",
142    "track_artist_name_desc": "artists.search_name DESC, search_name ASC",
143    "random": "RANDOM()",
144    "random_play_count": "RANDOM(), play_count ASC",
145}
146
147
148@dataclass(slots=True)
149class LibraryItemSyncDetails:
150    """
151    Lightweight snapshot of a library item with just the fields the library sync needs.
152
153    Used by the provider sync loops to detect (un)changed items without hydrating
154    full MediaItem objects from the database.
155    """
156
157    item_id: int
158    favorite: bool
159    date_added: datetime
160    provider_mappings: set[ProviderMapping]
161
162
163@dataclass(slots=True)
164class TrackSyncDetails(LibraryItemSyncDetails):
165    """Lightweight sync snapshot of a library track."""
166
167    has_album: bool
168
169
170@dataclass(slots=True)
171class AudiobookSyncDetails(LibraryItemSyncDetails):
172    """Lightweight sync snapshot of a library audiobook."""
173
174    author_is_str: bool
175    narrator_is_str: bool
176    fully_played: bool | None
177    resume_position_ms: int | None
178
179
180class MediaControllerBase[ItemCls: "MediaItemType"](metaclass=ABCMeta):
181    """Base model for controller managing a MediaType."""
182
183    media_type: MediaType
184    item_cls: type[MediaItemType]
185    summary_item_cls: type[MediaItemSummaryType]
186    db_table: str
187
188    def __init__(self, mass: MusicAssistant) -> None:
189        """Initialize class."""
190        self.mass = mass
191        self.logger = logging.getLogger(f"{MASS_LOGGER_NAME}.music.{self.media_type.value}")
192        # register (base) api handlers
193        self.api_base = api_base = f"{self.media_type}s"
194        self.mass.register_api_command(
195            f"music/{api_base}/count", self.library_count, required_scope=Scope.LIBRARY_READ
196        )
197        self.mass.register_api_command(
198            f"music/{api_base}/library_items",
199            self.library_items,
200            required_scope=Scope.LIBRARY_READ,
201            allow_impersonation=True,
202        )
203        self.mass.register_api_command(
204            f"music/{api_base}/get", self.get, required_scope=Scope.LIBRARY_READ
205        )
206        self.mass.register_api_command(
207            f"music/{api_base}/get_by_external_id",
208            self.get_library_item_by_external_id,
209            required_scope=Scope.LIBRARY_READ,
210        )
211        self.mass.register_api_command(
212            f"music/{api_base}/get_collection",
213            self.get_collection,
214            required_scope=Scope.LIBRARY_READ,
215            allow_impersonation=True,
216        )
217        # Backward compatibility alias - prefer the generic "get" endpoint
218        self.mass.register_api_command(
219            f"music/{api_base}/get_{self.media_type}",
220            self.get,
221            required_scope=Scope.LIBRARY_READ,
222            alias=True,
223        )
224        self.mass.register_api_command(
225            f"music/{api_base}/update",
226            self.update_item_in_library,
227            required_scope=Scope.LIBRARY_MANAGE,
228        )
229        self.mass.register_api_command(
230            f"music/{api_base}/remove",
231            self.remove_item_from_library,
232            required_scope=Scope.LIBRARY_MANAGE,
233        )
234        self._db_add_lock = asyncio.Lock()
235
236    @property
237    def translation_owner(self) -> str:
238        """Return the "core.music" namespace these media controllers' translation strings live under."""
239        return "core.music"
240
241    @property
242    def base_query(self) -> tuple[str, dict[str, Any]]:
243        """
244        Return the base SELECT query for this media type and its bound query params.
245
246        Override in a subclass to customize the query (extra joins/columns) and/or to
247        inject dynamic, parameterized filters.
248        """
249        query = f"""
250        SELECT
251            {self.db_table}.*,
252            {self._external_ids_query()} AS external_ids,
253            {self._provider_mappings_query()} AS provider_mappings
254            FROM {self.db_table} """
255        return query, {}
256
257    @property
258    def summary_query(self) -> tuple[str, dict[str, Any]]:
259        """
260        Return the slim SELECT query used for summary listings and its bound query params.
261
262        Selects only the columns needed to build summary items. Override in a subclass
263        to select additional per-type columns.
264        """
265        query = f"""
266        SELECT
267            {self._summary_base_columns()},
268            {self._provider_mappings_query()} AS provider_mappings
269            FROM {self.db_table} """
270        return query, {}
271
272    @final
273    async def add_item_to_library(
274        self,
275        item: ItemCls,
276        overwrite_existing: bool = False,
277    ) -> ItemCls:
278        """Add item to library and return the new (or updated) database item."""
279        new_item = False
280        # batch the many writes of an item add/update into a single commit
281        async with self.mass.music.database.deferred_commit():
282            # check for existing item first
283            if library_id := await self._get_library_item_by_match(item):
284                # update existing item
285                await self._update_library_item(library_id, item, overwrite=overwrite_existing)
286            else:
287                # actually add a new item in the library db
288                self.mass.music.match_provider_instances(item)
289                async with self._db_add_lock:
290                    # Another task may have inserted the same item while this task waited.
291                    if library_id := await self._get_library_item_by_match(item):
292                        await self._update_library_item(
293                            library_id, item, overwrite=overwrite_existing
294                        )
295                    else:
296                        library_id = await self._add_library_item(item)
297                        new_item = True
298        # return final library_item
299        library_item = await self.get_library_item(library_id)
300        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
301            self.mass.signal_event(
302                EventType.MEDIA_ITEM_ADDED if new_item else EventType.MEDIA_ITEM_UPDATED,
303                library_item.uri,
304                library_item,
305            )
306        return library_item
307
308    @final
309    async def update_item_in_library(
310        self, item_id: str | int, update: ItemCls, overwrite: bool = False
311    ) -> ItemCls:
312        """Update existing library record in the library database."""
313        self.mass.music.match_provider_instances(update)
314        # batch the many writes of an item update into a single commit
315        async with self.mass.music.database.deferred_commit():
316            await self._update_library_item(item_id, update, overwrite=overwrite)
317        # return the updated object
318        library_item = await self.get_library_item(item_id)
319        if SUPPRESS_MEDIA_ITEM_UPDATES.get():
320            # during a sync the update originates from the provider itself,
321            # so skip both the event and the write-back to that provider
322            return library_item
323        # drop cached artwork for the updated item so replaced art is served fresh
324        for img in library_item.metadata.images or []:
325            await self.mass.metadata.invalidate_image_cache(img.provider, img.path)
326        self.mass.signal_event(
327            EventType.MEDIA_ITEM_UPDATED,
328            library_item.uri,
329            library_item,
330        )
331        # notify music providers of the update so they can sync their own storage
332        for prov_mapping in library_item.provider_mappings:
333            if provider := self.mass.get_provider(prov_mapping.provider_instance):
334                if provider.type != ProviderType.MUSIC:
335                    continue
336                provider = cast("MusicProvider", provider)
337                await provider.on_item_updated(library_item)
338        return library_item
339
340    async def remove_item_from_library(self, item_id: str | int, recursive: bool = True) -> None:
341        """Delete library record from the database."""
342        db_id = int(item_id)  # ensure integer
343        library_item = await self.get_library_item(db_id)
344        assert library_item, f"Item does not exist: {db_id}"
345        # delete item
346        await self.mass.music.database.delete(
347            self.db_table,
348            {"item_id": db_id},
349        )
350        # update provider_mappings table
351        await self.mass.music.database.delete(
352            DB_TABLE_PROVIDER_MAPPINGS,
353            {"media_type": self.media_type.value, "item_id": db_id},
354        )
355        # cleanup external_id_lookup table
356        await self.mass.music.database.delete(
357            DB_TABLE_EXTERNAL_ID_LOOKUP,
358            {"media_type": self.media_type.value, "item_id": db_id},
359        )
360        # cleanup playlog table
361        await self.mass.music.database.delete(
362            DB_TABLE_PLAYLOG,
363            {
364                "media_type": self.media_type.value,
365                "item_id": db_id,
366                "provider": "library",
367            },
368        )
369        for prov_mapping in library_item.provider_mappings:
370            await self.mass.music.database.delete(
371                DB_TABLE_PLAYLOG,
372                {
373                    "media_type": self.media_type.value,
374                    "item_id": prov_mapping.item_id,
375                    "provider": prov_mapping.provider_instance,
376                },
377            )
378            # cleanup audio analysis rows for this provider mapping
379            for prov_key in (prov_mapping.provider_domain, prov_mapping.provider_instance):
380                await self.mass.music.database.delete(
381                    DB_TABLE_AUDIO_ANALYSIS,
382                    {
383                        "media_type": self.media_type.value,
384                        "item_id": prov_mapping.item_id,
385                        "provider": prov_key,
386                    },
387                )
388        # delete genre exclusions for this media item
389        await self.mass.music.database.delete(
390            DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
391            {"media_type": self.media_type.value, "media_id": db_id},
392        )
393        # NOTE: this does not delete any references to this item in other records,
394        # this is handled/overridden in the mediatype specific controllers
395        # drop cached artwork for the removed item
396        for img in library_item.metadata.images or []:
397            await self.mass.metadata.invalidate_image_cache(img.provider, img.path)
398        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
399            self.mass.signal_event(EventType.MEDIA_ITEM_DELETED, library_item.uri, library_item)
400        self.logger.debug("deleted item with id %s from database", db_id)
401
402    async def library_count(self, favorite_only: bool = False) -> int:
403        """
404        Return the number of items in the library.
405
406        Restricted to the providers the current user is allowed to see when that user
407        has a provider filter set.
408
409        :param favorite_only: Only count items marked as favorite.
410        """
411        query_parts: list[str] = []
412        query_params: dict[str, Any] = {}
413        if favorite_only:
414            query_parts.append("favorite = 1")
415        if provider_filter := self._ensure_provider_filter(None):
416            query_parts.append(
417                self._provider_filter_clause(query_params, provider_filter, in_library_only=True)
418            )
419        if not query_parts:
420            return await self.mass.music.database.get_count(self.db_table)
421        sql_query = f"SELECT item_id FROM {self.db_table} WHERE {' AND '.join(query_parts)}"
422        return await self.mass.music.database.get_count_from_query(sql_query, query_params)
423
424    if TYPE_CHECKING:
425
426        @overload
427        async def library_items(
428            self,
429            favorite: bool | None = None,
430            search: str | None = None,
431            limit: int = 500,
432            offset: int = 0,
433            order_by: str = "sort_name",
434            provider: str | list[str] | None = None,
435            genre: int | list[int] | None = None,
436            played_only: bool = False,
437            *,
438            summary: bool = True,
439            collapse_collections: Literal[False] = False,
440            reachable_via: list[str] | None = None,
441            **kwargs: Any,
442        ) -> list[ItemCls]: ...
443
444        @overload
445        async def library_items(
446            self,
447            favorite: bool | None = None,
448            search: str | None = None,
449            limit: int = 500,
450            offset: int = 0,
451            order_by: str = "sort_name",
452            provider: str | list[str] | None = None,
453            genre: int | list[int] | None = None,
454            played_only: bool = False,
455            *,
456            summary: bool = True,
457            collapse_collections: Literal[True],
458            reachable_via: list[str] | None = None,
459            **kwargs: Any,
460        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
461
462        @overload
463        async def library_items(
464            self,
465            favorite: bool | None = None,
466            search: str | None = None,
467            limit: int = 500,
468            offset: int = 0,
469            order_by: str = "sort_name",
470            provider: str | list[str] | None = None,
471            genre: int | list[int] | None = None,
472            played_only: bool = False,
473            *,
474            summary: bool = True,
475            collapse_collections: bool,
476            reachable_via: list[str] | None = None,
477            **kwargs: Any,
478        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
479
480    async def library_items(  # noqa: PLR0913
481        self,
482        favorite: bool | None = None,
483        search: str | None = None,
484        limit: int = 500,
485        offset: int = 0,
486        order_by: str = "sort_name",
487        provider: str | list[str] | None = None,
488        genre: int | list[int] | None = None,
489        played_only: bool = False,
490        *,
491        summary: bool = True,
492        collapse_collections: bool = False,
493        reachable_via: list[str] | None = None,
494        **kwargs: Any,
495    ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]:
496        """
497        Get the library items for this mediatype.
498
499        :param favorite: Filter by favorite status.
500        :param search: Filter by search query.
501        :param limit: Maximum number of items to return.
502        :param offset: Number of items to skip.
503        :param order_by: Order by field (e.g. 'sort_name', 'timestamp_added').
504        :param provider: Filter by provider instance ID (single string or list).
505        :param genre: Filter by genre id(s).
506        :param played_only: Only include items that have been played (last_played > 0).
507        :param summary: When True (default), return slim summary items containing only the
508            fields needed for a list view. Set to False to get fully hydrated items.
509        :param collapse_collections: Collapse available collections. Items in a collection won't
510            be returned individually.
511        :param reachable_via: Restrict results to items with a provider mapping reachable
512            through one of these provider instance ids (OR semantics), regardless of
513            whether that mapping is itself in that provider's own library. This is
514            independent of `provider`, which instead requires the *matched* mapping to
515            be in-library. None applies no filter; an explicit empty list, or a list
516            with no currently loaded/allowed instance, returns no items.
517        """
518        reachable_via = self._resolve_reachable_via(reachable_via)
519        if reachable_via is not None and not reachable_via:
520            return []
521        items = await self.get_library_items_by_query(
522            favorite=favorite,
523            search=search,
524            limit=limit,
525            offset=offset,
526            order_by=order_by,
527            provider_filter=self._provider_filter_considering_reachability(provider, reachable_via),
528            genre_ids=genre,
529            played_only=played_only,
530            in_library_only=True,
531            summary=summary,
532            collapse_collections=collapse_collections,
533            reachable_via=reachable_via,
534        )
535        if (
536            kwargs.get("_localized_fallback", True)
537            and search
538            and not items
539            and self.media_type in (MediaType.GENRE, MediaType.PLAYLIST)
540        ):
541            return await self._localized_search_fallback(
542                search,
543                limit=limit,
544                offset=offset,
545                favorite=favorite,
546                order_by=order_by,
547                provider=provider,
548                genre=genre,
549                summary=summary,
550                reachable_via=reachable_via,
551            )
552        return items
553
554    async def iter_library_items(
555        self,
556        favorite: bool | None = None,
557        search: str | None = None,
558        order_by: str = "sort_name",
559        provider: str | list[str] | None = None,
560        genre: int | list[int] | None = None,
561        library_items_only: bool = True,
562    ) -> AsyncGenerator[ItemCls]:
563        """Iterate all in-database items."""
564        limit: int = 500
565        offset: int = 0
566        if provider is not None:
567            provider_filter = provider if isinstance(provider, list) else [provider]
568        else:
569            provider_filter = None
570        while True:
571            next_items = await self.get_library_items_by_query(
572                favorite=favorite,
573                search=search,
574                genre_ids=genre,
575                limit=limit,
576                offset=offset,
577                order_by=order_by,
578                provider_filter=provider_filter,
579                in_library_only=library_items_only,
580            )
581            for item in next_items:
582                yield item
583            if len(next_items) < limit:
584                break
585            offset += limit
586
587    async def get(
588        self,
589        item_id: str,
590        provider_instance_id_or_domain: str,
591        allow_update_metadata: bool = True,
592    ) -> ItemCls:
593        """
594        Return (full) details for a single media item.
595
596        Tries to find the item in the library first, falling back to
597        fetching directly from the provider if not found.
598
599        :param item_id: The provider item id to fetch.
600        :param provider_instance_id_or_domain: The provider instance id or
601            domain to fetch the item from.
602        :param allow_update_metadata: Schedule a metadata refresh on access.
603            Set to False when fetching items in bulk (e.g. provider sync).
604        """
605        # always prefer the full library item if we have it
606        if library_item := await self.get_library_item_by_prov_id(
607            item_id,
608            provider_instance_id_or_domain,
609        ):
610            # schedule a refresh of the metadata on access of the item
611            # e.g. the item is being played or opened in the UI
612            if allow_update_metadata:
613                assert library_item.uri is not None
614                self.mass.metadata.schedule_update_metadata(library_item)
615            return library_item
616        # grab full details from the provider
617        return await self.get_provider_item(
618            item_id,
619            provider_instance_id_or_domain,
620        )
621
622    async def search(
623        self,
624        search_query: str,
625        provider_instance_id_or_domain: str,
626        limit: int = 25,
627    ) -> list[ItemCls]:
628        """Search database or provider with given query."""
629        # create safe search string
630        search_query = search_query.replace("/", " ").replace("'", "")
631        if provider_instance_id_or_domain == "library":
632            return await self.library_items(
633                search=search_query, limit=limit, summary=False, collapse_collections=False
634            )
635        if not (prov := self.mass.get_provider(provider_instance_id_or_domain)):
636            return []
637        if prov.type != ProviderType.MUSIC:
638            return []
639        prov = cast("MusicProvider", prov)
640        if ProviderFeature.SEARCH not in prov.supported_features:
641            return []
642        if self.media_type not in prov.supported_media_types:
643            return []
644        searchresult = await prov.search(
645            search_query,
646            [self.media_type],
647            limit,
648        )
649        match self.media_type:
650            case MediaType.ARTIST:
651                return cast("list[ItemCls]", searchresult.artists)
652            case MediaType.ALBUM:
653                return cast("list[ItemCls]", searchresult.albums)
654            case MediaType.TRACK:
655                return cast("list[ItemCls]", searchresult.tracks)
656            case MediaType.PLAYLIST:
657                return cast("list[ItemCls]", searchresult.playlists)
658            case MediaType.AUDIOBOOK:
659                return cast("list[ItemCls]", searchresult.audiobooks)
660            case MediaType.PODCAST:
661                return cast("list[ItemCls]", searchresult.podcasts)
662            case MediaType.RADIO:
663                return cast("list[ItemCls]", searchresult.radio)
664            case _:
665                return []
666
667    async def get_collection(self, item_id: str) -> MediaCollection[ItemCls]:
668        """Get a single collection."""
669        name = get_collection_name_from_item_id(item_id)
670        query_params: dict[str, Any] = {"collection_name": name}
671        sql_query, base_query_params = self._build_final_query([], [], None, summary=False)
672        for key, value in base_query_params.items():
673            query_params.setdefault(key, value)
674        sql_query = await self._adapt_query_for_collections(
675            sql_query, query_params, summary=False, order_by=None, collection_name=name
676        )
677        db_rows = await self.mass.music.database.get_rows_from_query(
678            sql_query, query_params, limit=1, offset=0
679        )
680        if len(db_rows) != 1:
681            raise MediaNotFoundError(f"Collection {name} not found.")
682
683        return cast(
684            "MediaCollection[ItemCls]",
685            MediaCollection(
686                item_id=get_collection_item_id(db_rows[0]["name"], item_media_type=self.media_type),
687                name=db_rows[0]["name"],
688                provider="library",
689                provider_mappings=set(),
690                items=UniqueList(
691                    [
692                        self.item_cls.from_dict(self._parse_db_row(json_loads(x)))
693                        for x in json_loads(db_rows[0]["media_data"])
694                    ]
695                ),
696            ),
697        )
698
699    async def get_library_item(self, item_id: int | str) -> ItemCls:
700        """Get single library item by id."""
701        db_id = int(item_id)  # ensure integer
702        extra_query = f"WHERE {self.db_table}.item_id = :item_id"
703        for db_item in await self.get_library_items_by_query(
704            extra_query_parts=[extra_query],
705            extra_query_params={"item_id": db_id},
706            in_library_only=False,
707        ):
708            return db_item
709        msg = f"{self.media_type.value} not found in library: {db_id}"
710        raise MediaNotFoundError(msg)
711
712    async def get_library_item_by_prov_id(
713        self,
714        item_id: str,
715        provider_instance_id_or_domain: str,
716    ) -> ItemCls | None:
717        """Get the library item for the given provider item, if present."""
718        assert item_id
719        assert provider_instance_id_or_domain
720        if provider_instance_id_or_domain == "library":
721            try:
722                return await self.get_library_item(item_id)
723            except MediaNotFoundError:
724                return None
725        for item in await self.get_library_items_by_prov_id(
726            provider_instance_id_or_domain=provider_instance_id_or_domain,
727            provider_item_id=item_id,
728        ):
729            return item
730        return None
731
732    @final
733    async def get_library_item_by_prov_mappings(
734        self,
735        provider_mappings: Iterable[ProviderMapping],
736    ) -> ItemCls | None:
737        """Get the library item for the given provider_instance."""
738        # always prefer provider instance first
739        for mapping in provider_mappings:
740            for item in await self.get_library_items_by_prov_id(
741                provider_instance=mapping.provider_instance,
742                provider_item_id=mapping.item_id,
743            ):
744                return item
745        # check by domain too
746        for mapping in provider_mappings:
747            for item in await self.get_library_items_by_prov_id(
748                provider_domain=mapping.provider_domain,
749                provider_item_id=mapping.item_id,
750            ):
751                return item
752        return None
753
754    @final
755    async def get_library_item_sync_details(
756        self,
757        provider_mappings: Iterable[ProviderMapping],
758    ) -> LibraryItemSyncDetails | None:
759        """
760        Get a lightweight sync snapshot of the library item for the given provider mappings.
761
762        Returns only the scalar columns and raw provider mapping rows the library sync
763        needs for its change detection, without hydrating a full MediaItem object.
764        Resolution order matches get_library_item_by_prov_mappings (instance first,
765        then domain).
766        """
767        extra_columns, extra_joins, extra_params = self._sync_details_query_parts()
768        base_sql = f"""
769            SELECT
770                {self.db_table}.item_id,
771                {self.db_table}.favorite,
772                {self.db_table}.timestamp_added,
773                (SELECT JSON_GROUP_ARRAY(
774                    json_object(
775                        'item_id', pm.provider_item_id,
776                        'provider_domain', pm.provider_domain,
777                        'provider_instance', pm.provider_instance,
778                        'available', pm.available,
779                        'in_library', pm.in_library,
780                        'is_unique', pm.is_unique
781                    )) FROM provider_mappings pm WHERE pm.item_id = {self.db_table}.item_id
782                        AND pm.media_type = '{self.media_type.value}') AS provider_mappings
783                {extra_columns}
784            FROM {self.db_table}
785            {extra_joins}
786            WHERE {self.db_table}.item_id IN (
787                SELECT item_id FROM provider_mappings
788                WHERE provider_mappings.media_type = '{self.media_type.value}'
789                AND provider_mappings.{{prov_column}} = :prov_id
790                AND provider_mappings.provider_item_id = :prov_item_id
791            )
792        """
793        # always prefer provider instance first, then domain
794        # (same resolution order as get_library_item_by_prov_mappings)
795        for prov_column in ("provider_instance", "provider_domain"):
796            for mapping in provider_mappings:
797                for db_row in await self.mass.music.database.get_rows_from_query(
798                    base_sql.format(prov_column=prov_column),
799                    {
800                        **extra_params,
801                        "prov_id": getattr(mapping, prov_column),
802                        "prov_item_id": mapping.item_id,
803                    },
804                    limit=1,
805                ):
806                    return self._parse_sync_details_row(db_row)
807        return None
808
809    @final
810    async def get_library_items_by_external_id(
811        self,
812        external_id: str,
813        external_id_type: ExternalID | None = None,
814        *,
815        limit: int | None,
816    ) -> list[ItemCls]:
817        """
818        Get library items for the given external identifier.
819
820        :param external_id: External identifier value to look up.
821        :param external_id_type: Optional identifier type.
822        :param limit: Maximum number of library items to return, or None for all matches.
823        """
824        if external_id_type:
825            lookup_values = external_id_lookup_values(external_id_type, external_id)
826        else:
827            lookup_values = external_id_lookup_values_untyped(external_id)
828        subquery_parts = [
829            "media_type = :ext_id_media_type",
830            "external_id IN :external_ids",
831        ]
832        query_params: dict[str, Any] = {
833            "ext_id_media_type": self.media_type.value,
834            "external_ids": lookup_values,
835        }
836        if external_id_type:
837            subquery_parts.append("external_id_type = :external_id_type")
838            query_params["external_id_type"] = str(external_id_type)
839        subquery = (
840            f"SELECT item_id FROM {DB_TABLE_EXTERNAL_ID_LOOKUP} "
841            f"WHERE {' AND '.join(subquery_parts)}"
842        )
843        query = f"{self.db_table}.item_id IN ({subquery})"
844        if limit is not None:
845            limited_items = await self.get_library_items_by_query(
846                limit=limit,
847                extra_query_parts=[query],
848                extra_query_params=query_params,
849            )
850            return sorted(limited_items, key=lambda item: int(item.item_id))
851
852        all_items: list[ItemCls] = []
853        offset = 0
854        page_size = 500
855        while page := await self.get_library_items_by_query(
856            limit=page_size,
857            offset=offset,
858            extra_query_parts=[query],
859            extra_query_params=query_params,
860        ):
861            all_items.extend(page)
862            if len(page) < page_size:
863                break
864            offset += page_size
865        return sorted(all_items, key=lambda item: int(item.item_id))
866
867    @final
868    async def get_library_item_by_external_id(
869        self, external_id: str, external_id_type: ExternalID | None = None
870    ) -> ItemCls | None:
871        """Get the first library item for the given external id, if present."""
872        items = await self.get_library_items_by_external_id(external_id, external_id_type, limit=1)
873        return items[0] if items else None
874
875    @final
876    async def get_library_items_by_external_ids(
877        self, external_ids: set[tuple[ExternalID, str]]
878    ) -> list[ItemCls]:
879        """Get all library items matching any of the given external identifiers."""
880        result: dict[str, ItemCls] = {}
881        for external_id_type, external_id in sorted(external_ids, key=external_id_sort_key):
882            for item in await self.get_library_items_by_external_id(
883                external_id, external_id_type, limit=None
884            ):
885                result.setdefault(item.item_id, item)
886        return list(result.values())
887
888    @final
889    async def get_library_item_by_external_ids(
890        self, external_ids: set[tuple[ExternalID, str]]
891    ) -> ItemCls | None:
892        """Get the library item for (one of) the given external ids."""
893        items = await self.get_library_items_by_external_ids(external_ids)
894        return items[0] if items else None
895
896    @final
897    async def get_library_items_by_prov_id(
898        self,
899        provider_domain: str | None = None,
900        provider_instance: str | None = None,
901        provider_instance_id_or_domain: str | None = None,
902        provider_item_id: str | None = None,
903        provider_item_ids: list[str] | None = None,
904        limit: int = 500,
905        offset: int = 0,
906    ) -> list[ItemCls]:
907        """
908        Fetch all records from library for given provider.
909
910        :param provider_item_ids: When given, batch-match this list of provider
911            item ids in a single query (the plural form of provider_item_id);
912            takes precedence over provider_item_id when both are passed. An
913            empty list matches nothing (distinct from None, which applies no
914            item-id filter).
915        """
916        assert provider_instance_id_or_domain != "library"
917        assert provider_domain != "library"
918        assert provider_instance != "library"
919        if provider_item_ids is not None and not provider_item_ids:
920            return []
921        subquery_parts: list[str] = []
922        query_params: dict[str, Any] = {}
923        if provider_instance:
924            query_params = {"prov_id": provider_instance}
925            subquery_parts.append("provider_mappings.provider_instance = :prov_id")
926        elif provider_domain:
927            query_params = {"prov_id": provider_domain}
928            subquery_parts.append("provider_mappings.provider_domain = :prov_id")
929        else:
930            query_params = {"prov_id": provider_instance_id_or_domain}
931            subquery_parts.append(
932                "(provider_mappings.provider_instance = :prov_id "
933                "OR provider_mappings.provider_domain = :prov_id)"
934            )
935        if provider_item_ids:
936            placeholders = ", ".join(f":item_id_{i}" for i in range(len(provider_item_ids)))
937            subquery_parts.append(f"provider_mappings.provider_item_id IN ({placeholders})")
938            for i, item_id in enumerate(provider_item_ids):
939                query_params[f"item_id_{i}"] = item_id
940        elif provider_item_id:
941            subquery_parts.append("provider_mappings.provider_item_id = :item_id")
942            query_params["item_id"] = provider_item_id
943        subquery = f"SELECT item_id FROM provider_mappings WHERE {' AND '.join(subquery_parts)}"
944        query = f"WHERE {self.db_table}.item_id IN ({subquery})"
945        return await self.get_library_items_by_query(
946            limit=limit,
947            offset=offset,
948            extra_query_parts=[query],
949            extra_query_params=query_params,
950            in_library_only=False,
951        )
952
953    @final
954    async def iter_library_items_by_prov_id(
955        self,
956        provider_instance_id_or_domain: str,
957        provider_item_id: str | None = None,
958    ) -> AsyncGenerator[ItemCls]:
959        """Iterate all records from database for given provider."""
960        limit: int = 500
961        offset: int = 0
962        while True:
963            next_items = await self.get_library_items_by_prov_id(
964                provider_instance_id_or_domain=provider_instance_id_or_domain,
965                provider_item_id=provider_item_id,
966                limit=limit,
967                offset=offset,
968            )
969            for item in next_items:
970                yield item
971            if len(next_items) < limit:
972                break
973            offset += limit
974
975    @final
976    async def set_favorite(self, item_id: str | int, favorite: bool) -> None:
977        """Set the favorite bool on a database item."""
978        db_id = int(item_id)  # ensure integer
979        library_item = await self.get_library_item(db_id)
980        if library_item.favorite == favorite:
981            return
982        match = {"item_id": db_id}
983        await self.mass.music.database.update(self.db_table, match, {"favorite": favorite})
984        library_item = await self.get_library_item(db_id)
985        self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, library_item.uri, library_item)
986
987    @guard_single_request
988    @final
989    async def get_provider_item(
990        self,
991        item_id: str,
992        provider_instance_id_or_domain: str,
993        force_refresh: bool = False,
994        fallback: ItemMapping | ItemCls | None = None,
995    ) -> ItemCls:
996        """Return item details for the given provider item id."""
997        if provider_instance_id_or_domain == "library":
998            return await self.get_library_item(item_id)
999        if not (provider := self.mass.get_provider(provider_instance_id_or_domain)):
1000            raise ProviderUnavailableError(f"{provider_instance_id_or_domain} is not available")
1001        if provider := self.mass.get_provider(provider_instance_id_or_domain):
1002            provider = cast("MusicProvider | PluginProvider", provider)
1003            with suppress(MediaNotFoundError):
1004                async with self.mass.cache.handle_refresh(force_refresh):
1005                    if self.media_type == MediaType.PLAYLIST:
1006                        return cast("ItemCls", await provider.get_playlist(item_id))
1007                    music_prov = cast("MusicProvider", provider)
1008                    if self.media_type == MediaType.ARTIST:
1009                        return cast("ItemCls", await music_prov.get_artist(item_id))
1010                    if self.media_type == MediaType.ALBUM:
1011                        return cast("ItemCls", await music_prov.get_album(item_id))
1012                    if self.media_type == MediaType.TRACK:
1013                        return cast("ItemCls", await music_prov.get_track(item_id))
1014                    if self.media_type == MediaType.RADIO:
1015                        return cast("ItemCls", await music_prov.get_radio(item_id))
1016                    if self.media_type == MediaType.AUDIOBOOK:
1017                        return cast("ItemCls", await music_prov.get_audiobook(item_id))
1018                    if self.media_type == MediaType.PODCAST:
1019                        return cast("ItemCls", await music_prov.get_podcast(item_id))
1020        # if we reach this point all possibilities failed and the item could not be found.
1021        # There is a possibility that the (streaming) provider changed the id of the item
1022        # so we return the previous details (if we have any) marked as unavailable, so
1023        # at least we have the possibility to sort out the new id through matching logic.
1024        fallback = fallback or await self.get_library_item_by_prov_id(
1025            item_id, provider_instance_id_or_domain
1026        )
1027        if (
1028            fallback
1029            and isinstance(fallback, ItemMapping)
1030            and (fallback_provider := self.mass.get_provider(fallback.provider))
1031        ):
1032            # fallback is a ItemMapping, try to convert to full item
1033            with suppress(LookupError, TypeError, ValueError):
1034                return cast(
1035                    "ItemCls",
1036                    self.item_cls.from_dict(
1037                        {
1038                            **fallback.to_dict(),
1039                            "provider_mappings": [
1040                                {
1041                                    "item_id": fallback.item_id,
1042                                    "provider_domain": fallback_provider.domain,
1043                                    "provider_instance": fallback_provider.instance_id,
1044                                    "available": fallback.available,
1045                                }
1046                            ],
1047                        }
1048                    ),
1049                )
1050        if fallback:
1051            # simply return the fallback item
1052            return cast("ItemCls", fallback)
1053        # all options exhausted, we really can not find this item
1054        msg = (
1055            f"{self.media_type.value}://{item_id} not "
1056            f"found on provider {provider_instance_id_or_domain}"
1057        )
1058        raise MediaNotFoundError(msg)
1059
1060    @final
1061    async def add_provider_mapping(
1062        self, item_id: str | int, provider_mapping: ProviderMapping
1063    ) -> None:
1064        """Add provider mapping to existing library item."""
1065        await self.add_provider_mappings(item_id, [provider_mapping])
1066
1067    @final
1068    async def merge_library_items(
1069        self, target_item_id: str | int, source_item_id: str | int
1070    ) -> ItemCls:
1071        """
1072        Merge one library item into another and return the target item.
1073
1074        The explicit target is the deterministic winner. Its current values stay authoritative
1075        where the normal non-overwrite model update keeps them; the source is merged as the
1076        incoming update. All source state is transferred before the source row is deleted.
1077
1078        :param target_item_id: Library ID of the item that remains after the merge.
1079        :param source_item_id: Library ID of the duplicate item that is removed after transfer.
1080        :raises InvalidDataError: When the IDs are identical or do not belong to this media type.
1081        """
1082        target_id = int(target_item_id)
1083        source_id = int(source_item_id)
1084        if target_id == source_id:
1085            msg = "Cannot merge a library item into itself"
1086            raise InvalidDataError(msg)
1087        async with self._db_add_lock:
1088            return await self._merge_library_items_batched(target_id, source_id)
1089
1090    @final
1091    async def add_provider_mappings(
1092        self, item_id: str | int, provider_mappings: Iterable[ProviderMapping]
1093    ) -> None:
1094        """
1095        Add provider mappings to existing library item.
1096
1097        :param item_id: The library item ID to add mappings to.
1098        :param provider_mappings: The provider mappings to add.
1099        """
1100        db_id = int(item_id)  # ensure integer
1101        mappings = set(provider_mappings)
1102        if not mappings:
1103            return
1104        async with self._db_add_lock:
1105            library_item = await self.get_library_item(db_id)
1106            while True:
1107                conflicting_item = None
1108                for mapping in mappings:
1109                    existing_item = await self.get_library_item_by_prov_id(
1110                        mapping.item_id, mapping.provider_instance
1111                    )
1112                    if existing_item and int(existing_item.item_id) != db_id:
1113                        conflicting_item = existing_item
1114                        break
1115                if conflicting_item is None:
1116                    break
1117                self.logger.debug(
1118                    "merging item id %s into item id %s based on provider mapping",
1119                    conflicting_item.item_id,
1120                    library_item.item_id,
1121                )
1122                library_item = await self._merge_library_items_batched(
1123                    db_id, int(conflicting_item.item_id)
1124                )
1125
1126            new_mappings = mappings.difference(library_item.provider_mappings)
1127            if not new_mappings:
1128                return
1129            library_item.provider_mappings.update(new_mappings)
1130            self.mass.music.match_provider_instances(library_item)
1131            await self.set_provider_mappings(db_id, library_item.provider_mappings)
1132            self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, library_item.uri, library_item)
1133
1134    @final
1135    async def update_provider_mapping(
1136        self,
1137        item_id: str | int,
1138        provider_instance_id: str,
1139        provider_item_id: str,
1140        *,
1141        available: bool | Any = UNSET,
1142        in_library: bool | Any = UNSET,
1143        is_unique: bool | None | Any = UNSET,
1144        url: str | None | Any = UNSET,
1145        details: str | None | Any = UNSET,
1146        audio_format: AudioFormat | Any = UNSET,
1147    ) -> None:
1148        """Update an existing provider mapping for a library item."""
1149        db_id = int(item_id)  # ensure integer
1150        library_item = await self.get_library_item(db_id)
1151
1152        # find the current mapping (strictly by provider instance + provider item id)
1153        cur_mapping: ProviderMapping | None = None
1154        for mapping in library_item.provider_mappings:
1155            if (
1156                mapping.provider_instance == provider_instance_id
1157                and mapping.item_id == provider_item_id
1158            ):
1159                cur_mapping = mapping
1160                break
1161        if cur_mapping is None:
1162            msg = (
1163                f"Provider mapping {provider_instance_id}/{provider_item_id} "
1164                f"not found for item {db_id}"
1165            )
1166            raise MediaNotFoundError(msg)
1167
1168        # guard against nulls for NOT NULL columns
1169        if available is None:
1170            available = UNSET
1171        if in_library is None:
1172            in_library = UNSET
1173
1174        updates: dict[str, Any] = {}
1175        if available is not UNSET:
1176            updates["available"] = bool(available)
1177        if in_library is not UNSET:
1178            updates["in_library"] = bool(in_library)
1179        if is_unique is not UNSET:
1180            updates["is_unique"] = is_unique
1181        if url is not UNSET:
1182            updates["url"] = url
1183        if details is not UNSET:
1184            updates["details"] = details
1185        if audio_format is not UNSET:
1186            updates["audio_format"] = serialize_to_json(audio_format)
1187
1188        if not updates:
1189            return
1190
1191        match = {
1192            "media_type": self.media_type.value,
1193            "item_id": db_id,
1194            "provider_instance": provider_instance_id,
1195            "provider_item_id": provider_item_id,
1196        }
1197        await self.mass.music.database.update(DB_TABLE_PROVIDER_MAPPINGS, match, updates)
1198
1199        # Re-fetch the updated item so the event payload reflects persisted DB state.
1200        updated_item = await self.get_library_item(db_id)
1201        self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, updated_item.uri, updated_item)
1202
1203    @final
1204    async def remove_provider_mapping(
1205        self, item_id: str | int, provider_instance_id: str, provider_item_id: str
1206    ) -> None:
1207        """Remove provider mapping(s) from item."""
1208        db_id = int(item_id)  # ensure integer
1209        try:
1210            library_item = await self.get_library_item(db_id)
1211        except MediaNotFoundError:
1212            # edge case: already deleted / race condition
1213            return
1214
1215        remaining_mappings = {
1216            x
1217            for x in library_item.provider_mappings
1218            if not (x.provider_instance == provider_instance_id and x.item_id == provider_item_id)
1219        }
1220        if not remaining_mappings:
1221            # this was the last mapping, so remove the entire library item, which also
1222            # clears its provider mapping rows. Dropping those rows up front would leave
1223            # the item behind without any mappings if the removal itself fails.
1224            with suppress(MediaNotFoundError):
1225                await self.remove_item_from_library(db_id)
1226            return
1227
1228        # update provider_mappings table
1229        await self.mass.music.database.delete(
1230            DB_TABLE_PROVIDER_MAPPINGS,
1231            {
1232                "media_type": self.media_type.value,
1233                "item_id": db_id,
1234                "provider_instance": provider_instance_id,
1235                "provider_item_id": provider_item_id,
1236            },
1237        )
1238        # cleanup playlog table
1239        await self.mass.music.database.delete(
1240            DB_TABLE_PLAYLOG,
1241            {
1242                "media_type": self.media_type.value,
1243                "item_id": provider_item_id,
1244                "provider": provider_instance_id,
1245            },
1246        )
1247        library_item.provider_mappings = remaining_mappings
1248        # if this was the last mapping for the provider instance, strip any artwork
1249        # that belonged to it (e.g. local file paths that are no longer resolvable)
1250        images_changed = not any(
1251            x.provider_instance == provider_instance_id for x in remaining_mappings
1252        ) and await self._remove_provider_images(db_id, provider_instance_id)
1253        self.logger.debug(
1254            "removed provider_mapping %s/%s from item id %s",
1255            provider_instance_id,
1256            provider_item_id,
1257            db_id,
1258        )
1259        # the removed provider mapping is itself a change to the item, so always notify
1260        # (unless suppressed during a bulk cleanup); re-fetch first when images were
1261        # stripped so the event payload stays accurate
1262        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
1263            event_item = await self.get_library_item(db_id) if images_changed else library_item
1264            self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, event_item.uri, event_item)
1265
1266    @final
1267    async def remove_provider_mappings(self, item_id: str | int, provider_instance_id: str) -> None:
1268        """Remove all provider mappings from an item."""
1269        db_id = int(item_id)  # ensure integer
1270        try:
1271            library_item = await self.get_library_item(db_id)
1272        except MediaNotFoundError:
1273            # edge case: already deleted / race condition, just drop any leftover rows
1274            await self.mass.music.database.delete(
1275                DB_TABLE_PROVIDER_MAPPINGS,
1276                {
1277                    "media_type": self.media_type.value,
1278                    "item_id": db_id,
1279                    "provider_instance": provider_instance_id,
1280                },
1281            )
1282            return
1283
1284        remaining_mappings = {
1285            x for x in library_item.provider_mappings if x.provider_instance != provider_instance_id
1286        }
1287        if not remaining_mappings:
1288            # these were the last mappings, so remove the entire library item, which also
1289            # clears its provider mapping rows. Dropping those rows up front would leave
1290            # the item behind without any mappings if the removal itself fails.
1291            with suppress(MediaNotFoundError):
1292                await self.remove_item_from_library(db_id)
1293            return
1294
1295        # update provider_mappings table
1296        await self.mass.music.database.delete(
1297            DB_TABLE_PROVIDER_MAPPINGS,
1298            {
1299                "media_type": self.media_type.value,
1300                "item_id": db_id,
1301                "provider_instance": provider_instance_id,
1302            },
1303        )
1304        library_item.provider_mappings = remaining_mappings
1305        # the item is kept (it still has other providers), but it may carry artwork
1306        # that belonged to the removed provider (e.g. local file paths that are no
1307        # longer resolvable), so strip those images from the stored metadata
1308        images_changed = await self._remove_provider_images(db_id, provider_instance_id)
1309        self.logger.debug(
1310            "removed all provider mappings for provider %s from item id %s",
1311            provider_instance_id,
1312            db_id,
1313        )
1314        # the removed provider mapping(s) are themselves a change to the item, so
1315        # always notify (unless suppressed during a bulk cleanup); re-fetch first when
1316        # images were stripped so the event payload stays accurate
1317        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
1318            event_item = await self.get_library_item(db_id) if images_changed else library_item
1319            self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, event_item.uri, event_item)
1320
1321    @final
1322    async def set_provider_mappings(
1323        self,
1324        item_id: str | int,
1325        provider_mappings: Iterable[ProviderMapping],
1326        overwrite: bool = False,
1327    ) -> None:
1328        """
1329        Update the provider_mappings table for the media item.
1330
1331        An empty set of mappings never clears the stored rows: an item without any
1332        mapping can not be played or resolved.
1333        """
1334        db_id = int(item_id)  # ensure integer
1335        prov_map_objs: list[dict[str, Any]] = []
1336        for provider_mapping in provider_mappings:
1337            prov_map_obj = {
1338                "media_type": self.media_type.value,
1339                "item_id": db_id,
1340                "provider_domain": provider_mapping.provider_domain,
1341                "provider_instance": provider_mapping.provider_instance,
1342                "provider_item_id": provider_mapping.item_id,
1343                "available": provider_mapping.available,
1344                "audio_format": serialize_to_json(provider_mapping.audio_format),
1345            }
1346            for key in ("url", "details", "in_library", "is_unique"):
1347                if (value := getattr(provider_mapping, key, None)) is not None:
1348                    prov_map_obj[key] = value
1349            prov_map_objs.append(prov_map_obj)
1350        if not prov_map_objs:
1351            if overwrite:
1352                # a caller asking to replace all mappings with none is a bug,
1353                # so keep the stored rows and make the attempt visible
1354                self.logger.warning(
1355                    "Ignoring request to clear all provider mappings of %s item id %s",
1356                    self.media_type.value,
1357                    db_id,
1358                )
1359            return
1360        if overwrite:
1361            # on overwrite, clear the provider_mappings table first
1362            # this is done for filesystem provider changing the path (and thus item_id)
1363            await self.mass.music.database.delete(
1364                DB_TABLE_PROVIDER_MAPPINGS,
1365                {"media_type": self.media_type.value, "item_id": db_id},
1366            )
1367        await self.mass.music.database.upsert_many(
1368            DB_TABLE_PROVIDER_MAPPINGS,
1369            prov_map_objs,
1370        )
1371
1372    @final
1373    async def set_external_ids(
1374        self,
1375        item_id: str | int,
1376        external_ids: Iterable[tuple[ExternalID, str]],
1377    ) -> None:
1378        """Update the external_id_lookup table rows for the media item."""
1379        db_id = int(item_id)  # ensure integer
1380        await self.mass.music.database.delete(
1381            DB_TABLE_EXTERNAL_ID_LOOKUP,
1382            {"media_type": self.media_type.value, "item_id": db_id},
1383        )
1384        external_ids = normalize_external_ids(external_ids)
1385        if lookup_rows := [
1386            {
1387                "media_type": self.media_type.value,
1388                "external_id_type": external_id_type,
1389                "external_id": external_id,
1390                "item_id": db_id,
1391            }
1392            for external_id_type, external_id in external_ids
1393        ]:
1394            await self.mass.music.database.upsert_many(DB_TABLE_EXTERNAL_ID_LOOKUP, lookup_rows)
1395
1396    @abstractmethod
1397    async def match_providers(self, db_item: ItemCls) -> None:
1398        """
1399        Try to find match on all (streaming) providers for the provided (database) item.
1400
1401        This is used to link objects of different providers/qualities together.
1402        """
1403
1404    if TYPE_CHECKING:
1405
1406        @overload
1407        async def get_library_items_by_query(
1408            self,
1409            favorite: bool | None = None,
1410            search: str | None = None,
1411            limit: int = 500,
1412            offset: int = 0,
1413            order_by: str | None = None,
1414            provider_filter: list[str] | None = None,
1415            extra_query_parts: list[str] | None = None,
1416            extra_query_params: dict[str, Any] | None = None,
1417            extra_join_parts: list[str] | None = None,
1418            genre_ids: int | list[int] | None = None,
1419            played_only: bool = False,
1420            in_library_only: bool = False,
1421            summary: bool = False,
1422            *,
1423            collapse_collections: Literal[True],
1424            reachable_via: list[str] | None = None,
1425        ) -> list[ItemCls | MediaCollection[ItemCls]]: ...
1426
1427        @overload
1428        async def get_library_items_by_query(
1429            self,
1430            favorite: bool | None = None,
1431            search: str | None = None,
1432            limit: int = 500,
1433            offset: int = 0,
1434            order_by: str | None = None,
1435            provider_filter: list[str] | None = None,
1436            extra_query_parts: list[str] | None = None,
1437            extra_query_params: dict[str, Any] | None = None,
1438            extra_join_parts: list[str] | None = None,
1439            genre_ids: int | list[int] | None = None,
1440            played_only: bool = False,
1441            in_library_only: bool = False,
1442            summary: bool = False,
1443            *,
1444            collapse_collections: Literal[False] = False,
1445            reachable_via: list[str] | None = None,
1446        ) -> list[ItemCls]: ...
1447
1448        @overload
1449        async def get_library_items_by_query(
1450            self,
1451            favorite: bool | None = None,
1452            search: str | None = None,
1453            limit: int = 500,
1454            offset: int = 0,
1455            order_by: str | None = None,
1456            provider_filter: list[str] | None = None,
1457            extra_query_parts: list[str] | None = None,
1458            extra_query_params: dict[str, Any] | None = None,
1459            extra_join_parts: list[str] | None = None,
1460            genre_ids: int | list[int] | None = None,
1461            played_only: bool = False,
1462            in_library_only: bool = False,
1463            summary: bool = False,
1464            *,
1465            collapse_collections: bool,
1466            reachable_via: list[str] | None = None,
1467        ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]: ...
1468
1469    @final
1470    async def get_library_items_by_query(  # noqa: PLR0913
1471        self,
1472        favorite: bool | None = None,
1473        search: str | None = None,
1474        limit: int = 500,
1475        offset: int = 0,
1476        order_by: str | None = None,
1477        provider_filter: list[str] | None = None,
1478        extra_query_parts: list[str] | None = None,
1479        extra_query_params: dict[str, Any] | None = None,
1480        extra_join_parts: list[str] | None = None,
1481        genre_ids: int | list[int] | None = None,
1482        played_only: bool = False,
1483        in_library_only: bool = False,
1484        summary: bool = False,
1485        *,
1486        collapse_collections: bool = False,
1487        reachable_via: list[str] | None = None,
1488    ) -> list[ItemCls] | list[ItemCls | MediaCollection[ItemCls]]:
1489        """Fetch MediaItem records from database by building the query."""
1490        query_params = dict(extra_query_params) if extra_query_params else {}
1491        query_parts: list[str] = list(extra_query_parts) if extra_query_parts else []
1492        join_parts: list[str] = list(extra_join_parts) if extra_join_parts else []
1493        search = self._preprocess_search(search)
1494        genre_ids = self._preprocess_genre_ids(genre_ids)
1495        # create special performant random query
1496        if order_by and order_by.startswith("random"):
1497            self._apply_random_subquery(
1498                query_parts=query_parts,
1499                query_params=query_params,
1500                join_parts=join_parts,
1501                favorite=favorite,
1502                search=search if not collapse_collections else None,
1503                genre_ids=genre_ids,
1504                provider_filter=provider_filter,
1505                played_only=played_only,
1506                limit=limit,
1507                in_library_only=in_library_only,
1508                reachable_via=reachable_via,
1509            )
1510        else:
1511            # apply filters
1512            self._apply_filters(
1513                query_parts=query_parts,
1514                query_params=query_params,
1515                favorite=favorite,
1516                search=search if not collapse_collections else None,
1517                genre_ids=genre_ids,
1518                provider_filter=provider_filter,
1519                played_only=played_only,
1520                in_library_only=in_library_only,
1521                reachable_via=reachable_via,
1522            )
1523        # build and execute final query
1524        sql_query, base_query_params = self._build_final_query(
1525            query_parts, join_parts, order_by, summary=summary
1526        )
1527        # base query params act as defaults: callers may override them via extra_query_params
1528        for key, value in base_query_params.items():
1529            query_params.setdefault(key, value)
1530
1531        if collapse_collections:
1532            if search:
1533                query_params["search"] = f"%{search}%"
1534            sql_query = await self._adapt_query_for_collections(
1535                sql_query, query_params, summary=summary, order_by=order_by, search=search
1536            )
1537
1538        db_rows = await self.mass.music.database.get_rows_from_query(
1539            sql_query, query_params, limit=limit, offset=offset
1540        )
1541        if collapse_collections:
1542            items: list[ItemCls | MediaCollection[ItemCls]] = []
1543
1544            def _parse_method(x: str) -> ItemCls:
1545                if summary:
1546                    return cast("ItemCls", self._parse_summary_row(json_loads(x)))
1547                return cast(
1548                    "ItemCls",
1549                    self.item_cls.from_dict(self._parse_db_row(json_loads(x))),
1550                )
1551
1552            for db_row in db_rows:
1553                if db_row["type"] == "single":
1554                    items.append(_parse_method(db_row["media_data"]))
1555                elif db_row["type"] == "collection":
1556                    items.append(
1557                        MediaCollection[ItemCls](
1558                            item_id=get_collection_item_id(
1559                                db_row["name"], item_media_type=self.media_type
1560                            ),
1561                            name=db_row["name"],
1562                            provider="library",
1563                            provider_mappings=set(),
1564                            items=UniqueList(
1565                                [_parse_method(x) for x in json_loads(db_row["media_data"])]
1566                            ),
1567                        )
1568                    )
1569            return items
1570        if summary:
1571            return [cast("ItemCls", self._parse_summary_row(db_row)) for db_row in db_rows]
1572        return [
1573            cast("ItemCls", self.item_cls.from_dict(self._parse_db_row(db_row)))
1574            for db_row in db_rows
1575        ]
1576
1577    @final
1578    async def _get_library_item_by_match(self, item: ItemCls | ItemMapping) -> int | None:
1579        if item.provider == "library":
1580            return int(item.item_id)
1581        # search by provider mappings if item is ItemMapping
1582        if isinstance(item, ItemMapping):
1583            if cur_item := await self.get_library_item_by_prov_id(item.item_id, item.provider):
1584                return int(cur_item.item_id)
1585
1586        # for all other items that are MediaItemType, check provider_mappings if it exists
1587        provider_mappings = getattr(item, "provider_mappings", None)
1588        if provider_mappings:
1589            if cur_item := await self.get_library_item_by_prov_mappings(provider_mappings):
1590                return int(cur_item.item_id)
1591        # fetch candidates per external id (best identifier first) and stop at the
1592        # first verified match; external identifiers may be reused, so verify
1593        # every candidate before accepting it
1594        seen_item_ids: set[str] = set()
1595        for external_id_type, external_id in sorted(item.external_ids, key=external_id_sort_key):
1596            for cur_item in await self.get_library_items_by_external_id(
1597                external_id, external_id_type, limit=None
1598            ):
1599                if cur_item.item_id in seen_item_ids:
1600                    continue
1601                seen_item_ids.add(cur_item.item_id)
1602                if await self._confirm_library_candidate(cur_item, item):
1603                    return int(cur_item.item_id)
1604        # search by normalized exact name match
1605        query = (
1606            f"{self.db_table}.search_name IN :search_names "
1607            f"OR {self.db_table}.search_sort_name = :search_sort_name"
1608        )
1609        query_params = {
1610            "search_names": self._library_match_names(item),
1611            "search_sort_name": create_safe_string(item.sort_name or "", True, True),
1612        }
1613        for db_item in await self.get_library_items_by_query(
1614            extra_query_parts=[query], extra_query_params=query_params
1615        ):
1616            if await self._confirm_library_candidate(db_item, item):
1617                return int(db_item.item_id)
1618        return None
1619
1620    def _library_match_names(self, item: ItemCls | ItemMapping) -> list[str]:
1621        """
1622        Return the normalized names a library row for this item may be stored under.
1623
1624        Override in a subclass when a media type's title carries formatting that the
1625        stored name keeps but its identity comparison ignores.
1626        """
1627        return [create_safe_string(item.name, True, True)]
1628
1629    async def _confirm_library_candidate(
1630        self, db_item: ItemCls, item: ItemCls | ItemMapping
1631    ) -> bool:
1632        """
1633        Return True if a library candidate is the same item as the one being added.
1634
1635        Override in a subclass to confirm a candidate that the items' own metadata
1636        cannot decide on with additional evidence.
1637
1638        :param db_item: Existing library item that matched on an external id or name.
1639        :param item: The (provider) item that is being added to the library.
1640        """
1641        return bool(compare_media_item(db_item, item, True))
1642
1643    def _external_ids_query(
1644        self, media_type: MediaType | None = None, table_alias: str | None = None
1645    ) -> str:
1646        """
1647        Return a subquery that selects the external ids of a media item as a JSON array.
1648
1649        :param media_type: Media type to select the external ids for, defaults to
1650            this controller's media type.
1651        :param table_alias: (Aliased) table name the subquery correlates against,
1652            defaults to this controller's table.
1653        """
1654        media_type = media_type or self.media_type
1655        table_alias = table_alias or self.db_table
1656        return (
1657            f"(SELECT JSON_GROUP_ARRAY(json_array("
1658            f"{DB_TABLE_EXTERNAL_ID_LOOKUP}.external_id_type, "
1659            f"{DB_TABLE_EXTERNAL_ID_LOOKUP}.external_id)) "
1660            f"FROM {DB_TABLE_EXTERNAL_ID_LOOKUP} "
1661            f"WHERE {DB_TABLE_EXTERNAL_ID_LOOKUP}.media_type = '{media_type.value}' "
1662            f"AND {DB_TABLE_EXTERNAL_ID_LOOKUP}.item_id = {table_alias}.item_id)"
1663        )
1664
1665    def _provider_mappings_query(self) -> str:
1666        """Return a subquery that selects the provider mappings of a media item as a JSON array."""
1667        return f"""(SELECT JSON_GROUP_ARRAY(
1668            json_object(
1669                'item_id', pm.provider_item_id,
1670                'provider_domain', pm.provider_domain,
1671                'provider_instance', pm.provider_instance,
1672                'available', pm.available,
1673                'audio_format', json(pm.audio_format),
1674                'url', pm.url,
1675                'details', pm.details,
1676                'in_library', pm.in_library,
1677                'is_unique', pm.is_unique
1678            )) FROM {DB_TABLE_PROVIDER_MAPPINGS} pm
1679            WHERE pm.item_id = {self.db_table}.item_id
1680            AND pm.media_type = '{self.media_type.value}')"""
1681
1682    def _artist_mappings_summary_query(
1683        self, m2m_table: str, m2m_key: str, include_artist_type: bool = False
1684    ) -> str:
1685        """
1686        Return a subquery selecting the slim artist mappings JSON of a summary row.
1687
1688        :param m2m_table: The many-to-many table linking artists to this media type.
1689        :param m2m_key: The column in the m2m table referencing this media type's item id.
1690        :param include_artist_type: Also select the artist_type of each artist.
1691        """
1692        artist_type_part = ",\n                'artist_type', artists.artist_type"
1693        return f"""(SELECT JSON_GROUP_ARRAY(
1694            json_object(
1695                'item_id', artists.item_id,
1696                'name', artists.name,
1697                'sort_name', artists.sort_name{artist_type_part if include_artist_type else ""}
1698            )) FROM artists
1699            JOIN {m2m_table} ON artists.item_id = {m2m_table}.artist_id
1700            WHERE {m2m_table}.{m2m_key} = {self.db_table}.item_id)"""
1701
1702    def _summary_base_columns(self) -> str:
1703        """Return the SELECT columns shared by every summary query."""
1704        # the search/sort/statistics columns are selected so ORDER BY (see sort_keys)
1705        # resolves them from the result set, like the full query's SELECT * does
1706        return f"""
1707            {self.db_table}.item_id,
1708            {self.db_table}.name,
1709            {self.db_table}.sort_name,
1710            {self.db_table}.favorite,
1711            {self.db_table}.search_name AS search_name,
1712            {self.db_table}.search_sort_name AS search_sort_name,
1713            {self.db_table}.play_count AS play_count,
1714            {self.db_table}.last_played AS last_played,
1715            {self.db_table}.timestamp_added AS timestamp_added,
1716            {self.db_table}.timestamp_modified AS timestamp_modified,
1717            json_extract({self.db_table}.metadata, '$.images') AS images,
1718            json_extract({self.db_table}.metadata, '$.collections') AS collections"""
1719
1720    async def _localized_search_fallback(
1721        self, search_query: str, limit: int, offset: int = 0, **call_kwargs: Any
1722    ) -> list[ItemCls]:
1723        """
1724        Retry a library search using the canonical names behind a localized query.
1725
1726        For genre/playlist searches that return nothing literally, reverse-resolve the query to the
1727        canonical (English) names of matching localized items and search those, so an item is
1728        findable by the localized name the user sees. The caller's other filters (favorite,
1729        order_by, provider and any controller-specific kwargs) are forwarded unchanged so the retry
1730        behaves like the literal search; results are merged, de-duplicated and paginated here. See
1731        ``TranslationController.reverse_lookup_media_names``.
1732        """
1733        seen: set[Any] = set()
1734        merged: list[ItemCls] = []
1735        # iterate the canonical names in a stable order, and fetch each from the start so the
1736        # offset/limit window can be applied to the merged, de-duplicated result set
1737        for name in sorted(await self.mass.translations.reverse_lookup_media_names(search_query)):
1738            for item in await self.library_items(
1739                search=name,
1740                limit=limit + offset,
1741                offset=0,
1742                _localized_fallback=False,
1743                **call_kwargs,
1744            ):
1745                if item.item_id not in seen:
1746                    seen.add(item.item_id)
1747                    merged.append(item)
1748        return merged[offset : offset + limit]
1749
1750    @abstractmethod
1751    async def _add_library_item(
1752        self,
1753        item: ItemCls,
1754        overwrite_existing: bool = False,
1755    ) -> int:
1756        """Add item to library and return the database id."""
1757
1758    @abstractmethod
1759    async def _update_library_item(
1760        self, item_id: str | int, update: ItemCls, overwrite: bool = False
1761    ) -> None:
1762        """Update existing library record in the database."""
1763
1764    def _search_filter_clause(self, search: str, query_params: dict[str, Any]) -> str:
1765        """Return the SQL WHERE clause fragment used for search filtering."""
1766        return search_name_match_clause(self.db_table, search, "search", query_params)
1767
1768    @final
1769    def _preprocess_search(self, search: str | None) -> str | None:
1770        """Normalize the search string for use in the search filter clauses."""
1771        return create_safe_string(search, True, True) if search else search
1772
1773    @final
1774    @staticmethod
1775    def _preprocess_genre_ids(genre_ids: int | list[int] | None) -> list[int] | None:
1776        if genre_ids is None:
1777            return None
1778        if isinstance(genre_ids, list):
1779            normalized = [int(x) for x in genre_ids]
1780        else:
1781            normalized = [int(genre_ids)]
1782        return normalized or None
1783
1784    @final
1785    @staticmethod
1786    def _clean_query_parts(query_parts: list[str]) -> list[str]:
1787        """Clean the query parts list by removing duplicate where statements."""
1788        return [x[5:] if x.lower().startswith("where ") else x for x in query_parts]
1789
1790    @final
1791    def _apply_random_subquery(  # noqa: PLR0913
1792        self,
1793        query_parts: list[str],
1794        query_params: dict[str, Any],
1795        join_parts: list[str],
1796        favorite: bool | None,
1797        search: str | None,
1798        genre_ids: list[int] | None,
1799        provider_filter: list[str] | None,
1800        played_only: bool = False,
1801        limit: int = 500,
1802        in_library_only: bool = False,
1803        reachable_via: list[str] | None = None,
1804    ) -> None:
1805        """Build a fast random subquery with all filters applied."""
1806        sub_query_parts = query_parts.copy()
1807        sub_join_parts = join_parts.copy()
1808
1809        # Apply all filters to the subquery
1810        self._apply_filters(
1811            query_parts=sub_query_parts,
1812            query_params=query_params,
1813            favorite=favorite,
1814            search=search,
1815            genre_ids=genre_ids,
1816            provider_filter=provider_filter,
1817            played_only=played_only,
1818            in_library_only=in_library_only,
1819            reachable_via=reachable_via,
1820        )
1821
1822        # Build the subquery
1823        sub_query = f"SELECT {self.db_table}.item_id FROM {self.db_table}"
1824
1825        if sub_join_parts:
1826            sub_query += f" {' '.join(sub_join_parts)}"
1827
1828        if sub_query_parts:
1829            sub_query += " WHERE " + " AND ".join(self._clean_query_parts(sub_query_parts))
1830
1831        sub_query += f" ORDER BY RANDOM() LIMIT {limit}"
1832
1833        # The query now only consists of the random subquery, which applies all filters
1834        # within itself
1835        query_parts.clear()
1836        query_parts.append(f"{self.db_table}.item_id in ({sub_query})")
1837        join_parts.clear()
1838
1839    @final
1840    def _apply_filters(
1841        self,
1842        query_parts: list[str],
1843        query_params: dict[str, Any],
1844        favorite: bool | None,
1845        search: str | None,
1846        genre_ids: list[int] | None,
1847        provider_filter: list[str] | None,
1848        played_only: bool = False,
1849        in_library_only: bool = False,
1850        reachable_via: list[str] | None = None,
1851    ) -> None:
1852        """Apply search, favorite, and provider filters."""
1853        # handle search
1854        if search:
1855            query_parts.append(self._search_filter_clause(search, query_params))
1856        # handle favorite filter
1857        if favorite is not None:
1858            query_parts.append(f"{self.db_table}.favorite = :favorite")
1859            query_params["favorite"] = favorite
1860        # handle played_only filter
1861        if played_only:
1862            query_parts.append(f"{self.db_table}.last_played > 0")
1863        # handle genre filter
1864        if genre_ids:
1865            query_params["genre_ids"] = genre_ids
1866            query_params["genre_media_type"] = self.media_type.value
1867            query_parts.append(
1868                f"EXISTS("
1869                f"SELECT 1 FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING} gm "
1870                f"WHERE gm.media_id = {self.db_table}.item_id "
1871                "AND gm.media_type = :genre_media_type "
1872                "AND gm.genre_id IN :genre_ids)"
1873            )
1874        # Apply the provider filter
1875        if provider_filter or in_library_only:
1876            query_parts.append(
1877                self._provider_filter_clause(query_params, provider_filter, in_library_only)
1878            )
1879        # Apply the reachability filter, independent of the (in-library) provider filter above
1880        if reachable_via is not None:
1881            query_parts.append(self._reachability_filter_clause(query_params, reachable_via))
1882
1883    @final
1884    def _reachability_filter_clause(
1885        self, query_params: dict[str, Any], reachable_via: list[str]
1886    ) -> str:
1887        """
1888        Return the SQL clause that restricts items to those reachable via given providers.
1889
1890        Unlike `_provider_filter_clause`, this only checks that an available mapping to
1891        one of the given provider instances exists: it does not require that mapping to
1892        be in that provider's own library. This is used to answer "can this (already
1893        in-library) item be played through one of these providers", as opposed to
1894        "is this item favorited on one of these providers".
1895
1896        :param query_params: Query params dict; the clause's bound params are added to it.
1897        :param reachable_via: Only match items with an available mapping to one of these
1898            provider instances.
1899        """
1900        query_params["reachable_via_media_type"] = self.media_type.value
1901        query_params["reachable_via_providers"] = reachable_via
1902        return (
1903            f"EXISTS(SELECT 1 FROM {DB_TABLE_PROVIDER_MAPPINGS} reachable_mappings "
1904            f"WHERE reachable_mappings.item_id = {self.db_table}.item_id "
1905            "AND reachable_mappings.media_type = :reachable_via_media_type "
1906            "AND reachable_mappings.available = 1 "
1907            "AND reachable_mappings.provider_instance IN :reachable_via_providers)"
1908        )
1909
1910    @final
1911    def _provider_filter_clause(
1912        self,
1913        query_params: dict[str, Any],
1914        provider_filter: list[str] | None,
1915        in_library_only: bool = False,
1916    ) -> str:
1917        """
1918        Return the SQL clause that restricts items by their provider mappings.
1919
1920        At least one of provider_filter/in_library_only must be set, otherwise the
1921        returned clause only asserts that the item has any mapping at all.
1922
1923        :param query_params: Query params dict; the clause's bound params are added to it.
1924        :param provider_filter: Only match items mapped to one of these provider instances.
1925        :param in_library_only: Only match provider mappings that are in the provider's library.
1926        """
1927        # NOTE: provider mapping filters are applied as a correlated EXISTS subquery
1928        # instead of a JOIN + GROUP BY, so SQLite can stream results straight from the
1929        # sort index instead of materializing/sorting the whole (deduped) result set.
1930        query_params["provider_media_type"] = self.media_type.value
1931        conditions = [
1932            f"provider_mappings.item_id = {self.db_table}.item_id",
1933            "provider_mappings.media_type = :provider_media_type",
1934        ]
1935        if in_library_only:
1936            conditions.append("provider_mappings.in_library = 1")
1937        if provider_filter:
1938            provider_conditions = []
1939            for idx, prov in enumerate(provider_filter):
1940                param_name = f"provider_filter_{idx}"
1941                provider_conditions.append(f"provider_mappings.provider_instance = :{param_name}")
1942                query_params[param_name] = prov
1943            conditions.append(f"({' OR '.join(provider_conditions)})")
1944        return f"EXISTS(SELECT 1 FROM provider_mappings WHERE {' AND '.join(conditions)})"
1945
1946    @final
1947    def _build_final_query(
1948        self,
1949        query_parts: list[str],
1950        join_parts: list[str],
1951        order_by: str | None,
1952        summary: bool = False,
1953    ) -> tuple[str, dict[str, Any]]:
1954        """Build the final SQL query string and its (base) bound query params."""
1955        sql_query, base_query_params = self.summary_query if summary else self.base_query
1956
1957        # Add joins
1958        if join_parts:
1959            sql_query += f" {' '.join(join_parts)} "
1960
1961        # Add where clauses
1962        if query_parts:
1963            # prevent duplicate where statement
1964            sql_query += " WHERE " + " AND ".join(self._clean_query_parts(query_parts))
1965
1966        # Add grouping (only needed when caller-provided joins can fan out rows)
1967        # and ordering. Without a GROUP BY, SQLite can stream results directly
1968        # from the sort index instead of sorting the whole result set.
1969        if join_parts:
1970            sql_query += f" GROUP BY {self.db_table}.item_id"
1971
1972        if order_by:
1973            if sort_key := SORT_KEYS.get(order_by):
1974                sql_query += f" ORDER BY {sort_key}"
1975
1976        return sql_query, base_query_params
1977
1978    @final
1979    @staticmethod
1980    def _parse_db_row(db_row: Mapping[str, Any]) -> dict[str, Any]:
1981        """Parse raw db Mapping into a dict."""
1982        db_row_dict = dict(db_row)
1983        db_row_dict["provider"] = "library"
1984        db_row_dict["favorite"] = bool(db_row_dict["favorite"])
1985        db_row_dict["item_id"] = str(db_row_dict["item_id"])
1986        db_row_dict["date_added"] = datetime.fromtimestamp(
1987            db_row_dict["timestamp_added"], tz=UTC
1988        ).isoformat()
1989
1990        for key in JSON_KEYS:
1991            if key not in db_row_dict:
1992                continue
1993            if not (raw_value := db_row_dict[key]):
1994                continue
1995            db_row_dict[key] = json_loads(raw_value)
1996
1997        # parse "fully_played" as bool if present in the row
1998        if "fully_played" in db_row_dict:
1999            db_row_dict["fully_played"] = parse_optional_bool(db_row_dict["fully_played"])
2000
2001        # copy track_album --> album
2002        if track_album := db_row_dict.get("track_album"):
2003            db_row_dict["album"] = track_album
2004            db_row_dict["disc_number"] = track_album["disc_number"]
2005            db_row_dict["track_number"] = track_album["track_number"]
2006            # always prefer album image over track image
2007            if (album_images := track_album.get("images")) and (
2008                album_thumb := next((x for x in album_images if x["type"] == "thumb"), None)
2009            ):
2010                # copy album image to itemmapping single image (on the track)
2011                db_row_dict["image"] = album_thumb
2012                # also set image on the album dict for ItemMapping compatibility
2013                track_album["image"] = album_thumb
2014                if db_row_dict["metadata"].get("images"):
2015                    # merge album image with existing images
2016                    db_row_dict["metadata"]["images"] = [
2017                        album_thumb,
2018                        *db_row_dict["metadata"]["images"],
2019                    ]
2020                else:
2021                    db_row_dict["metadata"]["images"] = [album_thumb]
2022
2023        if audiobook_artists := db_row_dict.get("audiobook_artists"):
2024            _narrators = []
2025            _authors = []
2026            for artist in audiobook_artists:
2027                artist_type = artist.get("artist_type")
2028                if artist_type == "author":
2029                    _authors.append(artist)
2030                elif artist_type == "narrator":
2031                    _narrators.append(artist)
2032            if _authors:
2033                # prevent overwriting string values
2034                db_row_dict["authors"] = _authors
2035            if _narrators:
2036                # prevent overwriting string values
2037                db_row_dict["narrators"] = _narrators
2038
2039        return db_row_dict
2040
2041    @final
2042    def _ensure_provider_filter(
2043        self,
2044        provider: str | list[str] | None,
2045    ) -> list[str] | None:
2046        """Ensure the provider filter respects the current user's provider filter."""
2047        # Apply user provider filter if needed
2048        user = get_current_user()
2049        user_provider_filter = user.provider_filter if user and user.provider_filter else None
2050        final_provider_filter: list[str] | None = None
2051        if user_provider_filter:
2052            plugin_provider_instances = {
2053                prov.instance_id for prov in self.mass.providers if prov.type == ProviderType.PLUGIN
2054            }
2055            # User has a provider filter set
2056            if provider:
2057                # Explicit provider filter provided - validate against user's allowed providers
2058                requested_providers = [provider] if isinstance(provider, str) else provider
2059                # Only restrict access to music providers.
2060                final_provider_filter = [
2061                    p
2062                    for p in requested_providers
2063                    if p in user_provider_filter or p in plugin_provider_instances
2064                ]
2065                if not final_provider_filter:
2066                    # No overlap - user requested providers they don't have access to
2067                    raise InsufficientPermissions(
2068                        "User does not have permission to access the requested provider(s)."
2069                    )
2070            else:
2071                # No explicit filter - apply user music provider filter but keep plugin providers.
2072                final_provider_filter = list(
2073                    dict.fromkeys([*user_provider_filter, *plugin_provider_instances])
2074                )
2075        elif provider is not None:
2076            # No user filter - use the provided filter as is
2077            final_provider_filter = [provider] if isinstance(provider, str) else provider
2078        return final_provider_filter
2079
2080    @final
2081    def _resolve_reachable_via(self, reachable_via: list[str] | None) -> list[str] | None:
2082        """
2083        Resolve a `reachable_via` filter against currently loaded, user-allowed providers.
2084
2085        :param reachable_via: Requested provider instance ids, or None for no filter.
2086        :return: None if no filter should be applied. Otherwise, the subset of
2087            `reachable_via` that is currently active and allowed for the current user
2088            (per `MusicController.get_active_provider_instances`). An empty list means
2089            the filter cannot match anything; callers must then return no items rather
2090            than issue a query.
2091        """
2092        if reachable_via is None:
2093            return None
2094        if not reachable_via:
2095            return []
2096        allowed_providers = set(self.mass.music.get_active_provider_instances())
2097        return [p for p in reachable_via if p in allowed_providers]
2098
2099    @final
2100    def _provider_filter_considering_reachability(
2101        self,
2102        provider: str | list[str] | None,
2103        resolved_reachable_via: list[str] | None,
2104    ) -> list[str] | None:
2105        """
2106        Resolve the `provider` filter, deferring to an active `reachable_via` filter.
2107
2108        The current user's provider access is already enforced on `resolved_reachable_via`
2109        by `_resolve_reachable_via`. So when `reachable_via` is active and no explicit
2110        `provider` filter was requested, skip `_ensure_provider_filter`'s implicit
2111        injection of the user's provider filter: that would additionally require the
2112        item's in-library mapping itself to be on one of those providers, which is
2113        stricter than (and redundant with) what `reachable_via` already checks.
2114
2115        :param provider: The explicit provider filter, as passed to `library_items`.
2116        :param resolved_reachable_via: The already-resolved `reachable_via` filter (the
2117            return value of `_resolve_reachable_via`), or None if not active.
2118        """
2119        if resolved_reachable_via is not None and provider is None:
2120            return None
2121        return self._ensure_provider_filter(provider)
2122
2123    @final
2124    def _select_provider_id(self, library_item: ItemCls) -> tuple[str, str]:
2125        """Select the correct provider id to use for fetching the item."""
2126        if not library_item.provider_mappings:
2127            msg = (
2128                f"{self.media_type.value} {library_item.item_id} "
2129                "is no longer available on any provider"
2130            )
2131            raise MediaNotFoundError(msg)
2132        user = get_current_user()
2133        user_provider_filter = user.provider_filter if user and user.provider_filter else None
2134        if not user_provider_filter:
2135            mapping = next(iter(library_item.provider_mappings))
2136            return (mapping.provider_instance, mapping.item_id)
2137
2138        # First prefer music provider mappings that are explicitly allowed for this user.
2139        # prefer user provider filter if available
2140        for mapping in library_item.provider_mappings:
2141            provider = self.mass.get_provider(mapping.provider_instance)
2142            if provider and provider.type == ProviderType.MUSIC:
2143                if mapping.provider_instance in user_provider_filter:
2144                    return (mapping.provider_instance, mapping.item_id)
2145
2146        # If no allowed music mapping exists, fall back to plugin mappings.
2147        for mapping in library_item.provider_mappings:
2148            provider = self.mass.get_provider(mapping.provider_instance)
2149            if provider and provider.type == ProviderType.PLUGIN:
2150                return (mapping.provider_instance, mapping.item_id)
2151
2152        # As a final fallback, preserve previous behavior.
2153        for mapping in library_item.provider_mappings:
2154            if mapping.provider_instance in user_provider_filter:
2155                return (mapping.provider_instance, mapping.item_id)
2156
2157        # fallback to first mapping
2158        mapping = next(iter(library_item.provider_mappings))
2159        return (mapping.provider_instance, mapping.item_id)
2160
2161    async def _remove_provider_images(self, db_id: int, provider_instance_id: str) -> bool:
2162        """
2163        Remove images belonging to a provider from a library item's stored metadata.
2164
2165        :param db_id: The library (database) id of the item.
2166        :param provider_instance_id: The provider instance whose images should be removed.
2167        :return: True if any images were removed and the db record was updated.
2168        """
2169        # read the raw metadata straight from the db (instead of via get_library_item)
2170        # to avoid persisting any images that are only injected at read time (such as
2171        # the album thumb that gets merged into a track's images)
2172        db_row = await self.mass.music.database.get_row(self.db_table, {"item_id": db_id})
2173        if not db_row or not (raw_metadata := db_row["metadata"]):
2174            return False
2175        metadata = MediaItemMetadata.from_dict(json_loads(raw_metadata))
2176        if not metadata.images:
2177            return False
2178        remaining = UniqueList(
2179            img for img in metadata.images if img.provider != provider_instance_id
2180        )
2181        if len(remaining) == len(metadata.images):
2182            # nothing belonged to this provider
2183            return False
2184        metadata.images = remaining or None
2185        await self.mass.music.database.update(
2186            self.db_table,
2187            {"item_id": db_id},
2188            {"metadata": serialize_to_json(metadata)},
2189        )
2190        return True
2191
2192    def _sync_details_query_parts(self) -> tuple[str, str, dict[str, Any]]:
2193        """
2194        Return extra (columns, joins, params) for this media type's sync-details query.
2195
2196        Override in a subclass to select additional lightweight columns needed by the
2197        library sync change detection for this media type.
2198        """
2199        return "", "", {}
2200
2201    def _parse_sync_details_row(self, db_row: Mapping[str, Any]) -> LibraryItemSyncDetails:
2202        """Parse a raw sync-details db row into a LibraryItemSyncDetails object."""
2203        return LibraryItemSyncDetails(
2204            item_id=db_row["item_id"],
2205            favorite=bool(db_row["favorite"]),
2206            date_added=datetime.fromtimestamp(db_row["timestamp_added"], tz=UTC),
2207            provider_mappings=self._parse_sync_details_mappings(db_row),
2208        )
2209
2210    @final
2211    def _parse_sync_details_mappings(self, db_row: Mapping[str, Any]) -> set[ProviderMapping]:
2212        """Parse the aggregated raw provider mapping rows of a sync-details db row."""
2213        return {
2214            ProviderMapping(
2215                item_id=raw_mapping["item_id"],
2216                provider_domain=raw_mapping["provider_domain"],
2217                provider_instance=raw_mapping["provider_instance"],
2218                available=bool(raw_mapping["available"]),
2219                in_library=parse_optional_bool(raw_mapping["in_library"]),
2220                is_unique=parse_optional_bool(raw_mapping["is_unique"]),
2221            )
2222            for raw_mapping in json_loads(db_row["provider_mappings"])
2223        }
2224
2225    def _parse_summary_row(self, db_row: Mapping[str, Any]) -> MediaItemSummaryType:
2226        """
2227        Parse a raw summary db row into a summary item of this controller's media type.
2228
2229        Override in a subclass to fill additional per-type fields (selected by the
2230        subclass's summary_query).
2231        """
2232        provider_mappings = self._parse_summary_provider_mappings(db_row)
2233        return self.summary_item_cls(
2234            item_id=str(db_row["item_id"]),
2235            provider="library",
2236            name=db_row["name"],
2237            sort_name=db_row["sort_name"],
2238            favorite=bool(db_row["favorite"]),
2239            provider_mappings=provider_mappings,
2240            available=self._summary_available(provider_mappings),
2241            metadata=self._parse_summary_metadata(db_row),
2242        )
2243
2244    @final
2245    @staticmethod
2246    def _parse_summary_provider_mappings(db_row: Mapping[str, Any]) -> set[ProviderMapping]:
2247        """Hydrate the provider mappings of a summary row into ProviderMapping objects."""
2248        if not (raw_mappings := db_row["provider_mappings"]):
2249            return set()
2250        return {ProviderMapping.from_dict(x) for x in json_loads(raw_mappings)}
2251
2252    @final
2253    @staticmethod
2254    def _summary_available(provider_mappings: set[ProviderMapping]) -> bool:
2255        """Compute the availability flag from a summary item's provider mappings."""
2256        # same semantics as the MediaItem.available property
2257        if not (available_providers := get_global_cache_value("available_providers")):
2258            return any(x.available for x in provider_mappings)
2259        if TYPE_CHECKING:
2260            available_providers = cast("set[str]", available_providers)
2261        return any(
2262            x.available and x.provider_instance in available_providers for x in provider_mappings
2263        )
2264
2265    @final
2266    @staticmethod
2267    def _parse_summary_metadata(db_row: Mapping[str, Any]) -> MediaItemMetadataSummary:
2268        """Build the slim metadata of a summary row, carrying only the (first) thumb image."""
2269        thumb: MediaItemImage | None = None
2270        if raw_images := db_row["images"]:
2271            for image in json_loads(raw_images):
2272                if image["type"] != ImageType.THUMB.value:
2273                    continue
2274                thumb = MediaItemImage(
2275                    type=ImageType.THUMB,
2276                    path=image["path"],
2277                    provider=image["provider"],
2278                    remotely_accessible=image.get("remotely_accessible", False),
2279                )
2280                break
2281        return MediaItemMetadataSummary(images=UniqueList([thumb]) if thumb else None)
2282
2283    @final
2284    def _parse_summary_artist_mappings(
2285        self, db_row: Mapping[str, Any]
2286    ) -> UniqueList[ItemMappingSummary]:
2287        """Parse the aggregated slim artist mapping rows of a summary db row."""
2288        return UniqueList(
2289            ItemMappingSummary(
2290                media_type=MediaType.ARTIST,
2291                item_id=str(raw_mapping["item_id"]),
2292                provider="library",
2293                name=raw_mapping["name"],
2294                sort_name=raw_mapping["sort_name"],
2295            )
2296            for raw_mapping in json_loads(db_row["artists"])
2297        )
2298
2299    async def _adapt_query_for_collections(
2300        self,
2301        sql_query: str,
2302        query_params: dict[str, Any],
2303        summary: bool,
2304        order_by: str | None,
2305        collection_name: str | None = None,
2306        search: str | None = None,
2307    ) -> str:
2308        cache_key_json_object = f"collection_{self.api_base}"
2309        json_object = await self.mass.cache.get(key=cache_key_json_object, category=int(summary))
2310        if json_object is None:
2311            # get column names of base query
2312            db_rows = await self.mass.music.database.get_rows_from_query(
2313                sql_query, query_params, limit=1, offset=0
2314            )
2315            # create a sql json_object which queries all these columns
2316            if db_rows:
2317                json_object = (
2318                    "json_object(" + ",".join([f"'{x}',{x}" for x in db_rows[0].keys()]) + ")"  # noqa: SIM118
2319                )
2320                await self.mass.cache.set(
2321                    key=cache_key_json_object, category=int(summary), data=json_object
2322                )
2323            else:
2324                json_object = "json_object()"
2325
2326        collections_column = "collections" if summary else "json_extract(metadata, '$.collections')"
2327
2328        supported_order_keys = [
2329            "name",
2330            "name_desc",
2331            "sort_name",
2332            "sort_name_desc",
2333            "timestamp_added",
2334            "timestamp_added_desc",
2335            "timestamp_modified",
2336            "timestamp_modified_desc",
2337            "last_played",
2338            "last_played_desc",
2339            "play_count",
2340            "play_count_desc",
2341        ]
2342
2343        # additional order options subject to media type
2344        # single is targeting a single media item, collection the aggregated ones
2345        single_extra_order_keys = ""
2346        collection_extra_order_keys = ""
2347        if MediaType.AUDIOBOOK.value in self.api_base:
2348            single_extra_order_keys = "duration,"
2349            collection_extra_order_keys = "SUM(duration) as duration,"
2350            supported_order_keys += ["duration", "duration_desc"]
2351
2352        sql_query = f"""
2353        SELECT * FROM (
2354
2355            WITH
2356                joined_table as ({sql_query}),
2357                collection_extract as (
2358                    SELECT
2359                        name as media_name,
2360                        timestamp_added,
2361                        timestamp_modified,
2362                        last_played,
2363                        play_count,
2364                        {single_extra_order_keys}
2365                        json_extract(iter_coll.value, '$.title') as collection_title,
2366                        json_extract(iter_coll.value, '$.sequence') as collection_sequence,
2367                        json_extract(iter_coll.value, '$.search_title') as collection_search_title,
2368                        json_extract(iter_coll.value, '$.search_sort_title') as collection_search_sort_title,
2369                        CASE
2370                            WHEN json_type(iter_coll.value, '$.sequence') IN ('integer', 'real')
2371                            THEN 1
2372                            WHEN json_type(iter_coll.value, '$.sequence') = 'text'
2373                                AND json_valid(json_extract(iter_coll.value, '$.sequence'))
2374                            THEN CASE
2375                                WHEN json_type(json_extract(iter_coll.value, '$.sequence'))
2376                                    IN ('integer', 'real')
2377                                THEN 1
2378                                ELSE 0
2379                            END
2380                            ELSE 0
2381                        END as collection_sequence_is_numeric,
2382                        {json_object} as media_data
2383                    FROM (
2384                        SELECT * FROM joined_table
2385                    ), json_each({collections_column}) as iter_coll
2386                )
2387            SELECT
2388                'collection' as type,
2389                collection_title as name,
2390                COALESCE(MAX(collection_search_title), replace(lower(collection_title),' ','')) AS search_name,
2391                COALESCE(MAX(collection_search_sort_title), replace(lower(collection_title),' ','')) AS search_sort_name,
2392                MAX(timestamp_added) as timestamp_added,
2393                MAX(timestamp_modified) as timestamp_modified,
2394                MAX(last_played) as last_played,
2395                SUM(play_count) as play_count,
2396                {collection_extra_order_keys}
2397                json_group_array(media_data) as media_data
2398            FROM (
2399                SELECT * FROM collection_extract
2400                -- NOTE: The following ORDER_BY to control the aggregation order of json_group_array is undocumented sqlite behavior
2401                -- Confirmed working with sqlite 3.40.1 & 3.53
2402                -- Once our image moves to sqlite 3.44 we can and should make use of ORDER_BY in the aggregate itself
2403                ORDER BY collection_title,
2404                -- null case
2405                CASE WHEN collection_sequence IS NULL THEN 1 ELSE 0 END,
2406                -- numeric before text
2407                CASE WHEN collection_sequence_is_numeric THEN 0 ELSE 1 END,
2408                -- order NUMERIC
2409                CASE WHEN collection_sequence_is_numeric
2410                    THEN CAST(collection_sequence AS REAL)
2411                END,
2412                -- order TEXT
2413                CASE WHEN NOT collection_sequence_is_numeric
2414                    THEN collection_sequence
2415                END COLLATE NOCASE,
2416                -- order by media name if no sequence given
2417                CASE
2418                    WHEN collection_sequence IS NULL
2419                    THEN media_name
2420                END COLLATE NOCASE
2421            )
2422            GROUP BY collection_title
2423
2424            UNION ALL
2425
2426            SELECT 'single', name, search_name, search_sort_name,
2427                timestamp_added, timestamp_modified, last_played, play_count,
2428                {single_extra_order_keys}
2429                {json_object} FROM joined_table
2430                WHERE {collections_column} IS NULL
2431                    OR {collections_column} = '[]'
2432        )
2433        """
2434
2435        if collection_name:
2436            sql_query += " WHERE type = 'collection' AND name = :collection_name"
2437            return sql_query
2438
2439        if search:
2440            sql_query += " WHERE search_name LIKE :search"
2441
2442        if order_by:
2443            if order_by not in supported_order_keys:
2444                self.logger.warning("%s is not supported for order_by key in collections", order_by)
2445                order_by = "name"  # fallback
2446            if sort_key := SORT_KEYS.get(order_by):
2447                sql_query += f" ORDER BY {sort_key}"
2448
2449        return sql_query
2450
2451    async def _merge_library_items(self, target_id: int, source_id: int) -> tuple[ItemCls, ItemCls]:
2452        """Merge the source library item into the target while the controller lock is held."""
2453        target_item = await self.get_library_item(target_id)
2454        source_item = await self.get_library_item(source_id)
2455        await self._validate_library_item_merge(target_item, source_item)
2456        target_row = await self.mass.music.database.get_row(self.db_table, {"item_id": target_id})
2457        source_row = await self.mass.music.database.get_row(self.db_table, {"item_id": source_id})
2458        assert target_row is not None
2459        assert source_row is not None
2460        timestamps_added = tuple(
2461            timestamp
2462            for timestamp in (
2463                int(target_row["timestamp_added"] or 0),
2464                int(source_row["timestamp_added"] or 0),
2465            )
2466            if timestamp
2467        )
2468
2469        token = SUPPRESS_MEDIA_ITEM_UPDATES.set(True)
2470        try:
2471            source_mappings = source_item.provider_mappings
2472            source_item.provider_mappings = set()
2473            try:
2474                await self._update_library_item_for_merge(target_id, source_item)
2475            finally:
2476                source_item.provider_mappings = source_mappings
2477
2478            await self.mass.music.database.execute_write(
2479                f"""
2480                UPDATE {self.db_table}
2481                SET play_count = CASE item_id
2482                    WHEN :target_id THEN :merged_play_count
2483                    WHEN :source_id THEN 0
2484                END
2485                WHERE item_id IN (:target_id, :source_id)
2486                """,
2487                {
2488                    "target_id": target_id,
2489                    "source_id": source_id,
2490                    "merged_play_count": int(target_row["play_count"] or 0)
2491                    + int(source_row["play_count"] or 0),
2492                },
2493            )
2494            await self.mass.music.database.update(
2495                self.db_table,
2496                {"item_id": target_id},
2497                {
2498                    "favorite": bool(target_row["favorite"]) or bool(source_row["favorite"]),
2499                    "last_played": max(
2500                        int(target_row["last_played"] or 0), int(source_row["last_played"] or 0)
2501                    ),
2502                    "timestamp_added": min(timestamps_added) if timestamps_added else 0,
2503                },
2504            )
2505            await self._merge_genre_mappings(target_id, source_id)
2506            await self._merge_library_item_references(target_id, source_id)
2507            await self._merge_library_playlog(target_id, source_id)
2508            # the transfer commits in steps (see `deferred_commit`), so it is ordered to
2509            # leave the source repairable wherever it is cut short: relations are copied
2510            # rather than moved, and only dropped once the target holds them and the
2511            # provider mappings. A source that kept its relations stays a duplicate the
2512            # reconciliation pass can finish; one that lost its mappings is cleaned up.
2513            await self._copy_library_item_relations(target_id, source_id)
2514            await self.mass.music.database.execute_write(
2515                f"UPDATE {DB_TABLE_PROVIDER_MAPPINGS} SET item_id = :target_id "
2516                "WHERE media_type = :media_type AND item_id = :source_id",
2517                {
2518                    "target_id": target_id,
2519                    "source_id": source_id,
2520                    "media_type": self.media_type.value,
2521                },
2522            )
2523            await self._drop_library_item_relations(source_id)
2524            await MediaControllerBase.remove_item_from_library(self, source_id, recursive=False)
2525            merged_item = await self.get_library_item(target_id)
2526        finally:
2527            SUPPRESS_MEDIA_ITEM_UPDATES.reset(token)
2528
2529        return source_item, merged_item
2530
2531    async def _merge_library_items_batched(self, target_id: int, source_id: int) -> ItemCls:
2532        """Merge library items while batching the transfer's database writes."""
2533        async with self.mass.music.database.deferred_commit():
2534            source_item, merged_item = await self._merge_library_items(target_id, source_id)
2535        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
2536            self.mass.signal_event(EventType.MEDIA_ITEM_DELETED, source_item.uri, source_item)
2537            self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, merged_item.uri, merged_item)
2538        return merged_item
2539
2540    async def _validate_library_item_merge(self, target: ItemCls, source: ItemCls) -> None:
2541        """Validate that the target and source items can be merged."""
2542        if target.media_type != self.media_type or source.media_type != self.media_type:
2543            msg = "Library items must have the controller's media type"
2544            raise InvalidDataError(msg)
2545
2546    async def _update_library_item_for_merge(self, item_id: int, update: ItemCls) -> None:
2547        """Merge model state into an existing library item."""
2548        await self._update_library_item(item_id, update)
2549
2550    async def _copy_library_item_relations(self, target_id: int, source_id: int) -> None:
2551        """Copy the relations that reference the merged media item onto the target."""
2552        for table, item_column in self._library_item_relations():
2553            columns = RELATION_TABLE_COLUMNS[table]
2554            selected = ", ".join(
2555                ":target_id" if column == item_column else column for column in columns
2556            )
2557            await self.mass.music.database.execute_write(
2558                f"INSERT OR IGNORE INTO {table}({', '.join(columns)}) "
2559                f"SELECT {selected} FROM {table} WHERE {item_column} = :source_id",
2560                {"target_id": target_id, "source_id": source_id},
2561            )
2562
2563    async def _drop_library_item_relations(self, source_id: int) -> None:
2564        """Drop the relations of a merged media item once the target holds them."""
2565        for table, item_column in self._library_item_relations():
2566            await self.mass.music.database.delete(table, {item_column: source_id})
2567
2568    def _library_item_relations(self) -> tuple[tuple[str, str], ...]:
2569        """Return the (table, column) pairs holding relations to this controller's items."""
2570        if self.media_type == MediaType.ALBUM:
2571            return (
2572                (DB_TABLE_ALBUM_ARTISTS, "album_id"),
2573                (DB_TABLE_ALBUM_TRACKS, "album_id"),
2574            )
2575        if self.media_type == MediaType.ARTIST:
2576            return (
2577                (DB_TABLE_ALBUM_ARTISTS, "artist_id"),
2578                (DB_TABLE_AUDIOBOOK_ARTISTS, "artist_id"),
2579                (DB_TABLE_TRACK_ARTISTS, "artist_id"),
2580            )
2581        if self.media_type == MediaType.AUDIOBOOK:
2582            return ((DB_TABLE_AUDIOBOOK_ARTISTS, "audiobook_id"),)
2583        if self.media_type == MediaType.TRACK:
2584            return (
2585                (DB_TABLE_ALBUM_TRACKS, "track_id"),
2586                (DB_TABLE_TRACK_ARTISTS, "track_id"),
2587            )
2588        return ()
2589
2590    async def _merge_library_item_references(self, target_id: int, source_id: int) -> None:
2591        """Transfer references to the source item owned by specialized controllers."""
2592        return
2593
2594    async def _merge_genre_mappings(self, target_id: int, source_id: int) -> None:
2595        """Transfer genre mappings and exclusions to the target item."""
2596        values = {
2597            "target_id": target_id,
2598            "source_id": source_id,
2599            "media_type": self.media_type.value,
2600        }
2601        await self.mass.music.database.execute_write(
2602            f"""
2603            INSERT INTO {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING}(
2604                genre_id, media_id, media_type, alias, is_derived, is_manual
2605            )
2606            SELECT genre_id, :target_id, media_type, alias, is_derived, is_manual
2607            FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING}
2608            WHERE media_id = :source_id AND media_type = :media_type
2609            ON CONFLICT(genre_id, media_id, media_type) DO UPDATE SET
2610                alias = CASE
2611                    WHEN excluded.is_manual AND NOT is_manual
2612                    THEN COALESCE(excluded.alias, alias)
2613                    ELSE COALESCE(alias, excluded.alias)
2614                END,
2615                is_derived = is_derived OR excluded.is_derived,
2616                is_manual = is_manual OR excluded.is_manual
2617            """,
2618            values,
2619        )
2620        await self.mass.music.database.execute_write(
2621            f"""
2622            INSERT OR IGNORE INTO {DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION}(
2623                genre_id, media_id, media_type
2624            )
2625            SELECT genre_id, :target_id, media_type
2626            FROM {DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION}
2627            WHERE media_id = :source_id AND media_type = :media_type
2628            """,
2629            values,
2630        )
2631        await self.mass.music.database.delete(
2632            DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
2633            {"media_id": source_id, "media_type": self.media_type.value},
2634        )
2635        await self.mass.music.database.delete(
2636            DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
2637            {"media_id": source_id, "media_type": self.media_type.value},
2638        )
2639
2640    async def _merge_genre_references(self, target_id: int, source_id: int) -> None:
2641        """Transfer media mappings and exclusions that point to the source genre."""
2642        values = {"target_id": target_id, "source_id": source_id}
2643        await self.mass.music.database.execute_write(
2644            f"""
2645            INSERT INTO {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING}(
2646                genre_id, media_id, media_type, alias, is_derived, is_manual
2647            )
2648            SELECT :target_id, media_id, media_type, alias, is_derived, is_manual
2649            FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING}
2650            WHERE genre_id = :source_id
2651            ON CONFLICT(genre_id, media_id, media_type) DO UPDATE SET
2652                alias = CASE
2653                    WHEN excluded.is_manual AND NOT is_manual
2654                    THEN COALESCE(excluded.alias, alias)
2655                    ELSE COALESCE(alias, excluded.alias)
2656                END,
2657                is_derived = is_derived OR excluded.is_derived,
2658                is_manual = is_manual OR excluded.is_manual
2659            """,
2660            values,
2661        )
2662        await self.mass.music.database.execute_write(
2663            f"""
2664            INSERT OR IGNORE INTO {DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION}(
2665                genre_id, media_id, media_type
2666            )
2667            SELECT :target_id, media_id, media_type
2668            FROM {DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION}
2669            WHERE genre_id = :source_id
2670            """,
2671            values,
2672        )
2673        await self.mass.music.database.delete(
2674            DB_TABLE_GENRE_MEDIA_ITEM_MAPPING, {"genre_id": source_id}
2675        )
2676        await self.mass.music.database.delete(
2677            DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION, {"genre_id": source_id}
2678        )
2679
2680    async def _merge_library_playlog(self, target_id: int, source_id: int) -> None:
2681        """Transfer library-keyed playlog rows using the normal latest-entry semantics."""
2682        values = {
2683            "target_id": target_id,
2684            "source_id": source_id,
2685            "media_type": self.media_type.value,
2686        }
2687        await self.mass.music.database.execute_write(
2688            f"""
2689            INSERT INTO {DB_TABLE_PLAYLOG}(
2690                item_id, provider, media_type, name, image, artists, timestamp,
2691                fully_played, seconds_played, userid, queue_id, user_initiated, playback_speed
2692            )
2693            SELECT
2694                :target_id, provider, media_type, name, image, artists, timestamp,
2695                fully_played, seconds_played, userid, queue_id, user_initiated, playback_speed
2696            FROM {DB_TABLE_PLAYLOG}
2697            WHERE item_id = :source_id AND provider = 'library' AND media_type = :media_type
2698            ON CONFLICT(item_id, provider, media_type, userid) DO UPDATE SET
2699                name = CASE WHEN excluded.timestamp > timestamp THEN excluded.name ELSE name END,
2700                image = CASE WHEN excluded.timestamp > timestamp THEN excluded.image ELSE image END,
2701                artists = CASE WHEN excluded.timestamp > timestamp THEN excluded.artists ELSE artists END,
2702                timestamp = MAX(timestamp, excluded.timestamp),
2703                fully_played = CASE
2704                    WHEN excluded.timestamp > timestamp THEN excluded.fully_played ELSE fully_played
2705                END,
2706                seconds_played = CASE
2707                    WHEN excluded.timestamp > timestamp THEN excluded.seconds_played
2708                    ELSE seconds_played
2709                END,
2710                queue_id = CASE
2711                    WHEN excluded.timestamp > timestamp THEN excluded.queue_id ELSE queue_id
2712                END,
2713                user_initiated = user_initiated OR excluded.user_initiated,
2714                playback_speed = CASE
2715                    WHEN excluded.timestamp > timestamp THEN excluded.playback_speed
2716                    ELSE playback_speed
2717                END
2718            """,
2719            values,
2720        )
2721        await self.mass.music.database.delete(
2722            DB_TABLE_PLAYLOG,
2723            {
2724                "item_id": source_id,
2725                "provider": "library",
2726                "media_type": self.media_type.value,
2727            },
2728        )
2729