music-assistant-server

1"""Built-in/generic provider to handle media from files and (remote) urls."""
2
3from __future__ import annotations
4
5import asyncio
6import os
7import re
8from collections.abc import AsyncGenerator, Mapping
9from typing import TYPE_CHECKING, Final, cast
10from urllib.parse import urlparse
11
12import aiofiles
13from music_assistant_models.auth import Scope
14from music_assistant_models.background_task import TaskSchedule
15from music_assistant_models.enums import (
16    ContentType,
17    ExternalID,
18    ImageType,
19    MediaType,
20    ProviderFeature,
21    StreamType,
22)
23from music_assistant_models.errors import (
24    InvalidDataError,
25    InvalidProviderURI,
26    MediaNotFoundError,
27    ProviderUnavailableError,
28)
29from music_assistant_models.media_items import (
30    Artist,
31    AudioFormat,
32    MediaItem,
33    MediaItemImage,
34    MediaItemMetadata,
35    MediaItemType,
36    Playlist,
37    ProviderMapping,
38    Radio,
39    SoundEffect,
40    Track,
41    UniqueList,
42    media_from_dict,
43)
44from music_assistant_models.streamdetails import StreamDetails
45
46from music_assistant.constants import (
47    GENRE_ICONS_DIR_NAME,
48    MASS_LOGO,
49    PLAYLIST_MEDIA_TYPES,
50    RESOURCES_DIR,
51    VARIOUS_ARTISTS_FANART,
52    PlaylistPlayableItem,
53)
54from music_assistant.controllers.cache import use_cache
55from music_assistant.controllers.tasks.context import (
56    get_current_task_id,
57    report_current_task_failure,
58    update_current_task_progress_from_index,
59    update_current_task_progress_text,
60)
61from music_assistant.helpers.compare import compare_strings
62from music_assistant.helpers.external_ids import normalize_external_id
63from music_assistant.helpers.playlists import (
64    ImageInfo,
65    IsHLSPlaylist,
66    PlaylistItem,
67    ProviderMappingInfo,
68    construct_media_item_from_playlist_item,
69    fetch_playlist,
70    generate_m3u,
71    media_item_to_playlist_item,
72    parse_extinf_title,
73    parse_m3u,
74    parse_m3u_playlist_image,
75    parse_m3u_playlist_name,
76)
77from music_assistant.helpers.security import is_safe_path
78from music_assistant.helpers.tags import AudioTags, async_parse_tags
79from music_assistant.helpers.track_filter import filter_tracks, get_track_filter
80from music_assistant.helpers.uri import parse_uri
81from music_assistant.models.music_provider import MusicProvider
82
83from .constants import (
84    ALL_FAVORITE_TRACKS,
85    BUILTIN_PLAYLISTS,
86    BUILTIN_PLAYLISTS_ENTRIES,
87    COLLAGE_IMAGE_PLAYLISTS,
88    CONF_ENTRY_LIBRARY_SYNC_BACK_HIDDEN,
89    CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS_HIDDEN,
90    CONF_ENTRY_LIBRARY_SYNC_RADIOS_HIDDEN,
91    CONF_ENTRY_LIBRARY_SYNC_TRACKS_HIDDEN,
92    CONF_KEY_PLAYLISTS,
93    CONF_KEY_RADIOS,
94    CONF_KEY_TRACKS,
95    DEFAULT_FANART,
96    DEFAULT_THUMB,
97    DYNAMIC_BUILTIN_PLAYLISTS,
98    INFINITE_MIX,
99    INFINITE_MIX_FAVORITES,
100    RANDOM_ALBUM,
101    RANDOM_ARTIST,
102    RANDOM_TRACKS,
103    RECENTLY_ADDED_TRACKS,
104    RECENTLY_PLAYED,
105    StoredItem,
106)
107
108if TYPE_CHECKING:
109    from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
110    from music_assistant_models.provider import ProviderManifest
111
112    from music_assistant.mass import MusicAssistant
113    from music_assistant.models import ProviderInstanceType
114
115CACHE_CATEGORY_MEDIA_INFO: Final[int] = 1
116CACHE_CATEGORY_PLAYLISTS: Final[int] = 2
117
118SUPPORTED_FEATURES = {
119    ProviderFeature.BROWSE,
120    ProviderFeature.LIBRARY_TRACKS,
121    ProviderFeature.LIBRARY_RADIOS,
122    ProviderFeature.LIBRARY_PLAYLISTS,
123    ProviderFeature.LIBRARY_TRACKS_EDIT,
124    ProviderFeature.LIBRARY_RADIOS_EDIT,
125    ProviderFeature.LIBRARY_PLAYLISTS_EDIT,
126    ProviderFeature.PLAYLIST_CREATE,
127    ProviderFeature.PLAYLIST_CREATE_AUDIOBOOKS,
128    ProviderFeature.PLAYLIST_CREATE_PODCAST_EPISODES,
129    ProviderFeature.PLAYLIST_CREATE_RADIOS,
130    ProviderFeature.PLAYLIST_CREATE_MIXED,
131    ProviderFeature.PLAYLIST_TRACKS_EDIT,
132}
133
134
135async def setup(
136    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
137) -> ProviderInstanceType:
138    """Initialize provider(instance) with given configuration."""
139    return BuiltinProvider(mass, manifest, config, SUPPORTED_FEATURES)
140
141
142class BuiltinProvider(MusicProvider):
143    """Built-in/generic provider to handle (manually added) media from files and (remote) urls."""
144
145    _playlists_dir: str
146    _playlist_lock: asyncio.Lock
147    _playlist_locks: dict[str, asyncio.Lock]
148
149    async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
150        """Return Config entries to setup this provider."""
151        return (
152            *BUILTIN_PLAYLISTS_ENTRIES,
153            # hide some of the default (dynamic) entries for library management
154            CONF_ENTRY_LIBRARY_SYNC_TRACKS_HIDDEN,
155            CONF_ENTRY_LIBRARY_SYNC_PLAYLISTS_HIDDEN,
156            CONF_ENTRY_LIBRARY_SYNC_RADIOS_HIDDEN,
157            CONF_ENTRY_LIBRARY_SYNC_BACK_HIDDEN,
158        )
159
160    async def loaded_in_mass(self) -> None:
161        """Call after the provider has been loaded."""
162        self._playlist_lock = asyncio.Lock()
163        self._playlist_locks = {}
164        self._playlists_dir = os.path.join(self.mass.storage_path, "playlists")
165        if not await asyncio.to_thread(os.path.exists, self._playlists_dir):
166            await asyncio.to_thread(os.mkdir, self._playlists_dir)
167        await super().loaded_in_mass()
168        # run in the background to avoid blocking startup. besides migrating old-style
169        # playlists, this repairs entries whose manually set name or artwork no longer
170        # matches the builtin config, which is not a one-off.
171        # TODO: drop the config->M3U migration after MA 2.9, keep the repair pass
172        self.mass.tasks.register_scheduled_task(
173            task_id="migrate_builtin_playlists",
174            name="Builtin provider playlist migration",
175            handler=self._migrate_playlists,
176            schedule=TaskSchedule.hourly(every=24),
177            initial_delay=60,
178        )
179        # register API commands for manual item management
180        self.mass.register_api_command(
181            "builtin/add_radio", self.add_radio, required_scope=Scope.LIBRARY_WRITE
182        )
183        self.mass.register_api_command(
184            "builtin/add_track", self.add_track, required_scope=Scope.LIBRARY_WRITE
185        )
186
187    @property
188    def is_streaming_provider(self) -> bool:
189        """Return True if the provider is a streaming provider."""
190        return False
191
192    def get_default_library_sync_schedule(self, media_type: MediaType) -> TaskSchedule:
193        """Return the default recurring schedule for builtin library sync tasks."""
194        return TaskSchedule.hourly(every=3)
195
196    async def get_track(self, prov_track_id: str) -> Track:
197        """Get full track details by id."""
198        parsed_item = await self.parse_item(prov_track_id, requested_media_type=MediaType.TRACK)
199        assert isinstance(parsed_item, Track)
200        return parsed_item
201
202    async def get_radio(self, prov_radio_id: str) -> Radio:
203        """Get full radio details by id."""
204        parsed_item = await self.parse_item(prov_radio_id, force_radio=True)
205        assert isinstance(parsed_item, Radio)
206        return parsed_item
207
208    async def get_artist(self, prov_artist_id: str) -> Artist:
209        """Get full artist details by id."""
210        artist = prov_artist_id
211        # this is here for compatibility reasons only
212        return Artist(
213            item_id=artist,
214            provider=self.domain,
215            name=artist,
216            provider_mappings={
217                ProviderMapping(
218                    item_id=artist,
219                    provider_domain=self.domain,
220                    provider_instance=self.instance_id,
221                    available=False,
222                )
223            },
224        )
225
226    async def get_playlist(self, prov_playlist_id: str) -> Playlist:
227        """Get full playlist details by id."""
228        if prov_playlist_id in BUILTIN_PLAYLISTS:
229            # this is one of our builtin/default playlists
230            return Playlist(
231                item_id=prov_playlist_id,
232                provider=self.instance_id,
233                name=BUILTIN_PLAYLISTS[prov_playlist_id],
234                translation_key=prov_playlist_id,
235                provider_mappings={
236                    ProviderMapping(
237                        item_id=prov_playlist_id,
238                        provider_domain=self.domain,
239                        provider_instance=self.instance_id,
240                    )
241                },
242                owner="Music Assistant",
243                is_editable=False,
244                is_dynamic=prov_playlist_id in DYNAMIC_BUILTIN_PLAYLISTS,
245                metadata=MediaItemMetadata(
246                    images=UniqueList([DEFAULT_THUMB])
247                    if prov_playlist_id in COLLAGE_IMAGE_PLAYLISTS
248                    else UniqueList([DEFAULT_THUMB, DEFAULT_FANART]),
249                ),
250            )
251        # user created playlist - read from M3U file on disk
252        playlist_file = os.path.join(self._playlists_dir, f"{prov_playlist_id}.m3u")
253        if not await asyncio.to_thread(os.path.isfile, playlist_file):
254            raise MediaNotFoundError(f"Playlist file not found: {prov_playlist_id}")
255        # read playlist name and image from M3U
256        m3u_data = await self._read_m3u_file(prov_playlist_id)
257        playlist_name = parse_m3u_playlist_name(m3u_data) or prov_playlist_id
258        metadata = MediaItemMetadata()
259        if image_url := parse_m3u_playlist_image(m3u_data):
260            metadata.images = UniqueList(
261                [
262                    MediaItemImage(
263                        type=ImageType.THUMB,
264                        path=image_url,
265                        provider=self.domain,
266                        remotely_accessible=image_url.startswith("http"),
267                    )
268                ]
269            )
270        return Playlist(
271            item_id=prov_playlist_id,
272            provider=self.instance_id,
273            name=playlist_name,
274            provider_mappings={
275                ProviderMapping(
276                    item_id=prov_playlist_id,
277                    provider_domain=self.domain,
278                    provider_instance=self.instance_id,
279                )
280            },
281            owner="Music Assistant",
282            # MediaType.SOUND_EFFECT is deliberately left out here: clients that do not
283            # know this media type yet reject the entire playlist listing when they
284            # receive it. Sound effects can still be added to these playlists, as the
285            # builtin provider accepts any uri regardless of this (advisory) set.
286            supported_mediatypes={
287                MediaType.AUDIOBOOK,
288                MediaType.PODCAST_EPISODE,
289                MediaType.RADIO,
290                MediaType.TRACK,
291            },
292            is_editable=True,
293            metadata=metadata,
294        )
295
296    async def get_library_tracks(self) -> AsyncGenerator[Track]:
297        """Retrieve library tracks from the provider."""
298        stored_items: list[StoredItem] = self.mass.config.get(CONF_KEY_TRACKS, [])
299        for item in stored_items:
300            try:
301                yield await self.get_track(item["item_id"])
302            except MediaNotFoundError as err:
303                self.report_skipped_sync_item(MediaType.TRACK, item["item_id"], err)
304
305    async def get_library_playlists(self) -> AsyncGenerator[Playlist]:
306        """Retrieve library/subscribed playlists from the provider."""
307        # return user stored playlists from M3U files on disk
308        for filename in await asyncio.to_thread(os.listdir, self._playlists_dir):
309            if not filename.endswith(".m3u"):
310                continue
311            playlist_id = filename[:-4]  # strip .m3u extension
312            try:
313                yield await self.get_playlist(playlist_id)
314            except MediaNotFoundError as err:
315                self.report_skipped_sync_item(MediaType.PLAYLIST, playlist_id, err)
316        # return builtin playlists
317        for item_id in BUILTIN_PLAYLISTS:
318            if self.config.get_value(item_id) is False:
319                continue
320            yield await self.get_playlist(item_id)
321
322    async def get_library_radios(self) -> AsyncGenerator[Radio]:
323        """Retrieve library/subscribed radio stations from the provider."""
324        stored_items: list[StoredItem] = self.mass.config.get(CONF_KEY_RADIOS, [])
325        for item in stored_items:
326            try:
327                yield await self.get_radio(item["item_id"])
328            except (MediaNotFoundError, InvalidDataError) as err:
329                self.logger.warning("Radio station %s not found: %s", item, err)
330                yield Radio(
331                    item_id=item["item_id"],
332                    provider=self.instance_id,
333                    name=item["name"],
334                    provider_mappings={
335                        ProviderMapping(
336                            item_id=item["item_id"],
337                            provider_domain=self.domain,
338                            provider_instance=self.instance_id,
339                            available=False,
340                        )
341                    },
342                )
343
344    async def library_add(self, item: MediaItemType) -> bool:
345        """Add item to provider's library. Return true on success."""
346        if item.media_type == MediaType.TRACK:
347            key = CONF_KEY_TRACKS
348        elif item.media_type == MediaType.RADIO:
349            key = CONF_KEY_RADIOS
350        else:
351            return False
352        stored_item = StoredItem(item_id=item.item_id, name=item.name)
353        if item.image:
354            stored_item["image_url"] = item.image.path
355        stored_items: list[StoredItem] = self.mass.config.get(key, [])
356        # filter out existing
357        stored_items = [x for x in stored_items if x["item_id"] != item.item_id]
358        stored_items.append(stored_item)
359        self.mass.config.set(key, stored_items)
360        return True
361
362    async def library_remove(self, prov_item_id: str, media_type: MediaType) -> bool:
363        """Remove item from provider's library. Return true on success."""
364        if media_type == MediaType.PLAYLIST and prov_item_id in BUILTIN_PLAYLISTS:
365            # user wants to disable/remove one of our builtin playlists
366            # to prevent it comes back, we mark it as disabled in config
367            self._update_config_value(prov_item_id, False)
368            return True
369        if media_type == MediaType.TRACK:
370            # regular manual track URL/path
371            key = CONF_KEY_TRACKS
372        elif media_type == MediaType.RADIO:
373            # regular manual radio URL/path
374            key = CONF_KEY_RADIOS
375        elif media_type == MediaType.PLAYLIST:
376            # user-created playlist removal - delete the M3U file
377            playlist_file = os.path.join(self._playlists_dir, f"{prov_item_id}.m3u")
378            if await asyncio.to_thread(os.path.isfile, playlist_file):
379                async with self._playlist_lock:
380                    await asyncio.to_thread(os.remove, playlist_file)
381            return True
382        else:
383            return False
384        stored_items: list[StoredItem] = self.mass.config.get(key, [])
385        stored_items = [x for x in stored_items if x["item_id"] != prov_item_id]
386        self.mass.config.set(key, stored_items)
387        return True
388
389    async def on_item_updated(self, item: MediaItemType) -> None:
390        """
391        Update stored item config when a library item is edited.
392
393        :param item: The updated media item with new metadata.
394        """
395        # find the builtin provider mapping to get the item_id
396        builtin_mapping = next(
397            (pm for pm in item.provider_mappings if pm.provider_domain == self.domain),
398            None,
399        )
400        if not builtin_mapping:
401            return
402
403        if item.media_type == MediaType.PLAYLIST:
404            image_url = item.image.path if item.image else None
405            await self._update_playlist_metadata(builtin_mapping.item_id, item.name, image_url)
406            return
407
408        if item.media_type == MediaType.RADIO:
409            key = CONF_KEY_RADIOS
410        elif item.media_type == MediaType.TRACK:
411            key = CONF_KEY_TRACKS
412        else:
413            return
414
415        # TODO: also allow updating description and other image types
416        stored_items: list[StoredItem] = self.mass.config.get(key, [])
417        for stored_item in stored_items:
418            if stored_item["item_id"] == builtin_mapping.item_id:
419                stored_item["name"] = item.name
420                if item.image:
421                    stored_item["image_url"] = item.image.path
422                elif "image_url" in stored_item:
423                    del stored_item["image_url"]
424                break
425        self.mass.config.set(key, stored_items)
426
427    async def add_radio(self, url: str, name: str, image_url: str | None = None) -> Radio:
428        """
429        Add a radio station.
430
431        :param url: Stream URL.
432        :param name: Display name.
433        :param image_url: Image URL.
434        """
435        stored_items: list[StoredItem] = self.mass.config.get(CONF_KEY_RADIOS, [])
436        # Remove existing entry with same URL if present
437        stored_items = [x for x in stored_items if x["item_id"] != url]
438        stored_item = StoredItem(item_id=url, name=name)
439        if image_url:
440            stored_item["image_url"] = image_url
441        stored_items.append(stored_item)
442        self.mass.config.set(CONF_KEY_RADIOS, stored_items)
443        # Trigger library sync
444        self.mass.call_later(
445            1,
446            self.mass.music.start_sync,
447            [MediaType.RADIO],
448            [self.instance_id],
449        )
450        return await self.get_radio(url)
451
452    async def add_track(self, url: str, name: str, image_url: str | None = None) -> Track:
453        """
454        Add a track.
455
456        :param url: URL or local path.
457        :param name: Display name.
458        :param image_url: Image URL.
459        """
460        stored_items: list[StoredItem] = self.mass.config.get(CONF_KEY_TRACKS, [])
461        # Remove existing entry with same URL if present
462        stored_items = [x for x in stored_items if x["item_id"] != url]
463        stored_item = StoredItem(item_id=url, name=name)
464        if image_url:
465            stored_item["image_url"] = image_url
466        stored_items.append(stored_item)
467        self.mass.config.set(CONF_KEY_TRACKS, stored_items)
468        # Trigger library sync
469        self.mass.call_later(
470            1,
471            self.mass.music.start_sync,
472            [MediaType.TRACK],
473            [self.instance_id],
474        )
475        return await self.get_track(url)
476
477    async def get_playlist_tracks(
478        self, prov_playlist_id: str, page: int = 0
479    ) -> list[PlaylistPlayableItem]:
480        """Get playlist tracks (paginated, 500 items per page)."""
481        if prov_playlist_id in BUILTIN_PLAYLISTS:
482            if page > 0:
483                return []
484            return list(await self._get_builtin_playlist_tracks(prov_playlist_id))
485        return await self._get_user_playlist_tracks(prov_playlist_id, page)
486
487    async def add_playlist_tracks(self, prov_playlist_id: str, prov_track_ids: list[str]) -> None:
488        """Add track(s) to playlist with full metadata and deduplication."""
489        async with self._get_playlist_lock(prov_playlist_id):
490            m3u_data = await self._read_m3u_file(prov_playlist_id)
491            existing_items = parse_m3u(m3u_data)
492            # build dedup set from existing URIs and provider item_ids
493            existing_item_ids: set[str] = set()
494            for item in existing_items:
495                existing_item_ids.add(item.path)
496                for prov in item.providers:
497                    existing_item_ids.add(f"{prov.domain}:{prov.item_id}")
498            entries: list[PlaylistItem] = list(existing_items)
499            for uri in prov_track_ids:
500                if uri in existing_item_ids:
501                    continue
502                try:
503                    entry = await self._build_m3u_entry_from_uri(uri)
504                except MediaNotFoundError, InvalidDataError, ProviderUnavailableError:
505                    self.logger.warning("Can't add %s to playlist - item not found", uri)
506                    continue
507                # check dedup against the newly built entry's providers too
508                new_ids = {entry.path}
509                if entry.providers:
510                    new_ids.update(f"{p.domain}:{p.item_id}" for p in entry.providers)
511                if new_ids & existing_item_ids:
512                    continue
513                existing_item_ids.update(new_ids)
514                entries.append(entry)
515            # write updated M3U file
516            playlist = await self.get_playlist(prov_playlist_id)
517            await self._write_m3u_file(
518                prov_playlist_id,
519                playlist.name,
520                entries,
521                self._get_playlist_image_url(playlist),
522            )
523
524    async def remove_playlist_tracks(
525        self, prov_playlist_id: str, positions_to_remove: tuple[int, ...]
526    ) -> None:
527        """Remove track(s) from playlist."""
528        async with self._get_playlist_lock(prov_playlist_id):
529            m3u_data = await self._read_m3u_file(prov_playlist_id)
530            existing_items = parse_m3u(m3u_data)
531            # remove items by position (1-indexed)
532            for i in sorted(positions_to_remove, reverse=True):
533                del existing_items[i - 1]
534            playlist = await self.get_playlist(prov_playlist_id)
535            await self._write_m3u_file(
536                prov_playlist_id,
537                playlist.name,
538                list(existing_items),
539                self._get_playlist_image_url(playlist),
540            )
541
542    async def create_playlist(self, name: str, media_types: set[MediaType]) -> Playlist:
543        """
544        Create a new playlist on provider with given name.
545
546        The playlist name is used as the filename (sanitized for filesystem safety).
547        """
548        playlist_id = self._sanitize_playlist_id(name)
549        # ensure uniqueness
550        counter = 1
551        base_id = playlist_id
552        while await asyncio.to_thread(
553            os.path.isfile, os.path.join(self._playlists_dir, f"{playlist_id}.m3u")
554        ):
555            playlist_id = f"{base_id} ({counter})"
556            counter += 1
557        # create empty M3U file with header
558        await self._write_m3u_file(playlist_id, name, [])
559        return await self.get_playlist(playlist_id)
560
561    async def import_playlist(self, m3u_data: str) -> Playlist:
562        """
563        Import a playlist from M3U8 format.
564
565        Creates a new playlist and populates it with items from the M3U data.
566        Items with valid MA URIs are added directly. Plain URLs or unresolvable
567        URIs are stored as-is for later matching.
568
569        :param m3u_data: The M3U8 playlist data as a string.
570        """
571        parsed_items = parse_m3u(m3u_data)
572        if not parsed_items:
573            msg = "No items found in M3U data"
574            raise InvalidDataError(msg)
575        playlist_name = parse_m3u_playlist_name(m3u_data) or "Imported Playlist"
576        playlist = await self.create_playlist(
577            playlist_name,
578            media_types={MediaType.TRACK, MediaType.RADIO},
579        )
580        playlist_image_url = parse_m3u_playlist_image(m3u_data)
581        # Write the parsed items directly as the M3U file, preserving all
582        # metadata from the source. This avoids re-resolving items that
583        # already have rich metadata (e.g. exported from another MA instance).
584        await self._write_m3u_file(
585            playlist.item_id,
586            playlist_name,
587            parsed_items,
588            playlist_image_url,
589        )
590        return await self.get_playlist(playlist.item_id)
591
592    async def match_imported_playlist_tracks(
593        self,
594        prov_playlist_id: str,
595        match_providers: list[str] | None = None,
596    ) -> None:
597        """
598        Match imported playlist tracks against available providers.
599
600        Iterates through playlist items whose provider is unavailable,
601        searching other providers for matches using metadata stored in
602        the M3U file. Matched tracks are replaced in-place.
603
604        :param prov_playlist_id: The provider-side playlist ID.
605        :param match_providers: Optional list of provider instance IDs or
606            domains to search. When None, all providers are searched.
607        """
608        m3u_data = await self._read_m3u_file(prov_playlist_id)
609        parsed_items = parse_m3u(m3u_data)
610        if not parsed_items:
611            return
612
613        total = len(parsed_items)
614        matched_count = 0
615        unmatched_count = 0
616        changed = False
617
618        for index, item in enumerate(parsed_items):
619            update_current_task_progress_from_index(
620                index, total, f"Matching track {index + 1}/{total}"
621            )
622            if not item.title:
623                continue
624            # check if the URI's provider is available
625            needs_matching = False
626            media_type = MediaType.TRACK
627            try:
628                media_type, prov_instance, _item_id = await parse_uri(item.path)
629                if media_type == MediaType.RADIO:
630                    continue
631                if not self.mass.get_provider(prov_instance):
632                    needs_matching = True
633            except Exception:
634                needs_matching = True
635
636            if not needs_matching:
637                continue
638
639            matched_uri = await self._match_track_by_metadata(item, match_providers=match_providers)
640            if matched_uri:
641                # enrich the entry with full metadata (#EXTPROV etc.) so it resolves to a
642                # playable item - just storing the URI leaves it without provider mappings
643                try:
644                    parsed_items[index] = await self._build_m3u_entry_from_uri(matched_uri)
645                except MediaNotFoundError, InvalidDataError, ProviderUnavailableError:
646                    item.path = matched_uri
647                changed = True
648                matched_count += 1
649            else:
650                report_current_task_failure(f"No match found for: {item.title}")
651                unmatched_count += 1
652
653        if changed:
654            playlist = await self.get_playlist(prov_playlist_id)
655            await self._write_m3u_file(
656                prov_playlist_id,
657                playlist.name,
658                parsed_items,
659                self._get_playlist_image_url(playlist),
660            )
661
662        self.logger.info(
663            "Import matching: %d matched, %d unmatched out of %d items",
664            matched_count,
665            unmatched_count,
666            total,
667        )
668        update_current_task_progress_from_index(total, total, "Matching complete")
669
670    async def parse_item(
671        self,
672        url: str,
673        force_refresh: bool = False,
674        force_radio: bool = False,
675        requested_media_type: MediaType | None = None,
676    ) -> Track | Radio | SoundEffect:
677        """
678        Parse a plain URL to a Track, Radio, or SoundEffect item.
679
680        Without an explicitly requested media type, a URL carrying no music tags resolves to
681        a sound effect: a notification or TTS clip is a one-off, not music to build a queue
682        around.
683        """
684        media_info = await self._get_media_info(url, force_refresh)
685        is_radio = media_info.get("icyname") or not media_info.duration
686        provider_mappings = {
687            ProviderMapping(
688                item_id=url,
689                provider_domain=self.domain,
690                provider_instance=self.instance_id,
691                audio_format=AudioFormat(
692                    content_type=ContentType.try_parse(media_info.format),
693                    sample_rate=media_info.sample_rate,
694                    bit_depth=media_info.bits_per_sample,
695                    bit_rate=media_info.bit_rate,
696                ),
697            )
698        }
699        media_item: Track | Radio | SoundEffect
700        if requested_media_type == MediaType.SOUND_EFFECT or (
701            requested_media_type == MediaType.UNKNOWN
702            and not is_radio
703            and not _has_music_tags(media_info)
704        ):
705            media_item = SoundEffect(
706                item_id=url,
707                provider=self.domain,
708                name=media_info.title or url,
709                provider_mappings=provider_mappings,
710            )
711            if media_info.duration:
712                media_item.duration = int(media_info.duration or 0)
713        elif (is_radio or force_radio) and requested_media_type != MediaType.TRACK:
714            # treat as radio, unless a track was explicitly requested: such a track
715            # stays a track, also when its stream carries an ICY name or no duration
716            media_item = Radio(
717                item_id=url,
718                provider=self.domain,
719                name=media_info.get("icyname")
720                or media_info.get("programtitle")
721                or media_info.title
722                or url,
723                provider_mappings=provider_mappings,
724            )
725        else:
726            media_item = Track(
727                item_id=url,
728                provider=self.domain,
729                name=media_info.title or url,
730                duration=int(media_info.duration or 0),
731                artists=UniqueList(
732                    [await self.get_artist(artist) for artist in media_info.artists]
733                ),
734                provider_mappings=provider_mappings,
735            )
736
737        if media_info.has_cover_image:
738            media_item.metadata.images = UniqueList(
739                [
740                    MediaItemImage(
741                        type=ImageType.THUMB,
742                        path=url,
743                        provider=self.domain,
744                        remotely_accessible=False,
745                    )
746                ]
747            )
748        if isinstance(media_item, Track | Radio):
749            self._apply_stored_details(media_item)
750        return media_item
751
752    async def resolve_image(self, path: str) -> str | bytes:
753        """
754        Resolve an image from an image path.
755
756        This either returns (a generator to get) raw bytes of the image or
757        a string with an http(s) URL or local path that is accessible from the server.
758        """
759        if path == "logo.png":
760            return MASS_LOGO
761        if path in ("fanart.jpg", "fallback_fanart.jpeg"):
762            return VARIOUS_ARTISTS_FANART
763        if path.startswith(f"{GENRE_ICONS_DIR_NAME}/"):
764            icon_name = path[len(GENRE_ICONS_DIR_NAME) + 1 :]
765            icons_base = RESOURCES_DIR.joinpath(GENRE_ICONS_DIR_NAME)
766            if not is_safe_path(icon_name, str(icons_base)):
767                raise FileNotFoundError(f"Invalid genre icon reference: {path}")
768            return str(icons_base.joinpath(icon_name))
769        return path
770
771    async def get_stream_details(self, item_id: str, media_type: MediaType) -> StreamDetails:
772        """Get stream details for a track, radio stream, or sound effect."""
773        media_info = await self._get_media_info(item_id)
774        is_radio = media_info.get("icyname") or not media_info.duration
775        stream_media_type = (
776            MediaType.SOUND_EFFECT
777            if media_type == MediaType.SOUND_EFFECT
778            else MediaType.RADIO
779            if is_radio
780            else MediaType.TRACK
781        )
782        return StreamDetails(
783            provider=self.instance_id,
784            item_id=item_id,
785            audio_format=AudioFormat(
786                content_type=ContentType.try_parse(media_info.format),
787                sample_rate=media_info.sample_rate,
788                bit_depth=media_info.bits_per_sample,
789                channels=media_info.channels,
790            ),
791            media_type=stream_media_type,
792            stream_type=StreamType.HTTP,
793            path=item_id,
794            can_seek=not is_radio,
795            allow_seek=not is_radio,
796        )
797
798    @staticmethod
799    def _get_playlist_image_url(playlist: Playlist) -> str | None:
800        """Return the playlist-level image URL to persist in the M3U header."""
801        return playlist.image.path if playlist.image else None
802
803    async def _update_playlist_metadata(
804        self, playlist_id: str, new_name: str, image_url: str | None
805    ) -> None:
806        """Update the name and image of a playlist in its M3U file."""
807        if playlist_id in BUILTIN_PLAYLISTS:
808            # builtin playlists are not editable
809            return
810        m3u_data = await self._read_m3u_file(playlist_id)
811        if not m3u_data:
812            return
813        existing_items = parse_m3u(m3u_data)
814        try:
815            await self._write_m3u_file(playlist_id, new_name, list(existing_items), image_url)
816        except OSError as err:
817            self.logger.warning("Failed to update playlist metadata: %s", err)
818
819    async def _match_track_by_metadata(
820        self,
821        item: PlaylistItem,
822        match_providers: list[str] | None = None,
823    ) -> str | None:
824        """
825        Search providers for a track matching the given PlaylistItem metadata.
826
827        Uses ISRC/MusicBrainz ID for exact matching first, then falls back
828        to fuzzy title/artist/duration matching.
829
830        :param item: The PlaylistItem with metadata from the M3U file.
831        :param match_providers: Optional list of provider instance IDs or
832            domains to limit the search.
833        """
834        artist_name, track_name = parse_extinf_title(item.title)
835        if not track_name:
836            return None
837
838        search_query = f"{artist_name} - {track_name}" if artist_name else track_name
839
840        all_providers = self.mass.music.get_unique_providers()
841        if match_providers:
842            provider_domains: dict[str, str] = {}
843            for pid in all_providers:
844                prov = self.mass.get_provider(pid)
845                if prov:
846                    provider_domains[pid] = prov.domain
847            all_providers = [
848                pid
849                for pid in all_providers
850                if pid in match_providers or provider_domains.get(pid) in match_providers
851            ]
852
853        best_match: tuple[int, str] | None = None
854
855        for provider_id in all_providers:
856            try:
857                results = await self.mass.music.tracks.search(search_query, provider_id, limit=5)
858            except Exception:
859                self.logger.debug(
860                    "Search failed on provider %s for '%s'", provider_id, search_query
861                )
862                continue
863
864            for result in results:
865                if not result.uri:
866                    continue
867                score = self._score_track_match(result, item)
868                if score >= 10:
869                    self.logger.debug("Exact ID match for '%s' -> %s", search_query, result.uri)
870                    return result.uri
871                if score > 0 and (best_match is None or score > best_match[0]):
872                    best_match = (score, result.uri)
873
874        if best_match:
875            self.logger.debug(
876                "Matched '%s' -> %s (score=%d)",
877                search_query,
878                best_match[1],
879                best_match[0],
880            )
881            return best_match[1]
882
883        self.logger.info("No match found for '%s'", search_query)
884        return None
885
886    def _score_track_match(
887        self,
888        candidate: Track,
889        item: PlaylistItem,
890    ) -> int:
891        """
892        Score how well a candidate track matches the PlaylistItem metadata.
893
894        Returns 0 for no match, higher scores for better matches.
895        ISRC or MusicBrainz Recording ID match returns 10 (maximum).
896
897        :param candidate: The track from search results.
898        :param item: The PlaylistItem with metadata from the M3U file.
899        """
900        metadata = item.metadata or {}
901        artist_name, track_name = parse_extinf_title(item.title)
902        if not track_name:
903            return 0
904
905        isrc = metadata.get("isrc")
906        mbid = metadata.get("mbid")
907
908        # exact ID matches (cross-provider definitive match)
909        if isrc:
910            candidate_isrc = candidate.get_external_id(ExternalID.ISRC)
911            if candidate_isrc and normalize_external_id(
912                ExternalID.ISRC, candidate_isrc
913            ) == normalize_external_id(ExternalID.ISRC, isrc):
914                return 10
915        if mbid:
916            candidate_mbid = candidate.get_external_id(ExternalID.MB_RECORDING)
917            if candidate_mbid and normalize_external_id(
918                ExternalID.MB_RECORDING, candidate_mbid
919            ) == normalize_external_id(ExternalID.MB_RECORDING, mbid):
920                return 10
921
922        # media type gate
923        if metadata.get("media_type"):
924            candidate_type = getattr(candidate, "media_type", None)
925            if candidate_type and candidate_type.value != metadata["media_type"]:
926                return 0
927
928        return self._score_fuzzy_metadata(candidate, artist_name, track_name, metadata, item)
929
930    def _score_fuzzy_metadata(
931        self,
932        candidate: Track,
933        artist_name: str | None,
934        track_name: str,
935        metadata: dict[str, str],
936        item: PlaylistItem,
937    ) -> int:
938        """
939        Score fuzzy metadata fields (title, artist, album, duration, version).
940
941        :param candidate: The track from search results.
942        :param artist_name: Parsed artist name from EXTINF, or None.
943        :param track_name: Parsed track title from EXTINF.
944        :param metadata: The #EXTMA metadata dict.
945        :param item: The PlaylistItem (for duration from item.length).
946        """
947        if not compare_strings(candidate.name, track_name, strict=False):
948            return 0
949        score = 1
950
951        if artist_name:
952            candidate_artists = [a.name for a in candidate.artists] if candidate.artists else []
953            if not any(compare_strings(a, artist_name, strict=False) for a in candidate_artists):
954                return 0
955            score += 2
956
957        score += self._score_bonus_fields(candidate, metadata)
958        score += self._score_duration(candidate, item)
959        return score
960
961    @staticmethod
962    def _score_bonus_fields(candidate: Track, metadata: dict[str, str]) -> int:
963        """Score bonus metadata fields: podcast, authors, album, version."""
964        score = 0
965        if metadata.get("podcast"):
966            candidate_podcast = getattr(candidate, "podcast", None)
967            if candidate_podcast and hasattr(candidate_podcast, "name"):
968                if compare_strings(candidate_podcast.name, metadata["podcast"], strict=False):
969                    score += 2
970
971        if metadata.get("authors"):
972            candidate_authors = getattr(candidate, "authors", None)
973            if candidate_authors:
974                if compare_strings("; ".join(candidate_authors), metadata["authors"], strict=False):
975                    score += 2
976
977        if metadata.get("album"):
978            candidate_album = getattr(candidate, "album", None)
979            if candidate_album and hasattr(candidate_album, "name"):
980                if compare_strings(candidate_album.name, metadata["album"], strict=False):
981                    score += 1
982
983        if metadata.get("version"):
984            candidate_version = getattr(candidate, "version", None) or ""
985            if candidate_version and compare_strings(
986                candidate_version, metadata["version"], strict=False
987            ):
988                score += 1
989            elif candidate_version:
990                score -= 1
991
992        return score
993
994    @staticmethod
995    def _score_duration(candidate: Track, item: PlaylistItem) -> int:
996        """Score duration proximity between candidate and playlist item."""
997        try:
998            duration = int(item.length) if item.length else None
999        except ValueError:
1000            return 0
1001        if duration is None or duration <= 0 or candidate.duration <= 0:
1002            return 0
1003        diff = abs(candidate.duration - duration)
1004        if diff <= 2:
1005            return 2
1006        if diff <= 5:
1007            return 1
1008        return 0
1009
1010    def _get_stored_item(
1011        self, item: PlaylistItem, stored_by_media_type: Mapping[str, Mapping[str, StoredItem]]
1012    ) -> StoredItem | None:
1013        """
1014        Return the stored details of a manually added playlist entry, if it is one.
1015
1016        :param item: The playlist entry as parsed from an M3U file.
1017        :param stored_by_media_type: Stored items per media type, each keyed on item_id.
1018        """
1019        prov_mapping = next((x for x in item.providers if x.domain == self.domain), None)
1020        if prov_mapping is None:
1021            return None
1022        media_type = (item.metadata or {}).get("media_type", "")
1023        return stored_by_media_type.get(media_type, {}).get(prov_mapping.item_id)
1024
1025    def _stored_details_differ(
1026        self, item: PlaylistItem, stored_by_media_type: Mapping[str, Mapping[str, StoredItem]]
1027    ) -> bool:
1028        """
1029        Return True when a playlist entry no longer carries its manually set name or image.
1030
1031        :param item: The playlist entry as parsed from an M3U file.
1032        :param stored_by_media_type: Stored items per media type, each keyed on item_id.
1033        """
1034        stored_item = self._get_stored_item(item, stored_by_media_type)
1035        if stored_item is None:
1036            return False
1037        # an M3U file cannot hold the surrounding whitespace of a name, so comparing
1038        # against the raw stored name would report a difference that no rewrite can settle
1039        if stored_item["name"].strip() != (item.metadata or {}).get("name"):
1040            return True
1041        # a stored item without an image is not a difference: cover art from the stream
1042        # is a valid fallback for as long as the user has set no image of their own
1043        if image_url := stored_item.get("image_url"):
1044            # only the thumbnail counts: the same url as another image type still leaves
1045            # the stream's cover art as the one that shows
1046            return not any(
1047                image.type == ImageType.THUMB.value and image.path == image_url
1048                for image in item.images
1049            )
1050        return False
1051
1052    def _restore_stored_details(
1053        self, item: PlaylistItem, stored_by_media_type: Mapping[str, Mapping[str, StoredItem]]
1054    ) -> None:
1055        """
1056        Write the manually set name and image of a playlist entry back into it.
1057
1058        :param item: The playlist entry to update in place.
1059        :param stored_by_media_type: Stored items per media type, each keyed on item_id.
1060        """
1061        stored_item = self._get_stored_item(item, stored_by_media_type)
1062        if stored_item is None:
1063            return
1064        name = stored_item["name"]
1065        item.metadata = {**(item.metadata or {}), "name": name}
1066        # #EXTINF holds "<artists> - <name>" for a track and the plain name for a radio
1067        # station, which has no artists; mirror how the entry would have been written
1068        item.title = f"{', '.join(x.name for x in item.artists)} - {name}" if item.artists else name
1069        if image_url := stored_item.get("image_url"):
1070            item.images = [
1071                ImageInfo(
1072                    type=ImageType.THUMB.value,
1073                    path=image_url,
1074                    provider=self.domain,
1075                    remotely_accessible=image_url.startswith("http"),
1076                ),
1077                *(x for x in item.images if x.type != ImageType.THUMB.value),
1078            ]
1079
1080    def _apply_stored_details(self, media_item: Track | Radio) -> None:
1081        """Apply the name and image stored for a manually added track or radio station."""
1082        key = CONF_KEY_RADIOS if isinstance(media_item, Radio) else CONF_KEY_TRACKS
1083        stored_items: list[StoredItem] = self.mass.config.get(key, [])
1084        stored_item = next(
1085            (x for x in stored_items if x["item_id"] == media_item.item_id),
1086            None,
1087        )
1088        if stored_item is None:
1089            return
1090        media_item.name = stored_item["name"]
1091        if image_url := stored_item.get("image_url"):
1092            # the stored image replaces any cover art on the stream, so exactly one
1093            # thumbnail is left to serialise into a playlist entry
1094            media_item.metadata.images = UniqueList(
1095                [
1096                    MediaItemImage(
1097                        type=ImageType.THUMB,
1098                        path=image_url,
1099                        provider=self.domain,
1100                        remotely_accessible=image_url.startswith("http"),
1101                    ),
1102                    *(x for x in (media_item.metadata.images or []) if x.type != ImageType.THUMB),
1103                ]
1104            )
1105
1106    async def _resolve_url(self, url: str) -> str:
1107        """
1108        Resolve a URL to the actual stream URL.
1109
1110        ffprobe cannot analyze PLS/M3U files directly as it sees them as text.
1111        This method extracts the actual audio stream URL from playlist files.
1112
1113        :param url: The URL to check and potentially resolve.
1114        :returns: The resolved stream URL, or the original URL if not a playlist.
1115        """
1116        parsed = urlparse(url)
1117        path_lower = parsed.path.lower()
1118        is_playlist = path_lower.endswith(".pls")
1119        if not is_playlist:
1120            return url
1121
1122        try:
1123            playlist_items = await fetch_playlist(self.mass, url, raise_on_hls=False)
1124            for item in playlist_items:
1125                if item.is_url:
1126                    return item.path
1127        except (InvalidDataError, IsHLSPlaylist) as err:
1128            self.logger.debug("Failed to resolve playlist URL %s: %s", url, err)
1129
1130        return url
1131
1132    async def _get_media_info(self, url: str, force_refresh: bool = False) -> AudioTags:
1133        """Retrieve mediainfo for url."""
1134        # do we have some cached info for this url ?
1135        cached_info = await self.mass.cache.get(
1136            url, provider=self.instance_id, category=CACHE_CATEGORY_MEDIA_INFO
1137        )
1138        if cached_info and not force_refresh:
1139            return AudioTags.parse(cached_info)
1140        resolved_url = await self._resolve_url(url)
1141        # parse info with ffprobe (and store in cache)
1142        media_info = await async_parse_tags(resolved_url)
1143        if "authSig" in url:
1144            media_info.has_cover_image = False
1145        await self.mass.cache.set(
1146            url, media_info.raw, provider=self.instance_id, category=CACHE_CATEGORY_MEDIA_INFO
1147        )
1148        return media_info
1149
1150    @use_cache(expiration=120, category=CACHE_CATEGORY_PLAYLISTS)
1151    async def _get_builtin_playlist_random_favorite_tracks(self) -> list[Track]:
1152        result: list[Track] = []
1153        res = await self.mass.music.tracks.library_items(
1154            favorite=True, limit=250000, order_by="random_play_count", summary=False
1155        )
1156        for idx, item in enumerate(res, 1):
1157            item.position = idx
1158            result.append(item)
1159        return result
1160
1161    @use_cache(expiration=120, category=CACHE_CATEGORY_PLAYLISTS)
1162    async def _get_builtin_playlist_random_tracks(self) -> list[Track]:
1163        result: list[Track] = []
1164        res = await self.mass.music.tracks.library_items(
1165            limit=500, order_by="random_play_count", summary=False
1166        )
1167        for idx, item in enumerate(res, 1):
1168            item.position = idx
1169            result.append(item)
1170        return result
1171
1172    @use_cache(expiration=3600, category=CACHE_CATEGORY_PLAYLISTS)
1173    async def _get_builtin_playlist_random_album(self) -> list[Track]:
1174        for random_album in await self.mass.music.albums.get_library_items_by_query(
1175            limit=1,
1176            order_by="random",
1177            extra_query_parts=["album_type != :excluded_album_type"],
1178            extra_query_params={"excluded_album_type": "single"},
1179        ):
1180            tracks = await self.mass.music.albums.tracks(
1181                random_album.item_id, random_album.provider
1182            )
1183            for idx, track in enumerate(tracks, 1):
1184                track.position = idx
1185            return tracks
1186        return []
1187
1188    @use_cache(expiration=3600, category=CACHE_CATEGORY_PLAYLISTS)
1189    async def _get_builtin_playlist_random_artist(self) -> list[Track]:
1190        for source in ("library", "top"):
1191            for min_tracks_required in (25, 10, 5, 1):
1192                for random_artist in await self.mass.music.artists.library_items(
1193                    limit=25, order_by="random", summary=False
1194                ):
1195                    if source == "library":
1196                        tracks = await self.mass.music.artists.tracks(
1197                            random_artist.item_id, "library"
1198                        )
1199                    else:
1200                        tracks = await self.mass.music.artists.top_tracks(
1201                            random_artist.item_id, random_artist.provider
1202                        )
1203                    if len(tracks) < min_tracks_required:
1204                        continue
1205                    for idx, track in enumerate(tracks, 1):
1206                        track.position = idx
1207                    return tracks
1208        return []
1209
1210    async def _get_builtin_playlist_recently_played(self) -> list[Track]:
1211        result: list[Track] = []
1212        recent_tracks = await self.mass.music.recently_played(100, [MediaType.TRACK])
1213        # "library" rows have no provider instance to resolve, so read them from the db in one go
1214        library_ids = [int(x.item_id) for x in recent_tracks if x.provider == "library"]
1215        library_tracks: dict[str, Track] = {}
1216        if library_ids:
1217            library_tracks = {
1218                track.item_id: track
1219                for track in await self.mass.music.tracks.get_library_items_by_query(
1220                    extra_query_parts=["tracks.item_id IN :item_ids"],
1221                    extra_query_params={"item_ids": library_ids},
1222                    in_library_only=False,
1223                )
1224            }
1225        for idx, item in enumerate(recent_tracks, 1):
1226            if item.provider == "library":
1227                # pop so a track played by several users is listed once (newest play first)
1228                if track := library_tracks.pop(item.item_id, None):
1229                    track.position = idx
1230                    result.append(track)
1231                continue
1232            if not (item_provider := self.mass.get_provider(item.provider)):
1233                continue
1234            track = Track(
1235                item_id=item.item_id,
1236                provider=item.provider,
1237                name=item.name,
1238                provider_mappings={
1239                    ProviderMapping(
1240                        item_id=item.item_id,
1241                        provider_domain=item_provider.domain,
1242                        provider_instance=item_provider.instance_id,
1243                    )
1244                },
1245            )
1246            if item.image:
1247                track.metadata.add_image(item.image)
1248            track.position = idx
1249            result.append(track)
1250        return result
1251
1252    @use_cache(expiration=60, category=CACHE_CATEGORY_PLAYLISTS)
1253    async def _get_builtin_playlist_recently_added_tracks(self) -> list[Track]:
1254        result: list[Track] = []
1255        recent_tracks = await self.mass.music.recently_added_tracks(100)
1256        for idx, track in enumerate(recent_tracks, 1):
1257            track.position = idx
1258            result.append(track)
1259        return result
1260
1261    async def _get_builtin_playlist_infinite_mix(self) -> list[Track]:
1262        """Return 25 random library tracks for the Infinite Mix dynamic playlist."""
1263        return await self._infinite_mix_tracks(favorite=None)
1264
1265    async def _get_builtin_playlist_infinite_mix_favorites(self) -> list[Track]:
1266        """Return 25 random favorited tracks for the Infinite Mix (favorites) dynamic playlist."""
1267        return await self._infinite_mix_tracks(favorite=True)
1268
1269    async def _infinite_mix_tracks(self, *, favorite: bool | None) -> list[Track]:
1270        """
1271        Return up to 25 random (optionally favorited) library tracks for an Infinite Mix.
1272
1273        :param favorite: Restrict to favorited tracks when True; all library tracks when None.
1274        """
1275        # over-fetch when a recency filter is published so dropping recently-played tracks still
1276        # leaves a full mix; the pool is trimmed back to the mix size after filtering
1277        limit = 25 * 3 if get_track_filter() is not None else 25
1278        candidates = list(
1279            await self.mass.music.tracks.library_items(
1280                favorite=favorite, limit=limit, order_by="random", summary=False
1281            )
1282        )
1283        tracks = filter_tracks(candidates)[:25]
1284        for idx, track in enumerate(tracks, 1):
1285            track.position = idx
1286        return tracks
1287
1288    async def _get_builtin_playlist_tracks(
1289        self, builtin_playlist_id: str
1290    ) -> list[Track] | UniqueList[Track]:
1291        """Get all playlist tracks for given builtin playlist id."""
1292        try:
1293            return await {
1294                ALL_FAVORITE_TRACKS: self._get_builtin_playlist_random_favorite_tracks,
1295                RANDOM_TRACKS: self._get_builtin_playlist_random_tracks,
1296                RANDOM_ALBUM: self._get_builtin_playlist_random_album,
1297                RANDOM_ARTIST: self._get_builtin_playlist_random_artist,
1298                RECENTLY_PLAYED: self._get_builtin_playlist_recently_played,
1299                RECENTLY_ADDED_TRACKS: self._get_builtin_playlist_recently_added_tracks,
1300                INFINITE_MIX: self._get_builtin_playlist_infinite_mix,
1301                INFINITE_MIX_FAVORITES: self._get_builtin_playlist_infinite_mix_favorites,
1302            }[builtin_playlist_id]()
1303        except KeyError:
1304            raise MediaNotFoundError(f"No built in playlist: {builtin_playlist_id}")
1305
1306    async def _read_m3u_file(self, playlist_id: str) -> str:
1307        """Read the raw M3U file content for a playlist."""
1308        playlist_file = os.path.join(self._playlists_dir, f"{playlist_id}.m3u")
1309        if not await asyncio.to_thread(os.path.isfile, playlist_file):
1310            return ""
1311        async with (
1312            self._playlist_lock,
1313            aiofiles.open(playlist_file, encoding="utf-8") as _file,
1314        ):
1315            result: str = await _file.read()
1316            return result
1317
1318    async def _write_m3u_file(
1319        self,
1320        playlist_id: str,
1321        playlist_name: str,
1322        entries: list[PlaylistItem],
1323        playlist_image_url: str | None = None,
1324    ) -> None:
1325        """Write an M3U playlist file to disk."""
1326        m3u_content = generate_m3u(playlist_name, entries, playlist_image_url)
1327        playlist_file = os.path.join(self._playlists_dir, f"{playlist_id}.m3u")
1328        async with (
1329            self._playlist_lock,
1330            aiofiles.open(playlist_file, "w", encoding="utf-8") as _file,
1331        ):
1332            await _file.write(m3u_content)
1333
1334    def _get_playlist_lock(self, playlist_id: str) -> asyncio.Lock:
1335        """Get or create a per-playlist lock for concurrent access protection."""
1336        if playlist_id not in self._playlist_locks:
1337            self._playlist_locks[playlist_id] = asyncio.Lock()
1338        return self._playlist_locks[playlist_id]
1339
1340    async def _resolve_playlist_item(self, item: PlaylistItem) -> MediaItemType | None:
1341        """
1342        Resolve a PlaylistItem to a MediaItem.
1343
1344        Constructs from stored metadata first. If no providers are available,
1345        falls back to a library lookup by domain.
1346        """
1347        media_item = construct_media_item_from_playlist_item(item, self.mass)
1348        if media_item is None:
1349            return None
1350        # if at least one provider mapping is available, we're done
1351        if any(pm.available for pm in media_item.provider_mappings):
1352            return media_item
1353        # all stored provider instances are unavailable - try library lookup by domain
1354        media_type = MediaType((item.metadata or {}).get("media_type", "track"))
1355        if media_type == MediaType.SOUND_EFFECT:
1356            return media_item
1357        media_controller = self.mass.music.get_controller(media_type)
1358        for prov_info in item.providers:
1359            try:
1360                library_item = await media_controller.get_library_item_by_prov_id(
1361                    prov_info.item_id, prov_info.domain
1362                )
1363                if library_item is not None:
1364                    return library_item
1365            except InvalidDataError, KeyError, NotImplementedError:
1366                continue
1367        # return unresolved media item so the entry still shows in the playlist
1368        return media_item
1369
1370    async def _get_user_playlist_tracks(
1371        self, prov_playlist_id: str, page: int
1372    ) -> list[PlaylistPlayableItem]:
1373        """Get user-created playlist tracks with caching and parallel resolution."""
1374        playlist_file = os.path.join(self._playlists_dir, f"{prov_playlist_id}.m3u")
1375        # use file mtime as cache checksum so edits invalidate the cache
1376        try:
1377            stat = await asyncio.to_thread(os.stat, playlist_file)
1378            cache_checksum = str(int(stat.st_mtime))
1379        except OSError:
1380            cache_checksum = "0"
1381
1382        cache_key = f"playlist_tracks.{prov_playlist_id}.{page}"
1383        cached = await self.mass.cache.get(
1384            cache_key,
1385            provider=self.instance_id,
1386            checksum=cache_checksum,
1387            category=CACHE_CATEGORY_PLAYLISTS,
1388        )
1389        if cached is not None:
1390            # cached data is a list of dicts, deserialize back to media items
1391            return [
1392                cast("PlaylistPlayableItem", media_from_dict(item_dict))
1393                if isinstance(item_dict, dict)
1394                else item_dict
1395                for item_dict in cached
1396            ]
1397
1398        async with self._get_playlist_lock(prov_playlist_id):
1399            m3u_data = await self._read_m3u_file(prov_playlist_id)
1400        all_items = parse_m3u(m3u_data)
1401        page_size = 500
1402        start = page * page_size
1403        if start >= len(all_items):
1404            return []
1405        page_items = all_items[start : start + page_size]
1406
1407        # resolve items in parallel with bounded concurrency
1408        semaphore = asyncio.Semaphore(50)
1409
1410        async def _resolve(index: int, item: PlaylistItem) -> PlaylistPlayableItem | None:
1411            async with semaphore:
1412                try:
1413                    media_item = await self._resolve_playlist_item(item)
1414                    if media_item is None:
1415                        return None
1416                    if media_item.media_type not in PLAYLIST_MEDIA_TYPES:
1417                        self.logger.warning(
1418                            "Unsupported media type in playlist %s: %s",
1419                            prov_playlist_id,
1420                            type(media_item),
1421                        )
1422                        return None
1423                    playlist_item = cast("PlaylistPlayableItem", media_item)
1424                    playlist_item.position = index
1425                    return playlist_item
1426                except (
1427                    MediaNotFoundError,
1428                    InvalidDataError,
1429                    ProviderUnavailableError,
1430                ) as err:
1431                    self.logger.warning(
1432                        "Skipping %s in playlist %s: %s",
1433                        item.path,
1434                        prov_playlist_id,
1435                        str(err),
1436                    )
1437                return None
1438
1439        tasks = [_resolve(start + idx + 1, item) for idx, item in enumerate(page_items)]
1440        resolved = await asyncio.gather(*tasks)
1441        result = [item for item in resolved if item is not None]
1442
1443        await self.mass.cache.set(
1444            key=cache_key,
1445            data=result,
1446            expiration=3600 * 24,
1447            provider=self.instance_id,
1448            checksum=cache_checksum,
1449            category=CACHE_CATEGORY_PLAYLISTS,
1450        )
1451        return result
1452
1453    async def _build_m3u_entry_from_uri(self, uri: str) -> PlaylistItem:
1454        """Fetch a media item by URI and convert it to a PlaylistItem with full metadata."""
1455        full_item = await self.mass.music.get_item_by_uri(uri, allow_update_metadata=False)
1456        if not isinstance(full_item, MediaItem):
1457            msg = f"Unsupported media type for playlist: {uri}"
1458            raise InvalidDataError(msg)
1459        return media_item_to_playlist_item(full_item)
1460
1461    @staticmethod
1462    def _sanitize_playlist_id(name: str) -> str:
1463        """Sanitize a playlist name for use as a filename (without extension)."""
1464        # replace invalid filename characters
1465        sanitized = re.sub(r'[<>:"/\\|?*]', "_", name)
1466        # remove leading/trailing spaces and dots
1467        sanitized = sanitized.strip(" .")
1468        return sanitized or "untitled"
1469
1470    async def _migrate_playlists(self) -> None:  # noqa: PLR0915
1471        """
1472        Migrate old-style playlists to M3U files and repair incomplete or stale entries.
1473
1474        Raises RuntimeError when too many entries could not be resolved to keep a broken
1475        install from rewriting every playlist.
1476        """
1477        # migrate playlists stored in config to M3U files on disk with enriched metadata
1478        stored_items: list[StoredItem] = self.mass.config.get(CONF_KEY_PLAYLISTS, [])
1479        for stored_item in stored_items:
1480            # keep the original item_id as filename so library DB references stay valid
1481            playlist_id = stored_item["item_id"]
1482            playlist_name = stored_item["name"]
1483            self.logger.info("Migrating playlist '%s' to M3U format...", playlist_name)
1484            update_current_task_progress_text(
1485                f"Migrating playlist '{playlist_name}' to M3U format..."
1486            )
1487            old_file = os.path.join(self._playlists_dir, playlist_id)
1488            # read old URI file and enrich each entry with full metadata
1489            uris: list[str] = []
1490            if await asyncio.to_thread(os.path.isfile, old_file):
1491                async with aiofiles.open(old_file, encoding="utf-8") as _file:
1492                    lines = await _file.readlines()
1493                    uris = [line.strip() for line in lines if line.strip()]
1494            entries: list[PlaylistItem] = []
1495            for uri in uris:
1496                try:
1497                    entries.append(await self._build_m3u_entry_from_uri(uri))
1498                except (
1499                    MediaNotFoundError,
1500                    InvalidDataError,
1501                    InvalidProviderURI,
1502                    ProviderUnavailableError,
1503                ):
1504                    # parse URI for minimal provider info so the entry is resolvable later
1505                    entry = PlaylistItem(path=uri)
1506                    if "://" in uri:
1507                        try:
1508                            domain, rest = uri.split("://", 1)
1509                            media_type_str, item_id = rest.split("/", 1)
1510                            entry.metadata = {"media_type": media_type_str}
1511                            entry.providers = [ProviderMappingInfo(domain=domain, item_id=item_id)]
1512                        except ValueError:
1513                            pass
1514                    entries.append(entry)
1515                    self.logger.debug("Could not enrich migrated entry: %s", uri)
1516            # write as {item_id}.m3u with the display name in #PLAYLIST
1517            await self._write_m3u_file(playlist_id, playlist_name, entries)
1518            # clean up old file (without .m3u extension)
1519            if await asyncio.to_thread(os.path.isfile, old_file):
1520                await asyncio.to_thread(os.remove, old_file)
1521            self.logger.debug("Migrated playlist '%s' -> %s.m3u", playlist_name, playlist_id)
1522        # clear old config entries
1523        self.mass.config.remove(CONF_KEY_PLAYLISTS)
1524        # fix (already migrated) user playlists that have unresolved URIs, or entries whose
1525        # manually set name or artwork was lost, by re-saving them with enriched metadata
1526        errors = 0
1527        # built once: a lookup per entry would rescan the entire config list each time
1528        stored_by_media_type = {
1529            MediaType.RADIO.value: {
1530                x["item_id"]: x for x in self.mass.config.get(CONF_KEY_RADIOS, [])
1531            },
1532            MediaType.TRACK.value: {
1533                x["item_id"]: x for x in self.mass.config.get(CONF_KEY_TRACKS, [])
1534            },
1535        }
1536        for filename in await asyncio.to_thread(os.listdir, self._playlists_dir):
1537            if not filename.endswith(".m3u"):
1538                continue
1539            playlist_id = filename[:-4]  # strip .m3u extension
1540            m3u_data = await self._read_m3u_file(playlist_id)
1541            playlist = await self.get_playlist(playlist_id)
1542            self.logger.debug("Checking playlist '%s' for unresolved entries...", playlist.name)
1543            update_current_task_progress_text(f"Checking playlist '{playlist.name}'")
1544            all_items = parse_m3u(m3u_data)
1545            has_changes = False
1546            orphaned: set[int] = set()
1547            for index, item in enumerate(all_items):
1548                if _is_orphaned_entry_path(item.path):
1549                    # leftover text from a value that once contained a line break: it is no
1550                    # reference to anything and never will be, so drop it instead of failing
1551                    # this (and every future) migration run on it
1552                    self.logger.warning(
1553                        "Dropping unresolvable entry %s from playlist '%s'",
1554                        item.path,
1555                        playlist.name,
1556                    )
1557                    orphaned.add(index)
1558                    has_changes = True
1559                    continue
1560                force_migration = item.metadata and item.metadata.get("album") and not item.album
1561                unresolved = bool(force_migration) or not (
1562                    item.title and item.providers and item.metadata
1563                )
1564                if not unresolved and not self._stored_details_differ(item, stored_by_media_type):
1565                    continue
1566                self.logger.debug(
1567                    "Found %s entry in playlist '%s': %s",
1568                    "unresolved" if unresolved else "outdated",
1569                    playlist_id,
1570                    item.path,
1571                )
1572                try:
1573                    enriched = await self._build_m3u_entry_from_uri(item.path)
1574                    item.length = enriched.length
1575                    item.title = enriched.title
1576                    item.images = enriched.images
1577                    item.providers = enriched.providers
1578                    item.metadata = enriched.metadata
1579                    item.album = enriched.album
1580                    item.artists = enriched.artists
1581                    item.podcast = enriched.podcast
1582                except (
1583                    MediaNotFoundError,
1584                    InvalidDataError,
1585                    InvalidProviderURI,
1586                    ProviderUnavailableError,
1587                ) as err:
1588                    if unresolved:
1589                        self.logger.warning(
1590                            "Could not enrich playlist entry %s during migration: %s",
1591                            item.path,
1592                            err,
1593                        )
1594                        report_current_task_failure(f"Could not enrich playlist entry: {item.path}")
1595                        errors += 1
1596                        continue
1597                    # an outdated entry is still playable, so failing to reach the stream is
1598                    # no migration error; restore the stored details without any IO so a
1599                    # permanently unreachable stream keeps its name and image
1600                    self.logger.debug(
1601                        "Could not refresh playlist entry %s, restoring stored details: %s",
1602                        item.path,
1603                        err,
1604                    )
1605                    self._restore_stored_details(item, stored_by_media_type)
1606                else:
1607                    # writing an entry the refresh did not bring back in step would leave
1608                    # it outdated, and every later run would rewrite the file again
1609                    if self._stored_details_differ(item, stored_by_media_type):
1610                        self._restore_stored_details(item, stored_by_media_type)
1611                    self.logger.debug("Enriched playlist entry %s", item.path)
1612                has_changes = True
1613            if has_changes:
1614                await self._write_m3u_file(
1615                    playlist_id,
1616                    playlist.name,
1617                    [item for idx, item in enumerate(all_items) if idx not in orphaned],
1618                    self._get_playlist_image_url(playlist),
1619                )
1620                self.logger.info("Updated playlist '%s' with enriched metadata", playlist.name)
1621            if errors > 25:
1622                raise RuntimeError("Too many errors during playlist migration")
1623        self.logger.info("Playlist migration completed with %d errors", errors)
1624        # if there were no errors, we can safely unregister the migration task
1625        if errors == 0 and (current_task_id := get_current_task_id()):
1626            # defer unregistering the scheduled task to avoid cancelling the current task
1627            self.mass.call_later(0, self.mass.tasks.unregister_scheduled_task, current_task_id)
1628
1629
1630def _is_orphaned_entry_path(path: str) -> bool:
1631    """Return True if the path is leftover text rather than a reference to a media item."""
1632    # a URI, URL or file path always carries one of these separators, so a path without
1633    # any of them cannot resolve to anything - not now and not on a later run either
1634    return not any(sep in path for sep in ("/", "\\", ":"))
1635
1636
1637def _has_music_tags(media_info: AudioTags) -> bool:
1638    """Return True if the stream carries the tags a music file is expected to have."""
1639    # notification and TTS clips are untagged, which is what tells them apart from a music
1640    # file someone plays by URL. The artists/album properties fall back to the filename, so
1641    # the raw tags are what has to be checked here.
1642    return any(
1643        media_info.get(tag) for tag in ("artist", "artists", "albumartist", "albumartists", "album")
1644    )
1645