music-assistant-server

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