/
/
/
1"""Parse objects from py-opensonic into Music Assistant types."""
2
3from __future__ import annotations
4
5import logging
6from datetime import UTC, datetime
7from typing import TYPE_CHECKING
8
9from music_assistant_models.enums import ContentType, ImageType, LinkType, MediaType
10from music_assistant_models.errors import InvalidDataError, MediaNotFoundError
11from music_assistant_models.media_items import (
12 Album,
13 Artist,
14 AudioFormat,
15 ItemMapping,
16 MediaItemImage,
17 MediaItemLink,
18 MediaItemMetadata,
19 Playlist,
20 Podcast,
21 PodcastEpisode,
22 ProviderMapping,
23 Radio,
24 Track,
25)
26
27from music_assistant.constants import UNKNOWN_ARTIST
28from music_assistant.helpers.tags import clean_mbid
29from music_assistant.helpers.util import parse_title_and_version
30
31if TYPE_CHECKING:
32 from libopensonic.media import AlbumID3 as SonicAlbum
33 from libopensonic.media import AlbumInfo as SonicAlbumInfo
34 from libopensonic.media import ArtistID3 as SonicArtist
35 from libopensonic.media import ArtistInfo2 as SonicArtistInfo
36 from libopensonic.media import Child as SonicSong
37 from libopensonic.media import InternetRadioStation as SonicRadio
38 from libopensonic.media import Playlist as SonicPlaylist
39 from libopensonic.media import PodcastChannel as SonicPodcast
40 from libopensonic.media import PodcastEpisode as SonicEpisode
41 from libopensonic.media import StructuredLyrics
42
43
44UNKNOWN_ARTIST_ID = "fake_artist_unknown"
45
46
47# Because of some subsonic API weirdness, we have to lookup any podcast episode by finding it in
48# the list of episodes in a channel, to facilitate, we will use both the episode id and the
49# channel id concatenated as an episode id to MA
50EP_CHAN_SEP = "$!$"
51
52
53# We need the following prefix because of the way that Navidrome reports artists for individual
54# tracks on Various Artists albums, see the note in the _parse_track() method and the handling
55# in get_artist()
56NAVI_VARIOUS_PREFIX = "MA-NAVIDROME-"
57
58
59SUBSONIC_DOMAIN = "opensubsonic"
60
61
62def get_item_mapping(instance_id: str, media_type: MediaType, key: str, name: str) -> ItemMapping:
63 """Construct an ItemMapping for the specified media."""
64 return ItemMapping(
65 media_type=media_type,
66 item_id=key,
67 provider=instance_id,
68 name=name,
69 )
70
71
72def parse_track( # noqa: PLR0915
73 logger: logging.Logger,
74 instance_id: str,
75 sonic_song: SonicSong,
76 album: Album | ItemMapping | None = None,
77 lyrics: tuple[str, bool] | None = None,
78) -> Track:
79 """Parse an OpenSubsonic.Child into an MA Track."""
80 # Unfortunately, the Song response type is not defined in the open subsonic spec so we have
81 # implementations which disagree about where the album id for this song should be stored.
82 # We accept either song.ablum_id or song.parent but prefer album_id.
83 if not album:
84 if sonic_song.album_id and sonic_song.album:
85 album = get_item_mapping(
86 instance_id, MediaType.ALBUM, sonic_song.album_id, sonic_song.album
87 )
88 elif sonic_song.parent and sonic_song.album:
89 album = get_item_mapping(
90 instance_id, MediaType.ALBUM, sonic_song.parent, sonic_song.album
91 )
92
93 metadata: MediaItemMetadata = MediaItemMetadata()
94
95 if sonic_song.cover_art:
96 metadata.add_image(
97 MediaItemImage(
98 type=ImageType.THUMB,
99 path=sonic_song.cover_art,
100 provider=instance_id,
101 remotely_accessible=False,
102 )
103 )
104
105 if lyrics:
106 ly, synced = lyrics
107 if synced:
108 metadata.lrc_lyrics = ly
109 else:
110 metadata.lyrics = ly
111
112 if sonic_song.explicit_status and sonic_song.explicit_status != "clean":
113 metadata.explicit = True
114
115 if sonic_song.genre:
116 if not metadata.genres:
117 metadata.genres = set()
118 metadata.genres.add(sonic_song.genre)
119
120 if sonic_song.genres:
121 if not metadata.genres:
122 metadata.genres = set()
123 for g in sonic_song.genres:
124 metadata.genres.add(g.name)
125
126 if sonic_song.moods:
127 metadata.mood = sonic_song.moods[0]
128
129 if sonic_song.contributors:
130 if not metadata.performers:
131 metadata.performers = set()
132 for c in sonic_song.contributors:
133 metadata.performers.add(c.artist.name)
134
135 if isinstance(album, Album) and album.version:
136 name = sonic_song.title
137 version = album.version
138 else:
139 name, version = parse_title_and_version(sonic_song.title)
140
141 track = Track(
142 item_id=sonic_song.id,
143 provider=instance_id,
144 name=name,
145 version=version,
146 album=album,
147 duration=sonic_song.duration or 0,
148 disc_number=sonic_song.disc_number or 0,
149 favorite=bool(sonic_song.starred),
150 metadata=metadata,
151 provider_mappings={
152 ProviderMapping(
153 item_id=sonic_song.id,
154 provider_domain=SUBSONIC_DOMAIN,
155 provider_instance=instance_id,
156 available=True,
157 audio_format=AudioFormat(
158 content_type=ContentType.try_parse(sonic_song.content_type or "?"),
159 sample_rate=sonic_song.sampling_rate or 44100,
160 bit_depth=sonic_song.bit_depth or 16,
161 channels=sonic_song.channel_count or 2,
162 bit_rate=sonic_song.bit_rate,
163 ),
164 )
165 },
166 track_number=sonic_song.track or 0,
167 )
168
169 if mbid := clean_mbid(sonic_song.music_brainz_id, sonic_song.path or f"track {name}", logger):
170 track.mbid = mbid
171
172 if sonic_song.sort_name:
173 track.sort_name = sonic_song.sort_name
174
175 # We need to find an artist for this track but various implementations seem to disagree
176 # about where the artist with the valid ID needs to be found. We will add any artist with
177 # an ID and only use UNKNOWN if none are found.
178
179 if sonic_song.artist_id:
180 track.artists.append(
181 get_item_mapping(
182 instance_id,
183 MediaType.ARTIST,
184 sonic_song.artist_id,
185 sonic_song.artist or UNKNOWN_ARTIST,
186 )
187 )
188
189 if sonic_song.artists:
190 for entry in sonic_song.artists:
191 if entry.id == sonic_song.artist_id:
192 continue
193 if entry.id is not None and entry.name is not None:
194 track.artists.append(
195 get_item_mapping(instance_id, MediaType.ARTIST, entry.id, entry.name)
196 )
197
198 if not track.artists:
199 if sonic_song.artist and not sonic_song.artist_id:
200 # This is how Navidrome handles tracks from albums which are marked
201 # 'Various Artists'. Unfortunately, we cannot lookup this artist independently
202 # because it will not have an entry in the artists table so the best we can do it
203 # add a 'fake' id with the proper artist name and have get_artist() check for this
204 # id and handle it locally.
205 fake_id = f"{NAVI_VARIOUS_PREFIX}{sonic_song.artist}"
206 artist = Artist(
207 item_id=fake_id,
208 provider=SUBSONIC_DOMAIN,
209 name=sonic_song.artist,
210 provider_mappings={
211 ProviderMapping(
212 item_id=fake_id,
213 provider_domain=SUBSONIC_DOMAIN,
214 provider_instance=instance_id,
215 )
216 },
217 )
218 else:
219 logger.info(
220 "Unable to find artist ID for track '%s' with ID '%s'.",
221 sonic_song.title,
222 sonic_song.id,
223 )
224 artist = Artist(
225 item_id=UNKNOWN_ARTIST_ID,
226 name=UNKNOWN_ARTIST,
227 provider=instance_id,
228 provider_mappings={
229 ProviderMapping(
230 item_id=UNKNOWN_ARTIST_ID,
231 provider_domain=SUBSONIC_DOMAIN,
232 provider_instance=instance_id,
233 )
234 },
235 )
236
237 track.artists.append(artist)
238 return track
239
240
241def parse_artist(
242 instance_id: str,
243 sonic_artist: SonicArtist,
244 sonic_info: SonicArtistInfo | None = None,
245 logger: logging.Logger | None = None,
246) -> Artist:
247 """Parse artist and artistInfo into a Music Assistant Artist."""
248 metadata: MediaItemMetadata = MediaItemMetadata()
249
250 if sonic_artist.cover_art:
251 metadata.add_image(
252 MediaItemImage(
253 type=ImageType.THUMB,
254 path=sonic_artist.cover_art,
255 provider=instance_id,
256 remotely_accessible=False,
257 )
258 )
259
260 if sonic_artist.artist_image_url:
261 metadata.add_image(
262 MediaItemImage(
263 type=ImageType.THUMB,
264 path=sonic_artist.artist_image_url,
265 provider=instance_id,
266 remotely_accessible=True,
267 )
268 )
269
270 if sonic_info:
271 if sonic_info.biography:
272 metadata.description = sonic_info.biography
273 if sonic_info.small_image_url:
274 metadata.add_image(
275 MediaItemImage(
276 type=ImageType.THUMB,
277 path=sonic_info.small_image_url,
278 provider=instance_id,
279 remotely_accessible=True,
280 )
281 )
282
283 artist = Artist(
284 item_id=sonic_artist.id,
285 name=sonic_artist.name,
286 metadata=metadata,
287 provider=SUBSONIC_DOMAIN,
288 favorite=bool(sonic_artist.starred),
289 provider_mappings={
290 ProviderMapping(
291 item_id=sonic_artist.id,
292 provider_domain=SUBSONIC_DOMAIN,
293 provider_instance=instance_id,
294 )
295 },
296 sort_name=sonic_artist.sort_name,
297 )
298
299 if mbid := clean_mbid(sonic_artist.music_brainz_id, f"artist {sonic_artist.name}", logger):
300 artist.mbid = mbid
301
302 return artist
303
304
305def parse_album(
306 logger: logging.Logger,
307 instance_id: str,
308 sonic_album: SonicAlbum,
309 sonic_info: SonicAlbumInfo | None = None,
310) -> Album:
311 """Parse album and albumInfo into a Music Assistant Album."""
312 metadata: MediaItemMetadata = MediaItemMetadata()
313
314 if sonic_album.cover_art:
315 metadata.add_image(
316 MediaItemImage(
317 type=ImageType.THUMB,
318 path=sonic_album.cover_art,
319 provider=instance_id,
320 remotely_accessible=False,
321 ),
322 )
323
324 if sonic_info:
325 if sonic_info.small_image_url:
326 metadata.add_image(
327 MediaItemImage(
328 type=ImageType.THUMB,
329 path=sonic_info.small_image_url,
330 remotely_accessible=True,
331 provider=instance_id,
332 )
333 )
334 if sonic_info.notes:
335 metadata.description = sonic_info.notes
336
337 if sonic_album.genre:
338 if not metadata.genres:
339 metadata.genres = set()
340 metadata.genres.add(sonic_album.genre)
341
342 if sonic_album.genres:
343 if not metadata.genres:
344 metadata.genres = set()
345 for g in sonic_album.genres:
346 metadata.genres.add(g.name)
347
348 if sonic_album.moods:
349 metadata.mood = sonic_album.moods[0]
350
351 if sonic_album.version:
352 name = sonic_album.name
353 version = sonic_album.version
354 else:
355 name, version = parse_title_and_version(sonic_album.name)
356
357 album = Album(
358 item_id=sonic_album.id,
359 provider=SUBSONIC_DOMAIN,
360 metadata=metadata,
361 name=name,
362 version=version,
363 favorite=bool(sonic_album.starred),
364 provider_mappings={
365 ProviderMapping(
366 item_id=sonic_album.id,
367 provider_domain=SUBSONIC_DOMAIN,
368 provider_instance=instance_id,
369 )
370 },
371 year=sonic_album.year,
372 )
373
374 if sonic_album.sort_name:
375 album.sort_name = sonic_album.sort_name
376
377 if mbid := clean_mbid(sonic_album.music_brainz_id, f"album {sonic_album.name}", logger):
378 album.mbid = mbid
379
380 if sonic_album.artist_id:
381 album.artists.append(
382 ItemMapping(
383 media_type=MediaType.ARTIST,
384 item_id=sonic_album.artist_id,
385 provider=instance_id,
386 name=sonic_album.artist or UNKNOWN_ARTIST,
387 )
388 )
389 elif not sonic_album.artists:
390 logger.info(
391 "Unable to find an artist ID for album '%s' with ID '%s'.",
392 sonic_album.name,
393 sonic_album.id,
394 )
395 album.artists.append(
396 Artist(
397 item_id=UNKNOWN_ARTIST_ID,
398 name=UNKNOWN_ARTIST,
399 provider=instance_id,
400 provider_mappings={
401 ProviderMapping(
402 item_id=UNKNOWN_ARTIST_ID,
403 provider_domain=SUBSONIC_DOMAIN,
404 provider_instance=instance_id,
405 )
406 },
407 )
408 )
409
410 if sonic_album.artists:
411 for a in sonic_album.artists:
412 if a.id == sonic_album.artist_id:
413 continue
414 album.artists.append(
415 ItemMapping(
416 media_type=MediaType.ARTIST, item_id=a.id, provider=instance_id, name=a.name
417 )
418 )
419
420 return album
421
422
423def parse_radio(instance_id: str, sonic_station: SonicRadio) -> Radio:
424 """Parse an OpenSubsonic internet radio station into an MA Radio item."""
425 metadata: MediaItemMetadata = MediaItemMetadata()
426 if sonic_station.cover_art:
427 metadata.add_image(
428 MediaItemImage(
429 type=ImageType.THUMB,
430 path=sonic_station.cover_art,
431 provider=instance_id,
432 remotely_accessible=False,
433 )
434 )
435
436 radio = Radio(
437 item_id=sonic_station.id,
438 provider=instance_id,
439 name=sonic_station.name,
440 uri=sonic_station.stream_url,
441 metadata=metadata,
442 provider_mappings={
443 ProviderMapping(
444 item_id=sonic_station.id,
445 provider_domain=SUBSONIC_DOMAIN,
446 provider_instance=instance_id,
447 )
448 },
449 )
450
451 if sonic_station.home_page_url:
452 radio.metadata.links = {
453 MediaItemLink(type=LinkType.WEBSITE, url=sonic_station.home_page_url)
454 }
455
456 return radio
457
458
459def parse_playlist(instance_id: str, sonic_playlist: SonicPlaylist) -> Playlist:
460 """Parse subsonic Playlist into MA Playlist."""
461 playlist = Playlist(
462 item_id=sonic_playlist.id,
463 provider=SUBSONIC_DOMAIN,
464 name=sonic_playlist.name,
465 is_editable=True,
466 provider_mappings={
467 ProviderMapping(
468 item_id=sonic_playlist.id,
469 provider_domain=SUBSONIC_DOMAIN,
470 provider_instance=instance_id,
471 )
472 },
473 )
474
475 if sonic_playlist.cover_art:
476 playlist.metadata.add_image(
477 MediaItemImage(
478 type=ImageType.THUMB,
479 path=sonic_playlist.cover_art,
480 provider=instance_id,
481 remotely_accessible=False,
482 )
483 )
484
485 return playlist
486
487
488def parse_podcast(instance_id: str, sonic_podcast: SonicPodcast) -> Podcast:
489 """Parse Subsonic PodcastChannel into MA Podcast."""
490 if not sonic_podcast.title:
491 raise InvalidDataError(
492 f"Subsonic Podcast ({sonic_podcast.id})is missing required name field."
493 )
494 podcast = Podcast(
495 item_id=sonic_podcast.id,
496 provider=SUBSONIC_DOMAIN,
497 name=sonic_podcast.title,
498 uri=sonic_podcast.url,
499 total_episodes=len(sonic_podcast.episode) if sonic_podcast.episode else 0,
500 provider_mappings={
501 ProviderMapping(
502 item_id=sonic_podcast.id,
503 provider_domain=SUBSONIC_DOMAIN,
504 provider_instance=instance_id,
505 )
506 },
507 )
508
509 podcast.metadata.description = sonic_podcast.description
510
511 if sonic_podcast.cover_art:
512 podcast.metadata.add_image(
513 MediaItemImage(
514 type=ImageType.THUMB,
515 path=sonic_podcast.cover_art,
516 provider=instance_id,
517 remotely_accessible=False,
518 )
519 )
520
521 return podcast
522
523
524def parse_epsiode(
525 instance_id: str,
526 sonic_episode: SonicEpisode,
527 sonic_channel: SonicPodcast,
528 position: int = 0,
529) -> PodcastEpisode:
530 """
531 Parse an Open Subsonic Podcast Episode into an MA PodcastEpisode.
532
533 :param position: The episode's listing position. Defaults to 0 (unknown).
534 """
535 eid = f"{sonic_episode.channel_id}{EP_CHAN_SEP}{sonic_episode.id}"
536 if not sonic_channel.episode:
537 raise MediaNotFoundError(f"Podcast Channel '{sonic_channel.id}' missing episode list")
538
539 episode = PodcastEpisode(
540 item_id=eid,
541 provider=SUBSONIC_DOMAIN,
542 name=sonic_episode.title,
543 position=position,
544 podcast=parse_podcast(instance_id, sonic_channel),
545 provider_mappings={
546 ProviderMapping(
547 item_id=eid,
548 provider_domain=SUBSONIC_DOMAIN,
549 provider_instance=instance_id,
550 )
551 },
552 duration=sonic_episode.duration or 0,
553 )
554
555 if sonic_episode.publish_date:
556 episode.metadata.release_date = datetime.fromisoformat(sonic_episode.publish_date)
557
558 if sonic_episode.description:
559 episode.metadata.description = sonic_episode.description
560
561 if sonic_episode.cover_art:
562 episode.metadata.add_image(
563 MediaItemImage(
564 type=ImageType.THUMB,
565 path=sonic_episode.cover_art,
566 provider=instance_id,
567 remotely_accessible=False,
568 )
569 )
570 elif sonic_channel.cover_art:
571 episode.metadata.add_image(
572 MediaItemImage(
573 type=ImageType.THUMB,
574 path=sonic_channel.cover_art,
575 provider=instance_id,
576 remotely_accessible=False,
577 )
578 )
579
580 return episode
581
582
583def parse_structured_lyrics(lyrics: StructuredLyrics) -> tuple[str, bool]:
584 """Parse the Open Subsonic Structured lyrics objest into MA Lyrics."""
585 lines: list[str] = []
586 if lyrics.synced:
587 offset: int = int(lyrics.offset) if lyrics.offset else 0
588 for line in lyrics.line:
589 if line.start is None:
590 raise InvalidDataError("Open Subsonic Synced lyric missing time index")
591 ms = int(line.start) + offset
592 dt = datetime.fromtimestamp(ms / 1000, tz=UTC)
593 ts = dt.strftime("%M:%S.%f")[:-4]
594 lines.append(f"[{ts}]{line.value}")
595 else:
596 for line in lyrics.line:
597 lines.append(line.value)
598 return ("\n".join(lines), lyrics.synced)
599