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