music-assistant-server

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