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