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