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