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