/
/
/
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
31from music_assistant.helpers.compare import compare_album_name, 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 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# TheAudioDB field suffix -> ISO 639-1 language code. CN/JP/SE/NO/IL use country-style
87# codes that don't match the ISO language code, so the mapping is explicit.
88TADB_SUFFIX_TO_ISO: dict[str, str] = {
89 "DE": "de",
90 "FR": "fr",
91 "IT": "it",
92 "ES": "es",
93 "PT": "pt",
94 "NL": "nl",
95 "RU": "ru",
96 "PL": "pl",
97 "HU": "hu",
98 "CN": "zh",
99 "JP": "ja",
100 "SE": "sv",
101 "NO": "nb",
102 "IL": "he",
103}
104
105
106async def setup(
107 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
108) -> ProviderInstanceType:
109 """Initialize provider(instance) with given configuration."""
110 return AudioDbMetadataProvider(mass, manifest, config, SUPPORTED_FEATURES)
111
112
113class AudioDbMetadataProvider(MetadataProvider):
114 """The AudioDB Metadata provider."""
115
116 throttler: Throttler
117
118 @property
119 def priority(self) -> int:
120 """Priority for this provider (lower = more preferred)."""
121 return 20
122
123 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
124 """Return Config entries to configure this provider."""
125 return (
126 ConfigEntry(
127 key=CONF_ENABLE_ARTIST_METADATA,
128 type=ConfigEntryType.BOOLEAN,
129 default_value=True,
130 ),
131 ConfigEntry(
132 key=CONF_ENABLE_ALBUM_METADATA,
133 type=ConfigEntryType.BOOLEAN,
134 default_value=True,
135 ),
136 ConfigEntry(
137 key=CONF_ENABLE_TRACK_METADATA,
138 type=ConfigEntryType.BOOLEAN,
139 default_value=False,
140 ),
141 ConfigEntry(
142 key=CONF_ENABLE_IMAGES,
143 type=ConfigEntryType.BOOLEAN,
144 default_value=True,
145 ),
146 )
147
148 async def handle_async_init(self) -> None:
149 """Handle async initialization of the provider."""
150 self.cache = self.mass.cache
151 self.throttler = Throttler(rate_limit=1, period=1)
152
153 async def get_artist_metadata(self, artist: Artist) -> MediaItemMetadata | None:
154 """Retrieve metadata for artist on theaudiodb."""
155 if not self.config.get_value(CONF_ENABLE_ARTIST_METADATA):
156 return None
157 if not artist.mbid:
158 # for 100% accuracy we require the musicbrainz id for all lookups
159 return None
160 self.logger.debug("Fetching metadata for Artist %s on The Audio DB", artist.name)
161 if data := await self._get_data("artist-mb.php", i=artist.mbid):
162 if data.get("artists"):
163 metadata = self.__parse_artist(data["artists"][0])
164 if metadata.description:
165 self.logger.debug(
166 "Found bio for %s on TheAudioDB in %s",
167 artist.name,
168 metadata.description_language or "unknown",
169 )
170 return metadata
171 return None
172
173 async def get_album_metadata(self, album: Album) -> MediaItemMetadata | None:
174 """Retrieve metadata for album on theaudiodb."""
175 if not self.config.get_value(CONF_ENABLE_ALBUM_METADATA):
176 return None
177 self.logger.debug("Fetching metadata for Album %s on The Audio DB", album.name)
178 if mbid := album.get_external_id(ExternalID.MB_RELEASEGROUP):
179 result = await self._get_data("album-mb.php", i=mbid)
180 if result and result.get("album"):
181 adb_album = result["album"][0]
182 return await self.__parse_album(album, adb_album)
183 # if there was no match on mbid, there will certainly be no match by name
184 return None
185 # fallback if no musicbrainzid: lookup by name
186 for album_artist in album.artists:
187 # make sure to include the version in the album name
188 album_name = f"{album.name} {album.version}" if album.version else album.name
189 result = await self._get_data("searchalbum.php?", s=album_artist.name, a=album_name)
190 if result and result.get("album"):
191 for item in result["album"]:
192 # some safety checks
193 if album_artist.mbid:
194 if album_artist.mbid != item["strMusicBrainzArtistID"]:
195 continue
196 elif not compare_strings(album_artist.name, item["strArtist"]):
197 continue
198 if compare_strings(album_name, item["strAlbum"], strict=False):
199 # match found !
200 return await self.__parse_album(album, item)
201 return None
202
203 async def get_track_metadata(self, track: Track) -> MediaItemMetadata | None:
204 """Retrieve metadata for track on theaudiodb."""
205 if not self.config.get_value(CONF_ENABLE_TRACK_METADATA):
206 return None
207 if track.mbid:
208 result = await self._get_data("track-mb.php", i=track.mbid)
209 if result and result.get("track"):
210 return await self.__parse_track(track, result["track"][0])
211 # if there was no match on mbid, there will certainly be no match by name
212 return None
213 # fallback if no musicbrainzid: lookup by name
214 for track_artist in track.artists:
215 # make sure to include the version in the album name
216 track_name = f"{track.name} {track.version}" if track.version else track.name
217 result = await self._get_data("searchtrack.php?", s=track_artist.name, t=track_name)
218 if result and result.get("track"):
219 for item in result["track"]:
220 # some safety checks
221 if track_artist.mbid:
222 if track_artist.mbid != item["strMusicBrainzArtistID"]:
223 continue
224 elif not compare_strings(track_artist.name, item["strArtist"]):
225 continue
226 if (
227 track.album
228 and (mb_rgid := track.album.get_external_id(ExternalID.MB_RELEASEGROUP))
229 # AudioDb swapped MB Album ID and ReleaseGroup ID ?!
230 and mb_rgid != item["strMusicBrainzAlbumID"]
231 ):
232 continue
233 if track.album and not compare_strings(
234 track.album.name, item["strAlbum"], strict=False
235 ):
236 continue
237 if not compare_strings(track_name, item["strTrack"], strict=False):
238 continue
239 return await self.__parse_track(track, item)
240 return None
241
242 def __parse_artist(self, artist_obj: dict[str, Any]) -> MediaItemMetadata:
243 """Parse audiodb artist object to MediaItemMetadata."""
244 metadata = MediaItemMetadata()
245 # generic data
246 metadata.label = artist_obj.get("strLabel")
247 metadata.style = artist_obj.get("strStyle")
248 if genre := artist_obj.get("strGenre"):
249 metadata.genres = {genre}
250 metadata.mood = artist_obj.get("strMood")
251 # links
252 metadata.links = set()
253 for key, link_type in LINK_MAPPING.items():
254 if link := artist_obj.get(key):
255 metadata.links.add(MediaItemLink(type=link_type, url=link))
256 # description/biography
257 metadata.description, metadata.description_language = self._localized_field(
258 artist_obj, "strBiography"
259 )
260 # images
261 if not self.config.get_value(CONF_ENABLE_IMAGES):
262 return metadata
263 metadata.images = UniqueList()
264 for key, img_type in IMG_MAPPING.items():
265 for postfix in ("", "2", "3", "4", "5", "6", "7", "8", "9", "10"):
266 if img := artist_obj.get(f"{key}{postfix}"):
267 metadata.images.append(
268 MediaItemImage(
269 type=img_type,
270 path=img,
271 provider=self.instance_id,
272 remotely_accessible=True,
273 )
274 )
275 else:
276 break
277 return metadata
278
279 async def __parse_album(self, album: Album, adb_album: dict[str, Any]) -> MediaItemMetadata:
280 """Parse audiodb album object to MediaItemMetadata."""
281 metadata = MediaItemMetadata()
282 # generic data
283 metadata.label = adb_album.get("strLabel")
284 metadata.style = adb_album.get("strStyle")
285 if genre := adb_album.get("strGenre"):
286 metadata.genres = {genre}
287 metadata.mood = adb_album.get("strMood")
288 # links
289 metadata.links = set()
290 if link := adb_album.get("strWikipediaID"):
291 metadata.links.add(
292 MediaItemLink(type=LinkType.WIKIPEDIA, url=f"https://wikipedia.org/wiki/{link}")
293 )
294 if link := adb_album.get("strAllMusicID"):
295 metadata.links.add(
296 MediaItemLink(type=LinkType.ALLMUSIC, url=f"https://www.allmusic.com/album/{link}")
297 )
298
299 # description
300 metadata.description, metadata.description_language = self._localized_field(
301 adb_album, "strDescription"
302 )
303 metadata.review = adb_album.get("strReview")
304 # fill in some missing album info if needed
305 if not album.year:
306 album.year = int(adb_album.get("intYearReleased", "0"))
307 if album.album_type == AlbumType.UNKNOWN and adb_album.get("strReleaseFormat"):
308 releaseformat = cast("str", adb_album.get("strReleaseFormat"))
309 album.album_type = ALBUMTYPE_MAPPING.get(releaseformat, AlbumType.UNKNOWN)
310 # update the artist mbid while at it
311 for album_artist in album.artists:
312 if not compare_strings(album_artist.name, adb_album["strArtist"]):
313 continue
314 if not album_artist.mbid and album_artist.provider == "library":
315 if isinstance(album_artist, ItemMapping):
316 album_artist = self.mass.music.artists.artist_from_item_mapping(album_artist) # noqa: PLW2901
317 album_artist.mbid = adb_album["strMusicBrainzArtistID"]
318 await self.mass.music.artists.update_item_in_library(
319 album_artist.item_id,
320 album_artist,
321 )
322 # images
323 if not self.config.get_value(CONF_ENABLE_IMAGES):
324 return metadata
325 metadata.images = UniqueList()
326 for key, img_type in IMG_MAPPING.items():
327 for postfix in ("", "2", "3", "4", "5", "6", "7", "8", "9", "10"):
328 if img := adb_album.get(f"{key}{postfix}"):
329 metadata.images.append(
330 MediaItemImage(
331 type=img_type,
332 path=img,
333 provider=self.instance_id,
334 remotely_accessible=True,
335 )
336 )
337 else:
338 break
339 return metadata
340
341 async def __parse_track(self, track: Track, adb_track: dict[str, Any]) -> MediaItemMetadata:
342 """Parse audiodb track object to MediaItemMetadata."""
343 metadata = MediaItemMetadata()
344 # generic data
345 metadata.lyrics = adb_track.get("strTrackLyrics")
346 metadata.style = adb_track.get("strStyle")
347 if genre := adb_track.get("strGenre"):
348 metadata.genres = {genre}
349 metadata.mood = adb_track.get("strMood")
350 # description
351 metadata.description, metadata.description_language = self._localized_field(
352 adb_track, "strDescription"
353 )
354 # update the artist mbid while at it
355 for album_artist in track.artists:
356 if not compare_strings(album_artist.name, adb_track["strArtist"]):
357 continue
358 if not album_artist.mbid and album_artist.provider == "library":
359 if isinstance(album_artist, ItemMapping):
360 album_artist = self.mass.music.artists.artist_from_item_mapping(album_artist) # noqa: PLW2901
361 album_artist.mbid = adb_track["strMusicBrainzArtistID"]
362 await self.mass.music.artists.update_item_in_library(
363 album_artist.item_id,
364 album_artist,
365 )
366 # update the album mbid while at it
367 if (
368 track.album
369 and track.album.provider == "library"
370 and not track.album.get_external_id(ExternalID.MB_RELEASEGROUP)
371 # a recording is shared by every release it appears on, so the album id is
372 # only ours to take when the matched record is for this album as well
373 and compare_album_name(track.album.name, adb_track["strAlbum"])
374 ):
375 await self.mass.music.albums.set_release_group(
376 int(track.album.item_id), adb_track["strMusicBrainzAlbumID"]
377 )
378 # images
379 if not self.config.get_value(CONF_ENABLE_IMAGES):
380 return metadata
381 metadata.images = UniqueList([])
382 for key, img_type in IMG_MAPPING.items():
383 for postfix in ("", "2", "3", "4", "5", "6", "7", "8", "9", "10"):
384 if img := adb_track.get(f"{key}{postfix}"):
385 metadata.images.append(
386 MediaItemImage(
387 type=img_type,
388 path=img,
389 provider=self.instance_id,
390 remotely_accessible=True,
391 )
392 )
393 else:
394 break
395 return metadata
396
397 def _localized_field(self, obj: dict[str, Any], prefix: str) -> tuple[str | None, str | None]:
398 """
399 Return the best-matching localized text for ``prefix`` and its ISO 639-1 language.
400
401 :param obj: TheAudioDB response object to read fields from.
402 :param prefix: Field name prefix (e.g. ``"strBiography"`` or ``"strDescription"``).
403 """
404 # region-first covers TheAudioDB's CN/JP/SE/NO/IL country-style suffixes, then the
405 # language code, and finally the suffix-less field (TheAudioDB's English default)
406 parts = self.mass.metadata.locale.split("_", 1)
407 lang_code = parts[0].upper()
408 region_code = parts[1].upper() if len(parts) > 1 else ""
409 for suffix in (region_code, lang_code):
410 if not suffix:
411 continue
412 if value := obj.get(f"{prefix}{suffix}"):
413 return value, TADB_SUFFIX_TO_ISO.get(suffix)
414 # bare field is the English default
415 if value := obj.get(prefix):
416 return value, "en"
417 return None, None
418
419 # None here only signals a failed request (a miss still returns a body), so don't cache it
420 @use_cache(86400 * 90, persistent=True, cache_none=False) # Cache for 90 days
421 async def _get_data(self, endpoint: str, **kwargs: Any) -> dict[str, Any] | None:
422 """Get data from api."""
423 url = f"https://theaudiodb.com/api/v1/json/{app_var('theaudiodb_api_key')}/{endpoint}"
424 async with (
425 self.throttler,
426 self.mass.http_session.get(url, params=kwargs, ssl=False) as response,
427 ):
428 try:
429 result = cast("dict[str, Any]", await response.json())
430 except (
431 aiohttp.client_exceptions.ContentTypeError,
432 JSONDecodeError,
433 ):
434 self.logger.error("Failed to retrieve %s", endpoint)
435 text_result = await response.text()
436 self.logger.debug(text_result)
437 return None
438 except (
439 aiohttp.client_exceptions.ClientConnectorError,
440 aiohttp.client_exceptions.ServerDisconnectedError,
441 TimeoutError,
442 ):
443 self.logger.warning("Failed to retrieve %s", endpoint)
444 return None
445 if "error" in result and "limit" in result["error"]:
446 self.logger.warning(result["error"])
447 return None
448 return result
449