music-assistant-server

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