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            raw_data = await resp.content.read(64 * 1024)
304            encoding = await detect_charset(raw_data, preferred=resp.charset)
305            playlist_data = raw_data.decode(encoding, errors="replace")
306    except TimeoutError as err:
307        msg = f"Timeout while fetching playlist {url}"
308        raise InvalidDataError(msg) from err
309    except client_exceptions.ClientError as err:
310        msg = f"Error while fetching playlist {url}"
311        raise InvalidDataError(msg) from err
312
313    if (
314        raise_on_hls and "#EXT-X-VERSION:" in playlist_data
315    ) or "#EXT-X-STREAM-INF:" in playlist_data:
316        raise IsHLSPlaylist
317
318    if urlparse(url).path.endswith("pls") or "[playlist]" in playlist_data:
319        playlist = parse_pls(playlist_data)
320    else:
321        playlist = parse_m3u(playlist_data)
322
323    if not playlist:
324        msg = f"Empty playlist {url}"
325        raise InvalidDataError(msg)
326
327    return playlist
328
329
330# --------------------------------------------------------------------------- #
331#  Generation                                                                  #
332# --------------------------------------------------------------------------- #
333
334
335def generate_m3u(
336    playlist_name: str,
337    items: Sequence[PlaylistItem],
338    playlist_image_url: str | None = None,
339) -> str:
340    """
341    Generate an M3U8 playlist string from PlaylistItem entries.
342
343    :param playlist_name: Human-readable name (written as #PLAYLIST directive).
344    :param items: Entries to write. Only fields that are set are emitted.
345    :param playlist_image_url: Optional playlist cover image URL.
346    """
347    # Playlist-level image using #EXTIMG directive (de facto standard for playlist covers)
348    lines: list[str] = ["#EXTM3U"]
349    if playlist_image_url:
350        lines.append(f"#EXTIMG:{playlist_image_url}")
351    lines.append(f"#PLAYLIST:{playlist_name}")
352    sep = FIELD_SEPARATOR
353    for item in items:
354        if item.metadata:
355            pairs = sep.join(f"{k}={v}" for k, v in item.metadata.items())
356            lines.append(f"#EXTMA:{pairs}")
357        for prov in item.providers:
358            fields = [
359                prov.domain,
360                prov.item_id,
361                prov.instance_id,
362                prov.content_type,
363                str(prov.sample_rate),
364                str(prov.bit_depth),
365                str(prov.bit_rate),
366            ]
367            lines.append(f"#EXTPROV:{sep.join(fields)}")
368        for artist in item.artists:
369            fields = [artist.name, artist.provider_domain, artist.item_id, artist.provider_instance]
370            lines.append(f"#EXTARTIST:{sep.join(fields)}")
371        if item.album:
372            fields = [
373                item.album.name,
374                item.album.provider_domain,
375                item.album.item_id,
376                item.album.provider_instance,
377                item.album.version,
378            ]
379            lines.append(f"#EXTALBUM:{sep.join(fields)}")
380        if item.podcast:
381            fields = [
382                item.podcast.name,
383                item.podcast.provider_domain,
384                item.podcast.item_id,
385                item.podcast.provider_instance,
386            ]
387            lines.append(f"#EXTPODCAST:{sep.join(fields)}")
388        for img in item.images:
389            remotely = "true" if img.remotely_accessible else "false"
390            fields = [img.type, img.path, img.provider, remotely]
391            lines.append(f"#EXTIMG:{sep.join(fields)}")
392        if item.title is not None:
393            # -1 is the M3U convention for unknown length; without it an entry without
394            # duration (such as a radio station) loses its title on every rewrite
395            lines.append(f"#EXTINF:{item.length if item.length is not None else -1},{item.title}")
396        lines.append(item.path)
397    return "\n".join(lines) + "\n"
398
399
400# --------------------------------------------------------------------------- #
401#  Media item construction from playlist metadata                              #
402# --------------------------------------------------------------------------- #
403
404
405def construct_media_item_from_playlist_item(
406    item: PlaylistItem,
407    mass: MusicAssistant,
408    default_media_type: MediaType = MediaType.TRACK,
409) -> MediaItemType | None:
410    """
411    Construct a MediaItem from a PlaylistItem's stored metadata.
412
413    Resolves provider mappings by instance_id first (free dict lookup),
414    falling back to domain if the instance no longer exists.
415
416    :param item: Parsed PlaylistItem with metadata, providers, and images.
417    :param mass: MusicAssistant instance for provider resolution.
418    :param default_media_type: Media type to build when the entry's #EXTMA metadata
419        holds no usable media_type.
420    """
421    metadata = item.metadata or {}
422    try:
423        media_type = MediaType(metadata.get("media_type", default_media_type.value))
424    except ValueError:
425        media_type = default_media_type
426    name = metadata.get("name") or item.title or item.path
427    duration = try_parse_int(item.length, default=None) if item.length else None
428
429    provider_mappings = _resolve_provider_mappings(item, mass)
430    if not provider_mappings and item.path.startswith(BUILTIN_URL_SCHEMES):
431        # an item without any mapping never reaches library_add, leaving an unplayable
432        # library entry, so a plain stream URL is mapped to the provider that serves it
433        provider_mappings = _builtin_fallback_mappings(item.path, mass)
434    external_ids = _collect_external_ids(metadata)
435
436    # a set has no order, so sort to keep the chosen mapping stable across runs
437    sorted_mappings = sorted(provider_mappings, key=lambda pm: (pm.provider_instance, pm.item_id))
438    first_provider = next(
439        (pm for pm in sorted_mappings if pm.available),
440        next(iter(sorted_mappings), None),
441    )
442    # prefer the instance over the domain: a domain resolves to whichever instance of it
443    # happens to be loaded first, which is the wrong one when several are configured
444    item_provider = (
445        (first_provider.provider_instance or first_provider.provider_domain)
446        if first_provider
447        else "builtin"
448    )
449    item_id = first_provider.item_id if first_provider else item.path
450
451    media_item: MediaItemType
452    if media_type == MediaType.SOUND_EFFECT:
453        media_item = SoundEffect(
454            item_id=item_id,
455            provider=item_provider,
456            name=name,
457            provider_mappings=provider_mappings,
458            external_ids=external_ids,
459        )
460        if duration is not None:
461            media_item.duration = duration
462    elif media_type == MediaType.RADIO:
463        media_item = Radio(
464            item_id=item_id,
465            provider=item_provider,
466            name=name,
467            provider_mappings=provider_mappings,
468            external_ids=external_ids,
469        )
470    elif media_type == MediaType.PODCAST_EPISODE:
471        if not item.podcast:
472            return None
473        p_provider = item.podcast.provider_domain or item_provider
474        p_item_id = item.podcast.item_id or item.podcast.name
475        podcast_mapping = ItemMapping(
476            item_id=p_item_id,
477            provider=p_provider,
478            name=item.podcast.name,
479            media_type=MediaType.PODCAST,
480        )
481        media_item = PodcastEpisode(
482            item_id=item_id,
483            provider=item_provider,
484            name=name,
485            position=0,
486            provider_mappings=provider_mappings,
487            podcast=podcast_mapping,
488            external_ids=external_ids,
489        )
490        if duration is not None:
491            media_item.duration = duration
492    elif media_type == MediaType.AUDIOBOOK:
493        media_item = Audiobook(
494            item_id=item_id,
495            provider=item_provider,
496            name=name,
497            provider_mappings=provider_mappings,
498            authors=UniqueList(
499                metadata.get("authors", "").split("; ") if metadata.get("authors") else []
500            ),
501            narrators=UniqueList(
502                metadata.get("narrators", "").split("; ") if metadata.get("narrators") else []
503            ),
504            external_ids=external_ids,
505        )
506        if duration is not None:
507            media_item.duration = duration
508    else:
509        media_item = _construct_track(
510            item,
511            metadata,
512            item_id,
513            item_provider,
514            name,
515            duration,
516            provider_mappings,
517            external_ids,
518        )
519
520    for img in item.images:
521        try:
522            image_type = ImageType(img.type)
523        except ValueError:
524            continue
525        media_item.metadata.add_image(
526            MediaItemImage(
527                type=image_type,
528                path=img.path,
529                provider=img.provider,
530                remotely_accessible=img.remotely_accessible,
531            )
532        )
533    return media_item
534
535
536def _resolve_provider_mappings(item: PlaylistItem, mass: MusicAssistant) -> set[ProviderMapping]:
537    """Resolve provider mappings from playlist item, checking loaded providers."""
538    provider_mappings: set[ProviderMapping] = set()
539    for prov_info in item.providers:
540        prov = None
541        if prov_info.instance_id:
542            prov = mass.get_provider(prov_info.instance_id)
543        if not prov:
544            prov = mass.get_provider(prov_info.domain)
545        audio_format = AudioFormat(
546            content_type=ContentType.try_parse(prov_info.content_type),
547            sample_rate=prov_info.sample_rate,
548            bit_depth=prov_info.bit_depth,
549            bit_rate=prov_info.bit_rate,
550        )
551        provider_mappings.add(
552            ProviderMapping(
553                item_id=prov_info.item_id,
554                provider_domain=prov.domain if prov else prov_info.domain,
555                provider_instance=prov.instance_id if prov else prov_info.instance_id,
556                available=prov is not None,
557                audio_format=audio_format,
558            )
559        )
560    return provider_mappings
561
562
563def _builtin_fallback_mappings(path: str, mass: MusicAssistant) -> set[ProviderMapping]:
564    """Return the builtin mapping for a stream URL, which builtin takes as its item_id."""
565    prov = mass.get_provider("builtin")
566    return {
567        ProviderMapping(
568            item_id=path,
569            provider_domain="builtin",
570            provider_instance=prov.instance_id if prov else "builtin",
571            available=prov is not None,
572        )
573    }
574
575
576def _collect_external_ids(metadata: dict[str, str]) -> set[tuple[ExternalID, str]]:
577    """Collect external IDs (ISRC, MusicBrainz) from metadata dict."""
578    external_ids: set[tuple[ExternalID, str]] = set()
579    if isrc := metadata.get("isrc"):
580        external_ids.add((ExternalID.ISRC, isrc))
581    if mbid := metadata.get("mbid"):
582        external_ids.add((ExternalID.MB_RECORDING, mbid))
583    return external_ids
584
585
586def _construct_track(
587    item: PlaylistItem,
588    metadata: dict[str, str],
589    item_id: str,
590    item_provider: str,
591    name: str,
592    duration: int | None,
593    provider_mappings: set[ProviderMapping],
594    external_ids: set[tuple[ExternalID, str]],
595) -> Track:
596    """Construct a Track from playlist item data, including artists and album."""
597    artists: UniqueList[Artist | ItemMapping] = UniqueList()
598    for artist_info in item.artists:
599        artist_provider = artist_info.provider_domain or item_provider
600        artist_item_id = artist_info.item_id or artist_info.name
601        artists.append(
602            ItemMapping(
603                item_id=artist_item_id,
604                provider=artist_provider,
605                name=artist_info.name,
606                media_type=MediaType.ARTIST,
607            )
608        )
609    album_mapping: ItemMapping | None = None
610    if item.album:
611        album_provider = item.album.provider_domain or item_provider
612        album_item_id = item.album.item_id or item.album.name
613        album_mapping = ItemMapping(
614            item_id=album_item_id,
615            provider=album_provider,
616            name=item.album.name,
617            version=item.album.version,
618            media_type=MediaType.ALBUM,
619        )
620    track = Track(
621        item_id=item_id,
622        provider=item_provider,
623        name=name,
624        version=metadata.get("version", ""),
625        artists=artists,
626        album=album_mapping,
627        provider_mappings=provider_mappings,
628        external_ids=external_ids,
629    )
630    if duration is not None:
631        track.duration = duration
632    return track
633
634
635# --------------------------------------------------------------------------- #
636#  Media item to playlist dataclass helpers                                    #
637# --------------------------------------------------------------------------- #
638
639
640def collect_artist_infos(full_item: MediaItem) -> list[ArtistInfo]:
641    """Extract artist info from a media item for M3U serialization."""
642    artist_infos: list[ArtistInfo] = []
643    if not hasattr(full_item, "artists") or not full_item.artists:
644        return artist_infos
645    for artist in full_item.artists:
646        prov_mappings = getattr(artist, "provider_mappings", None)
647        if prov_mappings:
648            first_mapping = next(iter(prov_mappings))
649            a_domain = first_mapping.provider_domain
650            a_item_id = first_mapping.item_id
651            a_instance = first_mapping.provider_instance
652        else:
653            a_domain = artist.provider
654            a_item_id = artist.item_id
655            a_instance = artist.provider
656        artist_infos.append(
657            ArtistInfo(
658                name=artist.name,
659                provider_domain=a_domain,
660                item_id=a_item_id,
661                provider_instance=a_instance,
662            )
663        )
664    return artist_infos
665
666
667def collect_album_info(full_item: MediaItem) -> AlbumInfo | None:
668    """Extract album info from a media item for M3U serialization."""
669    if not hasattr(full_item, "album") or not full_item.album:
670        return None
671    album = full_item.album
672    prov_mappings = getattr(album, "provider_mappings", None)
673    if prov_mappings:
674        first_mapping = next(iter(prov_mappings))
675        al_domain = first_mapping.provider_domain
676        al_item_id = first_mapping.item_id
677        al_instance = first_mapping.provider_instance
678    else:
679        al_domain = album.provider
680        al_item_id = album.item_id
681        al_instance = album.provider
682    return AlbumInfo(
683        name=album.name,
684        provider_domain=al_domain,
685        item_id=al_item_id,
686        provider_instance=al_instance,
687        version=getattr(album, "version", "") or "",
688    )
689
690
691def collect_podcast_info(full_item: MediaItem) -> PodcastInfo | None:
692    """Extract podcast info from a media item for M3U serialization."""
693    if not hasattr(full_item, "podcast") or not full_item.podcast:
694        return None
695    podcast = full_item.podcast
696    prov_mappings = getattr(podcast, "provider_mappings", None)
697    if prov_mappings:
698        first_mapping = next(iter(prov_mappings))
699        p_domain = first_mapping.provider_domain
700        p_item_id = first_mapping.item_id
701        p_instance = first_mapping.provider_instance
702    else:
703        p_domain = podcast.provider
704        p_item_id = podcast.item_id
705        p_instance = podcast.provider
706    return PodcastInfo(
707        name=podcast.name,
708        provider_domain=p_domain,
709        item_id=p_item_id,
710        provider_instance=p_instance,
711    )
712
713
714def media_item_to_playlist_item(full_item: MediaItem) -> PlaylistItem:
715    """
716    Convert a MediaItem to a PlaylistItem with full M3U metadata.
717
718    Pure conversion — takes an already-fetched MediaItem and produces a
719    PlaylistItem suitable for ``generate_m3u``.
720
721    :param full_item: Any MediaItem (Track, Radio, PodcastEpisode, Audiobook, etc.).
722    """
723    # build M3U-compliant EXTINF title
724    if hasattr(full_item, "artists") and full_item.artists:
725        artist_names = ", ".join(a.name for a in full_item.artists)
726        title = f"{artist_names} - {full_item.name}"
727    elif hasattr(full_item, "podcast") and full_item.podcast:
728        title = f"{full_item.podcast.name} - {full_item.name}"
729    else:
730        title = full_item.name
731
732    duration = getattr(full_item, "duration", None)
733
734    # build EXTMA metadata
735    metadata: dict[str, str] = {
736        "media_type": full_item.media_type.value,
737        "name": full_item.name,
738    }
739    if hasattr(full_item, "authors") and full_item.authors:
740        metadata["authors"] = "; ".join(full_item.authors)
741    if hasattr(full_item, "narrators") and full_item.narrators:
742        metadata["narrators"] = "; ".join(full_item.narrators)
743    if full_item.version:
744        metadata["version"] = full_item.version
745    if isrc := full_item.get_external_id(ExternalID.ISRC):
746        metadata["isrc"] = isrc
747    if mbid := full_item.get_external_id(ExternalID.MB_RECORDING):
748        metadata["mbid"] = mbid
749
750    # collect one provider mapping per domain (highest quality)
751    prov_infos: list[ProviderMappingInfo] = []
752    seen_domains: set[str] = set()
753    sorted_mappings = sorted(full_item.provider_mappings, key=lambda x: x.quality, reverse=True)
754    for prov_mapping in sorted_mappings:
755        domain = prov_mapping.provider_domain
756        if domain in seen_domains:
757            continue
758        seen_domains.add(domain)
759        prov_infos.append(
760            ProviderMappingInfo(
761                domain=domain,
762                item_id=prov_mapping.item_id,
763                instance_id=prov_mapping.provider_instance,
764                content_type=prov_mapping.audio_format.content_type.value,
765                sample_rate=prov_mapping.audio_format.sample_rate,
766                bit_depth=prov_mapping.audio_format.bit_depth,
767                bit_rate=prov_mapping.audio_format.bit_rate or 0,
768            )
769        )
770
771    # primary URI = highest quality provider (or first if no mappings)
772    if prov_infos:
773        primary = prov_infos[0]
774        primary_uri = create_uri(full_item.media_type, primary.domain, primary.item_id)
775    else:
776        primary_uri = full_item.uri or ""
777
778    artist_infos = collect_artist_infos(full_item)
779    album_info = collect_album_info(full_item)
780    podcast_info = collect_podcast_info(full_item)
781
782    # collect images
783    images: list[ImageInfo] = []
784    if hasattr(full_item, "metadata") and full_item.metadata and full_item.metadata.images:
785        for img in full_item.metadata.images:
786            images.append(
787                ImageInfo(
788                    type=img.type.value,
789                    path=img.path,
790                    provider=img.provider,
791                    remotely_accessible=img.remotely_accessible,
792                )
793            )
794
795    return PlaylistItem(
796        path=primary_uri,
797        title=title,
798        length=str(duration) if duration else None,
799        metadata=metadata,
800        providers=prov_infos,
801        images=images,
802        artists=artist_infos,
803        album=album_info,
804        podcast=podcast_info,
805    )
806
807
808# --------------------------------------------------------------------------- #
809#  Internal line parsers                                                       #
810# --------------------------------------------------------------------------- #
811
812
813def _parse_hls_stream_info(line: str) -> dict[str, str]:
814    """Parse #EXT-X-STREAM-INF into a dict of stream properties."""
815    stream_info: dict[str, str] = {}
816    for part in line.replace("#EXT-X-STREAM-INF:", "").split(","):
817        if "=" not in part:
818            continue
819        kev_value_parts = part.strip().split("=")
820        stream_info[kev_value_parts[0]] = kev_value_parts[1]
821    return stream_info
822
823
824def _parse_hls_key(line: str, current_key: str | None) -> str | None:
825    """Parse #EXT-X-KEY and return the key URI, or None for METHOD=NONE."""
826    if "METHOD=NONE" in line:
827        return None
828    if ",URI=" in line:
829        return line.split(",URI=")[1].strip('"')
830    return current_key
831
832
833def _parse_extma_line(line: str) -> dict[str, str]:
834    """Parse #EXTMA into a metadata dict (double-pipe-separated key=value pairs)."""
835    raw = line.split("#EXTMA:", 1)[1]
836    metadata: dict[str, str] = {}
837    for pair in raw.split(FIELD_SEPARATOR):
838        if "=" not in pair:
839            continue
840        k, v = pair.split("=", 1)
841        metadata[k.strip()] = v.strip()
842    return metadata
843
844
845def _parse_extprov_line(line: str) -> ProviderMappingInfo | None:
846    """Parse #EXTPROV (double-pipe-separated, 2 required + 5 optional fields)."""
847    parts = line.split("#EXTPROV:", 1)[1].strip().split(FIELD_SEPARATOR)
848    if len(parts) < 2 or len(parts) > 7:
849        return None
850    try:
851        return ProviderMappingInfo(
852            domain=parts[0],
853            item_id=parts[1],
854            instance_id=parts[2] if len(parts) > 2 else "",
855            content_type=parts[3] if len(parts) > 3 else "",
856            sample_rate=int(parts[4]) if len(parts) > 4 and parts[4] else 0,
857            bit_depth=int(parts[5]) if len(parts) > 5 and parts[5] else 0,
858            bit_rate=int(parts[6]) if len(parts) > 6 and parts[6] else 0,
859        )
860    except ValueError:
861        return None
862
863
864def _parse_extimg_line(line: str) -> ImageInfo | None:
865    """Parse #EXTIMG (double-pipe-separated, expects 4 fields)."""
866    parts = line.split("#EXTIMG:", 1)[1].strip().split(FIELD_SEPARATOR)
867    if len(parts) != 4:
868        return None
869    return ImageInfo(
870        type=parts[0],
871        path=parts[1],
872        provider=parts[2],
873        remotely_accessible=parts[3].lower() == "true",
874    )
875
876
877def _parse_extartist_line(line: str) -> ArtistInfo | None:
878    """Parse #EXTARTIST (double-pipe-separated, expects 4 fields)."""
879    parts = line.split("#EXTARTIST:", 1)[1].strip().split(FIELD_SEPARATOR)
880    if len(parts) != 4 or not parts[0]:
881        return None
882    return ArtistInfo(
883        name=parts[0],
884        provider_domain=parts[1],
885        item_id=parts[2],
886        provider_instance=parts[3],
887    )
888
889
890def _parse_extalbum_line(line: str) -> AlbumInfo | None:
891    """Parse #EXTALBUM (double-pipe-separated, expects 5 fields)."""
892    parts = line.split("#EXTALBUM:", 1)[1].strip().split(FIELD_SEPARATOR)
893    if len(parts) != 5 or not parts[0]:
894        return None
895    return AlbumInfo(
896        name=parts[0],
897        provider_domain=parts[1],
898        item_id=parts[2],
899        provider_instance=parts[3],
900        version=parts[4],
901    )
902
903
904def _parse_extpodcast_line(line: str) -> PodcastInfo | None:
905    """Parse #EXTPODCAST (double-pipe-separated, expects 4 fields)."""
906    parts = line.split("#EXTPODCAST:", 1)[1].strip().split(FIELD_SEPARATOR)
907    if len(parts) != 4 or not parts[0]:
908        return None
909    return PodcastInfo(
910        name=parts[0],
911        provider_domain=parts[1],
912        item_id=parts[2],
913        provider_instance=parts[3],
914    )
915