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.cover_art:
225        metadata.add_image(
226            MediaItemImage(
227                type=ImageType.THUMB,
228                path=sonic_artist.cover_art,
229                provider=instance_id,
230                remotely_accessible=False,
231            )
232        )
233
234    if sonic_artist.artist_image_url:
235        metadata.add_image(
236            MediaItemImage(
237                type=ImageType.THUMB,
238                path=sonic_artist.artist_image_url,
239                provider=instance_id,
240                remotely_accessible=True,
241            )
242        )
243
244    if sonic_info:
245        if sonic_info.biography:
246            metadata.description = sonic_info.biography
247        if sonic_info.small_image_url:
248            metadata.add_image(
249                MediaItemImage(
250                    type=ImageType.THUMB,
251                    path=sonic_info.small_image_url,
252                    provider=instance_id,
253                    remotely_accessible=True,
254                )
255            )
256
257    artist = Artist(
258        item_id=sonic_artist.id,
259        name=sonic_artist.name,
260        metadata=metadata,
261        provider=SUBSONIC_DOMAIN,
262        favorite=bool(sonic_artist.starred),
263        provider_mappings={
264            ProviderMapping(
265                item_id=sonic_artist.id,
266                provider_domain=SUBSONIC_DOMAIN,
267                provider_instance=instance_id,
268            )
269        },
270        sort_name=sonic_artist.sort_name,
271    )
272
273    if sonic_artist.music_brainz_id:
274        artist.mbid = sonic_artist.music_brainz_id
275
276    return artist
277
278
279def parse_album(
280    logger: logging.Logger,
281    instance_id: str,
282    sonic_album: SonicAlbum,
283    sonic_info: SonicAlbumInfo | None = None,
284) -> Album:
285    """Parse album and albumInfo into a Music Assistant Album."""
286    metadata: MediaItemMetadata = MediaItemMetadata()
287
288    if sonic_album.cover_art:
289        metadata.add_image(
290            MediaItemImage(
291                type=ImageType.THUMB,
292                path=sonic_album.cover_art,
293                provider=instance_id,
294                remotely_accessible=False,
295            ),
296        )
297
298    if sonic_info:
299        if sonic_info.small_image_url:
300            metadata.add_image(
301                MediaItemImage(
302                    type=ImageType.THUMB,
303                    path=sonic_info.small_image_url,
304                    remotely_accessible=True,
305                    provider=instance_id,
306                )
307            )
308        if sonic_info.notes:
309            metadata.description = sonic_info.notes
310
311    if sonic_album.genre:
312        if not metadata.genres:
313            metadata.genres = set()
314        metadata.genres.add(sonic_album.genre)
315
316    if sonic_album.genres:
317        if not metadata.genres:
318            metadata.genres = set()
319        for g in sonic_album.genres:
320            metadata.genres.add(g.name)
321
322    if sonic_album.moods:
323        metadata.mood = sonic_album.moods[0]
324
325    if sonic_album.version:
326        name = sonic_album.name
327        version = sonic_album.version
328    else:
329        name, version = parse_title_and_version(sonic_album.name)
330
331    album = Album(
332        item_id=sonic_album.id,
333        provider=SUBSONIC_DOMAIN,
334        metadata=metadata,
335        name=name,
336        version=version,
337        favorite=bool(sonic_album.starred),
338        provider_mappings={
339            ProviderMapping(
340                item_id=sonic_album.id,
341                provider_domain=SUBSONIC_DOMAIN,
342                provider_instance=instance_id,
343            )
344        },
345        year=sonic_album.year,
346    )
347
348    if sonic_album.sort_name:
349        album.sort_name = sonic_album.sort_name
350
351    if sonic_album.music_brainz_id:
352        album.mbid = sonic_album.music_brainz_id
353
354    if sonic_album.artist_id:
355        album.artists.append(
356            ItemMapping(
357                media_type=MediaType.ARTIST,
358                item_id=sonic_album.artist_id,
359                provider=instance_id,
360                name=sonic_album.artist or UNKNOWN_ARTIST,
361            )
362        )
363    elif not sonic_album.artists:
364        logger.info(
365            "Unable to find an artist ID for album '%s' with ID '%s'.",
366            sonic_album.name,
367            sonic_album.id,
368        )
369        album.artists.append(
370            Artist(
371                item_id=UNKNOWN_ARTIST_ID,
372                name=UNKNOWN_ARTIST,
373                provider=instance_id,
374                provider_mappings={
375                    ProviderMapping(
376                        item_id=UNKNOWN_ARTIST_ID,
377                        provider_domain=SUBSONIC_DOMAIN,
378                        provider_instance=instance_id,
379                    )
380                },
381            )
382        )
383
384    if sonic_album.artists:
385        for a in sonic_album.artists:
386            if a.id == sonic_album.artist_id:
387                continue
388            album.artists.append(
389                ItemMapping(
390                    media_type=MediaType.ARTIST, item_id=a.id, provider=instance_id, name=a.name
391                )
392            )
393
394    return album
395
396
397def parse_playlist(instance_id: str, sonic_playlist: SonicPlaylist) -> Playlist:
398    """Parse subsonic Playlist into MA Playlist."""
399    playlist = Playlist(
400        item_id=sonic_playlist.id,
401        provider=SUBSONIC_DOMAIN,
402        name=sonic_playlist.name,
403        is_editable=True,
404        provider_mappings={
405            ProviderMapping(
406                item_id=sonic_playlist.id,
407                provider_domain=SUBSONIC_DOMAIN,
408                provider_instance=instance_id,
409            )
410        },
411    )
412
413    if sonic_playlist.cover_art:
414        playlist.metadata.add_image(
415            MediaItemImage(
416                type=ImageType.THUMB,
417                path=sonic_playlist.cover_art,
418                provider=instance_id,
419                remotely_accessible=False,
420            )
421        )
422
423    return playlist
424
425
426def parse_podcast(instance_id: str, sonic_podcast: SonicPodcast) -> Podcast:
427    """Parse Subsonic PodcastChannel into MA Podcast."""
428    if not sonic_podcast.title:
429        raise InvalidDataError(
430            f"Subsonic Podcast ({sonic_podcast.id})is missing required name field."
431        )
432    podcast = Podcast(
433        item_id=sonic_podcast.id,
434        provider=SUBSONIC_DOMAIN,
435        name=sonic_podcast.title,
436        uri=sonic_podcast.url,
437        total_episodes=len(sonic_podcast.episode) if sonic_podcast.episode else 0,
438        provider_mappings={
439            ProviderMapping(
440                item_id=sonic_podcast.id,
441                provider_domain=SUBSONIC_DOMAIN,
442                provider_instance=instance_id,
443            )
444        },
445    )
446
447    podcast.metadata.description = sonic_podcast.description
448
449    if sonic_podcast.cover_art:
450        podcast.metadata.add_image(
451            MediaItemImage(
452                type=ImageType.THUMB,
453                path=sonic_podcast.cover_art,
454                provider=instance_id,
455                remotely_accessible=False,
456            )
457        )
458
459    return podcast
460
461
462def parse_epsiode(
463    instance_id: str, sonic_episode: SonicEpisode, sonic_channel: SonicPodcast
464) -> PodcastEpisode:
465    """Parse an Open Subsonic Podcast Episode into an MA PodcastEpisode."""
466    eid = f"{sonic_episode.channel_id}{EP_CHAN_SEP}{sonic_episode.id}"
467    pos = 1
468    if not sonic_channel.episode:
469        raise MediaNotFoundError(f"Podcast Channel '{sonic_channel.id}' missing episode list")
470
471    for ep in sonic_channel.episode:
472        if ep.id == sonic_episode.id:
473            break
474        pos += 1
475
476    episode = PodcastEpisode(
477        item_id=eid,
478        provider=SUBSONIC_DOMAIN,
479        name=sonic_episode.title,
480        position=pos,
481        podcast=parse_podcast(instance_id, sonic_channel),
482        provider_mappings={
483            ProviderMapping(
484                item_id=eid,
485                provider_domain=SUBSONIC_DOMAIN,
486                provider_instance=instance_id,
487            )
488        },
489        duration=sonic_episode.duration or 0,
490    )
491
492    if sonic_episode.publish_date:
493        episode.metadata.release_date = datetime.fromisoformat(sonic_episode.publish_date)
494
495    if sonic_episode.description:
496        episode.metadata.description = sonic_episode.description
497
498    if sonic_episode.cover_art:
499        episode.metadata.add_image(
500            MediaItemImage(
501                type=ImageType.THUMB,
502                path=sonic_episode.cover_art,
503                provider=instance_id,
504                remotely_accessible=False,
505            )
506        )
507    elif sonic_channel.cover_art:
508        episode.metadata.add_image(
509            MediaItemImage(
510                type=ImageType.THUMB,
511                path=sonic_channel.cover_art,
512                provider=instance_id,
513                remotely_accessible=False,
514            )
515        )
516
517    return episode
518