/
/
/
1"""Fanart.tv 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 ConfigEntryType, ExternalID, ImageType, ProviderFeature
11from music_assistant_models.media_items import MediaItemImage, MediaItemMetadata, UniqueList
12
13from music_assistant.controllers.cache import use_cache
14from music_assistant.helpers.app_vars import app_var
15from music_assistant.helpers.throttle_retry import Throttler
16from music_assistant.models.metadata_provider import MetadataProvider
17
18if TYPE_CHECKING:
19 from music_assistant_models.config_entries import ProviderConfig
20 from music_assistant_models.media_items import Album, Artist
21 from music_assistant_models.provider import ProviderManifest
22
23 from music_assistant.mass import MusicAssistant
24 from music_assistant.models import ProviderInstanceType
25
26SUPPORTED_FEATURES = {
27 ProviderFeature.ARTIST_METADATA,
28 ProviderFeature.ALBUM_METADATA,
29}
30
31CONF_ENABLE_ARTIST_IMAGES = "enable_artist_images"
32CONF_ENABLE_ALBUM_IMAGES = "enable_album_images"
33CONF_CLIENT_KEY = "client_key"
34
35ARTIST_IMG_MAPPING = {
36 "artistthumb": ImageType.THUMB,
37 "hdmusiclogo": ImageType.LOGO,
38 "musicbanner": ImageType.BANNER,
39 "artistbackground": ImageType.FANART,
40}
41
42ALBUM_IMG_MAPPING = {
43 "albumcover": ImageType.THUMB,
44 "cdart": ImageType.DISCART,
45}
46
47
48async def setup(
49 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
50) -> ProviderInstanceType:
51 """Initialize provider(instance) with given configuration."""
52 return FanartTvMetadataProvider(mass, manifest, config, SUPPORTED_FEATURES)
53
54
55class FanartTvMetadataProvider(MetadataProvider):
56 """Fanart.tv Metadata provider."""
57
58 throttler: Throttler
59
60 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
61 """Return Config entries to setup this provider."""
62 return (
63 ConfigEntry(
64 key=CONF_ENABLE_ARTIST_IMAGES,
65 type=ConfigEntryType.BOOLEAN,
66 default_value=True,
67 ),
68 ConfigEntry(
69 key=CONF_ENABLE_ALBUM_IMAGES,
70 type=ConfigEntryType.BOOLEAN,
71 default_value=True,
72 ),
73 ConfigEntry(
74 key=CONF_CLIENT_KEY,
75 type=ConfigEntryType.SECURE_STRING,
76 required=False,
77 ),
78 )
79
80 @property
81 def priority(self) -> int:
82 """Priority for this provider (lower = more preferred)."""
83 return 10
84
85 async def handle_async_init(self) -> None:
86 """Handle async initialization of the provider."""
87 self.cache = self.mass.cache
88 if self.config.get_value(CONF_CLIENT_KEY):
89 # loosen the throttler when a personal client key is used
90 self.throttler = Throttler(rate_limit=1, period=1)
91 else:
92 self.throttler = Throttler(rate_limit=1, period=30)
93
94 async def get_artist_metadata(self, artist: Artist) -> MediaItemMetadata | None:
95 """Retrieve metadata for artist on fanart.tv."""
96 if not artist.mbid:
97 return None
98 if not self.config.get_value(CONF_ENABLE_ARTIST_IMAGES):
99 return None
100 self.logger.debug("Fetching metadata for Artist %s on Fanart.tv", artist.name)
101 if data := await self._get_data(f"music/{artist.mbid}"):
102 metadata = MediaItemMetadata()
103 for key, img_type in ARTIST_IMG_MAPPING.items():
104 items = data.get(key)
105 if not items:
106 continue
107 for item in items:
108 metadata.add_image(
109 MediaItemImage(
110 type=img_type,
111 path=item["url"],
112 provider=self.domain,
113 remotely_accessible=True,
114 )
115 )
116 if metadata.images:
117 self.logger.debug(
118 "Found %d image(s) for Artist %s on Fanart.tv",
119 len(metadata.images),
120 artist.name,
121 )
122 else:
123 self.logger.debug(
124 "No images found for Artist %s on Fanart.tv (available keys: %s)",
125 artist.name,
126 list(data.keys()),
127 )
128 return metadata
129 return None
130
131 async def get_album_metadata(self, album: Album) -> MediaItemMetadata | None:
132 """Retrieve metadata for album on fanart.tv."""
133 if (mbid := album.get_external_id(ExternalID.MB_RELEASEGROUP)) is None:
134 return None
135 if not self.config.get_value(CONF_ENABLE_ALBUM_IMAGES):
136 return None
137 self.logger.debug("Fetching metadata for Album %s on Fanart.tv", album.name)
138 if data := await self._get_data(f"music/albums/{mbid}"):
139 if data and data.get("albums"):
140 if album_data := data["albums"].get(mbid):
141 metadata = MediaItemMetadata()
142 metadata.images = UniqueList()
143 for key, img_type in ALBUM_IMG_MAPPING.items():
144 items = album_data.get(key)
145 if not items:
146 continue
147 for item in items:
148 metadata.images.append(
149 MediaItemImage(
150 type=img_type,
151 path=item["url"],
152 provider=self.domain,
153 remotely_accessible=True,
154 )
155 )
156 if metadata.images:
157 self.logger.debug(
158 "Found %d image(s) for Album %s on Fanart.tv",
159 len(metadata.images),
160 album.name,
161 )
162 else:
163 self.logger.debug(
164 "No images found for Album %s on Fanart.tv (available keys: %s)",
165 album.name,
166 list(album_data.keys()),
167 )
168 return metadata
169 return None
170
171 # None here only signals a failed or rate-limited request, so don't cache it
172 @use_cache(86400 * 60, cache_none=False) # Cache for 60 days
173 async def _get_data(self, endpoint: str, **kwargs: str) -> dict[str, Any] | None:
174 """Get data from api."""
175 url = f"http://webservice.fanart.tv/v3/{endpoint}"
176 headers = {
177 "api-key": app_var("fanarttv_api_key"),
178 }
179 if client_key := self.config.get_value(CONF_CLIENT_KEY):
180 headers["client_key"] = str(client_key)
181 async with (
182 self.throttler,
183 self.mass.http_session_no_ssl.get(
184 url, params=kwargs, headers=headers, ssl=False
185 ) as response,
186 ):
187 try:
188 result = await response.json()
189 except (
190 aiohttp.client_exceptions.ContentTypeError,
191 JSONDecodeError,
192 ):
193 self.logger.error("Failed to retrieve %s", endpoint)
194 text_result = await response.text()
195 self.logger.debug(text_result)
196 return None
197 except (
198 aiohttp.client_exceptions.ClientConnectorError,
199 aiohttp.client_exceptions.ServerDisconnectedError,
200 ):
201 self.logger.warning("Failed to retrieve %s", endpoint)
202 return None
203 if "error" in result and "limit" in result["error"]:
204 self.logger.warning(result["error"])
205 return None
206 return cast("dict[str, Any]", result)
207