music-assistant-server

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