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 cleanup_provider_shortcuts(self, provider_instance: str) -> None:
2151        """
2152        Clean up sidebar shortcuts after a music provider is removed.
2153
2154        :param provider_instance: The instance ID of the removed provider.
2155        """
2156
2157        async def _rewrite(uri: str) -> str | None:
2158            try:
2159                media_type, provider, item_id = await parse_uri(uri)
2160            except InvalidProviderURI, InvalidProviderID, KeyError, ValueError:
2161                return uri
2162            if provider != provider_instance:
2163                return uri
2164            try:
2165                ctrl = self.get_controller(media_type)
2166            except NotImplementedError:
2167                return None
2168            if library_item := await ctrl.get_library_item_by_prov_id(item_id, provider_instance):
2169                return f"library://{media_type.value}/{library_item.item_id}"
2170            return None
2171
2172        await self.mass.webserver.auth.cleanup_user_shortcuts(_rewrite)
2173
2174    async def cleanup_library_shortcuts(self) -> None:
2175        """Remove sidebar shortcuts whose library item no longer exists."""
2176
2177        async def _rewrite(uri: str) -> str | None:
2178            try:
2179                media_type, provider, item_id = await parse_uri(uri)
2180            except InvalidProviderURI, InvalidProviderID, KeyError, ValueError:
2181                return uri
2182            if provider != "library":
2183                return uri
2184            try:
2185                ctrl = self.get_controller(media_type)
2186            except NotImplementedError:
2187                return uri
2188            try:
2189                await ctrl.get_library_item(item_id)
2190            except MediaNotFoundError, ValueError:
2191                return None
2192            return uri
2193
2194        await self.mass.webserver.auth.cleanup_user_shortcuts(_rewrite)
2195
2196    async def schedule_provider_sync(self, provider_instance_id: str) -> None:
2197        """Schedule Library sync for given provider."""
2198        if not (
2199            provider := self.mass.get_provider(provider_instance_id, provider_type=MusicProvider)
2200        ):
2201            return
2202        await self.unschedule_provider_sync(provider.instance_id, clear_persisted_state=False)
2203        for media_type in MediaType:
2204            if not self.library_supported(provider, media_type):
2205                continue
2206            await self._schedule_provider_mediatype_sync(provider, media_type, True)
2207
2208    async def unschedule_provider_sync(
2209        self, provider_instance_id: str, clear_persisted_state: bool = True
2210    ) -> None:
2211        """
2212        Unschedule Library sync for given provider and wait for a running sync to stop.
2213
2214        Callers tear down provider state right after this (unloading the provider, or
2215        rescheduling its syncs), so all media types are cancelled first and then awaited
2216        together, keeping the bounded wait to one timeout instead of one per media type.
2217
2218        :param provider_instance_id: The provider instance id to unschedule.
2219        :param clear_persisted_state: Whether to remove persisted schedule state from config.
2220        """
2221        await asyncio.gather(
2222            *(
2223                self.mass.tasks.unregister_scheduled_task_and_wait(
2224                    self._get_sync_task_id(provider_instance_id, media_type),
2225                    clear_persisted_state=clear_persisted_state,
2226                )
2227                for media_type in MediaType
2228            )
2229        )
2230
2231    def get_provider_sync_schedule(
2232        self, provider_instance_id: str, media_type: MediaType
2233    ) -> TaskSchedule | None:
2234        """Return the effective schedule for a provider sync task, if any."""
2235        task_id = self._get_sync_task_id(provider_instance_id, media_type)
2236        with suppress(InvalidDataError):
2237            task = self.mass.tasks.get_task(task_id)
2238            return task.schedule
2239        if not (
2240            provider := self.mass.get_provider(provider_instance_id, provider_type=MusicProvider)
2241        ):
2242            return None
2243        if not self.library_supported(provider, media_type):
2244            return None
2245        return provider.get_default_library_sync_schedule(media_type)
2246
2247    def match_provider_instances(
2248        self,
2249        item: MediaItemType,
2250    ) -> bool:
2251        """Match all provider instances for the given item."""
2252        mappings_added = False
2253        for provider_mapping in list(item.provider_mappings):
2254            if provider_mapping.is_unique:
2255                # unique mapping, no need to map
2256                continue
2257            if not (provider := self.mass.get_provider(provider_mapping.provider_instance)):
2258                continue
2259            if not isinstance(provider, MusicProvider):
2260                continue
2261            if not provider.is_streaming_provider:
2262                continue
2263            provider_instances = self.get_provider_instances(
2264                provider.domain, return_unavailable=True
2265            )
2266            if len(provider_instances) <= 1:
2267                # only a single instance, no need to map
2268                continue
2269            for prov_instance in provider_instances:
2270                if prov_instance.instance_id == provider.instance_id:
2271                    continue
2272                if any(
2273                    pm.provider_instance == prov_instance.instance_id
2274                    for pm in item.provider_mappings
2275                ):
2276                    # mapping already exists
2277                    continue
2278                # create additional mapping for other provider instances of the same provider
2279                item.provider_mappings.add(
2280                    ProviderMapping(
2281                        item_id=provider_mapping.item_id,
2282                        provider_domain=provider.domain,
2283                        provider_instance=prov_instance.instance_id,
2284                        available=provider_mapping.available,
2285                        is_unique=provider_mapping.is_unique,
2286                        audio_format=provider_mapping.audio_format,
2287                        url=provider_mapping.url,
2288                        details=provider_mapping.details,
2289                        in_library=None,
2290                    )
2291                )
2292                mappings_added = True
2293        return mappings_added
2294
2295    @api_command("music/add_provider_mapping", required_scope=Scope.LIBRARY_MANAGE)
2296    async def add_provider_mapping(
2297        self, media_type: MediaType, db_id: str, mapping: ProviderMapping
2298    ) -> None:
2299        """Add provider mapping to the given library item."""
2300        ctrl = self.get_controller(media_type)
2301        await ctrl.add_provider_mappings(db_id, [mapping])
2302
2303    @api_command("music/remove_provider_mapping", required_scope=Scope.LIBRARY_MANAGE)
2304    async def remove_provider_mapping(
2305        self, media_type: MediaType, db_id: str, mapping: ProviderMapping
2306    ) -> None:
2307        """Remove provider mapping from the given library item."""
2308        ctrl = self.get_controller(media_type)
2309        await ctrl.remove_provider_mapping(db_id, mapping.provider_instance, mapping.item_id)
2310
2311    @api_command("music/match_providers", required_scope=Scope.LIBRARY_MANAGE)
2312    async def match_providers(self, media_type: MediaType, db_id: str) -> None:
2313        """Search for mappings on all providers for the given library item."""
2314        ctrl = self.get_controller(media_type)
2315        db_item = await ctrl.get_library_item(db_id)
2316        # ctrl is chosen by media_type, so it matches db_item's runtime type
2317        await cast("MediaControllerBase[MediaItemType]", ctrl).match_providers(db_item)
2318
2319    async def update_provider_mapping(
2320        self,
2321        media_type: MediaType,
2322        db_id: str | int,
2323        provider_instance_id: str,
2324        provider_item_id: str,
2325        *,
2326        available: bool | Any = UNSET,
2327        in_library: bool | Any = UNSET,
2328        is_unique: bool | None | Any = UNSET,
2329        url: str | None | Any = UNSET,
2330        details: str | None | Any = UNSET,
2331        audio_format: AudioFormat | Any = UNSET,
2332    ) -> None:
2333        """Update an existing provider mapping for a library item."""
2334        ctrl = self.get_controller(media_type)
2335        await ctrl.update_provider_mapping(
2336            item_id=db_id,
2337            provider_instance_id=provider_instance_id,
2338            provider_item_id=provider_item_id,
2339            available=available,
2340            in_library=in_library,
2341            is_unique=is_unique,
2342            url=url,
2343            details=details,
2344            audio_format=audio_format,
2345        )
2346
2347    def queue_provider_mapping_correction_task(self) -> BackgroundTask:
2348        """Queue the provider mapping correction as a managed background task."""
2349        self._register_provider_mapping_correction_task()
2350        return self.mass.tasks.run_task(PROVIDER_MAPPING_CORRECTION_TASK_ID)
2351
2352    async def correct_multi_instance_provider_mappings(self) -> None:
2353        """Correct provider mappings for multi-instance providers."""
2354        self.logger.debug("Correcting provider mappings for multi-instance providers...")
2355        multi_instance_providers: set[str] = set()
2356        for provider in self.providers:
2357            if len(self.get_provider_instances(provider.domain)) > 1:
2358                multi_instance_providers.add(provider.instance_id)
2359        if not multi_instance_providers:
2360            return  # no multi-instance providers found, nothing to do
2361
2362        for ctrl in (
2363            self.albums,
2364            self.artists,
2365            self.tracks,
2366            self.playlists,
2367            self.radio,
2368            self.audiobooks,
2369            self.podcasts,
2370        ):
2371            async for db_item in ctrl.iter_library_items(
2372                provider=list(multi_instance_providers), library_items_only=False
2373            ):
2374                if self.match_provider_instances(db_item):
2375                    # ctrl is the per-type controller, so it matches db_item's runtime type
2376                    await cast("MediaControllerBase[MediaItemType]", ctrl).update_item_in_library(
2377                        db_item.item_id, db_item
2378                    )
2379                # prevent overwhelming the event loop
2380                await asyncio.sleep(0.2)
2381        self.logger.debug("Provider mappings correction done")
2382
2383    def library_supported(self, provider: Provider, media_type: MediaType) -> bool:
2384        """Return whether the provider declares LIBRARY support for the given media type."""
2385        if provider.type != ProviderType.MUSIC:
2386            return False
2387        if (feature := LIBRARY_FEATURE_BY_MEDIA_TYPE.get(media_type)) is None:
2388            return False
2389        return provider.supports_feature(feature)
2390
2391    def library_edit_supported(self, provider: Provider, media_type: MediaType) -> bool:
2392        """Return whether the provider supports library add/remove for the given media type."""
2393        if provider.type != ProviderType.MUSIC:
2394            return False
2395        if media_type == MediaType.ARTIST:
2396            return provider.supports_feature(ProviderFeature.LIBRARY_ARTISTS_EDIT)
2397        if media_type == MediaType.ALBUM:
2398            return provider.supports_feature(ProviderFeature.LIBRARY_ALBUMS_EDIT)
2399        if media_type == MediaType.TRACK:
2400            return provider.supports_feature(ProviderFeature.LIBRARY_TRACKS_EDIT)
2401        if media_type == MediaType.PLAYLIST:
2402            return provider.supports_feature(ProviderFeature.LIBRARY_PLAYLISTS_EDIT)
2403        if media_type == MediaType.RADIO:
2404            return provider.supports_feature(ProviderFeature.LIBRARY_RADIOS_EDIT)
2405        if media_type == MediaType.AUDIOBOOK:
2406            return provider.supports_feature(ProviderFeature.LIBRARY_AUDIOBOOKS_EDIT)
2407        if media_type == MediaType.PODCAST:
2408            return provider.supports_feature(ProviderFeature.LIBRARY_PODCASTS_EDIT)
2409        return False
2410
2411    def library_favorites_edit_supported(self, provider: Provider, media_type: MediaType) -> bool:
2412        """Return whether the provider supports favorites add/remove for the given media type."""
2413        if provider.type != ProviderType.MUSIC:
2414            return False
2415        if media_type == MediaType.ARTIST:
2416            return provider.supports_feature(ProviderFeature.FAVORITE_ARTISTS_EDIT)
2417        if media_type == MediaType.ALBUM:
2418            return provider.supports_feature(ProviderFeature.FAVORITE_ALBUMS_EDIT)
2419        if media_type == MediaType.TRACK:
2420            return provider.supports_feature(ProviderFeature.FAVORITE_TRACKS_EDIT)
2421        if media_type == MediaType.PLAYLIST:
2422            return provider.supports_feature(ProviderFeature.FAVORITE_PLAYLISTS_EDIT)
2423        if media_type == MediaType.RADIO:
2424            return provider.supports_feature(ProviderFeature.FAVORITE_RADIOS_EDIT)
2425        if media_type == MediaType.AUDIOBOOK:
2426            return provider.supports_feature(ProviderFeature.FAVORITE_AUDIOBOOKS_EDIT)
2427        if media_type == MediaType.PODCAST:
2428            return provider.supports_feature(ProviderFeature.FAVORITE_PODCASTS_EDIT)
2429        return False
2430
2431    def library_sync_back_enabled(self, provider: Provider, media_type: MediaType) -> bool:
2432        """Return whether library sync back is enabled for the provider+media_type."""
2433        conf_value = provider.config.get_value(
2434            CONF_ENTRY_LIBRARY_SYNC_BACK.key, CONF_ENTRY_LIBRARY_SYNC_BACK.default_value
2435        )
2436        return bool(conf_value)
2437
2438    @api_command("music/item_by_name", required_scope=Scope.LIBRARY_READ, allow_impersonation=True)
2439    async def get_item_by_name(
2440        self,
2441        name: str,
2442        artist: str | None = None,
2443        album: str | None = None,
2444        media_type: MediaType | None = None,
2445    ) -> MediaItemType | ItemMapping | None:
2446        """Try to find a media item (such as a playlist) by name."""
2447        return await self._get_item_by_name(name, artist, album, media_type)
2448
2449    @api_command(
2450        "music/verify_item_uri", required_scope=Scope.LIBRARY_READ, allow_impersonation=True
2451    )
2452    async def verify_item_uri(self, uri: str) -> bool:
2453        """
2454        Verify whether a uri points to a valid, accessible item.
2455
2456        :param uri: The uri to verify.
2457        """
2458        return await self._handle_verify_item_uri(uri)
2459
2460    async def _get_plugin_audio_sources(
2461        self, provider: PluginProvider, player_id: str | None
2462    ) -> list[AudioSource]:
2463        """
2464        Return the AudioSources of a plugin to list, scoped to a player when given.
2465
2466        Player-bound plugins yield only the sources bound to the given player;
2467        without a player scope all their sources bound to a player the calling
2468        user may see are yielded. Player-unbound plugins always yield all sources.
2469        """
2470        # probing with the (possibly empty) scope tells bound and unbound apart:
2471        # a player-bound plugin returns a list for any player id, unbound returns None
2472        if provider.get_player_audio_sources(player_id or "") is None:
2473            return await provider.get_audio_sources()
2474        # bound sources honor the calling user's player access filter, so a
2475        # restricted user cannot discover sources of players hidden from them
2476        current_user = get_current_user()
2477        player_filter = (
2478            current_user.player_filter
2479            if current_user and not has_scope(current_user, Scope.ALL)
2480            else None
2481        )
2482        if player_id is not None:
2483            if player_filter and player_id not in player_filter:
2484                return []
2485            return provider.get_player_audio_sources(player_id) or []
2486        if not player_filter:
2487            return await provider.get_audio_sources()
2488        sources: list[AudioSource] = []
2489        for allowed_player_id in player_filter:
2490            sources.extend(provider.get_player_audio_sources(allowed_player_id) or [])
2491        return sources
2492
2493    def _apply_user_provider_filter(
2494        self,
2495        providers: Iterable[ProviderInstanceType],
2496    ) -> list[ProviderInstanceType]:
2497        """Filter providers by the current user's music provider filter."""
2498        user = get_current_user()
2499        user_provider_filter = user.provider_filter if user else None
2500        if not user_provider_filter:
2501            return list(providers)
2502        return [
2503            p
2504            for p in providers
2505            if p.type != ProviderType.MUSIC or p.instance_id in user_provider_filter
2506        ]
2507
2508    async def _search_shareable_url(self, search_query: str) -> SearchResults | None:
2509        """
2510        Handle a search query that is a streaming provider public shareable URL.
2511
2512        Returns None if the query is not such a URL and a regular search must be done.
2513        """
2514        try:
2515            media_type, provider_instance_id_or_domain, item_id = await parse_uri(
2516                search_query, validate_id=True
2517            )
2518        except InvalidProviderURI:
2519            return None
2520        except InvalidProviderID as err:
2521            self.logger.warning("%s", str(err))
2522            return SearchResults()
2523        if provider_instance_id_or_domain not in PROVIDERS_WITH_SHAREABLE_URLS:
2524            return None
2525        try:
2526            item = await self.get_item(
2527                media_type=media_type,
2528                item_id=item_id,
2529                provider_instance_id_or_domain=provider_instance_id_or_domain,
2530            )
2531        except MusicAssistantError as err:
2532            self.logger.warning("%s", str(err))
2533            return SearchResults()
2534        if media_type == MediaType.ARTIST:
2535            return SearchResults(artists=[cast("Artist", item)])
2536        if media_type == MediaType.ALBUM:
2537            return SearchResults(albums=[cast("Album", item)])
2538        if media_type == MediaType.TRACK:
2539            return SearchResults(tracks=[cast("Track", item)])
2540        if media_type == MediaType.PLAYLIST:
2541            return SearchResults(playlists=[cast("Playlist", item)])
2542        if media_type == MediaType.AUDIOBOOK:
2543            return SearchResults(audiobooks=[cast("Audiobook", item)])
2544        if media_type == MediaType.PODCAST:
2545            return SearchResults(podcasts=[cast("Podcast", item)])
2546        return SearchResults()
2547
2548    async def _search_provider(
2549        self,
2550        search_query: str,
2551        provider_instance_id_or_domain: str,
2552        media_types: list[MediaType],
2553        limit: int = 10,
2554        skip_item_ids: set[tuple[MediaType, str, str]] | None = None,
2555    ) -> SearchResults | None:
2556        """
2557        Perform search on given provider, returns None if the search failed or timed out.
2558
2559        :param search_query: Search query
2560        :param provider_instance_id_or_domain: instance_id or domain of the provider
2561                                               to perform the search on.
2562        :param media_types: A list of media_types to include.
2563        :param limit: number of items to return in the search (per type).
2564        :param skip_item_ids: Optional set of (media_type, provider_domain, item_id)
2565                              tuples to filter out of the results.
2566        """
2567        prov = self.mass.get_provider(provider_instance_id_or_domain, provider_type=MusicProvider)
2568        if not prov:
2569            return SearchResults()
2570        if ProviderFeature.SEARCH not in prov.supported_features:
2571            return SearchResults()
2572
2573        # create safe search string
2574        search_query = search_query.replace("/", " ").replace("'", "")
2575        # use the per-provider cache so repeated and overlapping searches
2576        # do not hit the provider again
2577        cache_key = f"{search_query}-{'-'.join(sorted([mt.value for mt in media_types]))}-{limit}"
2578        if (
2579            cache := await self.mass.cache.get(
2580                key=cache_key,
2581                provider=prov.instance_id,
2582                category=CACHE_CATEGORY_SEARCH_RESULTS,
2583                base_class=SearchResults,
2584            )
2585        ) is not None:
2586            return filter_search_results(cast("SearchResults", cache), prov.domain, skip_item_ids)
2587        # run the provider search as a separate task (deduplicated by task_id so
2588        # identical concurrent searches share a single provider call) and wait for
2589        # it a limited amount of time only: a slow provider then contributes no
2590        # results now, while its search continues in the background so the result
2591        # is cached and available for a next search request
2592        task = self.mass.create_task(
2593            self._execute_provider_search(prov, search_query, media_types, limit, cache_key),
2594            task_id=f"provider_search_{prov.instance_id}_{cache_key}",
2595        )
2596        try:
2597            async with asyncio.timeout(SEARCH_PROVIDER_SOFT_TIMEOUT):
2598                prov_search_results = await asyncio.shield(task)
2599        except TimeoutError:
2600            self.logger.warning(
2601                "Search on provider %s did not return in time, "
2602                "the search continues in the background",
2603                prov.name,
2604            )
2605            return None
2606        if prov_search_results is None:
2607            return None
2608        return filter_search_results(prov_search_results, prov.domain, skip_item_ids)
2609
2610    async def _execute_provider_search(
2611        self,
2612        prov: MusicProvider,
2613        search_query: str,
2614        media_types: list[MediaType],
2615        limit: int,
2616        cache_key: str,
2617    ) -> SearchResults | None:
2618        """
2619        Execute the actual search on a provider and cache the result.
2620
2621        Returns None if the provider search failed or timed out. All errors are
2622        handled here (and not raised) as this coroutine runs as a background task
2623        that may outlive the request that started it.
2624        """
2625        try:
2626            async with asyncio.timeout(SEARCH_PROVIDER_HARD_TIMEOUT):
2627                result = await prov.search(search_query, media_types, limit)
2628        except TimeoutError:
2629            self.logger.warning("Search on provider %s timed out", prov.name)
2630            return None
2631        except MusicAssistantError as err:
2632            self.logger.warning("Search on provider %s failed: %s", prov.name, str(err))
2633            return None
2634        except Exception as err:
2635            self.logger.error("Search on provider %s failed: %s", prov.name, str(err), exc_info=err)
2636            return None
2637        # only successful results are cached, so failed or timed out
2638        # provider searches are simply retried on a next search
2639        await self._cache_search_results(
2640            cache_key,
2641            result,
2642            # plugin providers do not declare is_streaming_provider,
2643            # treat them as local so their results only get the short expiration
2644            SEARCH_CACHE_EXPIRATION_STREAMING_PROVIDER
2645            if getattr(prov, "is_streaming_provider", False)
2646            else SEARCH_CACHE_EXPIRATION_LOCAL_PROVIDER,
2647            prov.instance_id,
2648        )
2649        return result
2650
2651    async def _cache_search_results(
2652        self, cache_key: str, result: SearchResults, expiration: int, provider: str
2653    ) -> None:
2654        """Store search results in the cache, logging (instead of raising) any cache errors."""
2655        try:
2656            await self.mass.cache.set(
2657                key=cache_key,
2658                data=result.to_dict(),
2659                expiration=expiration,
2660                provider=provider,
2661                category=CACHE_CATEGORY_SEARCH_RESULTS,
2662            )
2663        except Exception as err:
2664            self.logger.warning("Failed to cache search results for %s: %s", provider, str(err))
2665
2666    def _get_covered_media_types(
2667        self, library_results: SearchResults, search_query: str
2668    ) -> set[tuple[MediaType, str]]:
2669        """
2670        Return the (media_type, provider domain/instance) pairs covered by the library.
2671
2672        A pair is considered covered when the library holds a (near) exact name match
2673        for the search query that is mapped to that provider.
2674        """
2675        covered: set[tuple[MediaType, str]] = set()
2676        # extract the artist and title part in case the
2677        # query is formatted as "artist - title"
2678        if " - " in search_query:
2679            artist_part, title_part = search_query.split(" - ", 1)
2680        else:
2681            artist_part, title_part = None, search_query
2682        items: Sequence[MediaItemType | ItemMapping]
2683        for items in (
2684            library_results.artists,
2685            library_results.albums,
2686            library_results.tracks,
2687            library_results.playlists,
2688            library_results.radio,
2689            library_results.audiobooks,
2690            library_results.podcasts,
2691        ):
2692            for item in items:
2693                if compare_strings(item.name, search_query, strict=False):
2694                    pass
2695                elif artist_part and compare_strings(item.name, title_part, strict=False):
2696                    # the item name matches the title part only,
2697                    # so the artist part must match one of the item artists
2698                    if not any(
2699                        compare_strings(artist.name, artist_part, strict=False)
2700                        for artist in getattr(item, "artists", [])
2701                    ):
2702                        continue
2703                else:
2704                    continue
2705                for prov_mapping in cast("MediaItemType", item).provider_mappings:
2706                    if not prov_mapping.available:
2707                        continue
2708                    covered.add((item.media_type, prov_mapping.provider_domain))
2709                    covered.add((item.media_type, prov_mapping.provider_instance))
2710        return covered
2711
2712    def _import_album_tracks_if_enabled(self, album: Album) -> None:
2713        """Import all album tracks into the library for providers that have this enabled."""
2714        for prov_mapping in album.provider_mappings:
2715            # only consider mappings the album was actually added on; additional
2716            # mappings auto-created for other instances of the same provider
2717            # (via match_provider_instances) carry in_library=None and must be skipped
2718            if not prov_mapping.in_library:
2719                continue
2720            provider = self.mass.get_provider(prov_mapping.provider_instance)
2721            if not isinstance(provider, MusicProvider):
2722                continue
2723            if not provider.library_sync_album_tracks_enabled():
2724                continue
2725            self.mass.create_task(provider.import_album_tracks(prov_mapping.item_id, album))
2726
2727    async def _get_provider_sound_effects(self, provider: MusicProvider) -> list[SoundEffect]:
2728        """Return all sound effect items from a single provider."""
2729        try:
2730            return [item async for item in provider.get_sound_effects()]
2731        except Exception as err:
2732            self.logger.warning(
2733                "Error while fetching sound effects from %s: %s",
2734                provider.name,
2735                str(err),
2736                exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
2737            )
2738            return []
2739
2740    def _create_provider_sync_handler(
2741        self, provider: MusicProvider, media_type: MediaType
2742    ) -> Callable[[], Awaitable[None]]:
2743        """Create the coroutine used for a managed provider sync task."""
2744
2745        async def run_sync() -> None:
2746            try:
2747                async with self._sync_lock:
2748                    # suppress per-item events during sync; a large library would otherwise
2749                    # emit one (serialized per client) for every item. Subscribers refresh
2750                    # on MUSIC_SYNC_COMPLETED and track progress via TASKS_UPDATED instead.
2751                    token = SUPPRESS_MEDIA_ITEM_UPDATES.set(True)
2752                    try:
2753                        await provider.sync_library(media_type)
2754                    finally:
2755                        SUPPRESS_MEDIA_ITEM_UPDATES.reset(token)
2756            finally:
2757                self.mass.call_later(
2758                    0,
2759                    self._handle_sync_completion_check,
2760                    task_id=MUSIC_SYNC_COMPLETION_CHECK_TASK_ID,
2761                )
2762
2763        return run_sync
2764
2765    def _get_sync_task_id(self, provider: MusicProvider | str, media_type: MediaType) -> str:
2766        """Return deterministic task id for a provider sync."""
2767        provider_instance = (
2768            provider.instance_id if isinstance(provider, MusicProvider) else provider
2769        )
2770        return f"music_sync_{provider_instance}_{media_type.value}"
2771
2772    def _get_sync_task_name(self, provider: MusicProvider, media_type: MediaType) -> str:
2773        """Return display name for a provider sync task."""
2774        return f"Sync {provider.name} {media_type.value}s"
2775
2776    def _get_sync_task_translation_key(self, media_type: MediaType) -> str:
2777        """Return translation key for a provider sync task."""
2778        if media_type == MediaType.ARTIST:
2779            return "sync_provider_artists"
2780        if media_type == MediaType.ALBUM:
2781            return "sync_provider_albums"
2782        if media_type == MediaType.TRACK:
2783            return "sync_provider_tracks"
2784        if media_type == MediaType.PLAYLIST:
2785            return "sync_provider_playlists"
2786        if media_type == MediaType.RADIO:
2787            return "sync_provider_radios"
2788        if media_type == MediaType.AUDIOBOOK:
2789            return "sync_provider_audiobooks"
2790        if media_type == MediaType.PODCAST:
2791            return "sync_provider_podcasts"
2792        return "settings.sync"
2793
2794    def _get_sync_task_metadata(
2795        self, provider: MusicProvider, media_type: MediaType
2796    ) -> TaskMetadata:
2797        """Return metadata for a provider sync task."""
2798        return {
2799            "task_domain": "music_sync",
2800            "provider_domain": provider.domain,
2801            "provider_instance": provider.instance_id,
2802            "provider_name": provider.name,
2803            "media_type": media_type.value,
2804        }
2805
2806    def _handle_sync_completion_check(self) -> None:
2807        """Run follow-up maintenance when no provider sync tasks remain active."""
2808        if self.active_sync_tasks:
2809            return
2810        self.mass.signal_event(EventType.MUSIC_SYNC_COMPLETED)
2811        # freshly synced content is the only source of new duplicates, so the reconciliation
2812        # pass owes the library another walk; it starts once the current one reaches the end,
2813        # since rewinding right now would keep re-examining the same prefix forever
2814        self._set_track_reconciliation_state(self._track_reconciliation_cursor, True)
2815        self._queue_database_cleanup_task()
2816        # prune sidebar shortcuts for library items that no longer exist
2817        self.mass.create_task(self.cleanup_library_shortcuts())
2818
2819    def _register_database_cleanup_task(self) -> BackgroundTask:
2820        """Register the recurring database cleanup background task."""
2821        utc_hour, utc_minute = local_clock_time_to_utc(5, 0)
2822        desired_schedule = TaskSchedule.daily(hour=utc_hour, minute=utc_minute)
2823        return self.mass.tasks.register_scheduled_task(
2824            task_id=DATABASE_CLEANUP_TASK_ID,
2825            name="Database cleanup",
2826            handler=self._cleanup_database,
2827            schedule=desired_schedule,
2828            translation_key="database_cleanup",
2829            translation_owner=self.translation_owner,
2830            metadata={
2831                "task_domain": "music_database_cleanup",
2832            },
2833            allow_retry=True,
2834        )
2835
2836    def _register_provider_mapping_correction_task(self) -> BackgroundTask:
2837        """Register the recurring provider mapping correction background task."""
2838        utc_hour, utc_minute = local_clock_time_to_utc(4, 0)
2839        desired_schedule = TaskSchedule.daily(every=30, hour=utc_hour, minute=utc_minute)
2840        return self.mass.tasks.register_scheduled_task(
2841            task_id=PROVIDER_MAPPING_CORRECTION_TASK_ID,
2842            name="Correct provider mappings",
2843            handler=self.correct_multi_instance_provider_mappings,
2844            schedule=desired_schedule,
2845            translation_key="correct_provider_mappings",
2846            translation_owner=self.translation_owner,
2847            metadata={
2848                "task_domain": "music_provider_mapping_correction",
2849            },
2850            allow_retry=True,
2851        )
2852
2853    def _register_track_reconciliation_task(self) -> BackgroundTask:
2854        """Register the recurring duplicate track reconciliation background task."""
2855        # runs every hour rather than spread across the day: it is bounded to a small
2856        # batch of candidates per run and never leaves the local database
2857        return self.mass.tasks.register_scheduled_task(
2858            task_id=TRACK_RECONCILIATION_TASK_ID,
2859            name="Reconcile duplicate tracks",
2860            handler=self._reconcile_duplicate_tracks,
2861            schedule=TaskSchedule.hourly(),
2862            translation_key="reconcile_duplicate_tracks",
2863            translation_owner=self.translation_owner,
2864            metadata={
2865                "task_domain": "music_track_reconciliation",
2866            },
2867            allow_retry=True,
2868        )
2869
2870    async def _reconcile_duplicate_tracks(self) -> None:
2871        """Merge a small batch of library tracks that are held twice across providers."""
2872        if self.active_sync_tasks:
2873            # a sync is still filling in albums and mappings, so hold off rather than
2874            # judge duplicates against a half-populated library
2875            update_current_task_progress_text("Waiting for music sync completion")
2876            return
2877        self._start_next_pass_if_due()
2878        if (cursor := self._track_reconciliation_cursor) is None:
2879            # the library has been walked end to end and nothing has been synced since,
2880            # so there is nothing to look for: skip the query rather than scan for a miss
2881            update_current_task_progress_text("No duplicate tracks found")
2882            return
2883        update_current_task_progress_text("Searching for duplicate tracks")
2884        rows = await self.database.get_rows_from_query(
2885            _DUPLICATE_TRACK_CANDIDATES_QUERY,
2886            {
2887                "max_duration_delta": TRACK_RECONCILIATION_MAX_DURATION_DELTA,
2888                "cursor_item_id_1": cursor[0],
2889                "cursor_item_id_2": cursor[1],
2890            },
2891            limit=TRACK_RECONCILIATION_BATCH_SIZE,
2892        )
2893        if not rows:
2894            self._set_track_reconciliation_state(None, self._track_reconciliation_rescan_due)
2895            update_current_task_progress_text("No duplicate tracks found")
2896            return
2897        merged = 0
2898        retry_due = False
2899        examined = cursor
2900        try:
2901            for index, row in enumerate(rows, 1):
2902                update_current_task_progress_from_index(
2903                    index, len(rows), f"Checking duplicate track {index}/{len(rows)}"
2904                )
2905                try:
2906                    if await self._merge_duplicate_track_pair(
2907                        int(row["item_id_1"]), int(row["item_id_2"])
2908                    ):
2909                        merged += 1
2910                except MediaNotFoundError:
2911                    # an earlier merge in this batch already absorbed one of the two rows
2912                    pass
2913                except MusicAssistantError as err:
2914                    # a pair that failed on something transient deserves another look
2915                    retry_due = True
2916                    report_current_task_failure(str(err))
2917                    self.logger.warning(
2918                        "Error while reconciling duplicate tracks %s and %s: %s",
2919                        row["item_id_1"],
2920                        row["item_id_2"],
2921                        str(err),
2922                        exc_info=err if self.logger.isEnabledFor(logging.DEBUG) else None,
2923                    )
2924                examined = (int(row["item_id_1"]), int(row["item_id_2"]))
2925        finally:
2926            # resume after the pair examined last, so candidates this run refused can never
2927            # starve the ones behind them, not even a further pair of the same track that the
2928            # batch boundary cut off. Recording it even when the run is cut short keeps the
2929            # pairs it did not reach for the next run rather than skipping past them.
2930            walked_to_end = len(rows) < TRACK_RECONCILIATION_BATCH_SIZE and examined == (
2931                int(rows[-1]["item_id_1"]),
2932                int(rows[-1]["item_id_2"]),
2933            )
2934            # a merge moves album and artist relations onto the surviving row, which can make
2935            # it a duplicate of a row this walk has already passed, so ask for another pass
2936            self._set_track_reconciliation_state(
2937                None if walked_to_end else examined,
2938                self._track_reconciliation_rescan_due or merged > 0 or retry_due,
2939            )
2940        update_current_task_progress(100, f"Merged {merged} duplicate track(s)")
2941
2942    def _restore_track_reconciliation_state(self) -> None:
2943        """Pick the duplicate track walk back up where the previous run left it."""
2944        cursor = self.mass.config.get_raw_core_config_value(
2945            self.domain, CONF_TRACK_RECONCILIATION_CURSOR, [0, 0]
2946        )
2947        self._track_reconciliation_cursor = (
2948            (int(cursor[0]), int(cursor[1])) if len(cursor) == 2 else None
2949        )
2950        self._track_reconciliation_rescan_due = bool(
2951            self.mass.config.get_raw_core_config_value(
2952                self.domain, CONF_TRACK_RECONCILIATION_RESCAN_DUE, False
2953            )
2954        )
2955
2956    def _set_track_reconciliation_state(
2957        self, cursor: tuple[int, int] | None, rescan_due: bool
2958    ) -> None:
2959        """
2960        Record how far the duplicate track walk has come, surviving a restart.
2961
2962        :param cursor: The pair examined last, or None once the walk reached the end.
2963        :param rescan_due: Whether a completed sync still owes the library another pass.
2964        """
2965        self._track_reconciliation_cursor = cursor
2966        self._track_reconciliation_rescan_due = rescan_due
2967        self.mass.config.set_raw_core_config_value(
2968            self.domain, CONF_TRACK_RECONCILIATION_CURSOR, list(cursor) if cursor else []
2969        )
2970        self.mass.config.set_raw_core_config_value(
2971            self.domain, CONF_TRACK_RECONCILIATION_RESCAN_DUE, rescan_due
2972        )
2973
2974    def _start_next_pass_if_due(self) -> None:
2975        """Rewind the duplicate track walk if a sync has added content and the walk is done."""
2976        # rewinding a walk still in progress would keep re-examining the same first
2977        # candidates, so a pending rescan waits for the current one to reach the end
2978        if not self._track_reconciliation_rescan_due:
2979            return
2980        if self._track_reconciliation_cursor is not None:
2981            return
2982        self._set_track_reconciliation_state((0, 0), False)
2983
2984    async def _albums_agree_on_edition(self, item_id_1: int, item_id_2: int) -> bool:
2985        """
2986        Check that two tracks share an album whose edition matches as well as its title.
2987
2988        :param item_id_1: Library ID of the first track.
2989        :param item_id_2: Library ID of the second track.
2990        """
2991        # the query relates titles loosely so a spelled-out retail suffix cannot hide a
2992        # shared album, which leaves the identity for the album comparison to confirm. An
2993        # edition is held apart from the title: without that an original and its remaster or
2994        # deluxe edition look like the same album whenever neither track carries a version
2995        rows = await self.database.get_rows_from_query(
2996            _SHARED_ALBUM_EDITIONS_QUERY,
2997            {"item_id_1": item_id_1, "item_id_2": item_id_2},
2998        )
2999        return any(
3000            compare_album_name(row["name_1"], row["name_2"])
3001            and compare_version(row["version_1"], row["version_2"])
3002            for row in rows
3003        )
3004
3005    async def _merge_duplicate_track_pair(self, item_id_1: int, item_id_2: int) -> bool:
3006        """
3007        Merge two candidate rows if they are confirmed to be the same track.
3008
3009        :param item_id_1: Library ID of the lower-numbered candidate row.
3010        :param item_id_2: Library ID of the higher-numbered candidate row.
3011        :return: True when the rows were merged, False when they were left alone.
3012        """
3013        track_1 = await self.tracks.get_library_item(item_id_1)
3014        track_2 = await self.tracks.get_library_item(item_id_2)
3015        # the checks below establish that both rows sit at the same position on an equally
3016        # titled album, which is the album agreement strict mode looks for, so the remaining
3017        # check is run in non-strict mode. Its version check is reinstated here
3018        # explicitly: without it a remaster, remix or radio edit of equal length would be
3019        # accepted as the original.
3020        if not compare_version(track_1.version, track_2.version):
3021            return False
3022        if not await self._albums_agree_on_edition(item_id_1, item_id_2):
3023            return False
3024        if not compare_track(track_1, track_2, strict=False):
3025            return False
3026        # keep the row that carries the most provider mappings so the fewest mappings and
3027        # relations have to move; equal counts keep the oldest row, which the query orders first
3028        target, source = (
3029            (track_1, track_2)
3030            if len(track_1.provider_mappings) >= len(track_2.provider_mappings)
3031            else (track_2, track_1)
3032        )
3033        self.logger.debug(
3034            "Merging duplicate track %s (id %s) into id %s",
3035            target.name,
3036            source.item_id,
3037            target.item_id,
3038        )
3039        await self.tracks.merge_library_items(target.item_id, source.item_id)
3040        return True
3041
3042    def _queue_database_cleanup_task(self) -> BackgroundTask:
3043        """Queue the post-sync database cleanup as a managed background task."""
3044        self._register_database_cleanup_task()
3045        return self.mass.tasks.run_task(DATABASE_CLEANUP_TASK_ID)
3046
3047    async def _schedule_provider_mediatype_sync(
3048        self, provider: MusicProvider, media_type: MediaType, is_initial: bool = False
3049    ) -> None:
3050        """Schedule Library sync for given provider and media type."""
3051        # handle mediatype specific sync config
3052        conf_key = f"library_sync_{media_type}s"
3053        sync_conf: ConfigValueType = await self.mass.config.get_provider_config_value(
3054            provider.instance_id, conf_key
3055        )
3056        if not sync_conf:
3057            self.mass.tasks.unregister_scheduled_task(self._get_sync_task_id(provider, media_type))
3058            return
3059        self.mass.tasks.register_scheduled_task(
3060            task_id=self._get_sync_task_id(provider, media_type),
3061            name=self._get_sync_task_name(provider, media_type),
3062            handler=self._create_provider_sync_handler(provider, media_type),
3063            schedule=provider.get_default_library_sync_schedule(media_type),
3064            initial_delay=INITIAL_SYNC_DELAY if is_initial else None,
3065            translation_key=self._get_sync_task_translation_key(media_type),
3066            translation_args=[provider.name],
3067            translation_owner=self.translation_owner,
3068            metadata=self._get_sync_task_metadata(provider, media_type),
3069            allow_retry=True,
3070        )
3071
3072    async def _get_user_for_provider(
3073        self, provider_mappings_or_instance_id: Iterable[ProviderMapping] | str
3074    ) -> User | None:
3075        """Try to get the MA User based on provider mappings and provider filter."""
3076        all_users = await self.mass.webserver.auth.list_users()
3077        for mapping_or_instance_id in provider_mappings_or_instance_id:
3078            for user in all_users:
3079                if not user.provider_filter:
3080                    continue
3081                if isinstance(mapping_or_instance_id, str):
3082                    if provider_mappings_or_instance_id in user.provider_filter:
3083                        return user
3084                elif mapping_or_instance_id.provider_instance in user.provider_filter:
3085                    return user
3086        return None
3087
3088    async def _resolve_playlog_item(self, media_item: MediaItemType | ItemMapping) -> MediaItemType:
3089        """
3090        Return the full media item for a (possibly minimized) media item reference.
3091
3092        :param media_item: The media item to resolve, either full or an ItemMapping.
3093        """
3094        if not isinstance(media_item, ItemMapping):
3095            return media_item
3096        resolved = await self.get_item(
3097            media_item.media_type,
3098            media_item.item_id,
3099            media_item.provider,
3100            allow_update_metadata=False,
3101        )
3102        if isinstance(resolved, BrowseFolder):
3103            msg = f"{media_item.uri} does not resolve to a media item"
3104            raise MediaNotFoundError(msg)
3105        return resolved
3106
3107    async def _upsert_playlog(self, entry: dict[str, Any]) -> None:
3108        """
3109        Write a playlog row, updating the existing row for the item/user if there is one.
3110
3111        Columns left out of the entry keep whatever the existing row holds, and
3112        `user_initiated` is sticky: once a play was explicitly user-initiated it stays that
3113        way for the lifetime of the row, so a later side-effect credit (an autoplay replay,
3114        or a track crediting its album/artist) can never demote it and drop the item out of
3115        the "recently played" recommendations.
3116
3117        The generic `database.upsert()` cannot express either half of that: the sticky OR is
3118        playlog-specific, and it needs an explicit conflict target because the playlog carries
3119        more than one unique constraint.
3120
3121        :param entry: The playlog column values to write, including all of
3122            `PLAYLOG_CONFLICT_KEYS`.
3123        """
3124        columns = list(entry)
3125        updates = [
3126            f"user_initiated = {DB_TABLE_PLAYLOG}.user_initiated OR excluded.user_initiated"
3127            if column == "user_initiated"
3128            else f"{column} = excluded.{column}"
3129            for column in columns
3130            if column not in PLAYLOG_CONFLICT_KEYS
3131        ]
3132        await self.database.execute_write(
3133            f"INSERT INTO {DB_TABLE_PLAYLOG} ({', '.join(columns)}) "
3134            f"VALUES ({', '.join(f':{column}' for column in columns)}) "
3135            f"ON CONFLICT({', '.join(PLAYLOG_CONFLICT_KEYS)}) DO UPDATE SET {', '.join(updates)}",
3136            entry,
3137        )
3138
3139    def _signal_playlog_updated(
3140        self,
3141        item: MediaItemType | ItemMapping,
3142        *,
3143        fully_played: bool,
3144        seconds_played: int,
3145        userid: str | None,
3146    ) -> None:
3147        """
3148        Signal that the playlog entry of the given item changed.
3149
3150        :param item: The item as it is keyed in the playlog.
3151        :param fully_played: The new fully played state of the item.
3152        :param seconds_played: The new resume position of the item.
3153        :param userid: The user the change applies to, or None for all users.
3154        """
3155        assert item.uri is not None
3156        self.mass.signal_event(
3157            EventType.PLAYLOG_UPDATED,
3158            object_id=item.uri,
3159            data=PlaylogUpdate(
3160                uri=item.uri,
3161                media_type=item.media_type,
3162                fully_played=fully_played,
3163                seconds_played=seconds_played,
3164                userid=userid,
3165            ),
3166        )
3167
3168    async def _credit_artist_plays(
3169        self,
3170        artists: Iterable[Artist | ItemMapping],
3171        *,
3172        timestamp: float,
3173        user_ids: list[str],
3174        queue_id: str | None,
3175        skip_ids: set[str],
3176    ) -> None:
3177        """Credit each (library-resolvable) artist with a play, skipping skip_ids."""
3178        for artist in artists:
3179            db_artist = await self.artists.get_library_item_by_prov_id(
3180                artist.item_id, artist.provider
3181            )
3182            if db_artist is None:
3183                continue
3184            if db_artist.item_id in skip_ids:
3185                self.logger.debug("Skipping already-credited artist '%s'", db_artist.name)
3186                continue
3187            await self.database.execute(
3188                f"UPDATE {self.artists.db_table} SET play_count = play_count + 1, "
3189                f"last_played = {timestamp} WHERE item_id = {db_artist.item_id}"
3190            )
3191            self.logger.debug("Credited play for artist '%s'", db_artist.name)
3192            playlog_entry: dict[str, Any] = {
3193                "item_id": db_artist.item_id,
3194                "provider": "library",
3195                "media_type": MediaType.ARTIST.value,
3196                "name": db_artist.name,
3197                "image": serialize_to_json(db_artist.image.to_dict()) if db_artist.image else None,
3198                "fully_played": True,
3199                "seconds_played": None,
3200                "timestamp": timestamp,
3201                "queue_id": queue_id,
3202                "user_initiated": False,
3203            }
3204            for user_id in user_ids:
3205                playlog_entry["userid"] = user_id
3206                await self._upsert_playlog(playlog_entry)
3207            self._signal_playlog_updated(
3208                db_artist,
3209                fully_played=True,
3210                seconds_played=0,
3211                userid=user_ids[0] if len(user_ids) == 1 else None,
3212            )
3213
3214    async def _credit_podcast_play(
3215        self,
3216        podcast: Podcast | ItemMapping,
3217        *,
3218        timestamp: float,
3219        user_ids: list[str],
3220        queue_id: str | None,
3221    ) -> None:
3222        """Credit the parent podcast with a play so the show surfaces in recently played."""
3223        # Resolve to the library item first, like _credit_artist_plays does, so an episode's
3224        # parent-podcast credit lands on the same library-scoped row as an explicit play of the
3225        # library show, instead of creating a separate provider-scoped duplicate.
3226        db_podcast = await self.podcasts.get_library_item_by_prov_id(
3227            podcast.item_id, podcast.provider
3228        )
3229        credited_podcast: Podcast | ItemMapping = db_podcast if db_podcast else podcast
3230        playlog_entry: dict[str, Any] = {
3231            "item_id": credited_podcast.item_id,
3232            "provider": "library" if db_podcast else podcast.provider,
3233            "media_type": MediaType.PODCAST.value,
3234            "name": credited_podcast.name,
3235            "image": serialize_to_json(credited_podcast.image.to_dict())
3236            if credited_podcast.image
3237            else None,
3238            "fully_played": True,
3239            "seconds_played": None,
3240            "timestamp": timestamp,
3241            "queue_id": queue_id,
3242            "user_initiated": False,
3243        }
3244        for user_id in user_ids:
3245            playlog_entry["userid"] = user_id
3246            await self._upsert_playlog(playlog_entry)
3247        self._signal_playlog_updated(
3248            credited_podcast,
3249            fully_played=True,
3250            seconds_played=0,
3251            userid=user_ids[0] if len(user_ids) == 1 else None,
3252        )
3253
3254    async def _get_item_by_name(
3255        self,
3256        name: str,
3257        artist: str | None = None,
3258        album: str | None = None,
3259        media_type: MediaType | None = None,
3260    ) -> MediaItemType | ItemMapping | None:
3261        """Try to find a media item (such as a playlist) by name."""
3262        # Future todo: enhance this method with AI capabilities to allow typos and
3263        # natural language.
3264        searchname = name.lower()
3265        allowed_media_types = [
3266            MediaType.PLAYLIST,
3267            MediaType.RADIO,
3268            MediaType.TRACK,
3269            MediaType.ALBUM,
3270            MediaType.ARTIST,
3271            MediaType.AUDIOBOOK,
3272            MediaType.PODCAST,
3273        ]
3274        if media_type in (None, MediaType.UNKNOWN):
3275            media_types = allowed_media_types
3276        elif media_type not in allowed_media_types:
3277            raise InvalidDataError(
3278                f"{media_type} is not a supported media_type. "
3279                f"Supported media_types are {allowed_media_types}"
3280            )
3281        else:
3282            media_types = [media_type]
3283        library_functions = [
3284            self.get_controller(media_type).library_items for media_type in media_types
3285        ]
3286        # prefer (exact) lookup in the library by name
3287        for func in library_functions:
3288            result = await func(search=searchname)
3289            for item in result:
3290                # handle optional artist filter
3291                if (
3292                    artist
3293                    and (artists := getattr(item, "artists", None))
3294                    and not any(x for x in artists if x.name.lower() == artist.lower())
3295                ):
3296                    continue
3297                # handle optional album filter
3298                if (
3299                    album
3300                    and (item_album := getattr(item, "album", None))
3301                    and item_album.name.lower() != album.lower()
3302                ):
3303                    continue
3304                if searchname == item.name.lower():
3305                    return item
3306        # nothing found in the library, fallback to global search
3307        search_name = name
3308        if album and artist:
3309            search_name = f"{artist} - {album} - {name}"
3310        elif album:
3311            search_name = f"{album} - {name}"
3312        elif artist:
3313            search_name = f"{artist} - {name}"
3314        search_results = await self.search(
3315            search_query=search_name,
3316            media_types=[media_type]
3317            if media_type and media_type != MediaType.UNKNOWN
3318            else MediaType.ALL,
3319            limit=8,
3320        )
3321        for results in (
3322            search_results.tracks,
3323            search_results.albums,
3324            search_results.playlists,
3325            search_results.artists,
3326            search_results.radio,
3327            search_results.audiobooks,
3328            search_results.podcasts,
3329        ):
3330            for _item in results:
3331                # simply return the first item because search is already sorted by best match
3332                return _item
3333        return None
3334
3335    async def _handle_verify_item_uri(self, uri: str) -> bool:
3336        user = get_current_user()
3337
3338        try:
3339            media_type, provider_instance_id_or_domain, item_id = await parse_uri(uri)
3340        except InvalidProviderURI, InvalidProviderID:
3341            return False
3342
3343        # fast return for a provider uri which is not part of a user with a provider filter
3344        if (
3345            provider_instance_id_or_domain != "library"
3346            and user
3347            and user.provider_filter
3348            and provider_instance_id_or_domain not in user.provider_filter
3349        ):
3350            return False
3351
3352        # verify that item itself exists
3353        try:
3354            item = await self.get_item(
3355                media_type=media_type,
3356                item_id=item_id,
3357                provider_instance_id_or_domain=provider_instance_id_or_domain,
3358                allow_update_metadata=False,  # no need trigger more methods
3359            )
3360        except MediaNotFoundError, NotImplementedError:
3361            # NotImplementedError: the uri has a valid format, but specifies an unknown media type
3362            return False
3363
3364        # non library item handling for users with no filter, or no user at all
3365        if (
3366            provider_instance_id_or_domain != "library"
3367            or not user
3368            or (user and not user.provider_filter)
3369            or isinstance(item, BrowseFolder)
3370        ):
3371            return True
3372
3373        # library item handling for users with provider filter
3374        for provider_mapping in item.provider_mappings:
3375            if provider_mapping.provider_instance in user.provider_filter:
3376                return True
3377
3378        return False
3379