music-assistant-server

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