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