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 compare_media_item(item, cur_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 compare_media_item(db_item, item, True):
1606                return int(db_item.item_id)
1607        return None
1608
1609    def _external_ids_query(
1610        self, media_type: MediaType | None = None, table_alias: str | None = None
1611    ) -> str:
1612        """
1613        Return a subquery that selects the external ids of a media item as a JSON array.
1614
1615        :param media_type: Media type to select the external ids for, defaults to
1616            this controller's media type.
1617        :param table_alias: (Aliased) table name the subquery correlates against,
1618            defaults to this controller's table.
1619        """
1620        media_type = media_type or self.media_type
1621        table_alias = table_alias or self.db_table
1622        return (
1623            f"(SELECT JSON_GROUP_ARRAY(json_array("
1624            f"{DB_TABLE_EXTERNAL_ID_LOOKUP}.external_id_type, "
1625            f"{DB_TABLE_EXTERNAL_ID_LOOKUP}.external_id)) "
1626            f"FROM {DB_TABLE_EXTERNAL_ID_LOOKUP} "
1627            f"WHERE {DB_TABLE_EXTERNAL_ID_LOOKUP}.media_type = '{media_type.value}' "
1628            f"AND {DB_TABLE_EXTERNAL_ID_LOOKUP}.item_id = {table_alias}.item_id)"
1629        )
1630
1631    def _provider_mappings_query(self) -> str:
1632        """Return a subquery that selects the provider mappings of a media item as a JSON array."""
1633        return f"""(SELECT JSON_GROUP_ARRAY(
1634            json_object(
1635                'item_id', pm.provider_item_id,
1636                'provider_domain', pm.provider_domain,
1637                'provider_instance', pm.provider_instance,
1638                'available', pm.available,
1639                'audio_format', json(pm.audio_format),
1640                'url', pm.url,
1641                'details', pm.details,
1642                'in_library', pm.in_library,
1643                'is_unique', pm.is_unique
1644            )) FROM {DB_TABLE_PROVIDER_MAPPINGS} pm
1645            WHERE pm.item_id = {self.db_table}.item_id
1646            AND pm.media_type = '{self.media_type.value}')"""
1647
1648    def _artist_mappings_summary_query(
1649        self, m2m_table: str, m2m_key: str, include_artist_type: bool = False
1650    ) -> str:
1651        """
1652        Return a subquery selecting the slim artist mappings JSON of a summary row.
1653
1654        :param m2m_table: The many-to-many table linking artists to this media type.
1655        :param m2m_key: The column in the m2m table referencing this media type's item id.
1656        :param include_artist_type: Also select the artist_type of each artist.
1657        """
1658        artist_type_part = ",\n                'artist_type', artists.artist_type"
1659        return f"""(SELECT JSON_GROUP_ARRAY(
1660            json_object(
1661                'item_id', artists.item_id,
1662                'name', artists.name,
1663                'sort_name', artists.sort_name{artist_type_part if include_artist_type else ""}
1664            )) FROM artists
1665            JOIN {m2m_table} ON artists.item_id = {m2m_table}.artist_id
1666            WHERE {m2m_table}.{m2m_key} = {self.db_table}.item_id)"""
1667
1668    def _summary_base_columns(self) -> str:
1669        """Return the SELECT columns shared by every summary query."""
1670        # the search/sort/statistics columns are selected so ORDER BY (see sort_keys)
1671        # resolves them from the result set, like the full query's SELECT * does
1672        return f"""
1673            {self.db_table}.item_id,
1674            {self.db_table}.name,
1675            {self.db_table}.sort_name,
1676            {self.db_table}.favorite,
1677            {self.db_table}.search_name AS search_name,
1678            {self.db_table}.search_sort_name AS search_sort_name,
1679            {self.db_table}.play_count AS play_count,
1680            {self.db_table}.last_played AS last_played,
1681            {self.db_table}.timestamp_added AS timestamp_added,
1682            {self.db_table}.timestamp_modified AS timestamp_modified,
1683            json_extract({self.db_table}.metadata, '$.images') AS images,
1684            json_extract({self.db_table}.metadata, '$.collections') AS collections"""
1685
1686    async def _localized_search_fallback(
1687        self, search_query: str, limit: int, offset: int = 0, **call_kwargs: Any
1688    ) -> list[ItemCls]:
1689        """
1690        Retry a library search using the canonical names behind a localized query.
1691
1692        For genre/playlist searches that return nothing literally, reverse-resolve the query to the
1693        canonical (English) names of matching localized items and search those, so an item is
1694        findable by the localized name the user sees. The caller's other filters (favorite,
1695        order_by, provider and any controller-specific kwargs) are forwarded unchanged so the retry
1696        behaves like the literal search; results are merged, de-duplicated and paginated here. See
1697        ``TranslationController.reverse_lookup_media_names``.
1698        """
1699        seen: set[Any] = set()
1700        merged: list[ItemCls] = []
1701        # iterate the canonical names in a stable order, and fetch each from the start so the
1702        # offset/limit window can be applied to the merged, de-duplicated result set
1703        for name in sorted(await self.mass.translations.reverse_lookup_media_names(search_query)):
1704            for item in await self.library_items(
1705                search=name,
1706                limit=limit + offset,
1707                offset=0,
1708                _localized_fallback=False,
1709                **call_kwargs,
1710            ):
1711                if item.item_id not in seen:
1712                    seen.add(item.item_id)
1713                    merged.append(item)
1714        return merged[offset : offset + limit]
1715
1716    @abstractmethod
1717    async def _add_library_item(
1718        self,
1719        item: ItemCls,
1720        overwrite_existing: bool = False,
1721    ) -> int:
1722        """Add item to library and return the database id."""
1723
1724    @abstractmethod
1725    async def _update_library_item(
1726        self, item_id: str | int, update: ItemCls, overwrite: bool = False
1727    ) -> None:
1728        """Update existing library record in the database."""
1729
1730    def _search_filter_clause(self, search: str, query_params: dict[str, Any]) -> str:
1731        """Return the SQL WHERE clause fragment used for search filtering."""
1732        return search_name_match_clause(self.db_table, search, "search", query_params)
1733
1734    @final
1735    def _preprocess_search(self, search: str | None) -> str | None:
1736        """Normalize the search string for use in the search filter clauses."""
1737        return create_safe_string(search, True, True) if search else search
1738
1739    @final
1740    @staticmethod
1741    def _preprocess_genre_ids(genre_ids: int | list[int] | None) -> list[int] | None:
1742        if genre_ids is None:
1743            return None
1744        if isinstance(genre_ids, list):
1745            normalized = [int(x) for x in genre_ids]
1746        else:
1747            normalized = [int(genre_ids)]
1748        return normalized or None
1749
1750    @final
1751    @staticmethod
1752    def _clean_query_parts(query_parts: list[str]) -> list[str]:
1753        """Clean the query parts list by removing duplicate where statements."""
1754        return [x[5:] if x.lower().startswith("where ") else x for x in query_parts]
1755
1756    @final
1757    def _apply_random_subquery(  # noqa: PLR0913
1758        self,
1759        query_parts: list[str],
1760        query_params: dict[str, Any],
1761        join_parts: list[str],
1762        favorite: bool | None,
1763        search: str | None,
1764        genre_ids: list[int] | None,
1765        provider_filter: list[str] | None,
1766        played_only: bool = False,
1767        limit: int = 500,
1768        in_library_only: bool = False,
1769        reachable_via: list[str] | None = None,
1770    ) -> None:
1771        """Build a fast random subquery with all filters applied."""
1772        sub_query_parts = query_parts.copy()
1773        sub_join_parts = join_parts.copy()
1774
1775        # Apply all filters to the subquery
1776        self._apply_filters(
1777            query_parts=sub_query_parts,
1778            query_params=query_params,
1779            favorite=favorite,
1780            search=search,
1781            genre_ids=genre_ids,
1782            provider_filter=provider_filter,
1783            played_only=played_only,
1784            in_library_only=in_library_only,
1785            reachable_via=reachable_via,
1786        )
1787
1788        # Build the subquery
1789        sub_query = f"SELECT {self.db_table}.item_id FROM {self.db_table}"
1790
1791        if sub_join_parts:
1792            sub_query += f" {' '.join(sub_join_parts)}"
1793
1794        if sub_query_parts:
1795            sub_query += " WHERE " + " AND ".join(self._clean_query_parts(sub_query_parts))
1796
1797        sub_query += f" ORDER BY RANDOM() LIMIT {limit}"
1798
1799        # The query now only consists of the random subquery, which applies all filters
1800        # within itself
1801        query_parts.clear()
1802        query_parts.append(f"{self.db_table}.item_id in ({sub_query})")
1803        join_parts.clear()
1804
1805    @final
1806    def _apply_filters(
1807        self,
1808        query_parts: list[str],
1809        query_params: dict[str, Any],
1810        favorite: bool | None,
1811        search: str | None,
1812        genre_ids: list[int] | None,
1813        provider_filter: list[str] | None,
1814        played_only: bool = False,
1815        in_library_only: bool = False,
1816        reachable_via: list[str] | None = None,
1817    ) -> None:
1818        """Apply search, favorite, and provider filters."""
1819        # handle search
1820        if search:
1821            query_parts.append(self._search_filter_clause(search, query_params))
1822        # handle favorite filter
1823        if favorite is not None:
1824            query_parts.append(f"{self.db_table}.favorite = :favorite")
1825            query_params["favorite"] = favorite
1826        # handle played_only filter
1827        if played_only:
1828            query_parts.append(f"{self.db_table}.last_played > 0")
1829        # handle genre filter
1830        if genre_ids:
1831            query_params["genre_ids"] = genre_ids
1832            query_params["genre_media_type"] = self.media_type.value
1833            query_parts.append(
1834                f"EXISTS("
1835                f"SELECT 1 FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING} gm "
1836                f"WHERE gm.media_id = {self.db_table}.item_id "
1837                "AND gm.media_type = :genre_media_type "
1838                "AND gm.genre_id IN :genre_ids)"
1839            )
1840        # Apply the provider filter
1841        if provider_filter or in_library_only:
1842            query_parts.append(
1843                self._provider_filter_clause(query_params, provider_filter, in_library_only)
1844            )
1845        # Apply the reachability filter, independent of the (in-library) provider filter above
1846        if reachable_via is not None:
1847            query_parts.append(self._reachability_filter_clause(query_params, reachable_via))
1848
1849    @final
1850    def _reachability_filter_clause(
1851        self, query_params: dict[str, Any], reachable_via: list[str]
1852    ) -> str:
1853        """
1854        Return the SQL clause that restricts items to those reachable via given providers.
1855
1856        Unlike `_provider_filter_clause`, this only checks that an available mapping to
1857        one of the given provider instances exists: it does not require that mapping to
1858        be in that provider's own library. This is used to answer "can this (already
1859        in-library) item be played through one of these providers", as opposed to
1860        "is this item favorited on one of these providers".
1861
1862        :param query_params: Query params dict; the clause's bound params are added to it.
1863        :param reachable_via: Only match items with an available mapping to one of these
1864            provider instances.
1865        """
1866        query_params["reachable_via_media_type"] = self.media_type.value
1867        query_params["reachable_via_providers"] = reachable_via
1868        return (
1869            f"EXISTS(SELECT 1 FROM {DB_TABLE_PROVIDER_MAPPINGS} reachable_mappings "
1870            f"WHERE reachable_mappings.item_id = {self.db_table}.item_id "
1871            "AND reachable_mappings.media_type = :reachable_via_media_type "
1872            "AND reachable_mappings.available = 1 "
1873            "AND reachable_mappings.provider_instance IN :reachable_via_providers)"
1874        )
1875
1876    @final
1877    def _provider_filter_clause(
1878        self,
1879        query_params: dict[str, Any],
1880        provider_filter: list[str] | None,
1881        in_library_only: bool = False,
1882    ) -> str:
1883        """
1884        Return the SQL clause that restricts items by their provider mappings.
1885
1886        At least one of provider_filter/in_library_only must be set, otherwise the
1887        returned clause only asserts that the item has any mapping at all.
1888
1889        :param query_params: Query params dict; the clause's bound params are added to it.
1890        :param provider_filter: Only match items mapped to one of these provider instances.
1891        :param in_library_only: Only match provider mappings that are in the provider's library.
1892        """
1893        # NOTE: provider mapping filters are applied as a correlated EXISTS subquery
1894        # instead of a JOIN + GROUP BY, so SQLite can stream results straight from the
1895        # sort index instead of materializing/sorting the whole (deduped) result set.
1896        query_params["provider_media_type"] = self.media_type.value
1897        conditions = [
1898            f"provider_mappings.item_id = {self.db_table}.item_id",
1899            "provider_mappings.media_type = :provider_media_type",
1900        ]
1901        if in_library_only:
1902            conditions.append("provider_mappings.in_library = 1")
1903        if provider_filter:
1904            provider_conditions = []
1905            for idx, prov in enumerate(provider_filter):
1906                param_name = f"provider_filter_{idx}"
1907                provider_conditions.append(f"provider_mappings.provider_instance = :{param_name}")
1908                query_params[param_name] = prov
1909            conditions.append(f"({' OR '.join(provider_conditions)})")
1910        return f"EXISTS(SELECT 1 FROM provider_mappings WHERE {' AND '.join(conditions)})"
1911
1912    @final
1913    def _build_final_query(
1914        self,
1915        query_parts: list[str],
1916        join_parts: list[str],
1917        order_by: str | None,
1918        summary: bool = False,
1919    ) -> tuple[str, dict[str, Any]]:
1920        """Build the final SQL query string and its (base) bound query params."""
1921        sql_query, base_query_params = self.summary_query if summary else self.base_query
1922
1923        # Add joins
1924        if join_parts:
1925            sql_query += f" {' '.join(join_parts)} "
1926
1927        # Add where clauses
1928        if query_parts:
1929            # prevent duplicate where statement
1930            sql_query += " WHERE " + " AND ".join(self._clean_query_parts(query_parts))
1931
1932        # Add grouping (only needed when caller-provided joins can fan out rows)
1933        # and ordering. Without a GROUP BY, SQLite can stream results directly
1934        # from the sort index instead of sorting the whole result set.
1935        if join_parts:
1936            sql_query += f" GROUP BY {self.db_table}.item_id"
1937
1938        if order_by:
1939            if sort_key := SORT_KEYS.get(order_by):
1940                sql_query += f" ORDER BY {sort_key}"
1941
1942        return sql_query, base_query_params
1943
1944    @final
1945    @staticmethod
1946    def _parse_db_row(db_row: Mapping[str, Any]) -> dict[str, Any]:
1947        """Parse raw db Mapping into a dict."""
1948        db_row_dict = dict(db_row)
1949        db_row_dict["provider"] = "library"
1950        db_row_dict["favorite"] = bool(db_row_dict["favorite"])
1951        db_row_dict["item_id"] = str(db_row_dict["item_id"])
1952        db_row_dict["date_added"] = datetime.fromtimestamp(
1953            db_row_dict["timestamp_added"], tz=UTC
1954        ).isoformat()
1955
1956        for key in JSON_KEYS:
1957            if key not in db_row_dict:
1958                continue
1959            if not (raw_value := db_row_dict[key]):
1960                continue
1961            db_row_dict[key] = json_loads(raw_value)
1962
1963        # parse "fully_played" as bool if present in the row
1964        if "fully_played" in db_row_dict:
1965            db_row_dict["fully_played"] = parse_optional_bool(db_row_dict["fully_played"])
1966
1967        # copy track_album --> album
1968        if track_album := db_row_dict.get("track_album"):
1969            db_row_dict["album"] = track_album
1970            db_row_dict["disc_number"] = track_album["disc_number"]
1971            db_row_dict["track_number"] = track_album["track_number"]
1972            # always prefer album image over track image
1973            if (album_images := track_album.get("images")) and (
1974                album_thumb := next((x for x in album_images if x["type"] == "thumb"), None)
1975            ):
1976                # copy album image to itemmapping single image (on the track)
1977                db_row_dict["image"] = album_thumb
1978                # also set image on the album dict for ItemMapping compatibility
1979                track_album["image"] = album_thumb
1980                if db_row_dict["metadata"].get("images"):
1981                    # merge album image with existing images
1982                    db_row_dict["metadata"]["images"] = [
1983                        album_thumb,
1984                        *db_row_dict["metadata"]["images"],
1985                    ]
1986                else:
1987                    db_row_dict["metadata"]["images"] = [album_thumb]
1988
1989        if audiobook_artists := db_row_dict.get("audiobook_artists"):
1990            _narrators = []
1991            _authors = []
1992            for artist in audiobook_artists:
1993                artist_type = artist.get("artist_type")
1994                if artist_type == "author":
1995                    _authors.append(artist)
1996                elif artist_type == "narrator":
1997                    _narrators.append(artist)
1998            if _authors:
1999                # prevent overwriting string values
2000                db_row_dict["authors"] = _authors
2001            if _narrators:
2002                # prevent overwriting string values
2003                db_row_dict["narrators"] = _narrators
2004
2005        return db_row_dict
2006
2007    @final
2008    def _ensure_provider_filter(
2009        self,
2010        provider: str | list[str] | None,
2011    ) -> list[str] | None:
2012        """Ensure the provider filter respects the current user's provider filter."""
2013        # Apply user provider filter if needed
2014        user = get_current_user()
2015        user_provider_filter = user.provider_filter if user and user.provider_filter else None
2016        final_provider_filter: list[str] | None = None
2017        if user_provider_filter:
2018            plugin_provider_instances = {
2019                prov.instance_id for prov in self.mass.providers if prov.type == ProviderType.PLUGIN
2020            }
2021            # User has a provider filter set
2022            if provider:
2023                # Explicit provider filter provided - validate against user's allowed providers
2024                requested_providers = [provider] if isinstance(provider, str) else provider
2025                # Only restrict access to music providers.
2026                final_provider_filter = [
2027                    p
2028                    for p in requested_providers
2029                    if p in user_provider_filter or p in plugin_provider_instances
2030                ]
2031                if not final_provider_filter:
2032                    # No overlap - user requested providers they don't have access to
2033                    raise InsufficientPermissions(
2034                        "User does not have permission to access the requested provider(s)."
2035                    )
2036            else:
2037                # No explicit filter - apply user music provider filter but keep plugin providers.
2038                final_provider_filter = list(
2039                    dict.fromkeys([*user_provider_filter, *plugin_provider_instances])
2040                )
2041        elif provider is not None:
2042            # No user filter - use the provided filter as is
2043            final_provider_filter = [provider] if isinstance(provider, str) else provider
2044        return final_provider_filter
2045
2046    @final
2047    def _resolve_reachable_via(self, reachable_via: list[str] | None) -> list[str] | None:
2048        """
2049        Resolve a `reachable_via` filter against currently loaded, user-allowed providers.
2050
2051        :param reachable_via: Requested provider instance ids, or None for no filter.
2052        :return: None if no filter should be applied. Otherwise, the subset of
2053            `reachable_via` that is currently active and allowed for the current user
2054            (per `MusicController.get_active_provider_instances`). An empty list means
2055            the filter cannot match anything; callers must then return no items rather
2056            than issue a query.
2057        """
2058        if reachable_via is None:
2059            return None
2060        if not reachable_via:
2061            return []
2062        allowed_providers = set(self.mass.music.get_active_provider_instances())
2063        return [p for p in reachable_via if p in allowed_providers]
2064
2065    @final
2066    def _provider_filter_considering_reachability(
2067        self,
2068        provider: str | list[str] | None,
2069        resolved_reachable_via: list[str] | None,
2070    ) -> list[str] | None:
2071        """
2072        Resolve the `provider` filter, deferring to an active `reachable_via` filter.
2073
2074        The current user's provider access is already enforced on `resolved_reachable_via`
2075        by `_resolve_reachable_via`. So when `reachable_via` is active and no explicit
2076        `provider` filter was requested, skip `_ensure_provider_filter`'s implicit
2077        injection of the user's provider filter: that would additionally require the
2078        item's in-library mapping itself to be on one of those providers, which is
2079        stricter than (and redundant with) what `reachable_via` already checks.
2080
2081        :param provider: The explicit provider filter, as passed to `library_items`.
2082        :param resolved_reachable_via: The already-resolved `reachable_via` filter (the
2083            return value of `_resolve_reachable_via`), or None if not active.
2084        """
2085        if resolved_reachable_via is not None and provider is None:
2086            return None
2087        return self._ensure_provider_filter(provider)
2088
2089    @final
2090    def _select_provider_id(self, library_item: ItemCls) -> tuple[str, str]:
2091        """Select the correct provider id to use for fetching the item."""
2092        if not library_item.provider_mappings:
2093            msg = (
2094                f"{self.media_type.value} {library_item.item_id} "
2095                "is no longer available on any provider"
2096            )
2097            raise MediaNotFoundError(msg)
2098        user = get_current_user()
2099        user_provider_filter = user.provider_filter if user and user.provider_filter else None
2100        if not user_provider_filter:
2101            mapping = next(iter(library_item.provider_mappings))
2102            return (mapping.provider_instance, mapping.item_id)
2103
2104        # First prefer music provider mappings that are explicitly allowed for this user.
2105        # prefer user provider filter if available
2106        for mapping in library_item.provider_mappings:
2107            provider = self.mass.get_provider(mapping.provider_instance)
2108            if provider and provider.type == ProviderType.MUSIC:
2109                if mapping.provider_instance in user_provider_filter:
2110                    return (mapping.provider_instance, mapping.item_id)
2111
2112        # If no allowed music mapping exists, fall back to plugin mappings.
2113        for mapping in library_item.provider_mappings:
2114            provider = self.mass.get_provider(mapping.provider_instance)
2115            if provider and provider.type == ProviderType.PLUGIN:
2116                return (mapping.provider_instance, mapping.item_id)
2117
2118        # As a final fallback, preserve previous behavior.
2119        for mapping in library_item.provider_mappings:
2120            if mapping.provider_instance in user_provider_filter:
2121                return (mapping.provider_instance, mapping.item_id)
2122
2123        # fallback to first mapping
2124        mapping = next(iter(library_item.provider_mappings))
2125        return (mapping.provider_instance, mapping.item_id)
2126
2127    async def _remove_provider_images(self, db_id: int, provider_instance_id: str) -> bool:
2128        """
2129        Remove images belonging to a provider from a library item's stored metadata.
2130
2131        :param db_id: The library (database) id of the item.
2132        :param provider_instance_id: The provider instance whose images should be removed.
2133        :return: True if any images were removed and the db record was updated.
2134        """
2135        # read the raw metadata straight from the db (instead of via get_library_item)
2136        # to avoid persisting any images that are only injected at read time (such as
2137        # the album thumb that gets merged into a track's images)
2138        db_row = await self.mass.music.database.get_row(self.db_table, {"item_id": db_id})
2139        if not db_row or not (raw_metadata := db_row["metadata"]):
2140            return False
2141        metadata = MediaItemMetadata.from_dict(json_loads(raw_metadata))
2142        if not metadata.images:
2143            return False
2144        remaining = UniqueList(
2145            img for img in metadata.images if img.provider != provider_instance_id
2146        )
2147        if len(remaining) == len(metadata.images):
2148            # nothing belonged to this provider
2149            return False
2150        metadata.images = remaining or None
2151        await self.mass.music.database.update(
2152            self.db_table,
2153            {"item_id": db_id},
2154            {"metadata": serialize_to_json(metadata)},
2155        )
2156        return True
2157
2158    def _sync_details_query_parts(self) -> tuple[str, str, dict[str, Any]]:
2159        """
2160        Return extra (columns, joins, params) for this media type's sync-details query.
2161
2162        Override in a subclass to select additional lightweight columns needed by the
2163        library sync change detection for this media type.
2164        """
2165        return "", "", {}
2166
2167    def _parse_sync_details_row(self, db_row: Mapping[str, Any]) -> LibraryItemSyncDetails:
2168        """Parse a raw sync-details db row into a LibraryItemSyncDetails object."""
2169        return LibraryItemSyncDetails(
2170            item_id=db_row["item_id"],
2171            favorite=bool(db_row["favorite"]),
2172            date_added=datetime.fromtimestamp(db_row["timestamp_added"], tz=UTC),
2173            provider_mappings=self._parse_sync_details_mappings(db_row),
2174        )
2175
2176    @final
2177    def _parse_sync_details_mappings(self, db_row: Mapping[str, Any]) -> set[ProviderMapping]:
2178        """Parse the aggregated raw provider mapping rows of a sync-details db row."""
2179        return {
2180            ProviderMapping(
2181                item_id=raw_mapping["item_id"],
2182                provider_domain=raw_mapping["provider_domain"],
2183                provider_instance=raw_mapping["provider_instance"],
2184                available=bool(raw_mapping["available"]),
2185                in_library=parse_optional_bool(raw_mapping["in_library"]),
2186                is_unique=parse_optional_bool(raw_mapping["is_unique"]),
2187            )
2188            for raw_mapping in json_loads(db_row["provider_mappings"])
2189        }
2190
2191    def _parse_summary_row(self, db_row: Mapping[str, Any]) -> MediaItemSummaryType:
2192        """
2193        Parse a raw summary db row into a summary item of this controller's media type.
2194
2195        Override in a subclass to fill additional per-type fields (selected by the
2196        subclass's summary_query).
2197        """
2198        provider_mappings = self._parse_summary_provider_mappings(db_row)
2199        return self.summary_item_cls(
2200            item_id=str(db_row["item_id"]),
2201            provider="library",
2202            name=db_row["name"],
2203            sort_name=db_row["sort_name"],
2204            favorite=bool(db_row["favorite"]),
2205            provider_mappings=provider_mappings,
2206            available=self._summary_available(provider_mappings),
2207            metadata=self._parse_summary_metadata(db_row),
2208        )
2209
2210    @final
2211    @staticmethod
2212    def _parse_summary_provider_mappings(db_row: Mapping[str, Any]) -> set[ProviderMapping]:
2213        """Hydrate the provider mappings of a summary row into ProviderMapping objects."""
2214        if not (raw_mappings := db_row["provider_mappings"]):
2215            return set()
2216        return {ProviderMapping.from_dict(x) for x in json_loads(raw_mappings)}
2217
2218    @final
2219    @staticmethod
2220    def _summary_available(provider_mappings: set[ProviderMapping]) -> bool:
2221        """Compute the availability flag from a summary item's provider mappings."""
2222        # same semantics as the MediaItem.available property
2223        if not (available_providers := get_global_cache_value("available_providers")):
2224            return any(x.available for x in provider_mappings)
2225        if TYPE_CHECKING:
2226            available_providers = cast("set[str]", available_providers)
2227        return any(
2228            x.available and x.provider_instance in available_providers for x in provider_mappings
2229        )
2230
2231    @final
2232    @staticmethod
2233    def _parse_summary_metadata(db_row: Mapping[str, Any]) -> MediaItemMetadataSummary:
2234        """Build the slim metadata of a summary row, carrying only the (first) thumb image."""
2235        thumb: MediaItemImage | None = None
2236        if raw_images := db_row["images"]:
2237            for image in json_loads(raw_images):
2238                if image["type"] != ImageType.THUMB.value:
2239                    continue
2240                thumb = MediaItemImage(
2241                    type=ImageType.THUMB,
2242                    path=image["path"],
2243                    provider=image["provider"],
2244                    remotely_accessible=image.get("remotely_accessible", False),
2245                )
2246                break
2247        return MediaItemMetadataSummary(images=UniqueList([thumb]) if thumb else None)
2248
2249    @final
2250    def _parse_summary_artist_mappings(
2251        self, db_row: Mapping[str, Any]
2252    ) -> UniqueList[ItemMappingSummary]:
2253        """Parse the aggregated slim artist mapping rows of a summary db row."""
2254        return UniqueList(
2255            ItemMappingSummary(
2256                media_type=MediaType.ARTIST,
2257                item_id=str(raw_mapping["item_id"]),
2258                provider="library",
2259                name=raw_mapping["name"],
2260                sort_name=raw_mapping["sort_name"],
2261            )
2262            for raw_mapping in json_loads(db_row["artists"])
2263        )
2264
2265    async def _adapt_query_for_collections(
2266        self,
2267        sql_query: str,
2268        query_params: dict[str, Any],
2269        summary: bool,
2270        order_by: str | None,
2271        collection_name: str | None = None,
2272        search: str | None = None,
2273    ) -> str:
2274        cache_key_json_object = f"collection_{self.api_base}"
2275        json_object = await self.mass.cache.get(key=cache_key_json_object, category=int(summary))
2276        if json_object is None:
2277            # get column names of base query
2278            db_rows = await self.mass.music.database.get_rows_from_query(
2279                sql_query, query_params, limit=1, offset=0
2280            )
2281            # create a sql json_object which queries all these columns
2282            if db_rows:
2283                json_object = (
2284                    "json_object(" + ",".join([f"'{x}',{x}" for x in db_rows[0].keys()]) + ")"  # noqa: SIM118
2285                )
2286                await self.mass.cache.set(
2287                    key=cache_key_json_object, category=int(summary), data=json_object
2288                )
2289            else:
2290                json_object = "json_object()"
2291
2292        collections_column = "collections" if summary else "json_extract(metadata, '$.collections')"
2293
2294        supported_order_keys = [
2295            "name",
2296            "name_desc",
2297            "sort_name",
2298            "sort_name_desc",
2299            "timestamp_added",
2300            "timestamp_added_desc",
2301            "timestamp_modified",
2302            "timestamp_modified_desc",
2303            "last_played",
2304            "last_played_desc",
2305            "play_count",
2306            "play_count_desc",
2307        ]
2308
2309        # additional order options subject to media type
2310        # single is targeting a single media item, collection the aggregated ones
2311        single_extra_order_keys = ""
2312        collection_extra_order_keys = ""
2313        if MediaType.AUDIOBOOK.value in self.api_base:
2314            single_extra_order_keys = "duration,"
2315            collection_extra_order_keys = "SUM(duration) as duration,"
2316            supported_order_keys += ["duration", "duration_desc"]
2317
2318        sql_query = f"""
2319        SELECT * FROM (
2320
2321            WITH
2322                joined_table as ({sql_query}),
2323                collection_extract as (
2324                    SELECT
2325                        name as media_name,
2326                        timestamp_added,
2327                        timestamp_modified,
2328                        last_played,
2329                        play_count,
2330                        {single_extra_order_keys}
2331                        json_extract(iter_coll.value, '$.title') as collection_title,
2332                        json_extract(iter_coll.value, '$.sequence') as collection_sequence,
2333                        json_extract(iter_coll.value, '$.search_title') as collection_search_title,
2334                        json_extract(iter_coll.value, '$.search_sort_title') as collection_search_sort_title,
2335                        CASE
2336                            WHEN json_type(iter_coll.value, '$.sequence') IN ('integer', 'real')
2337                            THEN 1
2338                            WHEN json_type(iter_coll.value, '$.sequence') = 'text'
2339                                AND json_valid(json_extract(iter_coll.value, '$.sequence'))
2340                            THEN CASE
2341                                WHEN json_type(json_extract(iter_coll.value, '$.sequence'))
2342                                    IN ('integer', 'real')
2343                                THEN 1
2344                                ELSE 0
2345                            END
2346                            ELSE 0
2347                        END as collection_sequence_is_numeric,
2348                        {json_object} as media_data
2349                    FROM (
2350                        SELECT * FROM joined_table
2351                    ), json_each({collections_column}) as iter_coll
2352                )
2353            SELECT
2354                'collection' as type,
2355                collection_title as name,
2356                COALESCE(MAX(collection_search_title), replace(lower(collection_title),' ','')) AS search_name,
2357                COALESCE(MAX(collection_search_sort_title), replace(lower(collection_title),' ','')) AS search_sort_name,
2358                MAX(timestamp_added) as timestamp_added,
2359                MAX(timestamp_modified) as timestamp_modified,
2360                MAX(last_played) as last_played,
2361                SUM(play_count) as play_count,
2362                {collection_extra_order_keys}
2363                json_group_array(media_data) as media_data
2364            FROM (
2365                SELECT * FROM collection_extract
2366                -- NOTE: The following ORDER_BY to control the aggregation order of json_group_array is undocumented sqlite behavior
2367                -- Confirmed working with sqlite 3.40.1 & 3.53
2368                -- Once our image moves to sqlite 3.44 we can and should make use of ORDER_BY in the aggregate itself
2369                ORDER BY collection_title,
2370                -- null case
2371                CASE WHEN collection_sequence IS NULL THEN 1 ELSE 0 END,
2372                -- numeric before text
2373                CASE WHEN collection_sequence_is_numeric THEN 0 ELSE 1 END,
2374                -- order NUMERIC
2375                CASE WHEN collection_sequence_is_numeric
2376                    THEN CAST(collection_sequence AS REAL)
2377                END,
2378                -- order TEXT
2379                CASE WHEN NOT collection_sequence_is_numeric
2380                    THEN collection_sequence
2381                END COLLATE NOCASE,
2382                -- order by media name if no sequence given
2383                CASE
2384                    WHEN collection_sequence IS NULL
2385                    THEN media_name
2386                END COLLATE NOCASE
2387            )
2388            GROUP BY collection_title
2389
2390            UNION ALL
2391
2392            SELECT 'single', name, search_name, search_sort_name,
2393                timestamp_added, timestamp_modified, last_played, play_count,
2394                {single_extra_order_keys}
2395                {json_object} FROM joined_table
2396                WHERE {collections_column} IS NULL
2397                    OR {collections_column} = '[]'
2398        )
2399        """
2400
2401        if collection_name:
2402            sql_query += " WHERE type = 'collection' AND name = :collection_name"
2403            return sql_query
2404
2405        if search:
2406            sql_query += " WHERE search_name LIKE :search"
2407
2408        if order_by:
2409            if order_by not in supported_order_keys:
2410                self.logger.warning("%s is not supported for order_by key in collections", order_by)
2411                order_by = "name"  # fallback
2412            if sort_key := SORT_KEYS.get(order_by):
2413                sql_query += f" ORDER BY {sort_key}"
2414
2415        return sql_query
2416
2417    async def _merge_library_items(self, target_id: int, source_id: int) -> tuple[ItemCls, ItemCls]:
2418        """Merge the source library item into the target while the controller lock is held."""
2419        target_item = await self.get_library_item(target_id)
2420        source_item = await self.get_library_item(source_id)
2421        await self._validate_library_item_merge(target_item, source_item)
2422        target_row = await self.mass.music.database.get_row(self.db_table, {"item_id": target_id})
2423        source_row = await self.mass.music.database.get_row(self.db_table, {"item_id": source_id})
2424        assert target_row is not None
2425        assert source_row is not None
2426        timestamps_added = tuple(
2427            timestamp
2428            for timestamp in (
2429                int(target_row["timestamp_added"] or 0),
2430                int(source_row["timestamp_added"] or 0),
2431            )
2432            if timestamp
2433        )
2434
2435        token = SUPPRESS_MEDIA_ITEM_UPDATES.set(True)
2436        try:
2437            source_mappings = source_item.provider_mappings
2438            source_item.provider_mappings = set()
2439            try:
2440                await self._update_library_item_for_merge(target_id, source_item)
2441            finally:
2442                source_item.provider_mappings = source_mappings
2443
2444            await self.mass.music.database.execute_write(
2445                f"""
2446                UPDATE {self.db_table}
2447                SET play_count = CASE item_id
2448                    WHEN :target_id THEN :merged_play_count
2449                    WHEN :source_id THEN 0
2450                END
2451                WHERE item_id IN (:target_id, :source_id)
2452                """,
2453                {
2454                    "target_id": target_id,
2455                    "source_id": source_id,
2456                    "merged_play_count": int(target_row["play_count"] or 0)
2457                    + int(source_row["play_count"] or 0),
2458                },
2459            )
2460            await self.mass.music.database.update(
2461                self.db_table,
2462                {"item_id": target_id},
2463                {
2464                    "favorite": bool(target_row["favorite"]) or bool(source_row["favorite"]),
2465                    "last_played": max(
2466                        int(target_row["last_played"] or 0), int(source_row["last_played"] or 0)
2467                    ),
2468                    "timestamp_added": min(timestamps_added) if timestamps_added else 0,
2469                },
2470            )
2471            await self._merge_genre_mappings(target_id, source_id)
2472            await self._merge_library_item_references(target_id, source_id)
2473            await self._merge_library_playlog(target_id, source_id)
2474            await self._merge_library_item_relations(target_id, source_id)
2475            await self.mass.music.database.execute_write(
2476                f"UPDATE {DB_TABLE_PROVIDER_MAPPINGS} SET item_id = :target_id "
2477                "WHERE media_type = :media_type AND item_id = :source_id",
2478                {
2479                    "target_id": target_id,
2480                    "source_id": source_id,
2481                    "media_type": self.media_type.value,
2482                },
2483            )
2484            await MediaControllerBase.remove_item_from_library(self, source_id, recursive=False)
2485            merged_item = await self.get_library_item(target_id)
2486        finally:
2487            SUPPRESS_MEDIA_ITEM_UPDATES.reset(token)
2488
2489        return source_item, merged_item
2490
2491    async def _merge_library_items_batched(self, target_id: int, source_id: int) -> ItemCls:
2492        """Merge library items while batching the transfer's database writes."""
2493        async with self.mass.music.database.deferred_commit():
2494            source_item, merged_item = await self._merge_library_items(target_id, source_id)
2495        if not SUPPRESS_MEDIA_ITEM_UPDATES.get():
2496            self.mass.signal_event(EventType.MEDIA_ITEM_DELETED, source_item.uri, source_item)
2497            self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, merged_item.uri, merged_item)
2498        return merged_item
2499
2500    async def _validate_library_item_merge(self, target: ItemCls, source: ItemCls) -> None:
2501        """Validate that the target and source items can be merged."""
2502        if target.media_type != self.media_type or source.media_type != self.media_type:
2503            msg = "Library items must have the controller's media type"
2504            raise InvalidDataError(msg)
2505
2506    async def _update_library_item_for_merge(self, item_id: int, update: ItemCls) -> None:
2507        """Merge model state into an existing library item."""
2508        await self._update_library_item(item_id, update)
2509
2510    async def _merge_library_item_relations(self, target_id: int, source_id: int) -> None:
2511        """Transfer relations that reference the merged media item."""
2512        if self.media_type == MediaType.ALBUM:
2513            await self._move_relation_rows(DB_TABLE_ALBUM_ARTISTS, "album_id", target_id, source_id)
2514            await self._move_relation_rows(DB_TABLE_ALBUM_TRACKS, "album_id", target_id, source_id)
2515        elif self.media_type == MediaType.ARTIST:
2516            await self._move_relation_rows(
2517                DB_TABLE_ALBUM_ARTISTS, "artist_id", target_id, source_id
2518            )
2519            await self._move_relation_rows(
2520                DB_TABLE_AUDIOBOOK_ARTISTS, "artist_id", target_id, source_id
2521            )
2522            await self._move_relation_rows(
2523                DB_TABLE_TRACK_ARTISTS, "artist_id", target_id, source_id
2524            )
2525        elif self.media_type == MediaType.AUDIOBOOK:
2526            await self._move_relation_rows(
2527                DB_TABLE_AUDIOBOOK_ARTISTS, "audiobook_id", target_id, source_id
2528            )
2529        elif self.media_type == MediaType.TRACK:
2530            await self._move_relation_rows(DB_TABLE_ALBUM_TRACKS, "track_id", target_id, source_id)
2531            await self._move_relation_rows(DB_TABLE_TRACK_ARTISTS, "track_id", target_id, source_id)
2532
2533    async def _merge_library_item_references(self, target_id: int, source_id: int) -> None:
2534        """Transfer references to the source item owned by specialized controllers."""
2535        return
2536
2537    async def _merge_genre_mappings(self, target_id: int, source_id: int) -> None:
2538        """Transfer genre mappings and exclusions to the target item."""
2539        values = {
2540            "target_id": target_id,
2541            "source_id": source_id,
2542            "media_type": self.media_type.value,
2543        }
2544        await self.mass.music.database.execute_write(
2545            f"""
2546            INSERT INTO {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING}(
2547                genre_id, media_id, media_type, alias, is_derived, is_manual
2548            )
2549            SELECT genre_id, :target_id, media_type, alias, is_derived, is_manual
2550            FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING}
2551            WHERE media_id = :source_id AND media_type = :media_type
2552            ON CONFLICT(genre_id, media_id, media_type) DO UPDATE SET
2553                alias = CASE
2554                    WHEN excluded.is_manual AND NOT is_manual
2555                    THEN COALESCE(excluded.alias, alias)
2556                    ELSE COALESCE(alias, excluded.alias)
2557                END,
2558                is_derived = is_derived OR excluded.is_derived,
2559                is_manual = is_manual OR excluded.is_manual
2560            """,
2561            values,
2562        )
2563        await self.mass.music.database.execute_write(
2564            f"""
2565            INSERT OR IGNORE INTO {DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION}(
2566                genre_id, media_id, media_type
2567            )
2568            SELECT genre_id, :target_id, media_type
2569            FROM {DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION}
2570            WHERE media_id = :source_id AND media_type = :media_type
2571            """,
2572            values,
2573        )
2574        await self.mass.music.database.delete(
2575            DB_TABLE_GENRE_MEDIA_ITEM_MAPPING,
2576            {"media_id": source_id, "media_type": self.media_type.value},
2577        )
2578        await self.mass.music.database.delete(
2579            DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION,
2580            {"media_id": source_id, "media_type": self.media_type.value},
2581        )
2582
2583    async def _merge_genre_references(self, target_id: int, source_id: int) -> None:
2584        """Transfer media mappings and exclusions that point to the source genre."""
2585        values = {"target_id": target_id, "source_id": source_id}
2586        await self.mass.music.database.execute_write(
2587            f"""
2588            INSERT INTO {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING}(
2589                genre_id, media_id, media_type, alias, is_derived, is_manual
2590            )
2591            SELECT :target_id, media_id, media_type, alias, is_derived, is_manual
2592            FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING}
2593            WHERE genre_id = :source_id
2594            ON CONFLICT(genre_id, media_id, media_type) DO UPDATE SET
2595                alias = CASE
2596                    WHEN excluded.is_manual AND NOT is_manual
2597                    THEN COALESCE(excluded.alias, alias)
2598                    ELSE COALESCE(alias, excluded.alias)
2599                END,
2600                is_derived = is_derived OR excluded.is_derived,
2601                is_manual = is_manual OR excluded.is_manual
2602            """,
2603            values,
2604        )
2605        await self.mass.music.database.execute_write(
2606            f"""
2607            INSERT OR IGNORE INTO {DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION}(
2608                genre_id, media_id, media_type
2609            )
2610            SELECT :target_id, media_id, media_type
2611            FROM {DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION}
2612            WHERE genre_id = :source_id
2613            """,
2614            values,
2615        )
2616        await self.mass.music.database.delete(
2617            DB_TABLE_GENRE_MEDIA_ITEM_MAPPING, {"genre_id": source_id}
2618        )
2619        await self.mass.music.database.delete(
2620            DB_TABLE_GENRE_MEDIA_ITEM_EXCLUSION, {"genre_id": source_id}
2621        )
2622
2623    async def _merge_library_playlog(self, target_id: int, source_id: int) -> None:
2624        """Transfer library-keyed playlog rows using the normal latest-entry semantics."""
2625        values = {
2626            "target_id": target_id,
2627            "source_id": source_id,
2628            "media_type": self.media_type.value,
2629        }
2630        await self.mass.music.database.execute_write(
2631            f"""
2632            INSERT INTO {DB_TABLE_PLAYLOG}(
2633                item_id, provider, media_type, name, image, artists, timestamp,
2634                fully_played, seconds_played, userid, queue_id, user_initiated, playback_speed
2635            )
2636            SELECT
2637                :target_id, provider, media_type, name, image, artists, timestamp,
2638                fully_played, seconds_played, userid, queue_id, user_initiated, playback_speed
2639            FROM {DB_TABLE_PLAYLOG}
2640            WHERE item_id = :source_id AND provider = 'library' AND media_type = :media_type
2641            ON CONFLICT(item_id, provider, media_type, userid) DO UPDATE SET
2642                name = CASE WHEN excluded.timestamp > timestamp THEN excluded.name ELSE name END,
2643                image = CASE WHEN excluded.timestamp > timestamp THEN excluded.image ELSE image END,
2644                artists = CASE WHEN excluded.timestamp > timestamp THEN excluded.artists ELSE artists END,
2645                timestamp = MAX(timestamp, excluded.timestamp),
2646                fully_played = CASE
2647                    WHEN excluded.timestamp > timestamp THEN excluded.fully_played ELSE fully_played
2648                END,
2649                seconds_played = CASE
2650                    WHEN excluded.timestamp > timestamp THEN excluded.seconds_played
2651                    ELSE seconds_played
2652                END,
2653                queue_id = CASE
2654                    WHEN excluded.timestamp > timestamp THEN excluded.queue_id ELSE queue_id
2655                END,
2656                user_initiated = user_initiated OR excluded.user_initiated,
2657                playback_speed = CASE
2658                    WHEN excluded.timestamp > timestamp THEN excluded.playback_speed
2659                    ELSE playback_speed
2660                END
2661            """,
2662            values,
2663        )
2664        await self.mass.music.database.delete(
2665            DB_TABLE_PLAYLOG,
2666            {
2667                "item_id": source_id,
2668                "provider": "library",
2669                "media_type": self.media_type.value,
2670            },
2671        )
2672
2673    async def _move_relation_rows(
2674        self, table: str, item_column: str, target_id: int, source_id: int
2675    ) -> None:
2676        """Move a relation column to the target item, retaining target duplicates."""
2677        values = {"target_id": target_id, "source_id": source_id}
2678        await self.mass.music.database.execute_write(
2679            f"UPDATE OR IGNORE {table} SET {item_column} = :target_id "
2680            f"WHERE {item_column} = :source_id",
2681            values,
2682        )
2683        await self.mass.music.database.delete(table, {item_column: source_id})
2684