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