music-assistant-server
11.6 KB•PY
parsers.py
11.6 KB • 325 lines • python
1"""Parse Jellyfin metadata into Music Assistant models."""
2
3from __future__ import annotations
4
5import logging
6from logging import Logger
7from typing import TYPE_CHECKING
8
9from aiojellyfin import ImageType as JellyImageType
10from music_assistant_models.enums import ContentType, ExternalID, ImageType, MediaType
11from music_assistant_models.errors import InvalidDataError
12from music_assistant_models.media_items import (
13 Album,
14 Artist,
15 AudioFormat,
16 ItemMapping,
17 MediaItemImage,
18 Playlist,
19 ProviderMapping,
20 Track,
21 UniqueList,
22)
23
24from music_assistant.helpers.util import parse_title_and_version
25
26from .const import (
27 DOMAIN,
28 ITEM_KEY_ALBUM,
29 ITEM_KEY_ALBUM_ARTIST,
30 ITEM_KEY_ALBUM_ARTISTS,
31 ITEM_KEY_ALBUM_ID,
32 ITEM_KEY_ARTIST_ITEMS,
33 ITEM_KEY_CAN_DOWNLOAD,
34 ITEM_KEY_ID,
35 ITEM_KEY_IMAGE_TAGS,
36 ITEM_KEY_MEDIA_CHANNELS,
37 ITEM_KEY_MEDIA_CODEC,
38 ITEM_KEY_MEDIA_STREAMS,
39 ITEM_KEY_MUSICBRAINZ_ALBUM,
40 ITEM_KEY_MUSICBRAINZ_ARTIST,
41 ITEM_KEY_MUSICBRAINZ_RELEASE_GROUP,
42 ITEM_KEY_MUSICBRAINZ_TRACK,
43 ITEM_KEY_NAME,
44 ITEM_KEY_OVERVIEW,
45 ITEM_KEY_PARENT_INDEX_NUM,
46 ITEM_KEY_PRODUCTION_YEAR,
47 ITEM_KEY_PROVIDER_IDS,
48 ITEM_KEY_RUNTIME_TICKS,
49 ITEM_KEY_SORT_NAME,
50 ITEM_KEY_USER_DATA,
51 MEDIA_IMAGE_TYPES,
52 UNKNOWN_ARTIST_MAPPING,
53 USER_DATA_KEY_IS_FAVORITE,
54)
55
56if TYPE_CHECKING:
57 from aiojellyfin import Album as JellyAlbum
58 from aiojellyfin import Artist as JellyArtist
59 from aiojellyfin import Connection
60 from aiojellyfin import MediaItem as JellyMediaItem
61 from aiojellyfin import Playlist as JellyPlaylist
62 from aiojellyfin import Track as JellyTrack
63
64
65def parse_album(
66 logger: Logger, instance_id: str, connection: Connection, jellyfin_album: JellyAlbum
67) -> Album:
68 """Parse a Jellyfin Album response to an Album model object."""
69 album_id = jellyfin_album[ITEM_KEY_ID]
70 name, version = parse_title_and_version(jellyfin_album[ITEM_KEY_NAME])
71 album = Album(
72 item_id=album_id,
73 provider=DOMAIN,
74 name=name,
75 version=version,
76 provider_mappings={
77 ProviderMapping(
78 item_id=str(album_id),
79 provider_domain=DOMAIN,
80 provider_instance=instance_id,
81 )
82 },
83 )
84 if ITEM_KEY_PRODUCTION_YEAR in jellyfin_album:
85 album.year = jellyfin_album[ITEM_KEY_PRODUCTION_YEAR]
86 album.metadata.images = _get_artwork(instance_id, connection, jellyfin_album)
87 if ITEM_KEY_OVERVIEW in jellyfin_album:
88 album.metadata.description = jellyfin_album[ITEM_KEY_OVERVIEW]
89 if ITEM_KEY_MUSICBRAINZ_ALBUM in jellyfin_album[ITEM_KEY_PROVIDER_IDS]:
90 try:
91 album.add_external_id(
92 ExternalID.MB_ALBUM,
93 jellyfin_album[ITEM_KEY_PROVIDER_IDS][ITEM_KEY_MUSICBRAINZ_ALBUM],
94 )
95 except InvalidDataError as error:
96 logger.warning(
97 "Jellyfin has an invalid musicbrainz album id for album %s",
98 album.name,
99 exc_info=error if logger.isEnabledFor(logging.DEBUG) else None,
100 )
101 if ITEM_KEY_MUSICBRAINZ_RELEASE_GROUP in jellyfin_album[ITEM_KEY_PROVIDER_IDS]:
102 try:
103 album.add_external_id(
104 ExternalID.MB_RELEASEGROUP,
105 jellyfin_album[ITEM_KEY_PROVIDER_IDS][ITEM_KEY_MUSICBRAINZ_RELEASE_GROUP],
106 )
107 except InvalidDataError as error:
108 logger.warning(
109 "Jellyfin has an invalid musicbrainz id for album %s",
110 album.name,
111 exc_info=error if logger.isEnabledFor(logging.DEBUG) else None,
112 )
113 if ITEM_KEY_SORT_NAME in jellyfin_album:
114 album.sort_name = jellyfin_album[ITEM_KEY_SORT_NAME]
115 if ITEM_KEY_ALBUM_ARTIST in jellyfin_album:
116 for album_artist in jellyfin_album[ITEM_KEY_ALBUM_ARTISTS]:
117 album.artists.append(
118 ItemMapping(
119 media_type=MediaType.ARTIST,
120 item_id=album_artist[ITEM_KEY_ID],
121 provider=instance_id,
122 name=album_artist[ITEM_KEY_NAME],
123 )
124 )
125 elif len(jellyfin_album.get(ITEM_KEY_ARTIST_ITEMS, [])) >= 1:
126 for artist_item in jellyfin_album[ITEM_KEY_ARTIST_ITEMS]:
127 album.artists.append(
128 ItemMapping(
129 media_type=MediaType.ARTIST,
130 item_id=artist_item[ITEM_KEY_ID],
131 provider=instance_id,
132 name=artist_item[ITEM_KEY_NAME],
133 )
134 )
135 else:
136 album.artists.append(UNKNOWN_ARTIST_MAPPING)
137
138 user_data = jellyfin_album.get(ITEM_KEY_USER_DATA, {})
139 album.favorite = user_data.get(USER_DATA_KEY_IS_FAVORITE, False)
140 return album
141
142
143def parse_artist(
144 logger: Logger, instance_id: str, connection: Connection, jellyfin_artist: JellyArtist
145) -> Artist:
146 """Parse a Jellyfin Artist response to Artist model object."""
147 artist_id = jellyfin_artist[ITEM_KEY_ID]
148 artist = Artist(
149 item_id=artist_id,
150 name=jellyfin_artist[ITEM_KEY_NAME],
151 provider=DOMAIN,
152 provider_mappings={
153 ProviderMapping(
154 item_id=str(artist_id),
155 provider_domain=DOMAIN,
156 provider_instance=instance_id,
157 )
158 },
159 )
160 if ITEM_KEY_OVERVIEW in jellyfin_artist:
161 artist.metadata.description = jellyfin_artist[ITEM_KEY_OVERVIEW]
162 if ITEM_KEY_MUSICBRAINZ_ARTIST in jellyfin_artist[ITEM_KEY_PROVIDER_IDS]:
163 try:
164 artist.mbid = jellyfin_artist[ITEM_KEY_PROVIDER_IDS][ITEM_KEY_MUSICBRAINZ_ARTIST]
165 except InvalidDataError as error:
166 logger.warning(
167 "Jellyfin has an invalid musicbrainz id for artist %s",
168 artist.name,
169 exc_info=error if logger.isEnabledFor(logging.DEBUG) else None,
170 )
171 if ITEM_KEY_SORT_NAME in jellyfin_artist:
172 artist.sort_name = jellyfin_artist[ITEM_KEY_SORT_NAME]
173 artist.metadata.images = _get_artwork(instance_id, connection, jellyfin_artist)
174 user_data = jellyfin_artist.get(ITEM_KEY_USER_DATA, {})
175 artist.favorite = user_data.get(USER_DATA_KEY_IS_FAVORITE, False)
176 return artist
177
178
179def audio_format(track: JellyTrack) -> AudioFormat:
180 """Build an AudioFormat model from a Jellyfin track."""
181 # Defensive: Handle missing or empty MediaStreams array
182 streams = track.get(ITEM_KEY_MEDIA_STREAMS, [])
183 if not streams:
184 return AudioFormat(content_type=ContentType.UNKNOWN)
185
186 stream = streams[0]
187 codec = stream.get(ITEM_KEY_MEDIA_CODEC)
188
189 return AudioFormat(
190 content_type=(ContentType.try_parse(codec) if codec else ContentType.UNKNOWN),
191 channels=stream.get(ITEM_KEY_MEDIA_CHANNELS, 2),
192 sample_rate=stream.get("SampleRate", 44100),
193 bit_rate=stream.get("BitRate"),
194 bit_depth=stream.get("BitDepth", 16),
195 )
196
197
198def parse_track(
199 logger: Logger, instance_id: str, client: Connection, jellyfin_track: JellyTrack
200) -> Track:
201 """Parse a Jellyfin Track response to a Track model object."""
202 available = jellyfin_track[ITEM_KEY_CAN_DOWNLOAD]
203 name, version = parse_title_and_version(jellyfin_track[ITEM_KEY_NAME])
204 track = Track(
205 item_id=jellyfin_track[ITEM_KEY_ID],
206 provider=instance_id,
207 name=name,
208 version=version,
209 provider_mappings={
210 ProviderMapping(
211 item_id=jellyfin_track[ITEM_KEY_ID],
212 provider_domain=DOMAIN,
213 provider_instance=instance_id,
214 available=available,
215 audio_format=audio_format(jellyfin_track),
216 url=client.audio_url(jellyfin_track[ITEM_KEY_ID]),
217 )
218 },
219 )
220
221 track.disc_number = jellyfin_track.get(ITEM_KEY_PARENT_INDEX_NUM, 0)
222 track.track_number = jellyfin_track.get("IndexNumber", 0)
223 if track.track_number is not None and track.track_number >= 0:
224 track.position = track.track_number
225
226 track.metadata.images = _get_artwork(instance_id, client, jellyfin_track)
227
228 if jellyfin_track[ITEM_KEY_ARTIST_ITEMS]:
229 for artist_item in jellyfin_track[ITEM_KEY_ARTIST_ITEMS]:
230 track.artists.append(
231 ItemMapping(
232 media_type=MediaType.ARTIST,
233 item_id=artist_item[ITEM_KEY_ID],
234 provider=instance_id,
235 name=artist_item[ITEM_KEY_NAME],
236 )
237 )
238 else:
239 track.artists.append(UNKNOWN_ARTIST_MAPPING)
240
241 if ITEM_KEY_ALBUM_ID in jellyfin_track:
242 if not (album_name := jellyfin_track.get(ITEM_KEY_ALBUM)):
243 logger.debug("Track %s has AlbumID but no AlbumName", track.name)
244 album_name = f"Unknown Album ({jellyfin_track[ITEM_KEY_ALBUM_ID]})"
245 track.album = ItemMapping(
246 media_type=MediaType.ALBUM,
247 item_id=jellyfin_track[ITEM_KEY_ALBUM_ID],
248 provider=instance_id,
249 name=album_name,
250 )
251
252 if ITEM_KEY_RUNTIME_TICKS in jellyfin_track:
253 track.duration = int(
254 jellyfin_track[ITEM_KEY_RUNTIME_TICKS] / 10000000
255 ) # 10000000 ticks per millisecond
256 if ITEM_KEY_MUSICBRAINZ_TRACK in jellyfin_track[ITEM_KEY_PROVIDER_IDS]:
257 track_mbid = jellyfin_track[ITEM_KEY_PROVIDER_IDS][ITEM_KEY_MUSICBRAINZ_TRACK]
258 try:
259 track.mbid = track_mbid
260 except InvalidDataError as error:
261 logger.warning(
262 "Jellyfin has an invalid musicbrainz id for track %s",
263 track.name,
264 exc_info=error if logger.isEnabledFor(logging.DEBUG) else None,
265 )
266 user_data = jellyfin_track.get(ITEM_KEY_USER_DATA, {})
267 track.favorite = user_data.get(USER_DATA_KEY_IS_FAVORITE, False)
268 return track
269
270
271def parse_playlist(
272 instance_id: str, client: Connection, jellyfin_playlist: JellyPlaylist
273) -> Playlist:
274 """Parse a Jellyfin Playlist response to a Playlist object."""
275 playlistid = jellyfin_playlist[ITEM_KEY_ID]
276 playlist = Playlist(
277 item_id=playlistid,
278 provider=DOMAIN,
279 name=jellyfin_playlist[ITEM_KEY_NAME],
280 provider_mappings={
281 ProviderMapping(
282 item_id=playlistid,
283 provider_domain=DOMAIN,
284 provider_instance=instance_id,
285 )
286 },
287 )
288 if ITEM_KEY_OVERVIEW in jellyfin_playlist:
289 playlist.metadata.description = jellyfin_playlist[ITEM_KEY_OVERVIEW]
290 playlist.metadata.images = _get_artwork(instance_id, client, jellyfin_playlist)
291 user_data = jellyfin_playlist.get(ITEM_KEY_USER_DATA, {})
292 playlist.favorite = user_data.get(USER_DATA_KEY_IS_FAVORITE, False)
293 playlist.is_editable = False
294 return playlist
295
296
297def _get_artwork(
298 instance_id: str, client: Connection, media_item: JellyMediaItem
299) -> UniqueList[MediaItemImage]:
300 images: UniqueList[MediaItemImage] = UniqueList()
301
302 for i, _ in enumerate(media_item.get("BackdropImageTags", [])):
303 images.append(
304 MediaItemImage(
305 type=ImageType.FANART,
306 path=client.artwork(media_item[ITEM_KEY_ID], JellyImageType.Backdrop, index=i),
307 provider=instance_id,
308 remotely_accessible=False,
309 )
310 )
311
312 image_tags = media_item[ITEM_KEY_IMAGE_TAGS]
313 for jelly_image_type, image_type in MEDIA_IMAGE_TYPES.items():
314 if jelly_image_type in image_tags:
315 images.append(
316 MediaItemImage(
317 type=image_type,
318 path=client.artwork(media_item[ITEM_KEY_ID], jelly_image_type),
319 provider=instance_id,
320 remotely_accessible=False,
321 )
322 )
323
324 return images
325