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