music-assistant-server

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