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