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