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