music-assistant-server

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