/
/
/
1"""
2Wikipedia Metadata provider for Music Assistant.
3
4Provides artist biographies in the user's preferred language, resolving the
5article through MusicBrainz URL relations and (where MusicBrainz lacks coverage)
6Wikidata sitelinks.
7"""
8
9from __future__ import annotations
10
11from json import JSONDecodeError
12from typing import TYPE_CHECKING, Any, cast
13from urllib.parse import unquote, urlparse
14
15import aiohttp
16from music_assistant_models.enums import ProviderFeature
17from music_assistant_models.errors import InvalidDataError, ResourceTemporarilyUnavailable
18from music_assistant_models.media_items import MediaItemMetadata
19
20from music_assistant.controllers.cache import use_cache
21from music_assistant.helpers.throttle_retry import Throttler
22from music_assistant.models.metadata_provider import MetadataProvider
23
24if TYPE_CHECKING:
25 from music_assistant_models.config_entries import ConfigEntry, ProviderConfig
26 from music_assistant_models.media_items import Artist
27 from music_assistant_models.provider import ProviderManifest
28
29 from music_assistant.mass import MusicAssistant
30 from music_assistant.models import ProviderInstanceType
31 from music_assistant.providers.musicbrainz import MusicbrainzProvider
32 from music_assistant.providers.musicbrainz.models import MusicBrainzRelation
33
34
35SUPPORTED_FEATURES: set[ProviderFeature] = {ProviderFeature.ARTIST_METADATA}
36
37WIKIDATA_API_URL = "https://www.wikidata.org/w/api.php"
38
39
40async def setup(
41 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
42) -> ProviderInstanceType:
43 """Initialize provider(instance) with given configuration."""
44 return WikipediaMetadataProvider(mass, manifest, config, SUPPORTED_FEATURES)
45
46
47class WikipediaMetadataProvider(MetadataProvider):
48 """Wikipedia Metadata provider."""
49
50 throttler: Throttler
51
52 @property
53 def priority(self) -> int:
54 """Priority for this provider (lower = more preferred)."""
55 return 25
56
57 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
58 """Return Config entries to configure this provider."""
59 return ()
60
61 async def handle_async_init(self) -> None:
62 """Handle async initialization of the provider."""
63 self.throttler = Throttler(rate_limit=1, period=1)
64
65 async def get_artist_metadata(self, artist: Artist) -> MediaItemMetadata | None:
66 """Retrieve metadata for an artist on Wikipedia."""
67 if not artist.mbid:
68 return None
69
70 preferred_lang = self.mass.metadata.preferred_language
71 languages: list[str] = [preferred_lang]
72 if preferred_lang != "en":
73 languages.append("en")
74
75 relations = await self._musicbrainz_relations(artist.mbid)
76 if not relations:
77 return None
78
79 titles_by_lang = _wiki_titles_by_lang(relations)
80
81 # only query Wikidata for languages MusicBrainz didn't already provide, saving a
82 # request whenever the MusicBrainz relations already cover the wanted languages
83 missing = [lang for lang in languages if lang not in titles_by_lang]
84 if missing and (qid := _wikidata_qid(relations)):
85 sitelinks = await self._wikidata_sitelinks(qid, tuple(sorted(missing)))
86 for lang, sitelink_title in sitelinks.items():
87 titles_by_lang.setdefault(lang, sitelink_title)
88
89 for lang in languages:
90 if title := titles_by_lang.get(lang):
91 if extract := await self._fetch_bio(lang, title):
92 self.logger.debug("Found bio for %s on Wikipedia in %s", artist.name, lang)
93 return MediaItemMetadata(description=extract, description_language=lang)
94 return None
95
96 async def _musicbrainz_relations(self, mbid: str) -> list[MusicBrainzRelation] | None:
97 """Return the MusicBrainz URL relations for an artist."""
98 mb_provider = cast("MusicbrainzProvider | None", self.mass.get_provider("musicbrainz"))
99 if mb_provider is None:
100 return None
101 try:
102 details = await mb_provider.get_artist_details(mbid)
103 except InvalidDataError:
104 return None
105 return details.relations
106
107 @use_cache(86400 * 90, persistent=True)
108 async def _wikidata_sitelinks(self, qid: str, languages: tuple[str, ...]) -> dict[str, str]:
109 """
110 Return Wikipedia article titles for a Wikidata entity keyed by language.
111
112 :param qid: Wikidata entity id (e.g. ``"Q12345"``).
113 :param languages: ISO 639-1 codes to request sitelinks for.
114 """
115 sitefilter = "|".join(f"{lang}wiki" for lang in languages)
116 data = await self._get_json(
117 WIKIDATA_API_URL,
118 params={
119 "action": "wbgetentities",
120 "ids": qid,
121 "props": "sitelinks",
122 "sitefilter": sitefilter,
123 "format": "json",
124 },
125 )
126 if not data:
127 return {}
128 sitelinks = data.get("entities", {}).get(qid, {}).get("sitelinks") or {}
129 result: dict[str, str] = {}
130 for site_key, payload in sitelinks.items():
131 if not site_key.endswith("wiki"):
132 continue
133 lang = site_key[: -len("wiki")]
134 title = payload.get("title")
135 if isinstance(title, str) and title:
136 result[lang] = title
137 return result
138
139 @use_cache(86400 * 90, persistent=True)
140 async def _fetch_bio(self, lang: str, title: str) -> str | None:
141 """
142 Return the plain-text lead section of a Wikipedia article.
143
144 :param lang: Wikipedia language edition (``"en"``, ``"de"``, ...).
145 :param title: Article title as it appears in the URL.
146 """
147 data = await self._get_json(
148 f"https://{lang}.wikipedia.org/w/api.php",
149 params={
150 "action": "query",
151 "prop": "extracts",
152 "exintro": "true",
153 "explaintext": "true",
154 "redirects": "1",
155 "titles": title,
156 "format": "json",
157 },
158 )
159 if not data:
160 return None
161 pages = data.get("query", {}).get("pages") or {}
162 if not pages:
163 return None
164 page = next(iter(pages.values()))
165 extract = page.get("extract")
166 if isinstance(extract, str) and extract.strip():
167 return extract
168 return None
169
170 async def _get_json(
171 self, url: str, params: dict[str, str] | None = None
172 ) -> dict[str, Any] | None:
173 """
174 Return the parsed JSON from a GET request, or None when the resource is absent.
175
176 Only a 404 yields None; a transient failure (network, another HTTP error or an
177 unparsable response) raises ResourceTemporarilyUnavailable so callers do not cache
178 it as a negative result.
179
180 :param url: Request URL.
181 :param params: Optional query parameters.
182 """
183 headers = {
184 "User-Agent": f"Music Assistant/{self.mass.version} (https://music-assistant.io)"
185 }
186 try:
187 async with (
188 self.throttler,
189 self.mass.http_session.get(url, params=params, headers=headers) as response,
190 ):
191 if response.status == 404:
192 return None
193 response.raise_for_status()
194 return cast("dict[str, Any]", await response.json())
195 except (aiohttp.ClientError, TimeoutError, JSONDecodeError) as err:
196 raise ResourceTemporarilyUnavailable("Wikipedia request failed") from err
197
198
199def _wiki_titles_by_lang(relations: list[MusicBrainzRelation]) -> dict[str, str]:
200 """Return Wikipedia article titles keyed by language from MusicBrainz URL relations."""
201 result: dict[str, str] = {}
202 for relation in relations:
203 if relation.type != "wikipedia" or not relation.url:
204 continue
205 parsed = urlparse(relation.url.resource)
206 host = parsed.netloc.lower()
207 if not host.endswith(".wikipedia.org"):
208 continue
209 lang = host.split(".", 1)[0]
210 if not parsed.path.startswith("/wiki/"):
211 continue
212 title = unquote(parsed.path[len("/wiki/") :])
213 if title:
214 result.setdefault(lang, title)
215 return result
216
217
218def _wikidata_qid(relations: list[MusicBrainzRelation]) -> str | None:
219 """Return the Wikidata entity id from MusicBrainz URL relations, if present."""
220 for relation in relations:
221 if relation.type != "wikidata" or not relation.url:
222 continue
223 candidate = relation.url.resource.rstrip("/").rsplit("/", 1)[-1]
224 if candidate.startswith("Q") and candidate[1:].isdigit():
225 return candidate
226 return None
227