music-assistant-server

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