music-assistant-server

1"""Helpers for parsing (online and offline) playlists."""
2
3from __future__ import annotations
4
5import configparser
6import logging
7from collections.abc import Sequence
8from dataclasses import dataclass, field
9from typing import TYPE_CHECKING
10from urllib.parse import urlparse
11
12from aiohttp import ClientTimeout, client_exceptions
13from music_assistant_models.enums import ContentType, ExternalID, ImageType, MediaType
14from music_assistant_models.errors import InvalidDataError
15from music_assistant_models.helpers import create_uri
16from music_assistant_models.media_items import (
17    Artist,
18    Audiobook,
19    AudioFormat,
20    ItemMapping,
21    MediaItem,
22    MediaItemImage,
23    MediaItemType,
24    PodcastEpisode,
25    ProviderMapping,
26    Radio,
27    SoundEffect,
28    Track,
29    UniqueList,
30)
31
32from music_assistant.helpers.aiohttp_client import encoded_request_url
33from music_assistant.helpers.uri import BUILTIN_URL_SCHEMES
34from music_assistant.helpers.util import detect_charset, try_parse_int
35
36if TYPE_CHECKING:
37    from music_assistant.mass import MusicAssistant
38
39
40LOGGER = logging.getLogger(__name__)
41HLS_CONTENT_TYPES = ("application/vnd.apple.mpegurl",)
42FIELD_SEPARATOR = "||"
43
44
45class IsHLSPlaylist(InvalidDataError):
46    """The playlist from an HLS stream and should not be parsed."""
47
48
49# --------------------------------------------------------------------------- #
50#  Dataclasses for M3U extended tags (shared between parsing and generation)   #
51# --------------------------------------------------------------------------- #
52
53
54@dataclass
55class ProviderMappingInfo:
56    """Provider mapping stored as #EXTPROV in M3U files."""
57
58    domain: str
59    item_id: str
60    instance_id: str = ""
61    content_type: str = ""
62    sample_rate: int = 0
63    bit_depth: int = 0
64    bit_rate: int = 0
65
66
67@dataclass
68class ImageInfo:
69    """Image metadata stored as #EXTIMG in M3U files."""
70
71    type: str
72    path: str
73    provider: str
74    remotely_accessible: bool = False
75
76
77@dataclass
78class ArtistInfo:
79    """Artist metadata stored as #EXTARTIST in M3U files."""
80
81    name: str
82    provider_domain: str
83    item_id: str
84    provider_instance: str
85
86
87@dataclass
88class AlbumInfo:
89    """Album metadata stored as #EXTALBUM in M3U files."""
90
91    name: str
92    provider_domain: str
93    item_id: str
94    provider_instance: str
95    version: str = ""
96
97
98@dataclass
99class PodcastInfo:
100    """Podcast metadata stored as #EXTPODCAST in M3U files."""
101
102    name: str
103    provider_domain: str
104    item_id: str
105    provider_instance: str
106
107
108@dataclass
109class PlaylistItem:
110    """Single entry in an M3U playlist. Used for both parsing and generation."""
111
112    path: str
113    length: str | None = None
114    title: str | None = None
115    stream_info: dict[str, str] | None = None
116    key: str | None = None
117    metadata: dict[str, str] | None = None
118    providers: list[ProviderMappingInfo] = field(default_factory=list)
119    images: list[ImageInfo] = field(default_factory=list)
120    artists: list[ArtistInfo] = field(default_factory=list)
121    album: AlbumInfo | None = None
122    podcast: PodcastInfo | None = None
123
124    @property
125    def is_url(self) -> bool:
126        """Return True if the path looks like a valid URL."""
127        result = urlparse(self.path)
128        return all([result.scheme, result.netloc])
129
130
131# --------------------------------------------------------------------------- #
132#  Parsing helpers                                                             #
133# --------------------------------------------------------------------------- #
134
135
136def parse_extinf_title(title: str | None) -> tuple[str | None, str | None]:
137    """Split "Artist - Title" into (artist, title). Returns (None, title) if no separator."""
138    if not title:
139        return None, None
140    if " - " in title:
141        artist, track_title = title.split(" - ", 1)
142        return artist.strip(), track_title.strip()
143    return None, title.strip()
144
145
146def parse_m3u(m3u_data: str) -> list[PlaylistItem]:  # noqa: PLR0915
147    """
148    Parse M3U/M3U8 data into a list of PlaylistItem entries.
149
150    Supports standard tags (#EXTINF, #EXT-X-STREAM-INF, #EXT-X-KEY) and
151    Music Assistant extensions (#EXTMA, #EXTPROV, #EXTIMG etc.).
152    """
153    m3u_lines = m3u_data.splitlines()
154    playlist: list[PlaylistItem] = []
155
156    # per-entry accumulators (reset after each path line)
157    length = None
158    title = None
159    stream_info: dict[str, str] | None = None
160    key = None
161    metadata: dict[str, str] | None = None
162    providers: list[ProviderMappingInfo] = []
163    images: list[ImageInfo] = []
164    artists: list[ArtistInfo] = []
165    album: AlbumInfo | None = None
166    podcast: PodcastInfo | None = None
167
168    for line in m3u_lines:
169        line = line.strip()  # noqa: PLW2901
170        if line.startswith("#EXTINF:"):
171            info = line.split("#EXTINF:")[1].split(",", 1)
172            if len(info) != 2:
173                continue
174            length = info[0].strip()
175            if length == "-1":
176                length = None
177            title = info[1].strip()
178        elif line.startswith("#EXT-X-STREAM-INF:"):
179            stream_info = _parse_hls_stream_info(line)
180        elif line.startswith("#EXT-X-KEY:"):
181            key = _parse_hls_key(line, key)
182        elif line.startswith("#EXTMA:"):
183            metadata = _parse_extma_line(line)
184        elif line.startswith("#EXTPROV:"):
185            if prov_info := _parse_extprov_line(line):
186                providers.append(prov_info)
187        elif line.startswith("#EXTIMG:"):
188            if img_info := _parse_extimg_line(line):
189                images.append(img_info)
190        elif line.startswith("#EXTARTIST:"):
191            if artist_info := _parse_extartist_line(line):
192                artists.append(artist_info)
193        elif line.startswith("#EXTALBUM:"):
194            album = _parse_extalbum_line(line)
195        elif line.startswith("#EXTPODCAST:"):
196            podcast = _parse_extpodcast_line(line)
197        elif line.startswith("#"):
198            continue
199        elif line:
200            filepath = line.replace("%20", " ").replace("\\", "/")
201            playlist.append(
202                PlaylistItem(
203                    path=filepath,
204                    length=length,
205                    title=title,
206                    stream_info=stream_info,
207                    key=key,
208                    metadata=metadata,
209                    providers=providers,
210                    images=images,
211                    artists=artists,
212                    album=album,
213                    podcast=podcast,
214                )
215            )
216            # reset accumulators
217            length = None
218            title = None
219            stream_info = None
220            metadata = None
221            providers = []
222            images = []
223            artists = []
224            album = None
225            podcast = None
226
227    return playlist
228
229
230def parse_m3u_playlist_name(m3u_data: str) -> str | None:
231    """Extract the playlist name from an M3U #PLAYLIST directive."""
232    for line in m3u_data.splitlines():
233        line = line.strip()  # noqa: PLW2901
234        if line.startswith("#PLAYLIST:"):
235            return line.split("#PLAYLIST:", 1)[1].strip()
236    return None
237
238
239def parse_m3u_playlist_image(m3u_data: str) -> str | None:
240    """
241    Extract the playlist cover image from an M3U #EXTIMG directive.
242
243    Looks for a simple #EXTIMG:url line before the first track entry.
244    This is a de facto standard for playlist-level cover images.
245    """
246    for line in m3u_data.splitlines():
247        line = line.strip()  # noqa: PLW2901
248        if not line or line.startswith(("#EXTM3U", "#PLAYLIST:")):
249            continue
250        # stop at first track entry or track-level metadata
251        if not line.startswith("#"):
252            break
253        if line.startswith(("#EXTINF:", "#EXTMA:", "#EXTPROV:")):
254            break
255        # simple playlist-level image (just URL, no field separator)
256        if line.startswith("#EXTIMG:") and FIELD_SEPARATOR not in line:
257            return line.split("#EXTIMG:", 1)[1].strip()
258    return None
259
260
261def parse_pls(pls_data: str) -> list[PlaylistItem]:
262    """Parse a PLS playlist file into PlaylistItem entries."""
263    pls_parser = configparser.ConfigParser(strict=False)
264    try:
265        pls_parser.read_string(pls_data, "playlist")
266    except configparser.Error as err:
267        raise InvalidDataError("Can't parse playlist") from err
268
269    if "playlist" not in pls_parser:
270        raise InvalidDataError("Invalid playlist")
271
272    try:
273        num_entries = pls_parser.getint("playlist", "NumberOfEntries")
274    except (configparser.NoOptionError, ValueError) as err:
275        raise InvalidDataError("Invalid NumberOfEntries in playlist") from err
276
277    playlist_section = pls_parser["playlist"]
278    playlist: list[PlaylistItem] = []
279    for entry in range(1, num_entries + 1):
280        file_option = f"File{entry}"
281        if file_option not in playlist_section:
282            continue
283        itempath = playlist_section[file_option]
284        length = playlist_section.get(f"Length{entry}")
285        playlist.append(
286            PlaylistItem(
287                length=length if length and length != "-1" else None,
288                title=playlist_section.get(f"Title{entry}"),
289                path=itempath,
290            )
291        )
292    return playlist
293
294
295async def fetch_playlist(
296    mass: MusicAssistant, url: str, raise_on_hls: bool = True
297) -> list[PlaylistItem]:
298    """Fetch and parse a remote M3U or PLS playlist."""
299    try:
300        async with mass.http_session.get(
301            encoded_request_url(url), allow_redirects=True, timeout=ClientTimeout(total=5)
302        ) as resp:
303            try:
304                raw_data = await resp.content.read(64 * 1024)
305                encoding = resp.charset or await detect_charset(raw_data)
306                playlist_data = raw_data.decode(encoding, errors="replace")
307            except (ValueError, UnicodeDecodeError) as err:
308                msg = f"Could not decode playlist {url}"
309                raise InvalidDataError(msg) from err
310    except TimeoutError as err:
311        msg = f"Timeout while fetching playlist {url}"
312        raise InvalidDataError(msg) from err
313    except client_exceptions.ClientError as err:
314        msg = f"Error while fetching playlist {url}"
315        raise InvalidDataError(msg) from err
316
317    if (
318        raise_on_hls and "#EXT-X-VERSION:" in playlist_data
319    ) or "#EXT-X-STREAM-INF:" in playlist_data:
320        raise IsHLSPlaylist
321
322    if urlparse(url).path.endswith("pls") or "[playlist]" in playlist_data:
323        playlist = parse_pls(playlist_data)
324    else:
325        playlist = parse_m3u(playlist_data)
326
327    if not playlist:
328        msg = f"Empty playlist {url}"
329        raise InvalidDataError(msg)
330
331    return playlist
332
333
334# --------------------------------------------------------------------------- #
335#  Generation                                                                  #
336# --------------------------------------------------------------------------- #
337
338
339def generate_m3u(
340    playlist_name: str,
341    items: Sequence[PlaylistItem],
342    playlist_image_url: str | None = None,
343) -> str:
344    """
345    Generate an M3U8 playlist string from PlaylistItem entries.
346
347    :param playlist_name: Human-readable name (written as #PLAYLIST directive).
348    :param items: Entries to write. Only fields that are set are emitted.
349    :param playlist_image_url: Optional playlist cover image URL.
350    """
351    # Playlist-level image using #EXTIMG directive (de facto standard for playlist covers)
352    lines: list[str] = ["#EXTM3U"]
353    if playlist_image_url:
354        lines.append(f"#EXTIMG:{playlist_image_url}")
355    lines.append(f"#PLAYLIST:{playlist_name}")
356    sep = FIELD_SEPARATOR
357    for item in items:
358        if item.metadata:
359            pairs = sep.join(f"{k}={v}" for k, v in item.metadata.items())
360            lines.append(f"#EXTMA:{pairs}")
361        for prov in item.providers:
362            fields = [
363                prov.domain,
364                prov.item_id,
365                prov.instance_id,
366                prov.content_type,
367                str(prov.sample_rate),
368                str(prov.bit_depth),
369                str(prov.bit_rate),
370            ]
371            lines.append(f"#EXTPROV:{sep.join(fields)}")
372        for artist in item.artists:
373            fields = [artist.name, artist.provider_domain, artist.item_id, artist.provider_instance]
374            lines.append(f"#EXTARTIST:{sep.join(fields)}")
375        if item.album:
376            fields = [
377                item.album.name,
378                item.album.provider_domain,
379                item.album.item_id,
380                item.album.provider_instance,
381                item.album.version,
382            ]
383            lines.append(f"#EXTALBUM:{sep.join(fields)}")
384        if item.podcast:
385            fields = [
386                item.podcast.name,
387                item.podcast.provider_domain,
388                item.podcast.item_id,
389                item.podcast.provider_instance,
390            ]
391            lines.append(f"#EXTPODCAST:{sep.join(fields)}")
392        for img in item.images:
393            remotely = "true" if img.remotely_accessible else "false"
394            fields = [img.type, img.path, img.provider, remotely]
395            lines.append(f"#EXTIMG:{sep.join(fields)}")
396        if item.title is not None:
397            # -1 is the M3U convention for unknown length; without it an entry without
398            # duration (such as a radio station) loses its title on every rewrite
399            lines.append(f"#EXTINF:{item.length if item.length is not None else -1},{item.title}")
400        lines.append(item.path)
401    return "\n".join(lines) + "\n"
402
403
404# --------------------------------------------------------------------------- #
405#  Media item construction from playlist metadata                              #
406# --------------------------------------------------------------------------- #
407
408
409def construct_media_item_from_playlist_item(
410    item: PlaylistItem,
411    mass: MusicAssistant,
412    default_media_type: MediaType = MediaType.TRACK,
413) -> MediaItemType | None:
414    """
415    Construct a MediaItem from a PlaylistItem's stored metadata.
416
417    Resolves provider mappings by instance_id first (free dict lookup),
418    falling back to domain if the instance no longer exists.
419
420    :param item: Parsed PlaylistItem with metadata, providers, and images.
421    :param mass: MusicAssistant instance for provider resolution.
422    :param default_media_type: Media type to build when the entry's #EXTMA metadata
423        holds no usable media_type.
424    """
425    metadata = item.metadata or {}
426    try:
427        media_type = MediaType(metadata.get("media_type", default_media_type.value))
428    except ValueError:
429        media_type = default_media_type
430    name = metadata.get("name") or item.title or item.path
431    duration = try_parse_int(item.length, default=None) if item.length else None
432
433    provider_mappings = _resolve_provider_mappings(item, mass)
434    if not provider_mappings and item.path.startswith(BUILTIN_URL_SCHEMES):
435        # an item without any mapping never reaches library_add, leaving an unplayable
436        # library entry, so a plain stream URL is mapped to the provider that serves it
437        provider_mappings = _builtin_fallback_mappings(item.path, mass)
438    external_ids = _collect_external_ids(metadata)
439
440    # a set has no order, so sort to keep the chosen mapping stable across runs
441    sorted_mappings = sorted(provider_mappings, key=lambda pm: (pm.provider_instance, pm.item_id))
442    first_provider = next(
443        (pm for pm in sorted_mappings if pm.available),
444        next(iter(sorted_mappings), None),
445    )
446    # prefer the instance over the domain: a domain resolves to whichever instance of it
447    # happens to be loaded first, which is the wrong one when several are configured
448    item_provider = (
449        (first_provider.provider_instance or first_provider.provider_domain)
450        if first_provider
451        else "builtin"
452    )
453    item_id = first_provider.item_id if first_provider else item.path
454
455    media_item: MediaItemType
456    if media_type == MediaType.SOUND_EFFECT:
457        media_item = SoundEffect(
458            item_id=item_id,
459            provider=item_provider,
460            name=name,
461            provider_mappings=provider_mappings,
462            external_ids=external_ids,
463        )
464        if duration is not None:
465            media_item.duration = duration
466    elif media_type == MediaType.RADIO:
467        media_item = Radio(
468            item_id=item_id,
469            provider=item_provider,
470            name=name,
471            provider_mappings=provider_mappings,
472            external_ids=external_ids,
473        )
474    elif media_type == MediaType.PODCAST_EPISODE:
475        if not item.podcast:
476            return None
477        p_provider = item.podcast.provider_domain or item_provider
478        p_item_id = item.podcast.item_id or item.podcast.name
479        podcast_mapping = ItemMapping(
480            item_id=p_item_id,
481            provider=p_provider,
482            name=item.podcast.name,
483            media_type=MediaType.PODCAST,
484        )
485        media_item = PodcastEpisode(
486            item_id=item_id,
487            provider=item_provider,
488            name=name,
489            position=0,
490            provider_mappings=provider_mappings,
491            podcast=podcast_mapping,
492            external_ids=external_ids,
493        )
494        if duration is not None:
495            media_item.duration = duration
496    elif media_type == MediaType.AUDIOBOOK:
497        media_item = Audiobook(
498            item_id=item_id,
499            provider=item_provider,
500            name=name,
501            provider_mappings=provider_mappings,
502            authors=UniqueList(
503                metadata.get("authors", "").split("; ") if metadata.get("authors") else []
504            ),
505            narrators=UniqueList(
506                metadata.get("narrators", "").split("; ") if metadata.get("narrators") else []
507            ),
508            external_ids=external_ids,
509        )
510        if duration is not None:
511            media_item.duration = duration
512    else:
513        media_item = _construct_track(
514            item,
515            metadata,
516            item_id,
517            item_provider,
518            name,
519            duration,
520            provider_mappings,
521            external_ids,
522        )
523
524    for img in item.images:
525        try:
526            image_type = ImageType(img.type)
527        except ValueError:
528            continue
529        media_item.metadata.add_image(
530            MediaItemImage(
531                type=image_type,
532                path=img.path,
533                provider=img.provider,
534                remotely_accessible=img.remotely_accessible,
535            )
536        )
537    return media_item
538
539
540def _resolve_provider_mappings(item: PlaylistItem, mass: MusicAssistant) -> set[ProviderMapping]:
541    """Resolve provider mappings from playlist item, checking loaded providers."""
542    provider_mappings: set[ProviderMapping] = set()
543    for prov_info in item.providers:
544        prov = None
545        if prov_info.instance_id:
546            prov = mass.get_provider(prov_info.instance_id)
547        if not prov:
548            prov = mass.get_provider(prov_info.domain)
549        audio_format = AudioFormat(
550            content_type=ContentType.try_parse(prov_info.content_type),
551            sample_rate=prov_info.sample_rate,
552            bit_depth=prov_info.bit_depth,
553            bit_rate=prov_info.bit_rate,
554        )
555        provider_mappings.add(
556            ProviderMapping(
557                item_id=prov_info.item_id,
558                provider_domain=prov.domain if prov else prov_info.domain,
559                provider_instance=prov.instance_id if prov else prov_info.instance_id,
560                available=prov is not None,
561                audio_format=audio_format,
562            )
563        )
564    return provider_mappings
565
566
567def _builtin_fallback_mappings(path: str, mass: MusicAssistant) -> set[ProviderMapping]:
568    """Return the builtin mapping for a stream URL, which builtin takes as its item_id."""
569    prov = mass.get_provider("builtin")
570    return {
571        ProviderMapping(
572            item_id=path,
573            provider_domain="builtin",
574            provider_instance=prov.instance_id if prov else "builtin",
575            available=prov is not None,
576        )
577    }
578
579
580def _collect_external_ids(metadata: dict[str, str]) -> set[tuple[ExternalID, str]]:
581    """Collect external IDs (ISRC, MusicBrainz) from metadata dict."""
582    external_ids: set[tuple[ExternalID, str]] = set()
583    if isrc := metadata.get("isrc"):
584        external_ids.add((ExternalID.ISRC, isrc))
585    if mbid := metadata.get("mbid"):
586        external_ids.add((ExternalID.MB_RECORDING, mbid))
587    return external_ids
588
589
590def _construct_track(
591    item: PlaylistItem,
592    metadata: dict[str, str],
593    item_id: str,
594    item_provider: str,
595    name: str,
596    duration: int | None,
597    provider_mappings: set[ProviderMapping],
598    external_ids: set[tuple[ExternalID, str]],
599) -> Track:
600    """Construct a Track from playlist item data, including artists and album."""
601    artists: UniqueList[Artist | ItemMapping] = UniqueList()
602    for artist_info in item.artists:
603        artist_provider = artist_info.provider_domain or item_provider
604        artist_item_id = artist_info.item_id or artist_info.name
605        artists.append(
606            ItemMapping(
607                item_id=artist_item_id,
608                provider=artist_provider,
609                name=artist_info.name,
610                media_type=MediaType.ARTIST,
611            )
612        )
613    album_mapping: ItemMapping | None = None
614    if item.album:
615        album_provider = item.album.provider_domain or item_provider
616        album_item_id = item.album.item_id or item.album.name
617        album_mapping = ItemMapping(
618            item_id=album_item_id,
619            provider=album_provider,
620            name=item.album.name,
621            version=item.album.version,
622            media_type=MediaType.ALBUM,
623        )
624    track = Track(
625        item_id=item_id,
626        provider=item_provider,
627        name=name,
628        version=metadata.get("version", ""),
629        artists=artists,
630        album=album_mapping,
631        provider_mappings=provider_mappings,
632        external_ids=external_ids,
633    )
634    if duration is not None:
635        track.duration = duration
636    return track
637
638
639# --------------------------------------------------------------------------- #
640#  Media item to playlist dataclass helpers                                    #
641# --------------------------------------------------------------------------- #
642
643
644def collect_artist_infos(full_item: MediaItem) -> list[ArtistInfo]:
645    """Extract artist info from a media item for M3U serialization."""
646    artist_infos: list[ArtistInfo] = []
647    if not hasattr(full_item, "artists") or not full_item.artists:
648        return artist_infos
649    for artist in full_item.artists:
650        prov_mappings = getattr(artist, "provider_mappings", None)
651        if prov_mappings:
652            first_mapping = next(iter(prov_mappings))
653            a_domain = first_mapping.provider_domain
654            a_item_id = first_mapping.item_id
655            a_instance = first_mapping.provider_instance
656        else:
657            a_domain = artist.provider
658            a_item_id = artist.item_id
659            a_instance = artist.provider
660        artist_infos.append(
661            ArtistInfo(
662                name=artist.name,
663                provider_domain=a_domain,
664                item_id=a_item_id,
665                provider_instance=a_instance,
666            )
667        )
668    return artist_infos
669
670
671def collect_album_info(full_item: MediaItem) -> AlbumInfo | None:
672    """Extract album info from a media item for M3U serialization."""
673    if not hasattr(full_item, "album") or not full_item.album:
674        return None
675    album = full_item.album
676    prov_mappings = getattr(album, "provider_mappings", None)
677    if prov_mappings:
678        first_mapping = next(iter(prov_mappings))
679        al_domain = first_mapping.provider_domain
680        al_item_id = first_mapping.item_id
681        al_instance = first_mapping.provider_instance
682    else:
683        al_domain = album.provider
684        al_item_id = album.item_id
685        al_instance = album.provider
686    return AlbumInfo(
687        name=album.name,
688        provider_domain=al_domain,
689        item_id=al_item_id,
690        provider_instance=al_instance,
691        version=getattr(album, "version", "") or "",
692    )
693
694
695def collect_podcast_info(full_item: MediaItem) -> PodcastInfo | None:
696    """Extract podcast info from a media item for M3U serialization."""
697    if not hasattr(full_item, "podcast") or not full_item.podcast:
698        return None
699    podcast = full_item.podcast
700    prov_mappings = getattr(podcast, "provider_mappings", None)
701    if prov_mappings:
702        first_mapping = next(iter(prov_mappings))
703        p_domain = first_mapping.provider_domain
704        p_item_id = first_mapping.item_id
705        p_instance = first_mapping.provider_instance
706    else:
707        p_domain = podcast.provider
708        p_item_id = podcast.item_id
709        p_instance = podcast.provider
710    return PodcastInfo(
711        name=podcast.name,
712        provider_domain=p_domain,
713        item_id=p_item_id,
714        provider_instance=p_instance,
715    )
716
717
718def media_item_to_playlist_item(full_item: MediaItem) -> PlaylistItem:
719    """
720    Convert a MediaItem to a PlaylistItem with full M3U metadata.
721
722    Pure conversion — takes an already-fetched MediaItem and produces a
723    PlaylistItem suitable for ``generate_m3u``.
724
725    :param full_item: Any MediaItem (Track, Radio, PodcastEpisode, Audiobook, etc.).
726    """
727    # build M3U-compliant EXTINF title
728    if hasattr(full_item, "artists") and full_item.artists:
729        artist_names = ", ".join(a.name for a in full_item.artists)
730        title = f"{artist_names} - {full_item.name}"
731    elif hasattr(full_item, "podcast") and full_item.podcast:
732        title = f"{full_item.podcast.name} - {full_item.name}"
733    else:
734        title = full_item.name
735
736    duration = getattr(full_item, "duration", None)
737
738    # build EXTMA metadata
739    metadata: dict[str, str] = {
740        "media_type": full_item.media_type.value,
741        "name": full_item.name,
742    }
743    if hasattr(full_item, "authors") and full_item.authors:
744        metadata["authors"] = "; ".join(full_item.authors)
745    if hasattr(full_item, "narrators") and full_item.narrators:
746        metadata["narrators"] = "; ".join(full_item.narrators)
747    if full_item.version:
748        metadata["version"] = full_item.version
749    if isrc := full_item.get_external_id(ExternalID.ISRC):
750        metadata["isrc"] = isrc
751    if mbid := full_item.get_external_id(ExternalID.MB_RECORDING):
752        metadata["mbid"] = mbid
753
754    # collect one provider mapping per domain (highest quality)
755    prov_infos: list[ProviderMappingInfo] = []
756    seen_domains: set[str] = set()
757    sorted_mappings = sorted(full_item.provider_mappings, key=lambda x: x.quality, reverse=True)
758    for prov_mapping in sorted_mappings:
759        domain = prov_mapping.provider_domain
760        if domain in seen_domains:
761            continue
762        seen_domains.add(domain)
763        prov_infos.append(
764            ProviderMappingInfo(
765                domain=domain,
766                item_id=prov_mapping.item_id,
767                instance_id=prov_mapping.provider_instance,
768                content_type=prov_mapping.audio_format.content_type.value,
769                sample_rate=prov_mapping.audio_format.sample_rate,
770                bit_depth=prov_mapping.audio_format.bit_depth,
771                bit_rate=prov_mapping.audio_format.bit_rate or 0,
772            )
773        )
774
775    # primary URI = highest quality provider (or first if no mappings)
776    if prov_infos:
777        primary = prov_infos[0]
778        primary_uri = create_uri(full_item.media_type, primary.domain, primary.item_id)
779    else:
780        primary_uri = full_item.uri or ""
781
782    artist_infos = collect_artist_infos(full_item)
783    album_info = collect_album_info(full_item)
784    podcast_info = collect_podcast_info(full_item)
785
786    # collect images
787    images: list[ImageInfo] = []
788    if hasattr(full_item, "metadata") and full_item.metadata and full_item.metadata.images:
789        for img in full_item.metadata.images:
790            images.append(
791                ImageInfo(
792                    type=img.type.value,
793                    path=img.path,
794                    provider=img.provider,
795                    remotely_accessible=img.remotely_accessible,
796                )
797            )
798
799    return PlaylistItem(
800        path=primary_uri,
801        title=title,
802        length=str(duration) if duration else None,
803        metadata=metadata,
804        providers=prov_infos,
805        images=images,
806        artists=artist_infos,
807        album=album_info,
808        podcast=podcast_info,
809    )
810
811
812# --------------------------------------------------------------------------- #
813#  Internal line parsers                                                       #
814# --------------------------------------------------------------------------- #
815
816
817def _parse_hls_stream_info(line: str) -> dict[str, str]:
818    """Parse #EXT-X-STREAM-INF into a dict of stream properties."""
819    stream_info: dict[str, str] = {}
820    for part in line.replace("#EXT-X-STREAM-INF:", "").split(","):
821        if "=" not in part:
822            continue
823        kev_value_parts = part.strip().split("=")
824        stream_info[kev_value_parts[0]] = kev_value_parts[1]
825    return stream_info
826
827
828def _parse_hls_key(line: str, current_key: str | None) -> str | None:
829    """Parse #EXT-X-KEY and return the key URI, or None for METHOD=NONE."""
830    if "METHOD=NONE" in line:
831        return None
832    if ",URI=" in line:
833        return line.split(",URI=")[1].strip('"')
834    return current_key
835
836
837def _parse_extma_line(line: str) -> dict[str, str]:
838    """Parse #EXTMA into a metadata dict (double-pipe-separated key=value pairs)."""
839    raw = line.split("#EXTMA:", 1)[1]
840    metadata: dict[str, str] = {}
841    for pair in raw.split(FIELD_SEPARATOR):
842        if "=" not in pair:
843            continue
844        k, v = pair.split("=", 1)
845        metadata[k.strip()] = v.strip()
846    return metadata
847
848
849def _parse_extprov_line(line: str) -> ProviderMappingInfo | None:
850    """Parse #EXTPROV (double-pipe-separated, 2 required + 5 optional fields)."""
851    parts = line.split("#EXTPROV:", 1)[1].strip().split(FIELD_SEPARATOR)
852    if len(parts) < 2 or len(parts) > 7:
853        return None
854    try:
855        return ProviderMappingInfo(
856            domain=parts[0],
857            item_id=parts[1],
858            instance_id=parts[2] if len(parts) > 2 else "",
859            content_type=parts[3] if len(parts) > 3 else "",
860            sample_rate=int(parts[4]) if len(parts) > 4 and parts[4] else 0,
861            bit_depth=int(parts[5]) if len(parts) > 5 and parts[5] else 0,
862            bit_rate=int(parts[6]) if len(parts) > 6 and parts[6] else 0,
863        )
864    except ValueError:
865        return None
866
867
868def _parse_extimg_line(line: str) -> ImageInfo | None:
869    """Parse #EXTIMG (double-pipe-separated, expects 4 fields)."""
870    parts = line.split("#EXTIMG:", 1)[1].strip().split(FIELD_SEPARATOR)
871    if len(parts) != 4:
872        return None
873    return ImageInfo(
874        type=parts[0],
875        path=parts[1],
876        provider=parts[2],
877        remotely_accessible=parts[3].lower() == "true",
878    )
879
880
881def _parse_extartist_line(line: str) -> ArtistInfo | None:
882    """Parse #EXTARTIST (double-pipe-separated, expects 4 fields)."""
883    parts = line.split("#EXTARTIST:", 1)[1].strip().split(FIELD_SEPARATOR)
884    if len(parts) != 4 or not parts[0]:
885        return None
886    return ArtistInfo(
887        name=parts[0],
888        provider_domain=parts[1],
889        item_id=parts[2],
890        provider_instance=parts[3],
891    )
892
893
894def _parse_extalbum_line(line: str) -> AlbumInfo | None:
895    """Parse #EXTALBUM (double-pipe-separated, expects 5 fields)."""
896    parts = line.split("#EXTALBUM:", 1)[1].strip().split(FIELD_SEPARATOR)
897    if len(parts) != 5 or not parts[0]:
898        return None
899    return AlbumInfo(
900        name=parts[0],
901        provider_domain=parts[1],
902        item_id=parts[2],
903        provider_instance=parts[3],
904        version=parts[4],
905    )
906
907
908def _parse_extpodcast_line(line: str) -> PodcastInfo | None:
909    """Parse #EXTPODCAST (double-pipe-separated, expects 4 fields)."""
910    parts = line.split("#EXTPODCAST:", 1)[1].strip().split(FIELD_SEPARATOR)
911    if len(parts) != 4 or not parts[0]:
912        return None
913    return PodcastInfo(
914        name=parts[0],
915        provider_domain=parts[1],
916        item_id=parts[2],
917        provider_instance=parts[3],
918    )
919