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