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