music-assistant-server

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