music-assistant-server

1"""MusicController: Orchestrates all data from music providers and sync to internal database."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from collections.abc import Awaitable, Callable, Coroutine, Iterable, Sequence
8from contextlib import suppress
9from copy import deepcopy
10from datetime import datetime
11from itertools import zip_longest
12from typing import TYPE_CHECKING, Any, NamedTuple, cast
13
14from music_assistant_models.auth import Scope
15from music_assistant_models.background_task import BackgroundTask, TaskMetadata, TaskSchedule
16from music_assistant_models.config_entries import (
17    ConfigActionResult,
18    ConfigEntry,
19    ConfigValueType,
20)
21from music_assistant_models.enums import (
22    ConfigEntryType,
23    EventType,
24    MediaType,
25    ProviderFeature,
26    ProviderType,
27    TaskStatus,
28)
29from music_assistant_models.errors import (
30    InvalidDataError,
31    InvalidProviderID,
32    InvalidProviderURI,
33    MediaNotFoundError,
34    MusicAssistantError,
35    UnsupportedFeaturedException,
36)
37from music_assistant_models.helpers import get_global_cache_value
38from music_assistant_models.media_items import (
39    Album,
40    Artist,
41    AudioFormat,
42    BrowseFolder,
43    Genre,
44    ItemMapping,
45    MediaItemType,
46    Playlist,
47    Podcast,
48    PodcastEpisode,
49    ProviderMapping,
50    SearchResults,
51    SoundEffect,
52    Track,
53)
54from music_assistant_models.media_items.media_item import MediaCollection
55
56from music_assistant.constants import (
57    CONF_ENTRY_LIBRARY_SYNC_BACK,
58    DB_TABLE_ALBUM_TRACKS,
59    DB_TABLE_ALBUMS,
60    DB_TABLE_PLAYLOG,
61    DB_TABLE_PROVIDER_MAPPINGS,
62    DB_TABLE_TRACK_ARTISTS,
63    DB_TABLE_TRACKS,
64    PROVIDERS_WITH_SHAREABLE_URLS,
65)
66from music_assistant.controllers.music.constants import (
67    CACHE_CATEGORY_SEARCH_RESULTS,
68    CONF_DELETED_PROVIDERS,
69    CONF_RESET_DB,
70    CONF_TRACK_RECONCILIATION_CURSOR,
71    CONF_TRACK_RECONCILIATION_RESCAN_DUE,
72    DATABASE_CLEANUP_TASK_ID,
73    DB_SCHEMA_VERSION,
74    INITIAL_SYNC_DELAY,
75    MUSIC_SYNC_COMPLETION_CHECK_TASK_ID,
76    PROVIDER_MAPPING_CORRECTION_TASK_ID,
77    SEARCH_CACHE_EXPIRATION_COMBINED,
78    SEARCH_CACHE_EXPIRATION_LOCAL_PROVIDER,
79    SEARCH_CACHE_EXPIRATION_STREAMING_PROVIDER,
80    SEARCH_PROVIDER_HARD_TIMEOUT,
81    SEARCH_PROVIDER_SOFT_TIMEOUT,
82    TRACK_RECONCILIATION_BATCH_SIZE,
83    TRACK_RECONCILIATION_MAX_DURATION_DELTA,
84    TRACK_RECONCILIATION_TASK_ID,
85)
86from music_assistant.controllers.music.database import (
87    PLAYLOG_CONFLICT_KEYS,
88    MusicDatabaseSetupMixin,
89)
90from music_assistant.controllers.music.helpers import filter_search_results, sort_search_result
91from music_assistant.controllers.music.media.albums import AlbumsController
92from music_assistant.controllers.music.media.artists import ArtistsController
93from music_assistant.controllers.music.media.audiobooks import AudiobooksController
94from music_assistant.controllers.music.media.base import SUPPRESS_MEDIA_ITEM_UPDATES
95from music_assistant.controllers.music.media.genres import GenreController
96from music_assistant.controllers.music.media.playlists import PlaylistController
97from music_assistant.controllers.music.media.podcasts import PodcastsController
98from music_assistant.controllers.music.media.radio import RadioController
99from music_assistant.controllers.music.media.tracks import TracksController
100from music_assistant.controllers.music.recency import RecencyEngine
101from music_assistant.controllers.music.recommendations.controller import (
102    RecommendationsController,
103)
104from music_assistant.controllers.tasks.context import (
105    report_current_task_failure,
106    update_current_task_progress,
107    update_current_task_progress_from_index,
108    update_current_task_progress_text,
109)
110from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
111from music_assistant.helpers.api import api_command
112from music_assistant.helpers.collections import get_collection_item_media_type_from_item_id
113from music_assistant.helpers.compare import (
114    ALBUM_RETAIL_SUFFIX_KEYS,
115    album_retail_suffix_sql_match,
116    compare_album_name,
117    compare_strings,
118    compare_track,
119    compare_version,
120)
121from music_assistant.helpers.database import UNSET, DatabaseConnection
122from music_assistant.helpers.datetime import (
123    from_utc_timestamp,
124    local_clock_time_to_utc,
125    utc_timestamp,
126)
127from music_assistant.helpers.json import json_loads, serialize_to_json
128from music_assistant.helpers.tags import split_artists
129from music_assistant.helpers.uri import parse_uri
130from music_assistant.helpers.util import parse_optional_bool, parse_title_and_version
131from music_assistant.models.core_controller import CoreController
132from music_assistant.models.music_provider import LIBRARY_FEATURE_BY_MEDIA_TYPE, MusicProvider
133from music_assistant.models.plugin import PluginProvider
134
135if TYPE_CHECKING:
136    from music_assistant_models.auth import User
137    from music_assistant_models.config_entries import CoreConfig
138    from music_assistant_models.media_items import Audiobook
139
140    from music_assistant import MusicAssistant
141    from music_assistant.controllers.music.media.base import MediaControllerBase
142    from music_assistant.helpers.json import SerializableType
143    from music_assistant.models import ProviderInstanceType
144    from music_assistant.models.provider import Provider
145    from music_assistant.providers.builtin import BuiltinProvider
146
147
148class RecentPlayedTrack(NamedTuple):
149    """A recently played track from the playlog, with the artists recorded at play time."""
150
151    track: ItemMapping
152    artists: list[ItemMapping]
153
154
155def _album_title_match(base: str, other: str) -> str:
156    """
157    Return a query part relating two album rows that may name the same album.
158
159    :param base: Alias of the album row the match is expressed against.
160    :param other: Alias of the album row related to it.
161    """
162    # a provider that spells out the retail suffix stores the album under the plain name
163    # plus that suffix, so the pair is related from either side. The raw title decides which
164    # side spelled it out, so an ordinary title that merely ends in those letters ("Step") is
165    # left alone. This relates more titles than the album comparison accepts, which is what
166    # confirms the pair afterwards.
167    matches = [f"{other}.search_name = {base}.search_name"]
168    for suffix in ALBUM_RETAIL_SUFFIX_KEYS:
169        matches.append(
170            f"({album_retail_suffix_sql_match(f'{other}.name', suffix)} "
171            f"AND {other}.search_name = {base}.search_name || '{suffix}')"
172        )
173        matches.append(
174            f"({album_retail_suffix_sql_match(f'{base}.name', suffix)} "
175            f"AND {other}.search_name = "
176            f"substr({base}.search_name, 1, length({base}.search_name) - {len(suffix)}))"
177        )
178    return " OR ".join(matches)
179
180
181# Selects pairs of library track rows that are likely the same recording held twice,
182# once per music provider. Both rows must carry the same normalized title, share a track
183# artist and sit within a few seconds of each other. The album term is the decisive one:
184# both rows must appear at the same position on an album with the same title, so the merge
185# always rests on two providers agreeing on where the track belongs rather than on title and
186# duration alone. Titles are related loosely enough to see past a spelled-out retail suffix,
187# leaving the identity for the album comparison the pair is then held to. Titles that
188# normalize to nothing (symbol-only album names) are excluded there, as they would match
189# every other such album. Rows that already share a provider are skipped, as a provider
190# listing the same recording twice is a separate (and far riskier) case.
191_DUPLICATE_TRACK_CANDIDATES_QUERY = f"""
192SELECT t1.item_id AS item_id_1, t2.item_id AS item_id_2
193FROM {DB_TABLE_TRACKS} t1
194JOIN {DB_TABLE_TRACKS} t2
195  ON t2.search_name = t1.search_name
196 AND t2.item_id > t1.item_id
197 AND abs(t2.duration - t1.duration) <= :max_duration_delta
198WHERE (t1.item_id > :cursor_item_id_1
199       OR (t1.item_id = :cursor_item_id_1 AND t2.item_id > :cursor_item_id_2))
200  AND EXISTS (
201    SELECT 1 FROM {DB_TABLE_TRACK_ARTISTS} ta1
202    JOIN {DB_TABLE_TRACK_ARTISTS} ta2
203      ON ta2.artist_id = ta1.artist_id AND ta2.track_id = t2.item_id
204    WHERE ta1.track_id = t1.item_id)
205  AND EXISTS (
206    SELECT 1 FROM {DB_TABLE_ALBUM_TRACKS} at1
207    JOIN {DB_TABLE_ALBUMS} al1 ON al1.item_id = at1.album_id
208    JOIN {DB_TABLE_ALBUM_TRACKS} at2 ON at2.track_id = t2.item_id
209    JOIN {DB_TABLE_ALBUMS} al2
210      ON al2.item_id = at2.album_id AND ({_album_title_match("al1", "al2")})
211    WHERE at1.track_id = t1.item_id
212      -- a title that is nothing but the suffix strips to nothing, which would relate it to
213      -- every symbol-only album, so neither side may normalize away
214      AND al1.search_name != ''
215      AND al2.search_name != ''
216      -- an unreported position is stored as 0, so two of those agree on nothing;
217      -- a missing disc number does read as disc 1, the way compare_track takes it
218      -- for local files that carry no disc tag
219      AND at1.track_number > 0
220      AND coalesce(nullif(at1.disc_number, 0), 1) = coalesce(nullif(at2.disc_number, 0), 1)
221      AND at1.track_number = at2.track_number)
222  AND NOT EXISTS (
223    SELECT 1 FROM {DB_TABLE_PROVIDER_MAPPINGS} pm1
224    JOIN {DB_TABLE_PROVIDER_MAPPINGS} pm2
225      ON pm2.provider_domain = pm1.provider_domain
226     AND pm2.media_type = 'track' AND pm2.item_id = t2.item_id
227    WHERE pm1.media_type = 'track' AND pm1.item_id = t1.item_id)
228ORDER BY t1.item_id, t2.item_id
229"""
230
231# Returns the title and edition of every album appearance that made the two tracks a
232# candidate, so the pair can be held to agreeing on both. The album terms mirror the candidate
233# query exactly: an appearance the pair does not share a position on says nothing about the
234# album of the one it does.
235_SHARED_ALBUM_EDITIONS_QUERY = f"""
236SELECT al1.name AS name_1, al2.name AS name_2,
237       al1.version AS version_1, al2.version AS version_2
238FROM {DB_TABLE_ALBUM_TRACKS} at1
239JOIN {DB_TABLE_ALBUMS} al1 ON al1.item_id = at1.album_id
240JOIN {DB_TABLE_ALBUM_TRACKS} at2 ON at2.track_id = :item_id_2
241JOIN {DB_TABLE_ALBUMS} al2
242  ON al2.item_id = at2.album_id AND ({_album_title_match("al1", "al2")})
243WHERE at1.track_id = :item_id_1
244  AND al1.search_name != ''
245  AND al2.search_name != ''
246  AND at1.track_number > 0
247  AND coalesce(nullif(at1.disc_number, 0), 1) = coalesce(nullif(at2.disc_number, 0), 1)
248  AND at1.track_number = at2.track_number
249"""
250
251
252class MusicController(MusicDatabaseSetupMixin, CoreController):
253    """Several helpers around the musicproviders."""
254
255    domain: str = "music"
256    config: CoreConfig
257    # where the duplicate track walk stands; restored from config on startup
258    _track_reconciliation_cursor: tuple[int, int] | None = (0, 0)
259    _track_reconciliation_rescan_due: bool = False
260
261    def __init__(self, mass: MusicAssistant) -> None:
262        """Initialize class."""
263        super().__init__(mass)
264        self.cache = self.mass.cache
265        self.artists = ArtistsController(self.mass)
266        self.albums = AlbumsController(self.mass)
267        self.tracks = TracksController(self.mass)
268        self.radio = RadioController(self.mass)
269        self.playlists = PlaylistController(self.mass)
270        self.audiobooks = AudiobooksController(self.mass)
271        self.podcasts = PodcastsController(self.mass)
272        self.genres = GenreController(self.mass)
273        self.recommendations = RecommendationsController(self.mass)
274        self.recency = RecencyEngine(self.mass)
275        self._database: DatabaseConnection | None = None
276        self._sync_lock = asyncio.Lock()
277        self.manifest.name = "Music controller"
278        self.manifest.description = (
279            "Music Assistant's core controller which manages all music from all providers."
280        )
281        self.manifest.icon = "archive-music"
282
283    @property
284    def database(self) -> DatabaseConnection:
285        """Return the database connection."""
286        if self._database is None:
287            raise RuntimeError("Database not initialized")
288        return self._database
289
290    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
291        """Return all Config Entries for this core module (if any)."""
292        return (
293            ConfigEntry(
294                key=CONF_RESET_DB,
295                type=ConfigEntryType.ACTION,
296                category="generic",
297                advanced=True,
298            ),
299        )
300
301    async def handle_config_action(
302        self, action: str
303    ) -> tuple[ConfigEntry, ...] | ConfigActionResult | None:
304        """Handle a one-shot action button press and report its outcome."""
305        if action == CONF_RESET_DB:
306            await self._reset_database()
307            await self.mass.cache.clear()
308            await self.start_sync()
309            return ConfigActionResult(translation_key=f"{CONF_RESET_DB}.result")
310        return await super().handle_config_action(action)
311
312    async def setup(self, config: CoreConfig) -> None:
313        """Async initialize of module."""
314        self.config = config
315        # setup library database
316        await self._setup_database()
317        # make sure to finish any removal jobs
318        for removed_provider in cast(
319            "list[str]",
320            self.mass.config.get_raw_core_config_value(self.domain, CONF_DELETED_PROVIDERS, []),
321        ):
322            await self.cleanup_provider(removed_provider)
323
324    async def post_setup(self) -> None:
325        """Handle logic after all core controllers have been set up."""
326        self._register_database_cleanup_task()
327        self._register_provider_mapping_correction_task()
328        self._restore_track_reconciliation_state()
329        self._register_track_reconciliation_task()
330        self.genres.register_scheduled_scan_task()
331
332    async def close(self) -> None:
333        """Cleanup on exit."""
334        if self._database:
335            await self._database.close()
336
337    async def get_diagnostics(self) -> dict[str, SerializableType]:
338        """Return diagnostics info for this controller to include in diagnostics reports."""
339        return {
340            "db_schema_version": DB_SCHEMA_VERSION,
341            "sync_tasks_active": len(self.active_sync_tasks),
342        }
343
344    async def on_provider_loaded(self, provider: MusicProvider) -> None:
345        """Handle logic when a provider is loaded."""
346        await self.schedule_provider_sync(provider.instance_id)
347
348    async def on_provider_unload(self, provider: MusicProvider) -> None:
349        """
350        Handle logic when a provider is (about to get) unloaded.
351
352        Sync tasks are unscheduled by MusicAssistant.unload_provider itself, which also
353        decides whether their persisted state is kept (reload) or cleared (removal).
354        """
355
356    @property
357    def providers(self) -> list[MusicProvider]:
358        """
359        Return all loaded/running MusicProviders (instances).
360
361        Note that this applies user provider filters (for all user types).
362        """
363        return cast(
364            "list[MusicProvider]",
365            [
366                x
367                for x in self._apply_user_provider_filter(self.mass.providers)
368                if x.type == ProviderType.MUSIC
369            ],
370        )
371
372    @api_command("music/sync", required_scope=Scope.LIBRARY_MANAGE)
373    async def start_sync(
374        self,
375        media_types: list[MediaType] | None = None,
376        providers: list[str] | None = None,
377    ) -> list[BackgroundTask]:
378        """
379        Start running the sync of (all or selected) musicproviders.
380
381        media_types: only sync these media types. None for all.
382        providers: only sync these provider instances. None for all.
383        """
384        tasks: list[BackgroundTask] = []
385        if media_types is None:
386            media_types = MediaType.ALL
387        if providers is None:
388            providers = [x.instance_id for x in self.providers]
389
390        for media_type in media_types:
391            for provider in self.providers:
392                if provider.instance_id not in providers:
393                    continue
394                if not self.library_supported(provider, media_type):
395                    continue
396                # handle mediatype specific sync config
397                conf_key = f"library_sync_{media_type}s"
398                sync_conf: ConfigValueType = await self.mass.config.get_provider_config_value(
399                    provider.instance_id, conf_key
400                )
401                if not sync_conf:
402                    continue
403                await self._schedule_provider_mediatype_sync(provider, media_type, True)
404                task_id = self._get_sync_task_id(provider, media_type)
405                try:
406                    tasks.append(self.mass.tasks.run_task(task_id))
407                except InvalidDataError:
408                    tasks.append(
409                        self.mass.tasks.run_background_task(
410                            task_id=task_id,
411                            name=self._get_sync_task_name(provider, media_type),
412                            handler=self._create_provider_sync_handler(provider, media_type),
413                            translation_key=self._get_sync_task_translation_key(media_type),
414                            translation_args=[provider.name],
415                            translation_owner=self.translation_owner,
416                            user_id=(user.user_id if (user := get_current_user()) else None),
417                            metadata=self._get_sync_task_metadata(provider, media_type),
418                            allow_retry=True,
419                            priority=True,
420                        )
421                    )
422        return tasks
423
424    @property
425    def active_sync_tasks(self) -> list[BackgroundTask]:
426        """Return provider sync tasks that are currently pending or running."""
427        return [
428            task
429            for task in self.mass.tasks.get_tasks_by_metadata(task_domain="music_sync")
430            if task.status in (TaskStatus.PENDING, TaskStatus.RUNNING)
431        ]
432
433    @api_command("music/search", required_scope=Scope.LIBRARY_READ, allow_impersonation=True)
434    async def search(
435        self,
436        search_query: str,
437        media_types: list[MediaType] = MediaType.ALL,
438        limit: int = 25,
439        library_only: bool = False,
440        providers: list[str] | None = None,
441    ) -> SearchResults:
442        """
443        Perform global search for media items on all providers.
444
445        :param search_query: Search query.
446        :param media_types: A list of media_types to include.
447        :param limit: number of items to return in the search (per type).
448        :param library_only: Deprecated - use providers=["library"] instead.
449        :param providers: Optionally restrict the search to the given providers
450            (by instance id or domain), where the special value "library" selects
451            the library. Omit to search the library and all available providers.
452        """
453        if not search_query.strip():
454            # several providers reject an empty query with a hard error
455            return SearchResults()
456        if not media_types:
457            media_types = MediaType.ALL
458        if library_only and providers is None:
459            # handle deprecated library_only flag
460            providers = ["library"]
461        # resolve the search targets: all (unique) music providers plus plugin
462        # providers with search support, optionally filtered by the providers argument
463        plugin_search_providers = [
464            p.instance_id
465            for p in self.mass.get_providers_supporting_feature(
466                ProviderFeature.SEARCH,
467                priority=(ProviderType.PLUGIN,),
468            )
469        ]
470        all_search_providers = sorted(self.get_unique_providers() + plugin_search_providers)
471        if providers is None:
472            include_library = True
473            search_providers = all_search_providers
474        else:
475            include_library = "library" in providers
476            requested_providers = set(providers)
477            search_providers = [
478                instance_id
479                for instance_id in all_search_providers
480                if (prov := self.mass.get_provider(instance_id))
481                and (prov.instance_id in requested_providers or prov.domain in requested_providers)
482            ]
483        # use cache to avoid repeated searches
484        cache_key = (
485            f"{search_query}-{'-'.join(sorted([mt.value for mt in media_types]))}-{limit}-"
486            f"{int(include_library)}-{','.join(search_providers)}"
487        )
488        if cache := await self.mass.cache.get(
489            key=cache_key,
490            provider=self.domain,
491            category=CACHE_CATEGORY_SEARCH_RESULTS,
492            base_class=SearchResults,
493        ):
494            return cast("SearchResults", cache)
495        # Check if the search query is a streaming provider public shareable URL
496        if (url_result := await self._search_shareable_url(search_query)) is not None:
497            return url_result
498        # handle normal global search by querying the library and all providers
499        # the library is always searched first: it is fast and its results are used
500        # to deduplicate provider results and to skip provider searches for media
501        # types that already have a (near) exact match in the library
502        library_results = await self.search_library(search_query, media_types, limit=limit)
503        results_per_provider: list[SearchResults] = []
504        if include_library:
505            results_per_provider.append(library_results)
506        all_results_complete = True
507        if search_providers:
508            # create a set of all provider item ids already in library
509            # this way we can avoid returning duplicates in the search results
510            all_prov_item_ids = {
511                (item.media_type, prov_mapping.provider_domain, prov_mapping.item_id)
512                for items in (
513                    library_results.artists,
514                    library_results.albums,
515                    library_results.tracks,
516                    library_results.playlists,
517                    library_results.audiobooks,
518                    library_results.podcasts,
519                )
520                for item in items
521                for prov_mapping in cast("MediaItemType", item).provider_mappings
522            }
523            # only apply the exact match shortcut on a regular global search;
524            # an explicit providers selection must always search those providers
525            covered_media_types = (
526                self._get_covered_media_types(library_results, search_query)
527                if providers is None
528                else set()
529            )
530            provider_searches: list[Coroutine[Any, Any, SearchResults | None]] = []
531            for provider_instance in search_providers:
532                if not (prov := self.mass.get_provider(provider_instance)):
533                    continue
534                # skip media types for which the library already holds a (near)
535                # exact match that is mapped to this provider: searching the
536                # provider again for that media type will not add anything new
537                prov_media_types = [
538                    mt
539                    for mt in media_types
540                    if (mt, prov.domain) not in covered_media_types
541                    and (mt, prov.instance_id) not in covered_media_types
542                ]
543                if not prov_media_types:
544                    continue
545                provider_searches.append(
546                    self._search_provider(
547                        search_query,
548                        provider_instance,
549                        prov_media_types,
550                        limit=limit,
551                        skip_item_ids=all_prov_item_ids,
552                    )
553                )
554            # include results from all (unique) music providers
555            # one failing provider must not break the entire search,
556            # so exceptions are logged and excluded from the results
557            gather_results = await asyncio.gather(*provider_searches, return_exceptions=True)
558            for res in gather_results:
559                if isinstance(res, SearchResults):
560                    results_per_provider.append(res)
561                    continue
562                # a provider that failed or timed out contributes no results
563                all_results_complete = False
564                if isinstance(res, BaseException):
565                    self.logger.error("Search on provider failed", exc_info=res)
566        # return result from all providers while keeping index
567        # so the result is sorted as each provider delivered
568        result = SearchResults(
569            artists=[
570                item
571                for sublist in zip_longest(*[x.artists for x in results_per_provider])
572                for item in sublist
573                if item is not None
574            ][:limit],
575            albums=[
576                item
577                for sublist in zip_longest(*[x.albums for x in results_per_provider])
578                for item in sublist
579                if item is not None
580            ][:limit],
581            genres=[
582                item
583                for sublist in zip_longest(*[x.genres for x in results_per_provider])
584                for item in sublist
585                if item is not None
586            ][:limit],
587            tracks=[
588                item
589                for sublist in zip_longest(*[x.tracks for x in results_per_provider])
590                for item in sublist
591                if item is not None
592            ][:limit],
593            playlists=[
594                item
595                for sublist in zip_longest(*[x.playlists for x in results_per_provider])
596                for item in sublist
597                if item is not None
598            ][:limit],
599            radio=[
600                item
601                for sublist in zip_longest(*[x.radio for x in results_per_provider])
602                for item in sublist
603                if item is not None
604            ][:limit],
605            audiobooks=[
606                item
607                for sublist in zip_longest(*[x.audiobooks for x in results_per_provider])
608                for item in sublist
609                if item is not None
610            ][:limit],
611            podcasts=[
612                item
613                for sublist in zip_longest(*[x.podcasts for x in results_per_provider])
614                for item in sublist
615                if item is not None
616            ][:limit],
617            sound_effects=[
618                item
619                for sublist in zip_longest(*[x.sound_effects for x in results_per_provider])
620                for item in sublist
621                if item is not None
622            ][:limit],
623        )
624
625        # the search results should already be sorted by relevance
626        # but we apply one extra round of sorting and that is to put exact name
627        # matches and library items first
628        for field in (
629            "artists",
630            "albums",
631            "genres",
632            "tracks",
633            "playlists",
634            "radio",
635            "audiobooks",
636            "podcasts",
637            "sound_effects",
638        ):
639            setattr(result, field, sort_search_result(search_query, getattr(result, field)))
640        # only cache the combined result if all providers contributed,
641        # so a failed or timed out provider is retried on a next search
642        if all_results_complete:
643            await self._cache_search_results(
644                cache_key, result, SEARCH_CACHE_EXPIRATION_COMBINED, self.domain
645            )
646        return result
647
648    async def search_library(
649        self,
650        search_query: str,
651        media_types: list[MediaType],
652        limit: int = 10,
653    ) -> SearchResults:
654        """
655        Perform search on the library.
656
657        :param search_query: Search query
658        :param media_types: A list of media_types to include.
659        :param limit: number of items to return in the search (per type).
660        """
661        result_fields: dict[MediaType, str] = {
662            MediaType.ARTIST: "artists",
663            MediaType.ALBUM: "albums",
664            MediaType.GENRE: "genres",
665            MediaType.TRACK: "tracks",
666            MediaType.PLAYLIST: "playlists",
667            MediaType.RADIO: "radio",
668            MediaType.AUDIOBOOK: "audiobooks",
669            MediaType.PODCAST: "podcasts",
670        }
671        result = SearchResults()
672        # search all media types in parallel, each is an independent db query
673        searchable_media_types = [x for x in media_types if x in result_fields]
674        search_results = await asyncio.gather(
675            *[
676                self.get_controller(media_type).search(search_query, "library", limit=limit)
677                for media_type in searchable_media_types
678            ]
679        )
680        for media_type, items in zip(searchable_media_types, search_results, strict=True):
681            if items:
682                setattr(result, result_fields[media_type], items)
683        return result
684
685    @api_command("music/browse", required_scope=Scope.LIBRARY_READ)
686    async def browse(
687        self, path: str | None = None
688    ) -> Sequence[MediaItemType | ItemMapping | BrowseFolder]:
689        """Browse Music providers."""
690        if not path or path == "root":
691            # root level; folder per provider that declares BROWSE
692            root_items: list[MediaItemType | BrowseFolder] = []
693            providers_with_browse = self.mass.get_providers_supporting_feature(
694                ProviderFeature.BROWSE
695            )
696            for prov in self._apply_user_provider_filter(providers_with_browse):
697                root_items.append(
698                    BrowseFolder(
699                        item_id="root",
700                        provider=prov.domain,
701                        path=f"{prov.instance_id}://",
702                        uri=f"{prov.instance_id}://",
703                        name=prov.name,
704                    )
705                )
706            # AudioSource providers surface at root like regular providers; a
707            # provider with a single user-initiable source is promoted to that
708            # source directly so it's playable in one tap.
709            audio_source_providers = self.mass.get_providers_supporting_feature(
710                ProviderFeature.AUDIO_SOURCE
711            )
712            for prov in self._apply_user_provider_filter(audio_source_providers):
713                if not isinstance(prov, PluginProvider):
714                    continue
715                initiable = [
716                    source for source in await prov.get_audio_sources() if source.can_initiate
717                ]
718                if not initiable:
719                    continue
720                if len(initiable) == 1:
721                    root_items.append(initiable[0])
722                else:
723                    root_items.append(
724                        BrowseFolder(
725                            item_id="root",
726                            provider=prov.domain,
727                            path=f"{prov.instance_id}://",
728                            uri=f"{prov.instance_id}://",
729                            name=prov.name,
730                        )
731                    )
732            return root_items
733
734        # provider level
735        prepend_items: list[BrowseFolder] = []
736        provider_instance, sub_path = path.split("://", 1)
737        browse_prov = self.mass.get_provider(provider_instance)
738        # handle regular provider listing, always add back folder first
739        if not browse_prov or not sub_path:
740            prepend_items.append(
741                BrowseFolder(item_id="root", provider="library", path="root", name="..")
742            )
743            if not browse_prov:
744                return prepend_items
745        else:
746            back_path = f"{provider_instance}://" + "/".join(sub_path.split("/")[:-1])
747            prepend_items.append(
748                BrowseFolder(
749                    item_id="back",
750                    provider=provider_instance,
751                    path=back_path,
752                    name="..",
753                )
754            )
755        # AudioSource providers don't implement browse(); list their initiable sources directly
756        if (
757            isinstance(browse_prov, PluginProvider)
758            and ProviderFeature.AUDIO_SOURCE in browse_prov.supported_features
759        ):
760            initiable_items: list[MediaItemType | BrowseFolder] = [
761                source for source in await browse_prov.get_audio_sources() if source.can_initiate
762            ]
763            return [*prepend_items, *initiable_items]
764        # limit -1 to account for the prepended items
765        prov_items = await cast("MusicProvider", browse_prov).browse(path=path)
766        return [*prepend_items, *prov_items]
767
768    @api_command("music/recently_played_items", required_scope=Scope.LIBRARY_READ)
769    async def recently_played(
770        self,
771        limit: int = 10,
772        media_types: list[MediaType] | None = None,
773        userid: str | None = None,
774        queue_id: str | None = None,
775        fully_played_only: bool = True,
776        user_initiated_only: bool = False,
777        played_after_timestamp: int | None = None,
778        providers: list[str] | None = None,
779        *,
780        always_include_media_types: list[MediaType] | None = None,
781    ) -> list[ItemMapping]:
782        """
783        Return a list of the last played items.
784
785        :param limit: Maximum number of items to return.
786        :param media_types: Filter by media types.
787        :param userid: Filter by specific user ID.
788        :param queue_id: Filter by specific queue ID.
789        :param fully_played_only: If True, only return fully played items.
790        :param user_initiated_only: If True, only return items initiated by the user.
791        :param played_after_timestamp: If set, only return items played at or after this
792            epoch-seconds timestamp.
793        :param providers: Restrict results to items reachable through one of these provider
794            instance ids (OR semantics). None applies no filter; an explicit empty list
795            returns no items.
796        :param always_include_media_types: Media types to include regardless of
797            user_initiated_only (e.g. podcasts/audiobooks, which have no user-initiated
798            container).
799        """
800        if providers is not None and not providers:
801            return []
802        if media_types is None:
803            media_types = MediaType.ALL
804        media_types_str = "(" + ",".join(f'"{x}"' for x in media_types) + ")"
805        available_providers = ("library", *self.get_active_provider_instances())
806        available_providers_str = "(" + ",".join(f'"{x}"' for x in available_providers) + ")"
807        # user_initiated_only constrains only `media_types`; always_include_media_types are
808        # included regardless (e.g. podcasts/audiobooks have no user-initiated container row).
809        media_type_clause = f"p.media_type in {media_types_str}"
810        if user_initiated_only:
811            media_type_clause += " AND p.user_initiated = 1"
812        media_type_clause = f"({media_type_clause})"
813        if always_include_media_types:
814            always_str = "(" + ",".join(f'"{x}"' for x in always_include_media_types) + ")"
815            media_type_clause = f"({media_type_clause} OR p.media_type in {always_str})"
816
817        params: dict[str, Any] = {}
818        user = get_current_user()
819        # a library row only needs resolving through its provider mappings when a filter
820        # (explicit or user-scoped) is actually active; otherwise every library row is
821        # kept, matching this method's unfiltered behavior.
822        if providers is not None or (user and user.provider_filter):
823            requested_clause = ""
824            direct_requested_clause = ""
825            if providers is not None:
826                params["requested_providers"] = providers
827                requested_clause = " AND m.provider_instance IN :requested_providers"
828                direct_requested_clause = " AND p.provider IN :requested_providers"
829            provider_clause = (
830                "(CASE WHEN p.provider = 'library' THEN "
831                f"EXISTS (SELECT 1 FROM {DB_TABLE_PROVIDER_MAPPINGS} m "
832                "WHERE m.item_id = p.item_id AND m.media_type = p.media_type "
833                f"AND m.available = 1 "
834                f"AND m.provider_instance IN {available_providers_str}{requested_clause}) "
835                f"ELSE (p.provider IN {available_providers_str}{direct_requested_clause}) END)"
836            )
837        else:
838            provider_clause = f"p.provider IN {available_providers_str}"
839        query = (
840            f"SELECT p.* FROM {DB_TABLE_PLAYLOG} p WHERE {media_type_clause} AND {provider_clause} "
841        )
842        if fully_played_only:
843            query += "AND p.fully_played = 1 "
844        if userid:
845            query += "AND p.userid = :userid "
846            params["userid"] = userid
847        elif user:
848            query += "AND p.userid = :userid "
849            params["userid"] = user.user_id
850        if queue_id:
851            query += "AND p.queue_id = :queue_id "
852            params["queue_id"] = queue_id
853        if played_after_timestamp is not None:
854            query += "AND p.timestamp >= :played_after_timestamp "
855            params["played_after_timestamp"] = played_after_timestamp
856        query += "ORDER BY p.timestamp DESC"
857        db_rows = await self.mass.music.database.get_rows_from_query(
858            query, params=params or None, limit=limit
859        )
860        result: list[ItemMapping] = []
861        available_providers = ("library", *get_global_cache_value("available_providers", []))
862        for db_row in db_rows:
863            provider = db_row["provider"]
864            result.append(
865                ItemMapping.from_dict(
866                    {
867                        "item_id": db_row["item_id"],
868                        "provider": provider,
869                        "media_type": db_row["media_type"],
870                        "name": db_row["name"],
871                        "image": json_loads(db_row["image"]) if db_row["image"] else None,
872                        "available": provider in available_providers,
873                    }
874                )
875            )
876        return result
877
878    async def recently_played_tracks(
879        self,
880        limit: int,
881        played_after_timestamp: int,
882        userid: str | None = None,
883    ) -> list[RecentPlayedTrack]:
884        """
885        Return recently played, fully played tracks with their recorded artists, newest first.
886
887        :param limit: Maximum number of plays to return.
888        :param played_after_timestamp: Only include plays at or after this epoch-seconds timestamp.
889        :param userid: Restrict to this user (defaults to the current session user, else all users).
890        """
891        query = (
892            f"SELECT item_id, provider, name, image, artists FROM {DB_TABLE_PLAYLOG} "
893            "WHERE media_type = 'track' AND fully_played = 1 "
894            "AND timestamp >= :played_after_timestamp "
895        )
896        params: dict[str, Any] = {"played_after_timestamp": played_after_timestamp}
897        if userid:
898            query += "AND userid = :userid "
899            params["userid"] = userid
900        elif user := get_current_user():
901            query += "AND userid = :userid "
902            params["userid"] = user.user_id
903        query += "ORDER BY timestamp DESC"
904        db_rows = await self.mass.music.database.get_rows_from_query(
905            query, params=params, limit=limit
906        )
907        available_providers = ("library", *get_global_cache_value("available_providers", []))
908        return [
909            RecentPlayedTrack(
910                track=ItemMapping.from_dict(
911                    {
912                        "item_id": db_row["item_id"],
913                        "provider": db_row["provider"],
914                        "media_type": "track",
915                        "name": db_row["name"],
916                        "image": json_loads(db_row["image"]) if db_row["image"] else None,
917                        "available": db_row["provider"] in available_providers,
918                    }
919                ),
920                artists=[ItemMapping.from_dict(artist) for artist in json_loads(db_row["artists"])]
921                if db_row["artists"]
922                else [],
923            )
924            for db_row in db_rows
925        ]
926
927    @api_command("music/recently_added_tracks", required_scope=Scope.LIBRARY_READ)
928    async def recently_added_tracks(self, limit: int = 10) -> list[Track]:
929        """Return a list of the last added tracks."""
930        return await self.tracks.library_items(
931            limit=limit, order_by="timestamp_added_desc", summary=False
932        )
933
934    @api_command("music/in_progress_items", required_scope=Scope.LIBRARY_READ)
935    async def in_progress_items(
936        self, limit: int = 10, all_users: bool = False, providers: list[str] | None = None
937    ) -> list[ItemMapping]:
938        """
939        Return a list of the Audiobooks and PodcastEpisodes that are in progress.
940
941        :param limit: Maximum number of items to return.
942        :param all_users: If True, include in-progress items across all users, not just
943            the current session's user.
944        :param providers: Restrict results to items reachable through one of these provider
945            instance ids (OR semantics). None applies no filter; an explicit empty list
946            returns no items.
947        """
948        if providers is not None and not providers:
949            return []
950        available_providers = ("library", *self.get_active_provider_instances())
951        available_providers_str = "(" + ",".join(f'"{x}"' for x in available_providers) + ")"
952        params: dict[str, Any] = {}
953        requested_clause = ""
954        direct_requested_clause = ""
955        if providers is not None:
956            params["requested_providers"] = providers
957            requested_clause = " AND m.provider_instance IN :requested_providers"
958            direct_requested_clause = " AND p.provider IN :requested_providers"
959
960        # An audiobook can be part of the library, in contrast to podcast episodes.
961        # We then need to check the provider mappings table.
962        one_week_ago = int(utc_timestamp()) - (7 * 86400)
963        query = (
964            "SELECT p.item_id, p.media_type, p.name, p.image, p.provider "
965            f"FROM {DB_TABLE_PLAYLOG} p "
966            "WHERE p.media_type IN ('audiobook', 'podcast_episode') "
967            "AND p.fully_played = 0 "
968            "AND p.seconds_played > 0 "
969            f"AND (p.media_type != 'podcast_episode' OR p.timestamp >= {one_week_ago}) "
970        )
971        query += (
972            "AND ( "
973            "CASE WHEN p.provider = 'library' THEN "
974            f"EXISTS (SELECT 1 FROM {DB_TABLE_PROVIDER_MAPPINGS} m "
975            "WHERE m.item_id = p.item_id AND m.media_type = p.media_type "
976            "AND m.available = 1 "
977        )
978        if not all_users and (user := get_current_user()):
979            filter_for_str = available_providers_str
980            if user.provider_filter:
981                filter_for_str = "(" + ",".join(f'"{x}"' for x in user.provider_filter) + ")"
982            query += (
983                f"AND m.provider_instance IN {filter_for_str} "
984                f"AND m.provider_instance IN {available_providers_str}"
985                f"{requested_clause} "
986                ") "
987                f"ELSE (p.provider IN {filter_for_str} AND p.provider IN {available_providers_str}"
988                f"{direct_requested_clause})"
989                "END "
990                ") "
991                f"AND p.userid = '{user.user_id}' "
992            )
993        else:
994            # for a library item, we still have to verify via the provider mapping table
995            # that the provider is available
996            query += (
997                f"AND m.provider_instance IN {available_providers_str}"
998                f"{requested_clause} "
999                ") "
1000                f"ELSE p.provider IN {available_providers_str}"
1001                f"{direct_requested_clause} "
1002                "END "
1003                ") "
1004            )
1005        query += "ORDER BY timestamp DESC"
1006
1007        db_rows = await self.mass.music.database.get_rows_from_query(
1008            query, params=params or None, limit=limit
1009        )
1010        result: list[ItemMapping] = []
1011        for db_row in db_rows:
1012            provider = db_row["provider"]
1013            result.append(
1014                ItemMapping.from_dict(
1015                    {
1016                        "item_id": db_row["item_id"],
1017                        "provider": provider,
1018                        "media_type": db_row["media_type"],
1019                        "name": db_row["name"],
1020                        "image": json_loads(db_row["image"]) if db_row["image"] else None,
1021                        "available": provider in available_providers,
1022                    }
1023                )
1024            )
1025        return result
1026
1027    async def get_playlog_provider_item_ids(
1028        self, provider_instance_id: str, limit: int = 0, userid: str | None = None
1029    ) -> list[tuple[MediaType, str]]:
1030        """Return a list of MediaType and provider_item_id of items in playlog of provider."""
1031        # check if there is a provider user
1032        # this method is not available in the frontend, so no need to check for session users.
1033        user: User | None = None
1034        if userid:
1035            # userid overridden by parameter
1036            user = await self.mass.webserver.auth.get_user(userid)
1037        elif provider_user := await self._get_user_for_provider(provider_instance_id):
1038            # based on configured provider filter we can try to find a user
1039            user = provider_user
1040
1041        query = (
1042            f"SELECT * FROM {DB_TABLE_PLAYLOG} "
1043            "WHERE media_type in ('audiobook', 'podcast_episode') "
1044            f"AND provider in ('library','{provider_instance_id}')"
1045        )
1046
1047        if user:
1048            # NOTE: if no user was found, we will return playlog items for all users
1049            query += f" AND userid = '{user.user_id}'"
1050        db_rows = await self.mass.music.database.get_rows_from_query(query, limit=limit)
1051
1052        result: list[tuple[MediaType, str]] = []
1053        for db_row in db_rows:
1054            if db_row["provider"] == "library":
1055                # If the provider is library, we need to make sure that the item
1056                # is part of the passed provider_instance_id.
1057                # A podcast_episode cannot be in the provider_mappings
1058                # so these entries must be audiobooks.
1059                subquery = (
1060                    f"SELECT * FROM {DB_TABLE_PROVIDER_MAPPINGS} "
1061                    f"WHERE media_type = 'audiobook' AND item_id = {db_row['item_id']} "
1062                    f"AND provider_instance = '{provider_instance_id}'"
1063                )
1064                subrow = await self.mass.music.database.get_rows_from_query(subquery)
1065                if len(subrow) != 1:
1066                    continue
1067                result.append((MediaType.AUDIOBOOK, subrow[0]["provider_item_id"]))
1068                continue
1069            # non library - item id is provider_item_id
1070            result.append((MediaType(db_row["media_type"]), db_row["item_id"]))
1071
1072        return result
1073
1074    @api_command("music/item_by_uri", required_scope=Scope.LIBRARY_READ)
1075    async def get_item_by_uri(
1076        self, uri: str, allow_update_metadata: bool = False
1077    ) -> MediaItemType | BrowseFolder:
1078        """Fetch MediaItem by uri."""
1079        media_type, provider_instance_id_or_domain, item_id = await parse_uri(uri)
1080        return await self.get_item(
1081            media_type=media_type,
1082            item_id=item_id,
1083            provider_instance_id_or_domain=provider_instance_id_or_domain,
1084            allow_update_metadata=allow_update_metadata,
1085        )
1086
1087    @api_command("music/sound_effects", required_scope=Scope.LIBRARY_READ)
1088    async def sound_effects(self) -> list[SoundEffect]:
1089        """Return all sound effect items from providers supporting them."""
1090        providers = self._apply_user_provider_filter(
1091            self.mass.get_providers_supporting_feature(ProviderFeature.SOUND_EFFECTS)
1092        )
1093        results_per_provider: list[list[SoundEffect]] = await asyncio.gather(
1094            *[
1095                self._get_provider_sound_effects(cast("MusicProvider", provider))
1096                for provider in providers
1097            ]
1098        )
1099        return [item for sublist in results_per_provider for item in sublist]
1100
1101    @api_command("music/item", required_scope=Scope.LIBRARY_READ)
1102    async def get_item(
1103        self,
1104        media_type: MediaType,
1105        item_id: str,
1106        provider_instance_id_or_domain: str,
1107        allow_update_metadata: bool = True,
1108    ) -> MediaItemType | BrowseFolder:
1109        """Get single music item by id and media type."""
1110        if provider_instance_id_or_domain == "database":
1111            # backwards compatibility - to remove when 2.0 stable is released
1112            provider_instance_id_or_domain = "library"
1113        provider = self.mass.get_provider(provider_instance_id_or_domain)
1114        if media_type in (
1115            MediaType.TRACK,
1116            MediaType.RADIO,
1117            MediaType.SOUND_EFFECT,
1118            MediaType.UNKNOWN,  # e.g. plain (HA) URLs, see helpers/uri.py
1119        ) and (
1120            provider_instance_id_or_domain == "builtin"
1121            or (provider and provider.domain == "builtin")
1122        ):
1123            # handle special case of 'builtin' MusicProvider which allows us to play regular url's
1124            builtin_prov = cast("BuiltinProvider", provider or self.mass.get_provider("builtin"))
1125            if media_type == MediaType.RADIO:
1126                # a radio station must stay a radio station, also when the stream
1127                # reports a duration or carries no ICY name
1128                return await builtin_prov.get_radio(item_id)
1129            if media_type == MediaType.TRACK:
1130                # and a track must stay a track, also when the stream carries an
1131                # ICY name or reports no duration
1132                return await builtin_prov.get_track(item_id)
1133            return await builtin_prov.parse_item(item_id, requested_media_type=media_type)
1134        if media_type == MediaType.PODCAST_EPISODE:
1135            # special case for podcast episodes
1136            return await self.podcasts.episode(item_id, provider_instance_id_or_domain)
1137        if media_type == MediaType.FOLDER:
1138            # special case for folders
1139            return BrowseFolder(
1140                item_id=item_id,
1141                provider=provider_instance_id_or_domain,
1142                name=item_id,
1143            )
1144        if media_type == MediaType.AUDIO_SOURCE:
1145            # AudioSources are not library-backed; resolve them through the owning
1146            # plugin provider's get_audio_sources() catalog. Returning the live
1147            # MediaItem lets play_media create a queue item the standard way.
1148            prov = self.mass.get_provider(provider_instance_id_or_domain)
1149            if isinstance(prov, PluginProvider):
1150                for source in await prov.get_audio_sources():
1151                    if source.item_id == item_id:
1152                        return source
1153            raise MediaNotFoundError(
1154                f"AudioSource {provider_instance_id_or_domain}/{item_id} not found"
1155            )
1156        if media_type == MediaType.SOUND_EFFECT:
1157            # Sound effects are not library-backed; resolve them live from the
1158            # owning music provider. Returning the live MediaItem lets play_media
1159            # create a queue item the standard way.
1160            prov = self.mass.get_provider(provider_instance_id_or_domain)
1161            if isinstance(prov, MusicProvider) and (
1162                ProviderFeature.SOUND_EFFECTS in prov.supported_features
1163            ):
1164                return await prov.get_sound_effect(item_id)
1165            raise MediaNotFoundError(
1166                f"SoundEffect {provider_instance_id_or_domain}/{item_id} not found"
1167            )
1168        if media_type == MediaType.COLLECTION:
1169            ctrl = self.get_controller_for_collection(item_id)
1170            return await ctrl.get_collection(item_id)
1171        ctrl = self.get_controller(media_type)
1172        return await ctrl.get(
1173            item_id=item_id,
1174            provider_instance_id_or_domain=provider_instance_id_or_domain,
1175            allow_update_metadata=allow_update_metadata,
1176        )
1177
1178    @api_command("music/get_library_item", required_scope=Scope.LIBRARY_READ)
1179    async def get_library_item_by_prov_id(
1180        self,
1181        media_type: MediaType,
1182        item_id: str,
1183        provider_instance_id_or_domain: str,
1184    ) -> MediaItemType | None:
1185        """Get the library item for the given provider item, if present."""
1186        ctrl = self.get_controller(media_type)
1187        return await ctrl.get_library_item_by_prov_id(
1188            item_id=item_id,
1189            provider_instance_id_or_domain=provider_instance_id_or_domain,
1190        )
1191
1192    @api_command("music/favorites/add_item", required_scope=Scope.LIBRARY_WRITE)
1193    async def add_item_to_favorites(
1194        self,
1195        item: str | MediaItemType | ItemMapping,
1196    ) -> None:
1197        """Add an item to the favorites."""
1198        if isinstance(item, str):
1199            # Inspect the URI's media_type first so a stale audio-source URI
1200            # whose plugin is unloaded gives the honest rejection error
1201            # instead of bubbling MediaNotFoundError from get_item_by_uri.
1202            try:
1203                uri_media_type, _, _ = await parse_uri(item)
1204            except InvalidProviderURI, InvalidProviderID:
1205                uri_media_type = None
1206            if uri_media_type in (MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
1207                raise UnsupportedFeaturedException(
1208                    f"{uri_media_type.value} items can not be favorites"
1209                )
1210            # a favorite URI always resolves to a media item, never a BrowseFolder
1211            item = cast("MediaItemType", await self.get_item_by_uri(item))
1212        if item.media_type in (MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
1213            # AudioSources and SoundEffects are live provider content (existence
1214            # depends on a loaded provider) and have no stable library identity,
1215            # so they can not be persisted as favorites.
1216            raise UnsupportedFeaturedException(
1217                f"{item.media_type.value} items can not be favorites"
1218            )
1219        # make sure we have a full library item
1220        # a favorite must always be in the library
1221        full_item = cast(
1222            "MediaItemType",
1223            await self.get_item(
1224                item.media_type,
1225                item.item_id,
1226                item.provider,
1227            ),
1228        )
1229        if full_item.provider != "library":
1230            full_item = await self.add_item_to_library(full_item)
1231        # set favorite in library db
1232        ctrl = self.get_controller(item.media_type)
1233        await ctrl.set_favorite(
1234            full_item.item_id,
1235            True,
1236        )
1237        # forward to provider(s) if needed
1238        for prov_mapping in full_item.provider_mappings:
1239            provider = self.mass.get_provider(
1240                prov_mapping.provider_instance, provider_type=MusicProvider
1241            )
1242            if not provider or not self.library_favorites_edit_supported(
1243                provider, full_item.media_type
1244            ):
1245                continue
1246            await provider.set_favorite(prov_mapping.item_id, full_item.media_type, True)
1247
1248    @api_command("music/favorites/remove_item", required_scope=Scope.LIBRARY_WRITE)
1249    async def remove_item_from_favorites(
1250        self,
1251        media_type: MediaType,
1252        library_item_id: str | int,
1253    ) -> None:
1254        """Remove (library) item from the favorites."""
1255        ctrl = self.get_controller(media_type)
1256        await ctrl.set_favorite(
1257            library_item_id,
1258            False,
1259        )
1260        # forward to provider(s) if needed
1261        full_item = await ctrl.get_library_item(library_item_id)
1262        for prov_mapping in full_item.provider_mappings:
1263            provider = self.mass.get_provider(
1264                prov_mapping.provider_instance, provider_type=MusicProvider
1265            )
1266            if not provider or not self.library_favorites_edit_supported(
1267                provider, full_item.media_type
1268            ):
1269                continue
1270            self.mass.create_task(provider.set_favorite(prov_mapping.item_id, media_type, False))
1271
1272    @api_command("music/library/remove_item", required_scope=Scope.LIBRARY_WRITE)
1273    async def remove_item_from_library(
1274        self, media_type: MediaType, library_item_id: str | int, recursive: bool = True
1275    ) -> None:
1276        """
1277        Remove item from the library.
1278
1279        Destructive! Will remove the item and all dependants.
1280        """
1281        ctrl = self.get_controller(media_type)
1282        # remove from provider(s) library
1283        full_item = await ctrl.get_library_item(library_item_id)
1284        for prov_mapping in full_item.provider_mappings:
1285            if not prov_mapping.in_library:
1286                continue
1287            provider = self.mass.get_provider(
1288                prov_mapping.provider_instance, provider_type=MusicProvider
1289            )
1290            if not provider or not self.library_edit_supported(provider, full_item.media_type):
1291                continue
1292            if not self.library_sync_back_enabled(provider, full_item.media_type):
1293                continue
1294            prov_mapping.in_library = False
1295            self.mass.create_task(provider.library_remove(prov_mapping.item_id, media_type))
1296        # remove from library
1297        await ctrl.remove_item_from_library(library_item_id, recursive)
1298
1299    @api_command("music/library/add_item", required_scope=Scope.LIBRARY_WRITE)
1300    async def add_item_to_library(
1301        self, item: str | MediaItemType | ItemMapping, overwrite_existing: bool = False
1302    ) -> MediaItemType:
1303        """Add item (uri or mediaitem) to the library."""
1304        if isinstance(item, ItemMapping):
1305            # handle browse results that are returned as ItemMappings
1306            # uri is always populated post-init, so it is never None here
1307            item = cast("str", item.uri)
1308        # ensure we have a full item
1309        if isinstance(item, str):
1310            # Inspect the URI's media_type first so a stale audio-source URI
1311            # whose plugin is unloaded gives the honest rejection error
1312            # instead of bubbling MediaNotFoundError from get_item_by_uri.
1313            # Mirrors the same guard in add_item_to_favorites.
1314            try:
1315                uri_media_type, _, _ = await parse_uri(item)
1316            except InvalidProviderURI, InvalidProviderID:
1317                uri_media_type = None
1318            if uri_media_type in (MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
1319                raise UnsupportedFeaturedException(
1320                    f"{uri_media_type.value} items can not be library items"
1321                )
1322            full_item = await self.get_item_by_uri(item)
1323        # For builtin provider (manual URLs), use the provided item directly
1324        # to preserve custom modifications (name, images, etc.)
1325        # For other providers, fetch fresh to ensure data validity
1326        elif item.provider == "builtin":
1327            full_item = item
1328        else:
1329            full_item = await self.get_item(
1330                item.media_type,
1331                item.item_id,
1332                item.provider,
1333            )
1334        full_item = cast("MediaItemType", full_item)
1335        if full_item.media_type in (MediaType.AUDIO_SOURCE, MediaType.SOUND_EFFECT):
1336            # AudioSources and SoundEffects are live provider content (existence
1337            # depends on a loaded provider) and have no stable library identity,
1338            # so they can not be persisted as library items.
1339            raise UnsupportedFeaturedException(
1340                f"{full_item.media_type.value} items can not be library items"
1341            )
1342        # add to provider(s) library first
1343        for prov_mapping in full_item.provider_mappings:
1344            # we optimistically set in library to True to prevent items
1345            # from disappearing when the provider doesn't support library edit
1346            # or 2-way sync is disabled.
1347            prov_mapping.in_library = True
1348            provider = self.mass.get_provider(
1349                prov_mapping.provider_instance, provider_type=MusicProvider
1350            )
1351            if not provider or not self.library_edit_supported(provider, full_item.media_type):
1352                continue
1353            if not self.library_sync_back_enabled(provider, full_item.media_type):
1354                continue
1355            prov_item = deepcopy(full_item) if full_item.provider == "library" else full_item
1356            prov_item.provider = prov_mapping.provider_instance
1357            prov_item.item_id = prov_mapping.item_id
1358            self.mass.create_task(provider.library_add(prov_item))
1359        # add (or overwrite) to library
1360        ctrl = self.get_controller(full_item.media_type)
1361        # ctrl is chosen by media_type, so it matches full_item's runtime type
1362        library_item = await cast("MediaControllerBase[MediaItemType]", ctrl).add_item_to_library(
1363            full_item, overwrite_existing
1364        )
1365        # optionally import all album tracks into the library, mirroring the behavior
1366        # of the library sync (which only triggers on a (scheduled) full sync run)
1367        if full_item.media_type == MediaType.ALBUM:
1368            self._import_album_tracks_if_enabled(cast("Album", library_item))
1369        # perform full metadata scan
1370        await self.mass.metadata.update_metadata(library_item, overwrite_existing)
1371        return library_item
1372
1373    @api_command("music/refresh_item", required_scope=Scope.LIBRARY_MANAGE)
1374    async def refresh_item(  # noqa: PLR0915
1375        self,
1376        media_item: str | MediaItemType,
1377    ) -> MediaItemType | None:
1378        """Try to refresh a mediaitem by requesting it's full object or search for substitutes."""
1379        if isinstance(media_item, str):
1380            # media item uri given
1381            # a refresh URI always resolves to a media item, never a BrowseFolder
1382            media_item = cast("MediaItemType", await self.get_item_by_uri(media_item))
1383
1384        media_type = media_item.media_type
1385        ctrl = self.get_controller(media_type)
1386
1387        # genres are library-only items with no provider mappings, nothing to refresh
1388        if media_type == MediaType.GENRE:
1389            return media_item
1390
1391        library_id = media_item.item_id if media_item.provider == "library" else None
1392
1393        # cache in_library state before the provider fetch overwrites media_item
1394        in_library_cache: dict[tuple[str, str], bool] = {}
1395        for m in media_item.provider_mappings:
1396            if m.in_library is not None:
1397                in_library_cache[(m.provider_instance, m.item_id)] = m.in_library
1398
1399        available_providers = get_global_cache_value("available_providers")
1400        if TYPE_CHECKING:
1401            available_providers = cast("set[str]", available_providers)
1402
1403        # fetch the first (available) provider item
1404        for prov_mapping in sorted(
1405            media_item.provider_mappings, key=lambda x: x.priority, reverse=True
1406        ):
1407            if not self.mass.get_provider(prov_mapping.provider_instance):
1408                # ignore unavailable providers
1409                continue
1410            with suppress(MediaNotFoundError):
1411                media_item = await ctrl.get_provider_item(
1412                    prov_mapping.item_id,
1413                    prov_mapping.provider_instance,
1414                    force_refresh=True,
1415                )
1416                provider = media_item.provider
1417                item_id = media_item.item_id
1418                break
1419        else:
1420            # try to find a substitute using search
1421            searchresult = await self.search(media_item.name, [media_item.media_type], 20)
1422            result: Sequence[MediaItemType | ItemMapping]
1423            if media_item.media_type == MediaType.ARTIST:
1424                result = searchresult.artists
1425            elif media_item.media_type == MediaType.ALBUM:
1426                result = searchresult.albums
1427            elif media_item.media_type == MediaType.TRACK:
1428                result = searchresult.tracks
1429            elif media_item.media_type == MediaType.PLAYLIST:
1430                result = searchresult.playlists
1431            elif media_item.media_type == MediaType.AUDIOBOOK:
1432                result = searchresult.audiobooks
1433            elif media_item.media_type == MediaType.PODCAST:
1434                result = searchresult.podcasts
1435            else:
1436                result = searchresult.radio
1437            for item in result:
1438                if item == media_item or item.provider == "library":
1439                    continue
1440                if item.available:
1441                    provider = item.provider
1442                    item_id = item.item_id
1443                    break
1444            else:
1445                # raise if we didn't find a substitute
1446                raise MediaNotFoundError(f"Could not find a substitute for {media_item.name}")
1447        # fetch full (provider) item
1448        media_item = await ctrl.get_provider_item(item_id, provider, force_refresh=True)
1449        # update library item if needed (including refresh of the metadata etc.)
1450        if library_id is None:
1451            return media_item
1452        # restore in_library state from before the refresh
1453        for prov_mapping in media_item.provider_mappings:
1454            key = (prov_mapping.provider_instance, prov_mapping.item_id)
1455            if prov_mapping.in_library is None and key in in_library_cache:
1456                prov_mapping.in_library = in_library_cache[key]
1457        # ctrl is chosen by media_type, so it matches media_item's runtime type
1458        library_item = await cast(
1459            "MediaControllerBase[MediaItemType]", ctrl
1460        ).update_item_in_library(library_id, media_item, overwrite=True)
1461        if library_item.media_type == MediaType.ALBUM:
1462            # update (local) album tracks
1463            for album_track in await self.albums.tracks(
1464                library_item.item_id, library_item.provider, True
1465            ):
1466                for prov_mapping in album_track.provider_mappings:
1467                    if not (prov := self.mass.get_provider(prov_mapping.provider_instance)):
1468                        continue
1469                    if not isinstance(prov, MusicProvider):
1470                        continue
1471                    if prov.is_streaming_provider:
1472                        continue
1473                    with suppress(MediaNotFoundError):
1474                        prov_track = await prov.get_track(prov_mapping.item_id)
1475                        await self.mass.music.tracks.update_item_in_library(
1476                            album_track.item_id, prov_track
1477                        )
1478        await cast("MediaControllerBase[MediaItemType]", ctrl).match_providers(library_item)
1479        await self.mass.metadata.update_metadata(library_item, force_refresh=True)
1480        return library_item
1481
1482    @api_command("music/mark_played", required_scope=Scope.LIBRARY_WRITE)
1483    async def mark_item_played(
1484        self,
1485        media_item: MediaItemType | ItemMapping,
1486        fully_played: bool = True,
1487        seconds_played: int | None = None,
1488        is_playing: bool = False,
1489        userid: str | None = None,
1490        queue_id: str | None = None,
1491        user_initiated: bool = True,
1492        skip_artist_ids: list[str] | None = None,
1493        playback_speed: float | None = None,
1494    ) -> None:
1495        """
1496        Mark item as played in playlog.
1497
1498        :param media_item: The media item to mark as played.
1499        :param fully_played: If True, mark the item as fully played.
1500        :param seconds_played: The number of seconds played.
1501        :param is_playing: If True, the item is currently playing.
1502        :param userid: The user ID to mark the item as played for (instead of the current user).
1503        :param queue_id: The queue ID where the item was played.
1504        :param user_initiated: If True, the playback was initiated by the user (e.g. enqueued).
1505            Sticky once set: a later report can promote a playlog row to user-initiated but
1506            never demote it, so a writer reporting playback it did not itself initiate
1507            (e.g. a provider sync) must pass False.
1508        :param skip_artist_ids: Library artist ids to skip when crediting an album's artists.
1509        :param playback_speed: The current playback speed to persist (audiobooks/podcasts).
1510            If None, any previously stored speed for the item is preserved.
1511        """
1512        timestamp = utc_timestamp()
1513        # we deliberately skip one-off items: sound effects and live inputs whoever owns
1514        # them, and everything the builtin provider plays (except playlists) is a one-off url
1515        if media_item.media_type in (MediaType.SOUND_EFFECT, MediaType.AUDIO_SOURCE):
1516            return
1517        if (
1518            media_item.provider.startswith("builtin")
1519            and media_item.media_type != MediaType.PLAYLIST
1520        ):
1521            return
1522        # the playlog is keyed by the identity the caller referenced, not the resolved one
1523        reference = media_item
1524        media_item = await self._resolve_playlog_item(media_item)
1525
1526        params = {
1527            "item_id": reference.item_id,
1528            "provider": reference.provider,
1529            "media_type": media_item.media_type.value,
1530            "name": media_item.name,
1531            "image": serialize_to_json(media_item.image.to_dict()) if media_item.image else None,
1532            # store lightweight artist mappings so playlog rows can later be matched or
1533            # resolved by artist without an extra provider lookup
1534            "artists": serialize_to_json(
1535                [ItemMapping.from_item(artist).to_dict() for artist in artists]
1536            )
1537            if (artists := getattr(media_item, "artists", None))
1538            else None,
1539            "fully_played": fully_played,
1540            "seconds_played": seconds_played,
1541            "timestamp": timestamp,
1542            "queue_id": queue_id,
1543            "user_initiated": user_initiated,
1544        }
1545        # try to figure out the user that triggered the action
1546        user: User | None = None
1547        if userid:
1548            # userid overridden by parameter
1549            user = await self.mass.webserver.auth.get_user(userid)
1550        elif session_user := get_current_user():
1551            # this is the active session user that triggered the action
1552            user = session_user
1553        elif provider_user := await self._get_user_for_provider(media_item.provider_mappings):
1554            # based on configured provider filter we can try to find a user
1555            user = provider_user
1556
1557        # update generic playlog table (when not playing)
1558        if not is_playing:
1559            if user:
1560                user_ids = [user.user_id]
1561            else:
1562                # NOTE: if no user was found, we will alter the playlog for all users
1563                user_ids = [user.user_id for user in await self.mass.webserver.auth.list_users()]
1564            # Leaving the speed out keeps whatever is already stored for this item/user
1565            # (a provider sync reporting progress has no speed to offer), and falls back to
1566            # the column default of 1.0 for a brand new row.
1567            if playback_speed is not None:
1568                params["playback_speed"] = playback_speed
1569            for user_id in user_ids:
1570                params["userid"] = user_id
1571                await self._upsert_playlog(params)
1572
1573        # Set seconds_played in accordance with fully_played, if the media_item has
1574        # a duration, before it is forwarded to music_providers
1575        if seconds_played is None:
1576            seconds_played = 0
1577            if (
1578                fully_played
1579                and not isinstance(
1580                    media_item, Album | Artist | Genre | Playlist | Podcast | MediaCollection
1581                )
1582                and isinstance(media_item.duration, int)  # for Radio duration can be None
1583            ):
1584                seconds_played = media_item.duration
1585
1586        # forward to provider(s) to sync resume state (e.g. for audiobooks)
1587        for prov_mapping in media_item.provider_mappings:
1588            if (
1589                user
1590                and user.provider_filter
1591                and prov_mapping.provider_instance not in user.provider_filter
1592            ):
1593                continue
1594            if music_prov := self.mass.get_provider(prov_mapping.provider_instance):
1595                if music_prov.type != ProviderType.MUSIC:
1596                    continue
1597                music_prov = cast("MusicProvider", music_prov)
1598                self.mass.create_task(
1599                    music_prov.on_played(
1600                        media_type=media_item.media_type,
1601                        prov_item_id=prov_mapping.item_id,
1602                        fully_played=fully_played,
1603                        position=seconds_played,
1604                        media_item=media_item,
1605                        is_playing=is_playing,
1606                    )
1607                )
1608
1609        # also update playcount in library table (if fully played)
1610        if not fully_played or is_playing:
1611            return
1612        try:
1613            ctrl = self.get_controller(media_item.media_type)
1614        except NotImplementedError:
1615            # skip non-library media types (e.g. AudioSource plugin sources)
1616            return
1617        db_item = await ctrl.get_library_item_by_prov_id(media_item.item_id, media_item.provider)
1618        if db_item:
1619            await self.database.execute(
1620                f"UPDATE {ctrl.db_table} SET play_count = play_count + 1, "
1621                f"last_played = {timestamp} WHERE item_id = {db_item.item_id}"
1622            )
1623            if isinstance(media_item, Track):
1624                self.logger.debug("Credited play for track '%s'", media_item.name)
1625        if isinstance(media_item, Track | Album):
1626            await self._credit_artist_plays(
1627                media_item.artists,
1628                timestamp=timestamp,
1629                user_ids=user_ids,
1630                queue_id=queue_id,
1631                skip_ids=set(skip_artist_ids or ()),
1632            )
1633        if isinstance(media_item, PodcastEpisode) and media_item.podcast:
1634            await self._credit_podcast_play(
1635                media_item.podcast,
1636                timestamp=timestamp,
1637                user_ids=user_ids,
1638                queue_id=queue_id,
1639            )
1640        await self.database.commit()
1641
1642    async def resolve_library_artist_ids(self, artists: Iterable[Artist | ItemMapping]) -> set[str]:
1643        """Resolve the given artist references to their library item ids (when present)."""
1644        ids: set[str] = set()
1645        for artist in artists:
1646            db_artist = await self.artists.get_library_item_by_prov_id(
1647                artist.item_id, artist.provider
1648            )
1649            if db_artist is not None:
1650                ids.add(db_artist.item_id)
1651        return ids
1652
1653    @api_command("music/mark_unplayed", required_scope=Scope.LIBRARY_WRITE)
1654    async def mark_item_unplayed(
1655        self,
1656        media_item: MediaItemType | ItemMapping,
1657        userid: str | None = None,
1658    ) -> None:
1659        """
1660        Mark item as unplayed in playlog.
1661
1662        :param media_item: The media item to mark as unplayed.
1663        :param all_users: If True, mark the item as unplayed for all users.
1664        :param userid: The user ID to mark the item as unplayed for (instead of the current user).
1665        """
1666        # the playlog is keyed by the identity the caller referenced, not the resolved one
1667        reference = media_item
1668        media_item = await self._resolve_playlog_item(media_item)
1669        params = {
1670            "item_id": reference.item_id,
1671            "provider": reference.provider,
1672            "media_type": media_item.media_type.value,
1673        }
1674        # try to figure out the user that triggered the action
1675        user: User | None = None
1676        if userid:
1677            # userid overridden by parameter
1678            user = await self.mass.webserver.auth.get_user(userid)
1679        elif session_user := get_current_user():
1680            # this is the active session user that triggered the action
1681            user = session_user
1682        elif provider_user := await self._get_user_for_provider(media_item.provider_mappings):
1683            # based on configured provider filter we can try to find a user
1684            user = provider_user
1685
1686        if user:
1687            user_ids = [user.user_id]
1688        else:
1689            # NOTE: if no user was found, we will alter the playlog for all users
1690            user_ids = [user.user_id for user in await self.mass.webserver.auth.list_users()]
1691        for user_id in user_ids:
1692            params["userid"] = user_id
1693            await self.database.delete(DB_TABLE_PLAYLOG, params)
1694
1695        # forward to provider(s) to sync resume state (e.g. for audiobooks)
1696        for prov_mapping in media_item.provider_mappings:
1697            if (
1698                user
1699                and user.provider_filter
1700                and prov_mapping.provider_instance not in user.provider_filter
1701            ):
1702                continue
1703            if music_prov := self.mass.get_provider(prov_mapping.provider_instance):
1704                if music_prov.type != ProviderType.MUSIC:
1705                    continue
1706                music_prov = cast("MusicProvider", music_prov)
1707                self.mass.create_task(
1708                    music_prov.on_played(
1709                        media_type=media_item.media_type,
1710                        prov_item_id=prov_mapping.item_id,
1711                        fully_played=False,
1712                        position=0,
1713                        media_item=media_item,
1714                    )
1715                )
1716        # also update playcount in library table
1717        ctrl = self.get_controller(media_item.media_type)
1718        db_item = await ctrl.get_library_item_by_prov_id(media_item.item_id, media_item.provider)
1719        if db_item:
1720            await self.database.execute(
1721                f"UPDATE {ctrl.db_table} SET play_count = play_count - 1, "
1722                f"last_played = 0 WHERE item_id = {db_item.item_id}"
1723            )
1724            await self.database.commit()
1725
1726    @api_command("music/track_by_name", required_scope=Scope.LIBRARY_READ)
1727    async def get_track_by_name(
1728        self,
1729        track_name: str,
1730        artist_name: str | None = None,
1731        album_name: str | None = None,
1732        track_version: str | None = None,
1733    ) -> Track | None:
1734        """Get a track by its name, optionally with artist and album."""
1735        if track_version is None:
1736            track_name, version = parse_title_and_version(track_name)
1737        search_query = f"{artist_name} - {track_name}" if artist_name else track_name
1738        search_result = await self.mass.music.search(
1739            search_query=search_query,
1740            media_types=[MediaType.TRACK],
1741        )
1742        for allow_item_mapping in (False, True):
1743            for search_track in search_result.tracks:
1744                if not allow_item_mapping and not isinstance(search_track, Track):
1745                    continue
1746                if not compare_strings(track_name, search_track.name):
1747                    continue
1748                if not compare_version(version, search_track.version):
1749                    continue
1750                # check optional artist(s)
1751                if artist_name and isinstance(search_track, Track):
1752                    for artist in search_track.artists:
1753                        if compare_strings(artist_name, artist.name, False):
1754                            break
1755                    else:
1756                        # no artist match found: abort
1757                        continue
1758                # check optional album
1759                if album_name and isinstance(search_track, Track):
1760                    track_album = search_track.album
1761                    # a track without album info can never match a requested album
1762                    if track_album is None or not compare_strings(
1763                        album_name, track_album.name, False
1764                    ):
1765                        # no album match found: abort
1766                        continue
1767                # if we reach this, we found a match
1768                if not isinstance(search_track, Track):
1769                    # ensure we return an actual Track object
1770                    return await self.mass.music.tracks.get(
1771                        item_id=search_track.item_id,
1772                        provider_instance_id_or_domain=search_track.provider,
1773                    )
1774                return search_track
1775
1776        # try to handle case where something is appended to the title
1777        for splitter in ("•", "-", "|", "(", "["):
1778            if splitter in track_name:
1779                return await self.get_track_by_name(
1780                    track_name=track_name.split(splitter)[0].strip(),
1781                    artist_name=artist_name,
1782                    album_name=None,
1783                    track_version=track_version,
1784                )
1785        # try to handle case where multiple artists are given as single string
1786        if artist_name and (artists := split_artists(artist_name, True)) and len(artists) > 1:
1787            for single_artist in artists:
1788                return await self.get_track_by_name(
1789                    track_name=track_name,
1790                    artist_name=single_artist.split(splitter)[0].strip(),
1791                    album_name=None,
1792                    track_version=track_version,
1793                )
1794        # allow non-exact album match as fallback
1795        if album_name:
1796            return await self.get_track_by_name(
1797                track_name=track_name,
1798                artist_name=artist_name,
1799                album_name=None,
1800                track_version=track_version,
1801            )
1802        # no match found
1803        return None
1804
1805    async def get_resume_position(
1806        self, media_item: Audiobook | PodcastEpisode, userid: str | None = None
1807    ) -> tuple[bool, int]:
1808        """
1809        Get progress (resume point) details for the given audiobook or episode.
1810
1811        This is a separate call to ensure the resume position is always up-to-date
1812        and because many providers have this info present on a dedicated endpoint.
1813
1814        Will be called right before playback starts to ensure the resume position is correct.
1815
1816        Returns a boolean with the fully_played status
1817        and an integer with the resume position in ms.
1818        """
1819        provider_fully_played = False
1820        provider_position_ms = 0
1821        provider_timestamp: datetime | None = None
1822
1823        user: User | None = None
1824        if userid:
1825            # userid overridden by parameter
1826            user = await self.mass.webserver.auth.get_user(userid)
1827        elif session_user := get_current_user():
1828            # this is the active session user that triggered the action
1829            user = session_user
1830        elif provider_user := await self._get_user_for_provider(media_item.provider_mappings):
1831            # based on configured provider filter we can try to find a user
1832            user = provider_user
1833
1834        provider_instances = {x.provider_instance for x in media_item.provider_mappings}
1835        if user and user.provider_filter:
1836            # only if the user has provider filters configured
1837            # otherwise we allow all providers
1838            preferred_provider_instances = provider_instances.intersection(user.provider_filter)
1839        else:
1840            preferred_provider_instances = provider_instances
1841
1842        preferred_providers = [
1843            x
1844            for x in media_item.provider_mappings
1845            if x.provider_instance in preferred_provider_instances
1846        ]
1847
1848        # Try to get position from providers
1849        for prov_mapping in preferred_providers:
1850            if not (
1851                provider := self.mass.get_provider(
1852                    prov_mapping.provider_instance, provider_type=MusicProvider
1853                )
1854            ):
1855                continue
1856            with suppress(NotImplementedError):
1857                (
1858                    provider_fully_played,
1859                    provider_position_ms,
1860                    provider_timestamp,
1861                ) = await provider.get_resume_position(prov_mapping.item_id, media_item.media_type)
1862                break  # Use first provider that returns data
1863
1864        # Get MA's internal position from playlog
1865        ma_fully_played = False
1866        ma_position_ms = 0
1867        ma_timestamp = from_utc_timestamp(0)
1868        params = {
1869            "media_type": media_item.media_type.value,
1870            "item_id": media_item.item_id,
1871            "provider": media_item.provider,
1872        }
1873        if userid:
1874            params["userid"] = userid
1875        elif user:
1876            params["userid"] = user.user_id
1877        if db_entry := await self.database.get_row(DB_TABLE_PLAYLOG, params):
1878            ma_position_ms = db_entry["seconds_played"] * 1000 if db_entry["seconds_played"] else 0
1879            # fully_played is a nullable column; treat an unknown (NULL) value as not played
1880            ma_fully_played = parse_optional_bool(db_entry["fully_played"]) or False
1881            ma_timestamp = from_utc_timestamp(db_entry["timestamp"])
1882
1883        if provider_timestamp is not None and provider_timestamp > ma_timestamp:
1884            return provider_fully_played, provider_position_ms
1885        # Return the higher position to ensure users never lose progress
1886        if ma_position_ms >= provider_position_ms:
1887            return ma_fully_played, ma_position_ms
1888        return provider_fully_played, provider_position_ms
1889
1890    async def get_playback_speed(
1891        self, media_item: Audiobook | PodcastEpisode, userid: str | None = None
1892    ) -> float:
1893        """
1894        Get the stored playback speed for the given audiobook or podcast episode.
1895
1896        Returns 1.0 (normal speed) when no custom speed was stored for the item,
1897        or when no user can be determined to scope the lookup.
1898
1899        :param media_item: The audiobook or podcast episode to look up.
1900        :param userid: The user ID to look up the speed for (instead of the current user).
1901        """
1902        if not userid:
1903            if session_user := get_current_user():
1904                userid = session_user.user_id
1905            elif provider_user := await self._get_user_for_provider(media_item.provider_mappings):
1906                userid = provider_user.user_id
1907            else:
1908                # the speed is stored per user; without one we can't scope the lookup
1909                return 1.0
1910        db_entry = await self.database.get_row(
1911            DB_TABLE_PLAYLOG,
1912            {
1913                "item_id": media_item.item_id,
1914                "provider": media_item.provider,
1915                "media_type": media_item.media_type.value,
1916                "userid": userid,
1917            },
1918        )
1919        if db_entry and (stored_speed := db_entry["playback_speed"]) is not None:
1920            return float(stored_speed)
1921        return 1.0
1922
1923    def get_controller(
1924        self, media_type: MediaType
1925    ) -> (
1926        ArtistsController
1927        | AlbumsController
1928        | TracksController
1929        | RadioController
1930        | PlaylistController
1931        | AudiobooksController
1932        | PodcastsController
1933        | GenreController
1934    ):
1935        """Return controller for MediaType."""
1936        if media_type == MediaType.ARTIST:
1937            return self.artists
1938        if media_type == MediaType.ALBUM:
1939            return self.albums
1940        if media_type == MediaType.TRACK:
1941            return self.tracks
1942        if media_type == MediaType.RADIO:
1943            return self.radio
1944        if media_type == MediaType.PLAYLIST:
1945            return self.playlists
1946        if media_type == MediaType.AUDIOBOOK:
1947            return self.audiobooks
1948        if media_type == MediaType.PODCAST:
1949            return self.podcasts
1950        if media_type == MediaType.PODCAST_EPISODE:
1951            return self.podcasts
1952        if media_type == MediaType.GENRE:
1953            return self.genres
1954        raise NotImplementedError(
1955            f"No media controller available for media type: {media_type.value}"
1956        )
1957
1958    def get_controller_for_collection(
1959        self, item_id: str
1960    ) -> (
1961        ArtistsController
1962        | AlbumsController
1963        | TracksController
1964        | RadioController
1965        | PlaylistController
1966        | AudiobooksController
1967        | PodcastsController
1968        | GenreController
1969    ):
1970        """Return controller for MediaType."""
1971        media_type = get_collection_item_media_type_from_item_id(item_id)
1972        controller = self.get_controller(media_type)
1973        if not isinstance(controller, AudiobooksController):
1974            # currently only supported for audiobooks
1975            raise NotImplementedError(
1976                f"No media controller available for media type: {media_type.value}"
1977            )
1978        return controller
1979
1980    def get_provider_instances(
1981        self, domain: str, return_unavailable: bool = False
1982    ) -> list[MusicProvider]:
1983        """
1984        Return all provider instances for a given domain.
1985
1986        Note that this skips user filters so may only be called from internal code.
1987        """
1988        return cast(
1989            "list[MusicProvider]",
1990            self.mass.get_provider_instances(domain, return_unavailable, ProviderType.MUSIC),
1991        )
1992
1993    def get_unique_providers(self) -> list[str]:
1994        """
1995        Return all unique MusicProvider (instance or domain) ids.
1996
1997        This will return a set of provider instance ids but will only return
1998        a single instance_id per streaming provider domain.
1999
2000        Applies user provider filters (for non-admin users).
2001        """
2002        processed_domains: set[str] = set()
2003        # Get user provider filter if set
2004        user = get_current_user()
2005        user_provider_filter = user.provider_filter if user and user.provider_filter else None
2006        result: list[str] = []
2007        for provider in self.providers:
2008            if provider.is_streaming_provider and provider.domain in processed_domains:
2009                continue
2010            if user_provider_filter and provider.instance_id not in user_provider_filter:
2011                continue
2012            result.append(provider.instance_id)
2013            processed_domains.add(provider.domain)
2014        return result
2015
2016    def get_active_provider_instances(self) -> list[str]:
2017        """
2018        Return the instance ids of all currently loaded, available MusicProviders.
2019
2020        Unlike `get_unique_providers`, this keeps every instance of a streaming
2021        provider's domain instead of collapsing to one per domain, so a caller
2022        validating a specific requested provider instance id isn't shadowed by
2023        another instance of the same domain. Applies the current user's provider
2024        filter (via the `providers` property) and excludes providers that are
2025        loaded but not currently available.
2026        """
2027        return [provider.instance_id for provider in self.providers if provider.available]
2028
2029    async def cleanup_provider(self, provider_instance: str) -> None:
2030        """Cleanup provider records from the database."""
2031        deleted_providers = self.mass.config.get_raw_core_config_value(
2032            self.domain, CONF_DELETED_PROVIDERS, []
2033        )
2034        # we add the provider to this hidden config setting just to make sure that
2035        # we can survive this over a restart to make sure that entries are cleaned up
2036        if provider_instance not in deleted_providers:
2037            deleted_providers.append(provider_instance)
2038            self.mass.config.set_raw_core_config_value(
2039                self.domain, CONF_DELETED_PROVIDERS, deleted_providers
2040            )
2041            self.mass.config.save(True)
2042
2043        # always clear cache when a provider is removed
2044        await self.mass.cache.clear()
2045
2046        # cleanup media items from db matched to deleted provider
2047        self.logger.info(
2048            "Removing provider %s from library, this can take a a while...",
2049            provider_instance,
2050        )
2051        errors = 0
2052        # suppress the per-item MEDIA_ITEM_UPDATED events during this bulk removal so we
2053        # don't flood subscribers; they refresh once via the PROVIDERS_UPDATED event
2054        token = SUPPRESS_MEDIA_ITEM_UPDATES.set(True)
2055        try:
2056            for ctrl in (
2057                # order is important here to recursively cleanup bottom up
2058                self.mass.music.radio,
2059                self.mass.music.playlists,
2060                self.mass.music.tracks,
2061                self.mass.music.albums,
2062                self.mass.music.artists,
2063                self.mass.music.podcasts,
2064                self.mass.music.audiobooks,
2065                # run main controllers twice to rule out relations
2066                self.mass.music.tracks,
2067                self.mass.music.albums,
2068                self.mass.music.artists,
2069            ):
2070                query = (
2071                    f"SELECT item_id FROM {DB_TABLE_PROVIDER_MAPPINGS} "
2072                    "WHERE media_type = :media_type "
2073                    "AND provider_instance = :provider_instance"
2074                )
2075                params = {
2076                    "media_type": ctrl.media_type.value,
2077                    "provider_instance": provider_instance,
2078                }
2079                for db_row in await self.database.get_rows_from_query(query, params, limit=100000):
2080                    try:
2081                        await ctrl.remove_provider_mappings(db_row["item_id"], provider_instance)
2082                    except Exception as err:
2083                        # we dont want the whole removal process to stall on one item
2084                        # so in case of an unexpected error, we log and move on.
2085                        self.logger.warning(
2086                            "Error while removing %s: %s",
2087                            db_row["item_id"],
2088                            str(err),
2089                            exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
2090                        )
2091                        errors += 1
2092        finally:
2093            SUPPRESS_MEDIA_ITEM_UPDATES.reset(token)
2094
2095        # remove all orphaned items (not in provider mappings table anymore)
2096        query = (
2097            f"SELECT item_id FROM {DB_TABLE_PROVIDER_MAPPINGS} "
2098            f"WHERE provider_instance = '{provider_instance}'"
2099        )
2100        if remaining_items_count := await self.database.get_count_from_query(query):
2101            errors += remaining_items_count
2102
2103        # cleanup playlog table
2104        await self.mass.music.database.delete(
2105            DB_TABLE_PLAYLOG,
2106            {
2107                "provider": provider_instance,
2108            },
2109        )
2110
2111        if errors == 0:
2112            # cleanup successful, remove from the deleted_providers setting
2113            self.logger.info("Provider %s removed from library", provider_instance)
2114            deleted_providers.remove(provider_instance)
2115            self.mass.config.set_raw_core_config_value(
2116                self.domain, CONF_DELETED_PROVIDERS, deleted_providers
2117            )
2118        else:
2119            self.logger.warning(
2120                "Provider %s was not not fully removed from library", provider_instance
2121            )
2122
2123    async def schedule_provider_sync(self, provider_instance_id: str) -> None:
2124        """Schedule Library sync for given provider."""
2125        if not (
2126            provider := self.mass.get_provider(provider_instance_id, provider_type=MusicProvider)
2127        ):
2128            return
2129        await self.unschedule_provider_sync(provider.instance_id, clear_persisted_state=False)
2130        for media_type in MediaType:
2131            if not self.library_supported(provider, media_type):
2132                continue
2133            await self._schedule_provider_mediatype_sync(provider, media_type, True)
2134
2135    async def unschedule_provider_sync(
2136        self, provider_instance_id: str, clear_persisted_state: bool = True
2137    ) -> None:
2138        """
2139        Unschedule Library sync for given provider and wait for a running sync to stop.
2140
2141        Callers tear down provider state right after this (unloading the provider, or
2142        rescheduling its syncs), so all media types are cancelled first and then awaited
2143        together, keeping the bounded wait to one timeout instead of one per media type.
2144
2145        :param provider_instance_id: The provider instance id to unschedule.
2146        :param clear_persisted_state: Whether to remove persisted schedule state from config.
2147        """
2148        await asyncio.gather(
2149            *(
2150                self.mass.tasks.unregister_scheduled_task_and_wait(
2151                    self._get_sync_task_id(provider_instance_id, media_type),
2152                    clear_persisted_state=clear_persisted_state,
2153                )
2154                for media_type in MediaType
2155            )
2156        )
2157
2158    def get_provider_sync_schedule(
2159        self, provider_instance_id: str, media_type: MediaType
2160    ) -> TaskSchedule | None:
2161        """Return the effective schedule for a provider sync task, if any."""
2162        task_id = self._get_sync_task_id(provider_instance_id, media_type)
2163        with suppress(InvalidDataError):
2164            task = self.mass.tasks.get_task(task_id)
2165            return task.schedule
2166        if not (
2167            provider := self.mass.get_provider(provider_instance_id, provider_type=MusicProvider)
2168        ):
2169            return None
2170        if not self.library_supported(provider, media_type):
2171            return None
2172        return provider.get_default_library_sync_schedule(media_type)
2173
2174    def match_provider_instances(
2175        self,
2176        item: MediaItemType,
2177    ) -> bool:
2178        """Match all provider instances for the given item."""
2179        mappings_added = False
2180        for provider_mapping in list(item.provider_mappings):
2181            if provider_mapping.is_unique:
2182                # unique mapping, no need to map
2183                continue
2184            if not (provider := self.mass.get_provider(provider_mapping.provider_instance)):
2185                continue
2186            if not isinstance(provider, MusicProvider):
2187                continue
2188            if not provider.is_streaming_provider:
2189                continue
2190            provider_instances = self.get_provider_instances(
2191                provider.domain, return_unavailable=True
2192            )
2193            if len(provider_instances) <= 1:
2194                # only a single instance, no need to map
2195                continue
2196            for prov_instance in provider_instances:
2197                if prov_instance.instance_id == provider.instance_id:
2198                    continue
2199                if any(
2200                    pm.provider_instance == prov_instance.instance_id
2201                    for pm in item.provider_mappings
2202                ):
2203                    # mapping already exists
2204                    continue
2205                # create additional mapping for other provider instances of the same provider
2206                item.provider_mappings.add(
2207                    ProviderMapping(
2208                        item_id=provider_mapping.item_id,
2209                        provider_domain=provider.domain,
2210                        provider_instance=prov_instance.instance_id,
2211                        available=provider_mapping.available,
2212                        is_unique=provider_mapping.is_unique,
2213                        audio_format=provider_mapping.audio_format,
2214                        url=provider_mapping.url,
2215                        details=provider_mapping.details,
2216                        in_library=None,
2217                    )
2218                )
2219                mappings_added = True
2220        return mappings_added
2221
2222    @api_command("music/add_provider_mapping", required_scope=Scope.LIBRARY_MANAGE)
2223    async def add_provider_mapping(
2224        self, media_type: MediaType, db_id: str, mapping: ProviderMapping
2225    ) -> None:
2226        """Add provider mapping to the given library item."""
2227        ctrl = self.get_controller(media_type)
2228        await ctrl.add_provider_mappings(db_id, [mapping])
2229
2230    @api_command("music/remove_provider_mapping", required_scope=Scope.LIBRARY_MANAGE)
2231    async def remove_provider_mapping(
2232        self, media_type: MediaType, db_id: str, mapping: ProviderMapping
2233    ) -> None:
2234        """Remove provider mapping from the given library item."""
2235        ctrl = self.get_controller(media_type)
2236        await ctrl.remove_provider_mapping(db_id, mapping.provider_instance, mapping.item_id)
2237
2238    @api_command("music/match_providers", required_scope=Scope.LIBRARY_MANAGE)
2239    async def match_providers(self, media_type: MediaType, db_id: str) -> None:
2240        """Search for mappings on all providers for the given library item."""
2241        ctrl = self.get_controller(media_type)
2242        db_item = await ctrl.get_library_item(db_id)
2243        # ctrl is chosen by media_type, so it matches db_item's runtime type
2244        await cast("MediaControllerBase[MediaItemType]", ctrl).match_providers(db_item)
2245
2246    async def update_provider_mapping(
2247        self,
2248        media_type: MediaType,
2249        db_id: str | int,
2250        provider_instance_id: str,
2251        provider_item_id: str,
2252        *,
2253        available: bool | Any = UNSET,
2254        in_library: bool | Any = UNSET,
2255        is_unique: bool | None | Any = UNSET,
2256        url: str | None | Any = UNSET,
2257        details: str | None | Any = UNSET,
2258        audio_format: AudioFormat | Any = UNSET,
2259    ) -> None:
2260        """Update an existing provider mapping for a library item."""
2261        ctrl = self.get_controller(media_type)
2262        await ctrl.update_provider_mapping(
2263            item_id=db_id,
2264            provider_instance_id=provider_instance_id,
2265            provider_item_id=provider_item_id,
2266            available=available,
2267            in_library=in_library,
2268            is_unique=is_unique,
2269            url=url,
2270            details=details,
2271            audio_format=audio_format,
2272        )
2273
2274    def queue_provider_mapping_correction_task(self) -> BackgroundTask:
2275        """Queue the provider mapping correction as a managed background task."""
2276        self._register_provider_mapping_correction_task()
2277        return self.mass.tasks.run_task(PROVIDER_MAPPING_CORRECTION_TASK_ID)
2278
2279    async def correct_multi_instance_provider_mappings(self) -> None:
2280        """Correct provider mappings for multi-instance providers."""
2281        self.logger.debug("Correcting provider mappings for multi-instance providers...")
2282        multi_instance_providers: set[str] = set()
2283        for provider in self.providers:
2284            if len(self.get_provider_instances(provider.domain)) > 1:
2285                multi_instance_providers.add(provider.instance_id)
2286        if not multi_instance_providers:
2287            return  # no multi-instance providers found, nothing to do
2288
2289        for ctrl in (
2290            self.albums,
2291            self.artists,
2292            self.tracks,
2293            self.playlists,
2294            self.radio,
2295            self.audiobooks,
2296            self.podcasts,
2297        ):
2298            async for db_item in ctrl.iter_library_items(
2299                provider=list(multi_instance_providers), library_items_only=False
2300            ):
2301                if self.match_provider_instances(db_item):
2302                    # ctrl is the per-type controller, so it matches db_item's runtime type
2303                    await cast("MediaControllerBase[MediaItemType]", ctrl).update_item_in_library(
2304                        db_item.item_id, db_item
2305                    )
2306                # prevent overwhelming the event loop
2307                await asyncio.sleep(0.2)
2308        self.logger.debug("Provider mappings correction done")
2309
2310    def library_supported(self, provider: Provider, media_type: MediaType) -> bool:
2311        """Return whether the provider declares LIBRARY support for the given media type."""
2312        if provider.type != ProviderType.MUSIC:
2313            return False
2314        if (feature := LIBRARY_FEATURE_BY_MEDIA_TYPE.get(media_type)) is None:
2315            return False
2316        return provider.supports_feature(feature)
2317
2318    def library_edit_supported(self, provider: Provider, media_type: MediaType) -> bool:
2319        """Return whether the provider supports library add/remove for the given media type."""
2320        if provider.type != ProviderType.MUSIC:
2321            return False
2322        if media_type == MediaType.ARTIST:
2323            return provider.supports_feature(ProviderFeature.LIBRARY_ARTISTS_EDIT)
2324        if media_type == MediaType.ALBUM:
2325            return provider.supports_feature(ProviderFeature.LIBRARY_ALBUMS_EDIT)
2326        if media_type == MediaType.TRACK:
2327            return provider.supports_feature(ProviderFeature.LIBRARY_TRACKS_EDIT)
2328        if media_type == MediaType.PLAYLIST:
2329            return provider.supports_feature(ProviderFeature.LIBRARY_PLAYLISTS_EDIT)
2330        if media_type == MediaType.RADIO:
2331            return provider.supports_feature(ProviderFeature.LIBRARY_RADIOS_EDIT)
2332        if media_type == MediaType.AUDIOBOOK:
2333            return provider.supports_feature(ProviderFeature.LIBRARY_AUDIOBOOKS_EDIT)
2334        if media_type == MediaType.PODCAST:
2335            return provider.supports_feature(ProviderFeature.LIBRARY_PODCASTS_EDIT)
2336        return False
2337
2338    def library_favorites_edit_supported(self, provider: Provider, media_type: MediaType) -> bool:
2339        """Return whether the provider supports favorites add/remove for the given media type."""
2340        if provider.type != ProviderType.MUSIC:
2341            return False
2342        if media_type == MediaType.ARTIST:
2343            return provider.supports_feature(ProviderFeature.FAVORITE_ARTISTS_EDIT)
2344        if media_type == MediaType.ALBUM:
2345            return provider.supports_feature(ProviderFeature.FAVORITE_ALBUMS_EDIT)
2346        if media_type == MediaType.TRACK:
2347            return provider.supports_feature(ProviderFeature.FAVORITE_TRACKS_EDIT)
2348        if media_type == MediaType.PLAYLIST:
2349            return provider.supports_feature(ProviderFeature.FAVORITE_PLAYLISTS_EDIT)
2350        if media_type == MediaType.RADIO:
2351            return provider.supports_feature(ProviderFeature.FAVORITE_RADIOS_EDIT)
2352        if media_type == MediaType.AUDIOBOOK:
2353            return provider.supports_feature(ProviderFeature.FAVORITE_AUDIOBOOKS_EDIT)
2354        if media_type == MediaType.PODCAST:
2355            return provider.supports_feature(ProviderFeature.FAVORITE_PODCASTS_EDIT)
2356        return False
2357
2358    def library_sync_back_enabled(self, provider: Provider, media_type: MediaType) -> bool:
2359        """Return whether library sync back is enabled for the provider+media_type."""
2360        conf_value = provider.config.get_value(
2361            CONF_ENTRY_LIBRARY_SYNC_BACK.key, CONF_ENTRY_LIBRARY_SYNC_BACK.default_value
2362        )
2363        return bool(conf_value)
2364
2365    @api_command("music/item_by_name", required_scope=Scope.LIBRARY_READ, allow_impersonation=True)
2366    async def get_item_by_name(
2367        self,
2368        name: str,
2369        artist: str | None = None,
2370        album: str | None = None,
2371        media_type: MediaType | None = None,
2372    ) -> MediaItemType | ItemMapping | None:
2373        """Try to find a media item (such as a playlist) by name."""
2374        return await self._get_item_by_name(name, artist, album, media_type)
2375
2376    @api_command(
2377        "music/verify_item_uri", required_scope=Scope.LIBRARY_READ, allow_impersonation=True
2378    )
2379    async def verify_item_uri(self, uri: str) -> bool:
2380        """
2381        Verify whether a uri points to a valid, accessible item.
2382
2383        :param uri: The uri to verify.
2384        """
2385        return await self._handle_verify_item_uri(uri)
2386
2387    def _apply_user_provider_filter(
2388        self,
2389        providers: Iterable[ProviderInstanceType],
2390    ) -> list[ProviderInstanceType]:
2391        """Filter providers by the current user's music provider filter."""
2392        user = get_current_user()
2393        user_provider_filter = user.provider_filter if user else None
2394        if not user_provider_filter:
2395            return list(providers)
2396        return [
2397            p
2398            for p in providers
2399            if p.type != ProviderType.MUSIC or p.instance_id in user_provider_filter
2400        ]
2401
2402    async def _search_shareable_url(self, search_query: str) -> SearchResults | None:
2403        """
2404        Handle a search query that is a streaming provider public shareable URL.
2405
2406        Returns None if the query is not such a URL and a regular search must be done.
2407        """
2408        try:
2409            media_type, provider_instance_id_or_domain, item_id = await parse_uri(
2410                search_query, validate_id=True
2411            )
2412        except InvalidProviderURI:
2413            return None
2414        except InvalidProviderID as err:
2415            self.logger.warning("%s", str(err))
2416            return SearchResults()
2417        if provider_instance_id_or_domain not in PROVIDERS_WITH_SHAREABLE_URLS:
2418            return None
2419        try:
2420            item = await self.get_item(
2421                media_type=media_type,
2422                item_id=item_id,
2423                provider_instance_id_or_domain=provider_instance_id_or_domain,
2424            )
2425        except MusicAssistantError as err:
2426            self.logger.warning("%s", str(err))
2427            return SearchResults()
2428        if media_type == MediaType.ARTIST:
2429            return SearchResults(artists=[cast("Artist", item)])
2430        if media_type == MediaType.ALBUM:
2431            return SearchResults(albums=[cast("Album", item)])
2432        if media_type == MediaType.TRACK:
2433            return SearchResults(tracks=[cast("Track", item)])
2434        if media_type == MediaType.PLAYLIST:
2435            return SearchResults(playlists=[cast("Playlist", item)])
2436        if media_type == MediaType.AUDIOBOOK:
2437            return SearchResults(audiobooks=[cast("Audiobook", item)])
2438        if media_type == MediaType.PODCAST:
2439            return SearchResults(podcasts=[cast("Podcast", item)])
2440        return SearchResults()
2441
2442    async def _search_provider(
2443        self,
2444        search_query: str,
2445        provider_instance_id_or_domain: str,
2446        media_types: list[MediaType],
2447        limit: int = 10,
2448        skip_item_ids: set[tuple[MediaType, str, str]] | None = None,
2449    ) -> SearchResults | None:
2450        """
2451        Perform search on given provider, returns None if the search failed or timed out.
2452
2453        :param search_query: Search query
2454        :param provider_instance_id_or_domain: instance_id or domain of the provider
2455                                               to perform the search on.
2456        :param media_types: A list of media_types to include.
2457        :param limit: number of items to return in the search (per type).
2458        :param skip_item_ids: Optional set of (media_type, provider_domain, item_id)
2459                              tuples to filter out of the results.
2460        """
2461        prov = self.mass.get_provider(provider_instance_id_or_domain, provider_type=MusicProvider)
2462        if not prov:
2463            return SearchResults()
2464        if ProviderFeature.SEARCH not in prov.supported_features:
2465            return SearchResults()
2466
2467        # create safe search string
2468        search_query = search_query.replace("/", " ").replace("'", "")
2469        # use the per-provider cache so repeated and overlapping searches
2470        # do not hit the provider again
2471        cache_key = f"{search_query}-{'-'.join(sorted([mt.value for mt in media_types]))}-{limit}"
2472        if (
2473            cache := await self.mass.cache.get(
2474                key=cache_key,
2475                provider=prov.instance_id,
2476                category=CACHE_CATEGORY_SEARCH_RESULTS,
2477                base_class=SearchResults,
2478            )
2479        ) is not None:
2480            return filter_search_results(cast("SearchResults", cache), prov.domain, skip_item_ids)
2481        # run the provider search as a separate task (deduplicated by task_id so
2482        # identical concurrent searches share a single provider call) and wait for
2483        # it a limited amount of time only: a slow provider then contributes no
2484        # results now, while its search continues in the background so the result
2485        # is cached and available for a next search request
2486        task = self.mass.create_task(
2487            self._execute_provider_search(prov, search_query, media_types, limit, cache_key),
2488            task_id=f"provider_search_{prov.instance_id}_{cache_key}",
2489        )
2490        try:
2491            async with asyncio.timeout(SEARCH_PROVIDER_SOFT_TIMEOUT):
2492                prov_search_results = await asyncio.shield(task)
2493        except TimeoutError:
2494            self.logger.warning(
2495                "Search on provider %s did not return in time, "
2496                "the search continues in the background",
2497                prov.name,
2498            )
2499            return None
2500        if prov_search_results is None:
2501            return None
2502        return filter_search_results(prov_search_results, prov.domain, skip_item_ids)
2503
2504    async def _execute_provider_search(
2505        self,
2506        prov: MusicProvider,
2507        search_query: str,
2508        media_types: list[MediaType],
2509        limit: int,
2510        cache_key: str,
2511    ) -> SearchResults | None:
2512        """
2513        Execute the actual search on a provider and cache the result.
2514
2515        Returns None if the provider search failed or timed out. All errors are
2516        handled here (and not raised) as this coroutine runs as a background task
2517        that may outlive the request that started it.
2518        """
2519        try:
2520            async with asyncio.timeout(SEARCH_PROVIDER_HARD_TIMEOUT):
2521                result = await prov.search(search_query, media_types, limit)
2522        except TimeoutError:
2523            self.logger.warning("Search on provider %s timed out", prov.name)
2524            return None
2525        except MusicAssistantError as err:
2526            self.logger.warning("Search on provider %s failed: %s", prov.name, str(err))
2527            return None
2528        except Exception as err:
2529            self.logger.error("Search on provider %s failed: %s", prov.name, str(err), exc_info=err)
2530            return None
2531        # only successful results are cached, so failed or timed out
2532        # provider searches are simply retried on a next search
2533        await self._cache_search_results(
2534            cache_key,
2535            result,
2536            # plugin providers do not declare is_streaming_provider,
2537            # treat them as local so their results only get the short expiration
2538            SEARCH_CACHE_EXPIRATION_STREAMING_PROVIDER
2539            if getattr(prov, "is_streaming_provider", False)
2540            else SEARCH_CACHE_EXPIRATION_LOCAL_PROVIDER,
2541            prov.instance_id,
2542        )
2543        return result
2544
2545    async def _cache_search_results(
2546        self, cache_key: str, result: SearchResults, expiration: int, provider: str
2547    ) -> None:
2548        """Store search results in the cache, logging (instead of raising) any cache errors."""
2549        try:
2550            await self.mass.cache.set(
2551                key=cache_key,
2552                data=result.to_dict(),
2553                expiration=expiration,
2554                provider=provider,
2555                category=CACHE_CATEGORY_SEARCH_RESULTS,
2556            )
2557        except Exception as err:
2558            self.logger.warning("Failed to cache search results for %s: %s", provider, str(err))
2559
2560    def _get_covered_media_types(
2561        self, library_results: SearchResults, search_query: str
2562    ) -> set[tuple[MediaType, str]]:
2563        """
2564        Return the (media_type, provider domain/instance) pairs covered by the library.
2565
2566        A pair is considered covered when the library holds a (near) exact name match
2567        for the search query that is mapped to that provider.
2568        """
2569        covered: set[tuple[MediaType, str]] = set()
2570        # extract the artist and title part in case the
2571        # query is formatted as "artist - title"
2572        if " - " in search_query:
2573            artist_part, title_part = search_query.split(" - ", 1)
2574        else:
2575            artist_part, title_part = None, search_query
2576        items: Sequence[MediaItemType | ItemMapping]
2577        for items in (
2578            library_results.artists,
2579            library_results.albums,
2580            library_results.tracks,
2581            library_results.playlists,
2582            library_results.radio,
2583            library_results.audiobooks,
2584            library_results.podcasts,
2585        ):
2586            for item in items:
2587                if compare_strings(item.name, search_query, strict=False):
2588                    pass
2589                elif artist_part and compare_strings(item.name, title_part, strict=False):
2590                    # the item name matches the title part only,
2591                    # so the artist part must match one of the item artists
2592                    if not any(
2593                        compare_strings(artist.name, artist_part, strict=False)
2594                        for artist in getattr(item, "artists", [])
2595                    ):
2596                        continue
2597                else:
2598                    continue
2599                for prov_mapping in cast("MediaItemType", item).provider_mappings:
2600                    if not prov_mapping.available:
2601                        continue
2602                    covered.add((item.media_type, prov_mapping.provider_domain))
2603                    covered.add((item.media_type, prov_mapping.provider_instance))
2604        return covered
2605
2606    def _import_album_tracks_if_enabled(self, album: Album) -> None:
2607        """Import all album tracks into the library for providers that have this enabled."""
2608        for prov_mapping in album.provider_mappings:
2609            # only consider mappings the album was actually added on; additional
2610            # mappings auto-created for other instances of the same provider
2611            # (via match_provider_instances) carry in_library=None and must be skipped
2612            if not prov_mapping.in_library:
2613                continue
2614            provider = self.mass.get_provider(prov_mapping.provider_instance)
2615            if not isinstance(provider, MusicProvider):
2616                continue
2617            if not provider.library_sync_album_tracks_enabled():
2618                continue
2619            self.mass.create_task(provider.import_album_tracks(prov_mapping.item_id, album.name))
2620
2621    async def _get_provider_sound_effects(self, provider: MusicProvider) -> list[SoundEffect]:
2622        """Return all sound effect items from a single provider."""
2623        try:
2624            return [item async for item in provider.get_sound_effects()]
2625        except Exception as err:
2626            self.logger.warning(
2627                "Error while fetching sound effects from %s: %s",
2628                provider.name,
2629                str(err),
2630                exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
2631            )
2632            return []
2633
2634    def _create_provider_sync_handler(
2635        self, provider: MusicProvider, media_type: MediaType
2636    ) -> Callable[[], Awaitable[None]]:
2637        """Create the coroutine used for a managed provider sync task."""
2638
2639        async def run_sync() -> None:
2640            try:
2641                async with self._sync_lock:
2642                    # suppress per-item events during sync; a large library would otherwise
2643                    # emit one (serialized per client) for every item. Subscribers refresh
2644                    # on MUSIC_SYNC_COMPLETED and track progress via TASKS_UPDATED instead.
2645                    token = SUPPRESS_MEDIA_ITEM_UPDATES.set(True)
2646                    try:
2647                        await provider.sync_library(media_type)
2648                    finally:
2649                        SUPPRESS_MEDIA_ITEM_UPDATES.reset(token)
2650            finally:
2651                self.mass.call_later(
2652                    0,
2653                    self._handle_sync_completion_check,
2654                    task_id=MUSIC_SYNC_COMPLETION_CHECK_TASK_ID,
2655                )
2656
2657        return run_sync
2658
2659    def _get_sync_task_id(self, provider: MusicProvider | str, media_type: MediaType) -> str:
2660        """Return deterministic task id for a provider sync."""
2661        provider_instance = (
2662            provider.instance_id if isinstance(provider, MusicProvider) else provider
2663        )
2664        return f"music_sync_{provider_instance}_{media_type.value}"
2665
2666    def _get_sync_task_name(self, provider: MusicProvider, media_type: MediaType) -> str:
2667        """Return display name for a provider sync task."""
2668        return f"Sync {provider.name} {media_type.value}s"
2669
2670    def _get_sync_task_translation_key(self, media_type: MediaType) -> str:
2671        """Return translation key for a provider sync task."""
2672        if media_type == MediaType.ARTIST:
2673            return "sync_provider_artists"
2674        if media_type == MediaType.ALBUM:
2675            return "sync_provider_albums"
2676        if media_type == MediaType.TRACK:
2677            return "sync_provider_tracks"
2678        if media_type == MediaType.PLAYLIST:
2679            return "sync_provider_playlists"
2680        if media_type == MediaType.RADIO:
2681            return "sync_provider_radios"
2682        if media_type == MediaType.AUDIOBOOK:
2683            return "sync_provider_audiobooks"
2684        if media_type == MediaType.PODCAST:
2685            return "sync_provider_podcasts"
2686        return "settings.sync"
2687
2688    def _get_sync_task_metadata(
2689        self, provider: MusicProvider, media_type: MediaType
2690    ) -> TaskMetadata:
2691        """Return metadata for a provider sync task."""
2692        return {
2693            "task_domain": "music_sync",
2694            "provider_domain": provider.domain,
2695            "provider_instance": provider.instance_id,
2696            "provider_name": provider.name,
2697            "media_type": media_type.value,
2698        }
2699
2700    def _handle_sync_completion_check(self) -> None:
2701        """Run follow-up maintenance when no provider sync tasks remain active."""
2702        if self.active_sync_tasks:
2703            return
2704        self.mass.signal_event(EventType.MUSIC_SYNC_COMPLETED)
2705        # freshly synced content is the only source of new duplicates, so the reconciliation
2706        # pass owes the library another walk; it starts once the current one reaches the end,
2707        # since rewinding right now would keep re-examining the same prefix forever
2708        self._set_track_reconciliation_state(self._track_reconciliation_cursor, True)
2709        self._queue_database_cleanup_task()
2710
2711    def _register_database_cleanup_task(self) -> BackgroundTask:
2712        """Register the recurring database cleanup background task."""
2713        utc_hour, utc_minute = local_clock_time_to_utc(5, 0)
2714        desired_schedule = TaskSchedule.daily(hour=utc_hour, minute=utc_minute)
2715        return self.mass.tasks.register_scheduled_task(
2716            task_id=DATABASE_CLEANUP_TASK_ID,
2717            name="Database cleanup",
2718            handler=self._cleanup_database,
2719            schedule=desired_schedule,
2720            translation_key="database_cleanup",
2721            translation_owner=self.translation_owner,
2722            metadata={
2723                "task_domain": "music_database_cleanup",
2724            },
2725            allow_retry=True,
2726        )
2727
2728    def _register_provider_mapping_correction_task(self) -> BackgroundTask:
2729        """Register the recurring provider mapping correction background task."""
2730        utc_hour, utc_minute = local_clock_time_to_utc(4, 0)
2731        desired_schedule = TaskSchedule.daily(every=30, hour=utc_hour, minute=utc_minute)
2732        return self.mass.tasks.register_scheduled_task(
2733            task_id=PROVIDER_MAPPING_CORRECTION_TASK_ID,
2734            name="Correct provider mappings",
2735            handler=self.correct_multi_instance_provider_mappings,
2736            schedule=desired_schedule,
2737            translation_key="correct_provider_mappings",
2738            translation_owner=self.translation_owner,
2739            metadata={
2740                "task_domain": "music_provider_mapping_correction",
2741            },
2742            allow_retry=True,
2743        )
2744
2745    def _register_track_reconciliation_task(self) -> BackgroundTask:
2746        """Register the recurring duplicate track reconciliation background task."""
2747        # runs every hour rather than spread across the day: it is bounded to a small
2748        # batch of candidates per run and never leaves the local database
2749        return self.mass.tasks.register_scheduled_task(
2750            task_id=TRACK_RECONCILIATION_TASK_ID,
2751            name="Reconcile duplicate tracks",
2752            handler=self._reconcile_duplicate_tracks,
2753            schedule=TaskSchedule.hourly(),
2754            translation_key="reconcile_duplicate_tracks",
2755            translation_owner=self.translation_owner,
2756            metadata={
2757                "task_domain": "music_track_reconciliation",
2758            },
2759            allow_retry=True,
2760        )
2761
2762    async def _reconcile_duplicate_tracks(self) -> None:
2763        """Merge a small batch of library tracks that are held twice across providers."""
2764        if self.active_sync_tasks:
2765            # a sync is still filling in albums and mappings, so hold off rather than
2766            # judge duplicates against a half-populated library
2767            update_current_task_progress_text("Waiting for music sync completion")
2768            return
2769        self._start_next_pass_if_due()
2770        if (cursor := self._track_reconciliation_cursor) is None:
2771            # the library has been walked end to end and nothing has been synced since,
2772            # so there is nothing to look for: skip the query rather than scan for a miss
2773            update_current_task_progress_text("No duplicate tracks found")
2774            return
2775        update_current_task_progress_text("Searching for duplicate tracks")
2776        rows = await self.database.get_rows_from_query(
2777            _DUPLICATE_TRACK_CANDIDATES_QUERY,
2778            {
2779                "max_duration_delta": TRACK_RECONCILIATION_MAX_DURATION_DELTA,
2780                "cursor_item_id_1": cursor[0],
2781                "cursor_item_id_2": cursor[1],
2782            },
2783            limit=TRACK_RECONCILIATION_BATCH_SIZE,
2784        )
2785        if not rows:
2786            self._set_track_reconciliation_state(None, self._track_reconciliation_rescan_due)
2787            update_current_task_progress_text("No duplicate tracks found")
2788            return
2789        merged = 0
2790        retry_due = False
2791        examined = cursor
2792        try:
2793            for index, row in enumerate(rows, 1):
2794                update_current_task_progress_from_index(
2795                    index, len(rows), f"Checking duplicate track {index}/{len(rows)}"
2796                )
2797                try:
2798                    if await self._merge_duplicate_track_pair(
2799                        int(row["item_id_1"]), int(row["item_id_2"])
2800                    ):
2801                        merged += 1
2802                except MediaNotFoundError:
2803                    # an earlier merge in this batch already absorbed one of the two rows
2804                    pass
2805                except MusicAssistantError as err:
2806                    # a pair that failed on something transient deserves another look
2807                    retry_due = True
2808                    report_current_task_failure(str(err))
2809                    self.logger.warning(
2810                        "Error while reconciling duplicate tracks %s and %s: %s",
2811                        row["item_id_1"],
2812                        row["item_id_2"],
2813                        str(err),
2814                        exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
2815                    )
2816                examined = (int(row["item_id_1"]), int(row["item_id_2"]))
2817        finally:
2818            # resume after the pair examined last, so candidates this run refused can never
2819            # starve the ones behind them, not even a further pair of the same track that the
2820            # batch boundary cut off. Recording it even when the run is cut short keeps the
2821            # pairs it did not reach for the next run rather than skipping past them.
2822            walked_to_end = len(rows) < TRACK_RECONCILIATION_BATCH_SIZE and examined == (
2823                int(rows[-1]["item_id_1"]),
2824                int(rows[-1]["item_id_2"]),
2825            )
2826            # a merge moves album and artist relations onto the surviving row, which can make
2827            # it a duplicate of a row this walk has already passed, so ask for another pass
2828            self._set_track_reconciliation_state(
2829                None if walked_to_end else examined,
2830                self._track_reconciliation_rescan_due or merged > 0 or retry_due,
2831            )
2832        update_current_task_progress(100, f"Merged {merged} duplicate track(s)")
2833
2834    def _restore_track_reconciliation_state(self) -> None:
2835        """Pick the duplicate track walk back up where the previous run left it."""
2836        cursor = self.mass.config.get_raw_core_config_value(
2837            self.domain, CONF_TRACK_RECONCILIATION_CURSOR, [0, 0]
2838        )
2839        self._track_reconciliation_cursor = (
2840            (int(cursor[0]), int(cursor[1])) if len(cursor) == 2 else None
2841        )
2842        self._track_reconciliation_rescan_due = bool(
2843            self.mass.config.get_raw_core_config_value(
2844                self.domain, CONF_TRACK_RECONCILIATION_RESCAN_DUE, False
2845            )
2846        )
2847
2848    def _set_track_reconciliation_state(
2849        self, cursor: tuple[int, int] | None, rescan_due: bool
2850    ) -> None:
2851        """
2852        Record how far the duplicate track walk has come, surviving a restart.
2853
2854        :param cursor: The pair examined last, or None once the walk reached the end.
2855        :param rescan_due: Whether a completed sync still owes the library another pass.
2856        """
2857        self._track_reconciliation_cursor = cursor
2858        self._track_reconciliation_rescan_due = rescan_due
2859        self.mass.config.set_raw_core_config_value(
2860            self.domain, CONF_TRACK_RECONCILIATION_CURSOR, list(cursor) if cursor else []
2861        )
2862        self.mass.config.set_raw_core_config_value(
2863            self.domain, CONF_TRACK_RECONCILIATION_RESCAN_DUE, rescan_due
2864        )
2865
2866    def _start_next_pass_if_due(self) -> None:
2867        """Rewind the duplicate track walk if a sync has added content and the walk is done."""
2868        # rewinding a walk still in progress would keep re-examining the same first
2869        # candidates, so a pending rescan waits for the current one to reach the end
2870        if not self._track_reconciliation_rescan_due:
2871            return
2872        if self._track_reconciliation_cursor is not None:
2873            return
2874        self._set_track_reconciliation_state((0, 0), False)
2875
2876    async def _albums_agree_on_edition(self, item_id_1: int, item_id_2: int) -> bool:
2877        """
2878        Check that two tracks share an album whose edition matches as well as its title.
2879
2880        :param item_id_1: Library ID of the first track.
2881        :param item_id_2: Library ID of the second track.
2882        """
2883        # the query relates titles loosely so a spelled-out retail suffix cannot hide a
2884        # shared album, which leaves the identity for the album comparison to confirm. An
2885        # edition is held apart from the title: without that an original and its remaster or
2886        # deluxe edition look like the same album whenever neither track carries a version
2887        rows = await self.database.get_rows_from_query(
2888            _SHARED_ALBUM_EDITIONS_QUERY,
2889            {"item_id_1": item_id_1, "item_id_2": item_id_2},
2890        )
2891        return any(
2892            compare_album_name(row["name_1"], row["name_2"])
2893            and compare_version(row["version_1"], row["version_2"])
2894            for row in rows
2895        )
2896
2897    async def _merge_duplicate_track_pair(self, item_id_1: int, item_id_2: int) -> bool:
2898        """
2899        Merge two candidate rows if they are confirmed to be the same track.
2900
2901        :param item_id_1: Library ID of the lower-numbered candidate row.
2902        :param item_id_2: Library ID of the higher-numbered candidate row.
2903        :return: True when the rows were merged, False when they were left alone.
2904        """
2905        track_1 = await self.tracks.get_library_item(item_id_1)
2906        track_2 = await self.tracks.get_library_item(item_id_2)
2907        # the checks below establish that both rows sit at the same position on an equally
2908        # titled album, which is the album agreement strict mode looks for, so the remaining
2909        # check is run in non-strict mode. Its version check is reinstated here
2910        # explicitly: without it a remaster, remix or radio edit of equal length would be
2911        # accepted as the original.
2912        if not compare_version(track_1.version, track_2.version):
2913            return False
2914        if not await self._albums_agree_on_edition(item_id_1, item_id_2):
2915            return False
2916        if not compare_track(track_1, track_2, strict=False):
2917            return False
2918        # keep the row that carries the most provider mappings so the fewest mappings and
2919        # relations have to move; equal counts keep the oldest row, which the query orders first
2920        target, source = (
2921            (track_1, track_2)
2922            if len(track_1.provider_mappings) >= len(track_2.provider_mappings)
2923            else (track_2, track_1)
2924        )
2925        self.logger.debug(
2926            "Merging duplicate track %s (id %s) into id %s",
2927            target.name,
2928            source.item_id,
2929            target.item_id,
2930        )
2931        await self.tracks.merge_library_items(target.item_id, source.item_id)
2932        return True
2933
2934    def _queue_database_cleanup_task(self) -> BackgroundTask:
2935        """Queue the post-sync database cleanup as a managed background task."""
2936        self._register_database_cleanup_task()
2937        return self.mass.tasks.run_task(DATABASE_CLEANUP_TASK_ID)
2938
2939    async def _schedule_provider_mediatype_sync(
2940        self, provider: MusicProvider, media_type: MediaType, is_initial: bool = False
2941    ) -> None:
2942        """Schedule Library sync for given provider and media type."""
2943        # handle mediatype specific sync config
2944        conf_key = f"library_sync_{media_type}s"
2945        sync_conf: ConfigValueType = await self.mass.config.get_provider_config_value(
2946            provider.instance_id, conf_key
2947        )
2948        if not sync_conf:
2949            self.mass.tasks.unregister_scheduled_task(self._get_sync_task_id(provider, media_type))
2950            return
2951        self.mass.tasks.register_scheduled_task(
2952            task_id=self._get_sync_task_id(provider, media_type),
2953            name=self._get_sync_task_name(provider, media_type),
2954            handler=self._create_provider_sync_handler(provider, media_type),
2955            schedule=provider.get_default_library_sync_schedule(media_type),
2956            initial_delay=INITIAL_SYNC_DELAY if is_initial else None,
2957            translation_key=self._get_sync_task_translation_key(media_type),
2958            translation_args=[provider.name],
2959            translation_owner=self.translation_owner,
2960            metadata=self._get_sync_task_metadata(provider, media_type),
2961            allow_retry=True,
2962        )
2963
2964    async def _get_user_for_provider(
2965        self, provider_mappings_or_instance_id: Iterable[ProviderMapping] | str
2966    ) -> User | None:
2967        """Try to get the MA User based on provider mappings and provider filter."""
2968        all_users = await self.mass.webserver.auth.list_users()
2969        for mapping_or_instance_id in provider_mappings_or_instance_id:
2970            for user in all_users:
2971                if not user.provider_filter:
2972                    continue
2973                if isinstance(mapping_or_instance_id, str):
2974                    if provider_mappings_or_instance_id in user.provider_filter:
2975                        return user
2976                elif mapping_or_instance_id.provider_instance in user.provider_filter:
2977                    return user
2978        return None
2979
2980    async def _resolve_playlog_item(self, media_item: MediaItemType | ItemMapping) -> MediaItemType:
2981        """
2982        Return the full media item for a (possibly minimized) media item reference.
2983
2984        :param media_item: The media item to resolve, either full or an ItemMapping.
2985        """
2986        if not isinstance(media_item, ItemMapping):
2987            return media_item
2988        resolved = await self.get_item(
2989            media_item.media_type,
2990            media_item.item_id,
2991            media_item.provider,
2992            allow_update_metadata=False,
2993        )
2994        if isinstance(resolved, BrowseFolder):
2995            msg = f"{media_item.uri} does not resolve to a media item"
2996            raise MediaNotFoundError(msg)
2997        return resolved
2998
2999    async def _upsert_playlog(self, entry: dict[str, Any]) -> None:
3000        """
3001        Write a playlog row, updating the existing row for the item/user if there is one.
3002
3003        Columns left out of the entry keep whatever the existing row holds, and
3004        `user_initiated` is sticky: once a play was explicitly user-initiated it stays that
3005        way for the lifetime of the row, so a later side-effect credit (an autoplay replay,
3006        or a track crediting its album/artist) can never demote it and drop the item out of
3007        the "recently played" recommendations.
3008
3009        The generic `database.upsert()` cannot express either half of that: the sticky OR is
3010        playlog-specific, and it needs an explicit conflict target because the playlog carries
3011        more than one unique constraint.
3012
3013        :param entry: The playlog column values to write, including all of
3014            `PLAYLOG_CONFLICT_KEYS`.
3015        """
3016        columns = list(entry)
3017        updates = [
3018            f"user_initiated = {DB_TABLE_PLAYLOG}.user_initiated OR excluded.user_initiated"
3019            if column == "user_initiated"
3020            else f"{column} = excluded.{column}"
3021            for column in columns
3022            if column not in PLAYLOG_CONFLICT_KEYS
3023        ]
3024        await self.database.execute_write(
3025            f"INSERT INTO {DB_TABLE_PLAYLOG} ({', '.join(columns)}) "
3026            f"VALUES ({', '.join(f':{column}' for column in columns)}) "
3027            f"ON CONFLICT({', '.join(PLAYLOG_CONFLICT_KEYS)}) DO UPDATE SET {', '.join(updates)}",
3028            entry,
3029        )
3030
3031    async def _credit_artist_plays(
3032        self,
3033        artists: Iterable[Artist | ItemMapping],
3034        *,
3035        timestamp: float,
3036        user_ids: list[str],
3037        queue_id: str | None,
3038        skip_ids: set[str],
3039    ) -> None:
3040        """Credit each (library-resolvable) artist with a play, skipping skip_ids."""
3041        for artist in artists:
3042            db_artist = await self.artists.get_library_item_by_prov_id(
3043                artist.item_id, artist.provider
3044            )
3045            if db_artist is None:
3046                continue
3047            if db_artist.item_id in skip_ids:
3048                self.logger.debug("Skipping already-credited artist '%s'", db_artist.name)
3049                continue
3050            await self.database.execute(
3051                f"UPDATE {self.artists.db_table} SET play_count = play_count + 1, "
3052                f"last_played = {timestamp} WHERE item_id = {db_artist.item_id}"
3053            )
3054            self.logger.debug("Credited play for artist '%s'", db_artist.name)
3055            playlog_entry: dict[str, Any] = {
3056                "item_id": db_artist.item_id,
3057                "provider": "library",
3058                "media_type": MediaType.ARTIST.value,
3059                "name": db_artist.name,
3060                "image": serialize_to_json(db_artist.image.to_dict()) if db_artist.image else None,
3061                "fully_played": True,
3062                "seconds_played": None,
3063                "timestamp": timestamp,
3064                "queue_id": queue_id,
3065                "user_initiated": False,
3066            }
3067            for user_id in user_ids:
3068                playlog_entry["userid"] = user_id
3069                await self._upsert_playlog(playlog_entry)
3070
3071    async def _credit_podcast_play(
3072        self,
3073        podcast: Podcast | ItemMapping,
3074        *,
3075        timestamp: float,
3076        user_ids: list[str],
3077        queue_id: str | None,
3078    ) -> None:
3079        """Credit the parent podcast with a play so the show surfaces in recently played."""
3080        # Resolve to the library item first, like _credit_artist_plays does, so an episode's
3081        # parent-podcast credit lands on the same library-scoped row as an explicit play of the
3082        # library show, instead of creating a separate provider-scoped duplicate.
3083        db_podcast = await self.podcasts.get_library_item_by_prov_id(
3084            podcast.item_id, podcast.provider
3085        )
3086        credited_podcast: Podcast | ItemMapping = db_podcast if db_podcast else podcast
3087        playlog_entry: dict[str, Any] = {
3088            "item_id": credited_podcast.item_id,
3089            "provider": "library" if db_podcast else podcast.provider,
3090            "media_type": MediaType.PODCAST.value,
3091            "name": credited_podcast.name,
3092            "image": serialize_to_json(credited_podcast.image.to_dict())
3093            if credited_podcast.image
3094            else None,
3095            "fully_played": True,
3096            "seconds_played": None,
3097            "timestamp": timestamp,
3098            "queue_id": queue_id,
3099            "user_initiated": False,
3100        }
3101        for user_id in user_ids:
3102            playlog_entry["userid"] = user_id
3103            await self._upsert_playlog(playlog_entry)
3104
3105    async def _get_item_by_name(
3106        self,
3107        name: str,
3108        artist: str | None = None,
3109        album: str | None = None,
3110        media_type: MediaType | None = None,
3111    ) -> MediaItemType | ItemMapping | None:
3112        """Try to find a media item (such as a playlist) by name."""
3113        # Future todo: enhance this method with AI capabilities to allow typos and
3114        # natural language.
3115        searchname = name.lower()
3116        allowed_media_types = [
3117            MediaType.PLAYLIST,
3118            MediaType.RADIO,
3119            MediaType.TRACK,
3120            MediaType.ALBUM,
3121            MediaType.ARTIST,
3122            MediaType.AUDIOBOOK,
3123            MediaType.PODCAST,
3124        ]
3125        if media_type in (None, MediaType.UNKNOWN):
3126            media_types = allowed_media_types
3127        elif media_type not in allowed_media_types:
3128            raise InvalidDataError(
3129                f"{media_type} is not a supported media_type. "
3130                f"Supported media_types are {allowed_media_types}"
3131            )
3132        else:
3133            media_types = [media_type]
3134        library_functions = [
3135            self.get_controller(media_type).library_items for media_type in media_types
3136        ]
3137        # prefer (exact) lookup in the library by name
3138        for func in library_functions:
3139            result = await func(search=searchname)
3140            for item in result:
3141                # handle optional artist filter
3142                if (
3143                    artist
3144                    and (artists := getattr(item, "artists", None))
3145                    and not any(x for x in artists if x.name.lower() == artist.lower())
3146                ):
3147                    continue
3148                # handle optional album filter
3149                if (
3150                    album
3151                    and (item_album := getattr(item, "album", None))
3152                    and item_album.name.lower() != album.lower()
3153                ):
3154                    continue
3155                if searchname == item.name.lower():
3156                    return item
3157        # nothing found in the library, fallback to global search
3158        search_name = name
3159        if album and artist:
3160            search_name = f"{artist} - {album} - {name}"
3161        elif album:
3162            search_name = f"{album} - {name}"
3163        elif artist:
3164            search_name = f"{artist} - {name}"
3165        search_results = await self.search(
3166            search_query=search_name,
3167            media_types=[media_type]
3168            if media_type and media_type != MediaType.UNKNOWN
3169            else MediaType.ALL,
3170            limit=8,
3171        )
3172        for results in (
3173            search_results.tracks,
3174            search_results.albums,
3175            search_results.playlists,
3176            search_results.artists,
3177            search_results.radio,
3178            search_results.audiobooks,
3179            search_results.podcasts,
3180        ):
3181            for _item in results:
3182                # simply return the first item because search is already sorted by best match
3183                return _item
3184        return None
3185
3186    async def _handle_verify_item_uri(self, uri: str) -> bool:
3187        user = get_current_user()
3188
3189        try:
3190            media_type, provider_instance_id_or_domain, item_id = await parse_uri(uri)
3191        except InvalidProviderURI, InvalidProviderID:
3192            return False
3193
3194        # fast return for a provider uri which is not part of a user with a provider filter
3195        if (
3196            provider_instance_id_or_domain != "library"
3197            and user
3198            and user.provider_filter
3199            and provider_instance_id_or_domain not in user.provider_filter
3200        ):
3201            return False
3202
3203        # verify that item itself exists
3204        try:
3205            item = await self.get_item(
3206                media_type=media_type,
3207                item_id=item_id,
3208                provider_instance_id_or_domain=provider_instance_id_or_domain,
3209                allow_update_metadata=False,  # no need trigger more methods
3210            )
3211        except MediaNotFoundError, NotImplementedError:
3212            # NotImplementedError: the uri has a valid format, but specifies an unknown media type
3213            return False
3214
3215        # non library item handling for users with no filter, or no user at all
3216        if (
3217            provider_instance_id_or_domain != "library"
3218            or not user
3219            or (user and not user.provider_filter)
3220            or isinstance(item, BrowseFolder)
3221        ):
3222            return True
3223
3224        # library item handling for users with provider filter
3225        for provider_mapping in item.provider_mappings:
3226            if provider_mapping.provider_instance in user.provider_filter:
3227                return True
3228
3229        return False
3230