/
/
/
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 # type: ignore[attr-defined]
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 ConfigValueType, 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
35IMG_MAPPING = {
36 "artistthumb": ImageType.THUMB,
37 "hdmusiclogo": ImageType.LOGO,
38 "musicbanner": ImageType.BANNER,
39 "artistbackground": ImageType.FANART,
40}
41
42
43async def setup(
44 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
45) -> ProviderInstanceType:
46 """Initialize provider(instance) with given configuration."""
47 return FanartTvMetadataProvider(mass, manifest, config, SUPPORTED_FEATURES)
48
49
50async def get_config_entries(
51 mass: MusicAssistant,
52 instance_id: str | None = None,
53 action: str | None = None,
54 values: dict[str, ConfigValueType] | None = None,
55) -> tuple[ConfigEntry, ...]:
56 """
57 Return Config entries to setup this provider.
58
59 instance_id: id of an existing provider instance (None if new instance setup).
60 action: [optional] action key called from config entries UI.
61 values: the (intermediate) raw values for config entries sent with the action.
62 """
63 # ruff: noqa: ARG001
64 return (
65 ConfigEntry(
66 key=CONF_ENABLE_ARTIST_IMAGES,
67 type=ConfigEntryType.BOOLEAN,
68 label="Enable retrieval of artist images.",
69 default_value=True,
70 ),
71 ConfigEntry(
72 key=CONF_ENABLE_ALBUM_IMAGES,
73 type=ConfigEntryType.BOOLEAN,
74 label="Enable retrieval of album image(s).",
75 default_value=True,
76 ),
77 ConfigEntry(
78 key=CONF_CLIENT_KEY,
79 type=ConfigEntryType.SECURE_STRING,
80 label="VIP Member Personal API Key (optional)",
81 description="Support this metadata provider by becoming a VIP Member, "
82 "resulting in higher rate limits and faster response times among other benefits. "
83 "See https://wiki.fanart.tv/General/personal%20api/ for more information.",
84 required=False,
85 ),
86 )
87
88
89class FanartTvMetadataProvider(MetadataProvider):
90 """Fanart.tv Metadata provider."""
91
92 throttler: Throttler
93
94 async def handle_async_init(self) -> None:
95 """Handle async initialization of the provider."""
96 self.cache = self.mass.cache
97 if self.config.get_value(CONF_CLIENT_KEY):
98 # loosen the throttler when a personal client key is used
99 self.throttler = Throttler(rate_limit=1, period=1)
100 else:
101 self.throttler = Throttler(rate_limit=1, period=30)
102
103 async def get_artist_metadata(self, artist: Artist) -> MediaItemMetadata | None:
104 """Retrieve metadata for artist on fanart.tv."""
105 if not artist.mbid:
106 return None
107 if not self.config.get_value(CONF_ENABLE_ARTIST_IMAGES):
108 return None
109 self.logger.debug("Fetching metadata for Artist %s on Fanart.tv", artist.name)
110 if data := await self._get_data(f"music/{artist.mbid}"):
111 metadata = MediaItemMetadata()
112 for key, img_type in IMG_MAPPING.items():
113 items = data.get(key)
114 if not items:
115 continue
116 for item in items:
117 metadata.add_image(
118 MediaItemImage(
119 type=img_type,
120 path=item["url"],
121 provider=self.domain,
122 remotely_accessible=True,
123 )
124 )
125 return metadata
126 return None
127
128 async def get_album_metadata(self, album: Album) -> MediaItemMetadata | None:
129 """Retrieve metadata for album on fanart.tv."""
130 if (mbid := album.get_external_id(ExternalID.MB_RELEASEGROUP)) is None:
131 return None
132 if not self.config.get_value(CONF_ENABLE_ALBUM_IMAGES):
133 return None
134 self.logger.debug("Fetching metadata for Album %s on Fanart.tv", album.name)
135 if data := await self._get_data(f"music/albums/{mbid}"):
136 if data and data.get("albums"):
137 if album := data["albums"][mbid]:
138 metadata = MediaItemMetadata()
139 metadata.images = UniqueList()
140 for key, img_type in IMG_MAPPING.items():
141 items = album.get(key)
142 if not items:
143 continue
144 for item in items:
145 metadata.images.append(
146 MediaItemImage(
147 type=img_type,
148 path=item["url"],
149 provider=self.domain,
150 remotely_accessible=True,
151 )
152 )
153 return metadata
154 return None
155
156 @use_cache(86400 * 60) # Cache for 60 days
157 async def _get_data(self, endpoint: str, **kwargs: str) -> dict[str, Any] | None:
158 """Get data from api."""
159 url = f"http://webservice.fanart.tv/v3/{endpoint}"
160 headers = {
161 "api-key": app_var(4),
162 }
163 if client_key := self.config.get_value(CONF_CLIENT_KEY):
164 headers["client_key"] = client_key
165 async with (
166 self.throttler,
167 self.mass.http_session_no_ssl.get(
168 url, params=kwargs, headers=headers, ssl=False
169 ) as response,
170 ):
171 try:
172 result = await response.json()
173 except (
174 aiohttp.client_exceptions.ContentTypeError,
175 JSONDecodeError,
176 ):
177 self.logger.error("Failed to retrieve %s", endpoint)
178 text_result = await response.text()
179 self.logger.debug(text_result)
180 return None
181 except (
182 aiohttp.client_exceptions.ClientConnectorError,
183 aiohttp.client_exceptions.ServerDisconnectedError,
184 ):
185 self.logger.warning("Failed to retrieve %s", endpoint)
186 return None
187 if "error" in result and "limit" in result["error"]:
188 self.logger.warning(result["error"])
189 return None
190 return cast("dict[str, Any]", result)
191