music-assistant-server
17.2 KB•PY
__init__.py
17.2 KB • 421 lines • python
1"""The AudioDB Metadata provider for Music Assistant."""
2
3from __future__ import annotations
4
5from json import JSONDecodeError
6from typing import TYPE_CHECKING, Any, cast
7
8import aiohttp.client_exceptions
9from music_assistant_models.config_entries import ConfigEntry
10from music_assistant_models.enums import (
11 AlbumType,
12 ConfigEntryType,
13 ExternalID,
14 ImageType,
15 LinkType,
16 ProviderFeature,
17)
18from music_assistant_models.media_items import (
19 Album,
20 Artist,
21 ItemMapping,
22 MediaItemImage,
23 MediaItemLink,
24 MediaItemMetadata,
25 Track,
26 UniqueList,
27)
28
29from music_assistant.controllers.cache import use_cache
30from music_assistant.helpers.app_vars import app_var # type: ignore[attr-defined]
31from music_assistant.helpers.compare import compare_strings
32from music_assistant.helpers.throttle_retry import Throttler
33from music_assistant.models.metadata_provider import MetadataProvider
34
35if TYPE_CHECKING:
36 from music_assistant_models.config_entries import ConfigValueType, ProviderConfig
37 from music_assistant_models.provider import ProviderManifest
38
39 from music_assistant.mass import MusicAssistant
40 from music_assistant.models import ProviderInstanceType
41
42SUPPORTED_FEATURES = {
43 ProviderFeature.ARTIST_METADATA,
44 ProviderFeature.ALBUM_METADATA,
45 ProviderFeature.TRACK_METADATA,
46}
47
48IMG_MAPPING = {
49 "strArtistThumb": ImageType.THUMB,
50 "strArtistLogo": ImageType.LOGO,
51 "strArtistCutout": ImageType.CUTOUT,
52 "strArtistClearart": ImageType.CLEARART,
53 "strArtistWideThumb": ImageType.LANDSCAPE,
54 "strArtistFanart": ImageType.FANART,
55 "strArtistBanner": ImageType.BANNER,
56 "strAlbumThumb": ImageType.THUMB,
57 "strAlbumThumbHQ": ImageType.THUMB,
58 "strAlbumCDart": ImageType.DISCART,
59 "strAlbum3DCase": ImageType.OTHER,
60 "strAlbum3DFlat": ImageType.OTHER,
61 "strAlbum3DFace": ImageType.OTHER,
62 "strAlbum3DThumb": ImageType.OTHER,
63 "strTrackThumb": ImageType.THUMB,
64 "strTrack3DCase": ImageType.OTHER,
65}
66
67LINK_MAPPING = {
68 "strWebsite": LinkType.WEBSITE,
69 "strFacebook": LinkType.FACEBOOK,
70 "strTwitter": LinkType.TWITTER,
71 "strLastFMChart": LinkType.LASTFM,
72}
73
74ALBUMTYPE_MAPPING = {
75 "Single": AlbumType.SINGLE,
76 "Compilation": AlbumType.COMPILATION,
77 "Album": AlbumType.ALBUM,
78 "EP": AlbumType.EP,
79}
80
81CONF_ENABLE_IMAGES = "enable_images"
82CONF_ENABLE_ARTIST_METADATA = "enable_artist_metadata"
83CONF_ENABLE_ALBUM_METADATA = "enable_album_metadata"
84CONF_ENABLE_TRACK_METADATA = "enable_track_metadata"
85
86
87async def setup(
88 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
89) -> ProviderInstanceType:
90 """Initialize provider(instance) with given configuration."""
91 return AudioDbMetadataProvider(mass, manifest, config, SUPPORTED_FEATURES)
92
93
94async def get_config_entries(
95 mass: MusicAssistant,
96 instance_id: str | None = None,
97 action: str | None = None,
98 values: dict[str, ConfigValueType] | None = None,
99) -> tuple[ConfigEntry, ...]:
100 """
101 Return Config entries to setup this provider.
102
103 instance_id: id of an existing provider instance (None if new instance setup).
104 action: [optional] action key called from config entries UI.
105 values: the (intermediate) raw values for config entries sent with the action.
106 """
107 # ruff: noqa: ARG001
108 return (
109 ConfigEntry(
110 key=CONF_ENABLE_ARTIST_METADATA,
111 type=ConfigEntryType.BOOLEAN,
112 label="Enable retrieval of artist metadata.",
113 default_value=True,
114 ),
115 ConfigEntry(
116 key=CONF_ENABLE_ALBUM_METADATA,
117 type=ConfigEntryType.BOOLEAN,
118 label="Enable retrieval of album metadata.",
119 default_value=True,
120 ),
121 ConfigEntry(
122 key=CONF_ENABLE_TRACK_METADATA,
123 type=ConfigEntryType.BOOLEAN,
124 label="Enable retrieval of track metadata.",
125 default_value=False,
126 ),
127 ConfigEntry(
128 key=CONF_ENABLE_IMAGES,
129 type=ConfigEntryType.BOOLEAN,
130 label="Enable retrieval of artist/album/track images",
131 default_value=True,
132 ),
133 )
134
135
136class AudioDbMetadataProvider(MetadataProvider):
137 """The AudioDB Metadata provider."""
138
139 throttler: Throttler
140
141 async def handle_async_init(self) -> None:
142 """Handle async initialization of the provider."""
143 self.cache = self.mass.cache
144 self.throttler = Throttler(rate_limit=1, period=1)
145
146 async def get_artist_metadata(self, artist: Artist) -> MediaItemMetadata | None:
147 """Retrieve metadata for artist on theaudiodb."""
148 if not self.config.get_value(CONF_ENABLE_ARTIST_METADATA):
149 return None
150 if not artist.mbid:
151 # for 100% accuracy we require the musicbrainz id for all lookups
152 return None
153 if data := await self._get_data("artist-mb.php", i=artist.mbid):
154 if data.get("artists"):
155 return self.__parse_artist(data["artists"][0])
156 return None
157
158 async def get_album_metadata(self, album: Album) -> MediaItemMetadata | None:
159 """Retrieve metadata for album on theaudiodb."""
160 if not self.config.get_value(CONF_ENABLE_ALBUM_METADATA):
161 return None
162 if mbid := album.get_external_id(ExternalID.MB_RELEASEGROUP):
163 result = await self._get_data("album-mb.php", i=mbid)
164 if result and result.get("album"):
165 adb_album = result["album"][0]
166 return await self.__parse_album(album, adb_album)
167 # if there was no match on mbid, there will certainly be no match by name
168 return None
169 # fallback if no musicbrainzid: lookup by name
170 for album_artist in album.artists:
171 # make sure to include the version in the album name
172 album_name = f"{album.name} {album.version}" if album.version else album.name
173 result = await self._get_data("searchalbum.php?", s=album_artist.name, a=album_name)
174 if result and result.get("album"):
175 for item in result["album"]:
176 # some safety checks
177 if album_artist.mbid:
178 if album_artist.mbid != item["strMusicBrainzArtistID"]:
179 continue
180 elif not compare_strings(album_artist.name, item["strArtist"]):
181 continue
182 if compare_strings(album_name, item["strAlbum"], strict=False):
183 # match found !
184 return await self.__parse_album(album, item)
185 return None
186
187 async def get_track_metadata(self, track: Track) -> MediaItemMetadata | None:
188 """Retrieve metadata for track on theaudiodb."""
189 if not self.config.get_value(CONF_ENABLE_TRACK_METADATA):
190 return None
191 if track.mbid:
192 result = await self._get_data("track-mb.php", i=track.mbid)
193 if result and result.get("track"):
194 return await self.__parse_track(track, result["track"][0])
195 # if there was no match on mbid, there will certainly be no match by name
196 return None
197 # fallback if no musicbrainzid: lookup by name
198 for track_artist in track.artists:
199 # make sure to include the version in the album name
200 track_name = f"{track.name} {track.version}" if track.version else track.name
201 result = await self._get_data("searchtrack.php?", s=track_artist.name, t=track_name)
202 if result and result.get("track"):
203 for item in result["track"]:
204 # some safety checks
205 if track_artist.mbid:
206 if track_artist.mbid != item["strMusicBrainzArtistID"]:
207 continue
208 elif not compare_strings(track_artist.name, item["strArtist"]):
209 continue
210 if (
211 track.album
212 and (mb_rgid := track.album.get_external_id(ExternalID.MB_RELEASEGROUP))
213 # AudioDb swapped MB Album ID and ReleaseGroup ID ?!
214 and mb_rgid != item["strMusicBrainzAlbumID"]
215 ):
216 continue
217 if track.album and not compare_strings(
218 track.album.name, item["strAlbum"], strict=False
219 ):
220 continue
221 if not compare_strings(track_name, item["strTrack"], strict=False):
222 continue
223 return await self.__parse_track(track, item)
224 return None
225
226 def __parse_artist(self, artist_obj: dict[str, Any]) -> MediaItemMetadata:
227 """Parse audiodb artist object to MediaItemMetadata."""
228 metadata = MediaItemMetadata()
229 # generic data
230 metadata.label = artist_obj.get("strLabel")
231 metadata.style = artist_obj.get("strStyle")
232 if genre := artist_obj.get("strGenre"):
233 metadata.genres = {genre}
234 metadata.mood = artist_obj.get("strMood")
235 # links
236 metadata.links = set()
237 for key, link_type in LINK_MAPPING.items():
238 if link := artist_obj.get(key):
239 metadata.links.add(MediaItemLink(type=link_type, url=link))
240 # description/biography
241 lang_code, lang_country = self.mass.metadata.locale.split("_")
242 if desc := artist_obj.get(f"strBiography{lang_country}") or (
243 desc := artist_obj.get(f"strBiography{lang_code.upper()}")
244 ):
245 metadata.description = desc
246 else:
247 metadata.description = artist_obj.get("strBiographyEN")
248 # images
249 if not self.config.get_value(CONF_ENABLE_IMAGES):
250 return metadata
251 metadata.images = UniqueList()
252 for key, img_type in IMG_MAPPING.items():
253 for postfix in ("", "2", "3", "4", "5", "6", "7", "8", "9", "10"):
254 if img := artist_obj.get(f"{key}{postfix}"):
255 metadata.images.append(
256 MediaItemImage(
257 type=img_type,
258 path=img,
259 provider=self.instance_id,
260 remotely_accessible=True,
261 )
262 )
263 else:
264 break
265 return metadata
266
267 async def __parse_album(self, album: Album, adb_album: dict[str, Any]) -> MediaItemMetadata:
268 """Parse audiodb album object to MediaItemMetadata."""
269 metadata = MediaItemMetadata()
270 # generic data
271 metadata.label = adb_album.get("strLabel")
272 metadata.style = adb_album.get("strStyle")
273 if genre := adb_album.get("strGenre"):
274 metadata.genres = {genre}
275 metadata.mood = adb_album.get("strMood")
276 # links
277 metadata.links = set()
278 if link := adb_album.get("strWikipediaID"):
279 metadata.links.add(
280 MediaItemLink(type=LinkType.WIKIPEDIA, url=f"https://wikipedia.org/wiki/{link}")
281 )
282 if link := adb_album.get("strAllMusicID"):
283 metadata.links.add(
284 MediaItemLink(type=LinkType.ALLMUSIC, url=f"https://www.allmusic.com/album/{link}")
285 )
286
287 # description
288 lang_code, lang_country = self.mass.metadata.locale.split("_")
289 if desc := adb_album.get(f"strDescription{lang_country}") or (
290 desc := adb_album.get(f"strDescription{lang_code.upper()}")
291 ):
292 metadata.description = desc
293 else:
294 metadata.description = adb_album.get("strDescriptionEN")
295 metadata.review = adb_album.get("strReview")
296 # images
297 if not self.config.get_value(CONF_ENABLE_IMAGES):
298 return metadata
299 metadata.images = UniqueList()
300 for key, img_type in IMG_MAPPING.items():
301 for postfix in ("", "2", "3", "4", "5", "6", "7", "8", "9", "10"):
302 if img := adb_album.get(f"{key}{postfix}"):
303 metadata.images.append(
304 MediaItemImage(
305 type=img_type,
306 path=img,
307 provider=self.instance_id,
308 remotely_accessible=True,
309 )
310 )
311 else:
312 break
313 # fill in some missing album info if needed
314 if not album.year:
315 album.year = int(adb_album.get("intYearReleased", "0"))
316 if album.album_type == AlbumType.UNKNOWN and adb_album.get("strReleaseFormat"):
317 releaseformat = cast("str", adb_album.get("strReleaseFormat"))
318 album.album_type = ALBUMTYPE_MAPPING.get(releaseformat, AlbumType.UNKNOWN)
319 # update the artist mbid while at it
320 for album_artist in album.artists:
321 if not compare_strings(album_artist.name, adb_album["strArtist"]):
322 continue
323 if not album_artist.mbid and album_artist.provider == "library":
324 if isinstance(album_artist, ItemMapping):
325 album_artist = self.mass.music.artists.artist_from_item_mapping(album_artist) # noqa: PLW2901
326 album_artist.mbid = adb_album["strMusicBrainzArtistID"]
327 await self.mass.music.artists.update_item_in_library(
328 album_artist.item_id,
329 album_artist,
330 )
331 return metadata
332
333 async def __parse_track(self, track: Track, adb_track: dict[str, Any]) -> MediaItemMetadata:
334 """Parse audiodb track object to MediaItemMetadata."""
335 metadata = MediaItemMetadata()
336 # generic data
337 metadata.lyrics = adb_track.get("strTrackLyrics")
338 metadata.style = adb_track.get("strStyle")
339 if genre := adb_track.get("strGenre"):
340 metadata.genres = {genre}
341 metadata.mood = adb_track.get("strMood")
342 # description
343 lang_code, lang_country = self.mass.metadata.locale.split("_")
344 if desc := adb_track.get(f"strDescription{lang_country}") or (
345 desc := adb_track.get(f"strDescription{lang_code.upper()}")
346 ):
347 metadata.description = desc
348 else:
349 metadata.description = adb_track.get("strDescriptionEN")
350 # images
351 if not self.config.get_value(CONF_ENABLE_IMAGES):
352 return metadata
353 metadata.images = UniqueList([])
354 for key, img_type in IMG_MAPPING.items():
355 for postfix in ("", "2", "3", "4", "5", "6", "7", "8", "9", "10"):
356 if img := adb_track.get(f"{key}{postfix}"):
357 metadata.images.append(
358 MediaItemImage(
359 type=img_type,
360 path=img,
361 provider=self.instance_id,
362 remotely_accessible=True,
363 )
364 )
365 else:
366 break
367 # update the artist mbid while at it
368 for album_artist in track.artists:
369 if not compare_strings(album_artist.name, adb_track["strArtist"]):
370 continue
371 if not album_artist.mbid and album_artist.provider == "library":
372 if isinstance(album_artist, ItemMapping):
373 album_artist = self.mass.music.artists.artist_from_item_mapping(album_artist) # noqa: PLW2901
374 album_artist.mbid = adb_track["strMusicBrainzArtistID"]
375 await self.mass.music.artists.update_item_in_library(
376 album_artist.item_id,
377 album_artist,
378 )
379 # update the album mbid while at it
380 if (
381 track.album
382 and not track.album.get_external_id(ExternalID.MB_RELEASEGROUP)
383 and track.album.provider == "library"
384 and isinstance(track.album, Album)
385 ):
386 track.album.add_external_id(
387 ExternalID.MB_RELEASEGROUP, adb_track["strMusicBrainzAlbumID"]
388 )
389 await self.mass.music.albums.update_item_in_library(track.album.item_id, track.album)
390 return metadata
391
392 @use_cache(86400 * 90, persistent=True) # Cache for 90 days
393 async def _get_data(self, endpoint: str, **kwargs: Any) -> dict[str, Any] | None:
394 """Get data from api."""
395 url = f"https://theaudiodb.com/api/v1/json/{app_var(3)}/{endpoint}"
396 async with (
397 self.throttler,
398 self.mass.http_session.get(url, params=kwargs, ssl=False) as response,
399 ):
400 try:
401 result = cast("dict[str, Any]", await response.json())
402 except (
403 aiohttp.client_exceptions.ContentTypeError,
404 JSONDecodeError,
405 ):
406 self.logger.error("Failed to retrieve %s", endpoint)
407 text_result = await response.text()
408 self.logger.debug(text_result)
409 return None
410 except (
411 aiohttp.client_exceptions.ClientConnectorError,
412 aiohttp.client_exceptions.ServerDisconnectedError,
413 TimeoutError,
414 ):
415 self.logger.warning("Failed to retrieve %s", endpoint)
416 return None
417 if "error" in result and "limit" in result["error"]:
418 self.logger.warning(result["error"])
419 return None
420 return result
421