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