/
/
/
1"""MusicBrainz metadata provider implementation."""
2
3from __future__ import annotations
4
5import re
6from contextlib import suppress
7from typing import TYPE_CHECKING, Any
8
9from mashumaro.exceptions import InvalidFieldValue, MissingField
10from music_assistant_models.config_entries import ConfigEntry
11from music_assistant_models.enums import ArtistEntityType, ConfigEntryType, ExternalID, LinkType
12from music_assistant_models.errors import InvalidDataError
13from music_assistant_models.media_items import MediaItemLink, MediaItemMetadata, UniqueList
14from music_assistant_models.media_items.metadata import LifeSpan
15
16from music_assistant.constants import VARIOUS_ARTISTS_MBID
17from music_assistant.controllers.cache import use_cache
18from music_assistant.helpers.compare import compare_strings
19from music_assistant.helpers.external_ids import (
20 external_id_lookup_values,
21 is_valid_barcode,
22 is_valid_isrc,
23 normalize_external_id,
24)
25from music_assistant.helpers.util import parse_title_and_version
26from music_assistant.models.metadata_provider import MetadataProvider
27
28from .api_client import MusicBrainzAPIClient
29from .constants import (
30 LUCENE_SPECIAL,
31 MIN_FIRST_RELEASE_CORRECTION_YEARS,
32 SOCIAL_HOST_MAPPING,
33 SUPPORTED_FEATURES,
34 URL_RELATION_TYPE_MAPPING,
35)
36from .models import (
37 MusicBrainzArtist,
38 MusicBrainzBarcodeRelease,
39 MusicBrainzRecording,
40 MusicBrainzRelation,
41 MusicBrainzRelease,
42 MusicBrainzReleaseGroup,
43)
44from .recommendations import MusicBrainzRecommendationManager
45
46if TYPE_CHECKING:
47 from music_assistant_models.config_entries import ProviderConfig
48 from music_assistant_models.media_items import (
49 Album,
50 Artist,
51 BrowseFolder,
52 ItemMapping,
53 MediaItemType,
54 RecommendationFolder,
55 Track,
56 )
57 from music_assistant_models.provider import ProviderManifest
58
59 from music_assistant.mass import MusicAssistant
60 from music_assistant.models import ProviderInstanceType
61
62# Config keys
63CONF_RECOMMENDATION_DAYS = "recommendation_days"
64
65
66async def setup(
67 mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
68) -> ProviderInstanceType:
69 """Initialize provider(instance) with given configuration."""
70 return MusicbrainzProvider(mass, manifest, config, SUPPORTED_FEATURES)
71
72
73class MusicbrainzProvider(MetadataProvider):
74 """The Musicbrainz Metadata provider."""
75
76 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
77 """Return Config entries to setup this provider."""
78 return (
79 ConfigEntry(
80 key=CONF_RECOMMENDATION_DAYS,
81 type=ConfigEntryType.INTEGER,
82 default_value=3,
83 range=(1, 15),
84 advanced=True,
85 ),
86 )
87
88 async def handle_async_init(self) -> None:
89 """Handle async initialization of the provider."""
90 self.cache = self.mass.cache
91 self._api_client = MusicBrainzAPIClient(self.mass)
92 self._recommendations = MusicBrainzRecommendationManager(self)
93
94 async def loaded_in_mass(self) -> None:
95 """Call after the provider has been loaded."""
96 await super().loaded_in_mass()
97 # Warm the recommendation cache in the background so the discover page is never
98 # blocked by the (rate-limited) initial MusicBrainz library scan.
99 self._recommendations.schedule_refresh()
100
101 async def unload(self, is_removed: bool = False) -> None:
102 """Handle unload/close of the provider."""
103 self._recommendations.cancel()
104
105 async def get_recommendations(self) -> list[RecommendationFolder]:
106 """Return MusicBrainz recommendation folders (artist birthdays/memorials and group founded/disbanded)."""
107 return await self._recommendations.get_recommendations()
108
109 async def get_recommendation_items(
110 self, item_id: str
111 ) -> UniqueList[MediaItemType | ItemMapping | BrowseFolder]:
112 """Get items for a MusicBrainz recommendation folder."""
113 return await self._recommendations.get_recommendation_items(item_id)
114
115 async def search(
116 self, artistname: str, albumname: str, trackname: str, trackversion: str | None = None
117 ) -> tuple[MusicBrainzArtist, MusicBrainzReleaseGroup, MusicBrainzRecording] | None:
118 """
119 Search MusicBrainz details by providing the artist, album and track name.
120
121 NOTE: The MusicBrainz objects returned are simplified objects without the optional data.
122 """
123 trackname, trackversion = parse_title_and_version(trackname, trackversion)
124 searchartist = re.sub(LUCENE_SPECIAL, r"\\\1", artistname)
125 searchalbum = re.sub(LUCENE_SPECIAL, r"\\\1", albumname)
126 searchtracks: list[str] = []
127 if trackversion:
128 searchtracks.append(f"{trackname} ({trackversion})")
129 searchtracks.append(trackname)
130 # the version is sometimes appended to the title and sometimes stored
131 # in disambiguation, so we try both
132 for strict in (True, False):
133 for searchtrack in searchtracks:
134 searchstr = re.sub(LUCENE_SPECIAL, r"\\\1", searchtrack)
135 result = await self._api_client.get_data(
136 "recording",
137 query=f'"{searchstr}" AND artist:"{searchartist}" AND release:"{searchalbum}"',
138 )
139 if not result or "recordings" not in result:
140 continue
141 for item in result["recordings"]:
142 # compare track title
143 if not compare_strings(item["title"], searchtrack, strict):
144 continue
145 # compare track version if needed
146 if (
147 trackversion
148 and trackversion not in searchtrack
149 and not compare_strings(item.get("disambiguation"), trackversion, strict)
150 ):
151 continue
152 # match (primary) track artist
153 artist_match: MusicBrainzArtist | None = None
154 for artist in item["artist-credit"]:
155 if compare_strings(artist["artist"]["name"], artistname, strict):
156 artist_match = MusicBrainzArtist.from_raw(artist["artist"])
157 else:
158 for alias in artist["artist"].get("aliases", []):
159 if compare_strings(alias["name"], artistname, strict):
160 artist_match = MusicBrainzArtist.from_raw(artist["artist"])
161 if not artist_match:
162 continue
163 # match album/release
164 album_match: MusicBrainzReleaseGroup | None = None
165 for release in item["releases"]:
166 if compare_strings(release["title"], albumname, strict) or compare_strings(
167 release["release-group"]["title"], albumname, strict
168 ):
169 album_match = MusicBrainzReleaseGroup.from_raw(release["release-group"])
170 break
171 else:
172 continue
173 # if we reach this point, we got a match on recording,
174 # artist and release(group)
175 recording = MusicBrainzRecording.from_raw(item)
176 return (artist_match, album_match, recording)
177
178 return None
179
180 async def get_artist_details(self, artist_id: str) -> MusicBrainzArtist:
181 """Get (full) Artist details by providing a MusicBrainz artist id."""
182 endpoint = (
183 f"artist/{artist_id}?inc=aliases+annotation+tags+ratings+genres+url-rels+work-rels"
184 )
185 if result := await self._api_client.get_data(endpoint):
186 if "id" not in result:
187 result["id"] = artist_id
188 try:
189 return MusicBrainzArtist.from_raw(result)
190 except MissingField as err:
191 raise InvalidDataError from err
192 msg = "Invalid MusicBrainz Artist ID provided"
193 raise InvalidDataError(msg)
194
195 async def resolve_artists_from_mbids(
196 self, mbids: tuple[str, ...]
197 ) -> list[tuple[str, str, str] | None]:
198 """
199 Look up canonical artist names for a sequence of MusicBrainz artist IDs.
200
201 Transient failures (MusicBrainz unreachable, retries exhausted) are left
202 to propagate so the caller can retry later rather than persist degraded
203 data; only a genuinely unresolvable MBID yields ``None``.
204
205 :param mbids: MusicBrainz artist IDs to look up.
206 :return: One entry per input MBID, in the same order, as a
207 ``(name, mbid, sort_name)`` tuple. ``None`` at a position means
208 that MBID could not be resolved.
209 """
210 results: list[tuple[str, str, str] | None] = []
211 for mbid in mbids:
212 try:
213 artist = await self.get_artist_details(mbid)
214 results.append((artist.name, mbid, artist.sort_name))
215 except InvalidDataError as err:
216 self.logger.warning("Failed to lookup MusicBrainz artist %s: %s", mbid, err)
217 results.append(None)
218 return results
219
220 async def get_artist_metadata(self, artist: Artist) -> MediaItemMetadata | None:
221 """Surface MusicBrainz artist type, life span, and URL relations."""
222 if not artist.mbid:
223 return None
224 try:
225 details = await self.get_artist_details(artist.mbid)
226 except InvalidDataError:
227 return None
228 artist_entity_type: ArtistEntityType | None = None
229 if details.type:
230 entity_type = ArtistEntityType(details.type.lower())
231 if entity_type != ArtistEntityType.UNKNOWN:
232 artist_entity_type = entity_type
233 life_span: LifeSpan | None = None
234 if details.life_span:
235 life_span = LifeSpan(
236 begin=details.life_span.begin,
237 end=details.life_span.end,
238 ended=details.life_span.ended,
239 )
240 links: set[MediaItemLink] = set()
241 if details.relations:
242 for relation in details.relations:
243 if not relation.url:
244 continue
245 if link_type := self._link_type_for_relation(relation):
246 links.add(MediaItemLink(type=link_type, url=relation.url.resource))
247 if not artist_entity_type and not life_span and not links:
248 return None
249 return MediaItemMetadata(
250 links=links,
251 artist_entity_type=artist_entity_type,
252 life_span=life_span,
253 )
254
255 async def get_recording_details(self, recording_id: str) -> MusicBrainzRecording:
256 """Get Recording details by providing a MusicBrainz Recording Id."""
257 if result := await self._api_client.get_data(
258 f"recording/{recording_id}?inc=artists+releases+isrcs"
259 ):
260 if "id" not in result:
261 result["id"] = recording_id
262 try:
263 return MusicBrainzRecording.from_raw(result)
264 except MissingField as err:
265 raise InvalidDataError from err
266 msg = "Invalid MusicBrainz recording ID provided"
267 raise InvalidDataError(msg)
268
269 @use_cache(86400 * 30)
270 async def get_isrcs_for_recording(self, recording_id: str) -> list[str]:
271 """
272 Get ISRCs for a MusicBrainz Recording ID.
273
274 :param recording_id: MusicBrainz recording ID, or a track ID as
275 handed out by e.g. Last.fm.
276 :return: List of ISRCs, or empty list if not found.
277 """
278 # the search response includes the ISRCs, so either ID kind costs one call
279 safe_id = re.sub(LUCENE_SPECIAL, r"\\\1", recording_id)
280 query = f"rid:{safe_id} OR tid:{safe_id}"
281 if (result := await self._api_client.get_data("recording", query=query)) and (
282 recordings := result.get("recordings")
283 ):
284 return recordings[0].get("isrcs") or []
285 # merged (redirected) recording MBIDs are absent from the search
286 # index but still resolve via direct lookup
287 with suppress(InvalidDataError):
288 recording = await self.get_recording_details(recording_id)
289 return recording.isrcs or []
290 return []
291
292 async def get_recordings_by_isrc(self, isrc: str) -> list[MusicBrainzRecording]:
293 """
294 Get the recordings MusicBrainz has on file for an ISRC.
295
296 Inverse of :meth:`get_isrcs_for_recording`: that one goes from a
297 recording to its ISRCs, this one goes from an ISRC back to recordings.
298
299 :param isrc: ISRC of the recording, with or without separators.
300 :return: Recordings tagged with this ISRC, or empty list if not found.
301 """
302 if not is_valid_isrc(isrc):
303 return []
304 safe_isrc = normalize_external_id(ExternalID.ISRC, isrc)
305 # the isrc resource rejects inc= parameters and already carries the release dates
306 result = await self._api_client.get_data(f"isrc/{safe_isrc}")
307 if not result or not (recordings := result.get("recordings")):
308 return []
309 parsed: list[MusicBrainzRecording] = []
310 for recording in recordings:
311 # a single malformed entry should not sink the recordings we did parse
312 with suppress(MissingField):
313 parsed.append(MusicBrainzRecording.from_raw(recording))
314 return parsed
315
316 async def get_release_year_by_isrc(self, isrc: str) -> int | None:
317 """
318 Get the year a recording was first released, by ISRC.
319
320 :param isrc: ISRC of the recording, with or without separators.
321 :return: The earliest known release year, or None if MusicBrainz does not know it.
322 """
323 recordings = await self.get_recordings_by_isrc(isrc)
324 # one ISRC can cover several recordings, the oldest one dates the song
325 years = [
326 year
327 for recording in recordings
328 if (year := _release_year(recording.first_release_date or "")) is not None
329 ]
330 return min(years, default=None)
331
332 async def get_release_details(self, album_id: str) -> MusicBrainzRelease:
333 """Get Release/Album details by providing a MusicBrainz Album id."""
334 endpoint = f"release/{album_id}?inc=artist-credits+aliases+labels"
335 if result := await self._api_client.get_data(endpoint):
336 if "id" not in result:
337 result["id"] = album_id
338 try:
339 return MusicBrainzRelease.from_raw(result)
340 except MissingField as err:
341 raise InvalidDataError from err
342 msg = "Invalid MusicBrainz Album ID provided"
343 raise InvalidDataError(msg)
344
345 async def get_releases_by_barcode(self, barcode: str) -> list[MusicBrainzBarcodeRelease]:
346 """
347 Get the releases MusicBrainz has on file for a barcode/UPC.
348
349 The result is complete: a truncated page or a release entry that cannot be parsed
350 raises :class:`InvalidDataError` so the caller abstains instead of mistaking a
351 partial release/group set for the whole set.
352
353 :param barcode: Album barcode (UPC/EAN/GTIN), with or without separators.
354 :return: Releases carrying this barcode, or an empty list if none are found.
355 """
356 if not is_valid_barcode(barcode):
357 return []
358 # a UPC-12 and its zero-padded EAN-13/GTIN forms are the same physical barcode,
359 # so query every compatible form to also resolve a provider's shorter/longer notation
360 barcodes = [
361 value
362 for value in external_id_lookup_values(ExternalID.BARCODE, barcode)
363 if value.isdigit()
364 ]
365 query = " OR ".join(f"barcode:{value}" for value in barcodes)
366 # a barcode identifies a single physical product, so one generously-sized page
367 # returns every release carrying it
368 result = await self._api_client.get_data("release", query=query, limit="100")
369 if not result or not (releases := result.get("releases")):
370 return []
371 if result.get("count", len(releases)) > len(releases):
372 msg = "MusicBrainz barcode result is truncated"
373 raise InvalidDataError(msg)
374 parsed: list[MusicBrainzBarcodeRelease] = []
375 for release in releases:
376 try:
377 parsed.append(MusicBrainzBarcodeRelease.from_raw(release))
378 except (MissingField, InvalidFieldValue) as err:
379 # dropping a malformed row would make the release/group set look complete
380 # when it is not, so the whole lookup is treated as unusable
381 msg = "MusicBrainz barcode result has an unparsable release"
382 raise InvalidDataError(msg) from err
383 return parsed
384
385 async def get_releasegroup_details(self, releasegroup_id: str) -> MusicBrainzReleaseGroup:
386 """Get ReleaseGroup details by providing a MusicBrainz ReleaseGroup id."""
387 endpoint = f"release-group/{releasegroup_id}?inc=artists+aliases"
388 if result := await self._api_client.get_data(endpoint):
389 if "id" not in result:
390 result["id"] = releasegroup_id
391 try:
392 return MusicBrainzReleaseGroup.from_raw(result)
393 except MissingField as err:
394 raise InvalidDataError from err
395 msg = "Invalid MusicBrainz ReleaseGroup ID provided"
396 raise InvalidDataError(msg)
397
398 async def get_artist_details_by_album(
399 self, artistname: str, ref_album: Album
400 ) -> MusicBrainzArtist | None:
401 """
402 Get musicbrainz artist details by providing the artist name and a reference album.
403
404 MusicBrainzArtist object that is returned does not contain the optional data.
405 """
406 result: MusicBrainzRelease | MusicBrainzReleaseGroup | None = None
407 if mb_id := ref_album.get_external_id(ExternalID.MB_RELEASEGROUP):
408 with suppress(InvalidDataError):
409 result = await self.get_releasegroup_details(mb_id)
410 elif mb_id := ref_album.get_external_id(ExternalID.MB_ALBUM):
411 with suppress(InvalidDataError):
412 result = await self.get_release_details(mb_id)
413 else:
414 return None
415 if not (result and result.artist_credit):
416 return None
417 for strict in (True, False):
418 for artist_credit in result.artist_credit:
419 if compare_strings(artist_credit.artist.name, artistname, strict):
420 return artist_credit.artist
421 for alias in artist_credit.artist.aliases or []:
422 if compare_strings(alias.name, artistname, strict):
423 return artist_credit.artist
424 return None
425
426 async def get_artist_details_by_track(
427 self, artistname: str, ref_track: Track
428 ) -> MusicBrainzArtist | None:
429 """
430 Get musicbrainz artist details by providing the artist name and a reference track.
431
432 MusicBrainzArtist object that is returned does not contain the optional data.
433 """
434 if not ref_track.mbid:
435 return None
436 result = None
437 with suppress(InvalidDataError):
438 result = await self.get_recording_details(ref_track.mbid)
439 if not (result and result.artist_credit):
440 return None
441 for strict in (True, False):
442 for artist_credit in result.artist_credit:
443 if compare_strings(artist_credit.artist.name, artistname, strict):
444 return artist_credit.artist
445 for alias in artist_credit.artist.aliases or []:
446 if compare_strings(alias.name, artistname, strict):
447 return artist_credit.artist
448 return None
449
450 async def get_artist_details_by_resource_url(
451 self, resource_url: str
452 ) -> MusicBrainzArtist | None:
453 """
454 Get musicbrainz artist details by providing a resource URL (e.g. Spotify share URL).
455
456 MusicBrainzArtist object that is returned does not contain the optional data.
457 """
458 if result := await self._api_client.get_data(
459 "url", resource=resource_url, inc="artist-rels"
460 ):
461 for relation in result.get("relations", []):
462 if not (artist := relation.get("artist")):
463 continue
464 return MusicBrainzArtist.from_raw(artist)
465 return None
466
467 async def get_release_group_by_track_name(
468 self, artist_name: str, track_name: str
469 ) -> tuple[MusicBrainzArtist, list[MusicBrainzReleaseGroup]] | None:
470 """
471 Find release groups for a track by searching MusicBrainz recordings.
472
473 Returns matching release groups sorted by release date,
474 prioritizing the earliest original recording to find the correct releases.
475
476 :param artist_name: Artist name to search for.
477 :param track_name: Track name to search for.
478 :returns: Tuple of (artist, release_groups) or None.
479 """
480 if not (result := await self._search_release_groups_by_track_name(artist_name, track_name)):
481 return None
482 artist, release_groups = result
483 return (MusicBrainzArtist.from_raw(artist), [rg for rg, _ in release_groups])
484
485 async def get_release_year_by_track_name(self, artist_name: str, track_name: str) -> int | None:
486 """
487 Get the year a song was first released, by artist and track name.
488
489 Weaker evidence than :meth:`get_release_year_by_isrc`, which identifies the exact
490 recording: this matches on name and only counts studio albums, soundtracks and
491 singles named after the song, so an ambiguous or unknown name yields no year
492 rather than a guess.
493 Costs up to two MusicBrainz requests.
494
495 :param artist_name: Name of the track's primary artist.
496 :param track_name: Name of the track.
497 :return: The earliest known release year, or None if MusicBrainz does not know it.
498 """
499 result = await self._search_release_groups_by_track_name(artist_name, track_name)
500 if not result or not (release_groups := result[1]):
501 return None
502 # the release groups are sorted oldest first, and undated ones sort last
503 release_year = _release_year(release_groups[0][1])
504 # the release found already dates the song, so a lookup that fails costs this song
505 # precision rather than the year the search already supplied
506 first_release_year: int | None = None
507 with suppress(Exception):
508 first_release_year = await self._earliest_first_release_year(
509 [release_group.id for release_group, _ in release_groups]
510 )
511 if first_release_year is None:
512 return release_year
513 if release_year is None:
514 return first_release_year
515 if release_year - first_release_year > MIN_FIRST_RELEASE_CORRECTION_YEARS:
516 return first_release_year
517 return release_year
518
519 @staticmethod
520 def _link_type_for_relation(relation: MusicBrainzRelation) -> LinkType | None:
521 if link_type := URL_RELATION_TYPE_MAPPING.get(relation.type):
522 return link_type
523 if relation.type == "social network" and relation.url:
524 url_lower = relation.url.resource.lower()
525 for host, link_type in SOCIAL_HOST_MAPPING:
526 if host in url_lower:
527 return link_type
528 return None
529
530 async def _search_release_groups_by_track_name(
531 self, artist_name: str, track_name: str
532 ) -> tuple[dict[str, Any], list[tuple[MusicBrainzReleaseGroup, str]]] | None:
533 """
534 Search recordings by artist and track name and aggregate their release groups.
535
536 :param artist_name: Artist name to search for.
537 :param track_name: Track name to search for.
538 :return: Tuple of (raw artist, release groups with their earliest release date sorted
539 oldest first), or None when no recording matched. The release groups are empty
540 when the matched recordings carry no studio album, soundtrack or same-named single.
541 """
542 search_artist = re.sub(LUCENE_SPECIAL, r"\\\1", artist_name)
543 search_track = re.sub(LUCENE_SPECIAL, r"\\\1", track_name)
544 result = await self._api_client.get_data(
545 "recording",
546 query=f'"{search_track}" AND artist:"{search_artist}"',
547 limit="100",
548 )
549 if not result or "recordings" not in result:
550 return None
551
552 # Collect all matching recordings with their artist and first-release-date
553 matches: list[tuple[dict[str, Any], dict[str, Any], str]] = []
554 for strict in (True, False):
555 for item in result["recordings"]:
556 if not compare_strings(item["title"], track_name, strict):
557 continue
558 for artist_credit in item.get("artist-credit", []):
559 artist = artist_credit.get("artist", {})
560 artist_matches = compare_strings(artist.get("name", ""), artist_name, strict)
561 if not artist_matches:
562 for alias in artist.get("aliases", []):
563 if compare_strings(alias.get("name", ""), artist_name, strict):
564 artist_matches = True
565 break
566 if artist_matches:
567 first_release = item.get("first-release-date", "") or ""
568 matches.append((item, artist, first_release))
569 break
570 if matches:
571 break
572
573 if not matches:
574 return None
575
576 # Sort by first-release-date to find the earliest (likely original studio recording)
577 matches.sort(key=lambda x: x[2] if x[2] else "9999")
578
579 # Aggregate release groups from ALL matching recordings
580 # This ensures we find albums even if the first recording only has singles
581 all_release_groups: dict[str, tuple[MusicBrainzReleaseGroup, str]] = {}
582 for recording, _, _ in matches:
583 for rg, release_date in self._get_release_groups_with_dates(recording, track_name):
584 rg_id = rg.id
585 if rg_id in all_release_groups:
586 existing_rg, existing_date = all_release_groups[rg_id]
587 if release_date and (not existing_date or release_date < existing_date):
588 if not rg.barcode:
589 rg.barcode = existing_rg.barcode
590 all_release_groups[rg_id] = (rg, release_date)
591 elif rg.barcode and not existing_rg.barcode:
592 existing_rg.barcode = rg.barcode
593 else:
594 all_release_groups[rg_id] = (rg, release_date)
595
596 if not all_release_groups:
597 # Fall back to the earliest recording (for artist lookup at least)
598 return (matches[0][1], [])
599 # Sort by release date
600 sorted_groups = sorted(all_release_groups.values(), key=lambda x: x[1] if x[1] else "9999")
601 return (matches[0][1], sorted_groups)
602
603 def _get_release_groups_with_dates(
604 self, recording: dict[str, Any], track_name: str
605 ) -> list[tuple[MusicBrainzReleaseGroup, str]]:
606 """
607 Collect release groups for a recording with their release dates.
608
609 Filters out compilations, live and other rereleases, including the compilations
610 credited to Various Artists rather than tagged as such. Soundtracks are kept,
611 since a song written for a film is first released on one.
612 For singles, only includes those where the title matches the track name.
613 Returns list of (release_group, release_date) tuples for singles and studio albums.
614
615 :param recording: MusicBrainz recording dict.
616 :param track_name: Track name to match against single titles.
617 """
618 releases = recording.get("releases", [])
619 if not releases:
620 return []
621
622 # Collect release groups with their earliest release date, deduplicating by ID
623 seen: dict[str, tuple[MusicBrainzReleaseGroup, str]] = {}
624
625 for release in releases:
626 # Skip bootleg and pseudo-releases
627 release_status = release.get("status", "")
628 if release_status in ("Bootleg", "Pseudo-Release"):
629 continue
630
631 # Plenty of hits compilations carry no Compilation secondary type, so they pass
632 # for studio albums. Their releases are credited to Various Artists, which an
633 # album or single of one artist never is.
634 if _is_various_artists_release(release):
635 continue
636
637 rg = release.get("release-group", {})
638 rg_id = rg.get("id")
639 if not rg_id:
640 continue
641
642 primary_type = rg.get("primary-type")
643 secondary_types = rg.get("secondary-types", [])
644
645 # Only include singles and studio albums (no compilations, live, etc.)
646 if primary_type not in ("Album", "Single"):
647 continue
648 # A song written for a film or musical is first released on its soundtrack, which
649 # MusicBrainz types as an album with a Soundtrack secondary type. Any other
650 # secondary type, alongside Soundtrack or not, means the release group is not
651 # where the song came out.
652 if secondary_types and secondary_types != ["Soundtrack"]:
653 continue
654
655 # For singles, only include if the title matches the track name
656 # (avoid B-sides and bonus tracks on unrelated singles)
657 if primary_type == "Single":
658 if not compare_strings(rg.get("title", ""), track_name, strict=False):
659 continue
660
661 release_date = release.get("date", "") or ""
662 barcode = release.get("barcode") or None
663
664 # Keep the earliest release date per release group
665 if rg_id in seen:
666 existing_rg, existing_date = seen[rg_id]
667 if release_date and (not existing_date or release_date < existing_date):
668 mb_rg = MusicBrainzReleaseGroup.from_raw(rg)
669 mb_rg.barcode = barcode or existing_rg.barcode
670 seen[rg_id] = (mb_rg, release_date)
671 elif barcode and not existing_rg.barcode:
672 existing_rg.barcode = barcode
673 else:
674 mb_rg = MusicBrainzReleaseGroup.from_raw(rg)
675 mb_rg.barcode = barcode
676 seen[rg_id] = (mb_rg, release_date)
677
678 return list(seen.values())
679
680 async def _earliest_first_release_year(self, release_group_ids: list[str]) -> int | None:
681 """
682 Get the year the oldest of the given release groups was first released.
683
684 :param release_group_ids: MusicBrainz release group IDs to look up.
685 :return: The earliest known first release year, or None if MusicBrainz knows none.
686 """
687 # a recording search never yields more than a handful of release groups, so one
688 # query covers them all
689 safe_ids = (re.sub(LUCENE_SPECIAL, r"\\\1", rg_id) for rg_id in release_group_ids)
690 result = await self._api_client.get_data(
691 "release-group",
692 query=f"rgid:({' OR '.join(safe_ids)})",
693 limit="100",
694 )
695 if not result:
696 return None
697 years = [
698 year
699 for release_group in result.get("release-groups", [])
700 if (year := _release_year(release_group.get("first-release-date") or "")) is not None
701 ]
702 return min(years, default=None)
703
704
705def _is_various_artists_release(release: dict[str, Any]) -> bool:
706 """
707 Return whether a MusicBrainz release is credited to Various Artists.
708
709 :param release: MusicBrainz release dict from a recording search.
710 """
711 # MusicBrainz always credits the Various Artists entity by id, while its display name is
712 # localized and other artists are named after it, so only the id identifies it.
713 return any(
714 (credit.get("artist") or {}).get("id") == VARIOUS_ARTISTS_MBID
715 for credit in release.get("artist-credit") or ()
716 )
717
718
719def _release_year(release_date: str) -> int | None:
720 """
721 Read the year off a MusicBrainz date of any precision.
722
723 :param release_date: MusicBrainz date, as a year, year-month or full date.
724 :return: The year, or None if the date is absent or unparsable.
725 """
726 return int(year) if (year := release_date[:4]).isdigit() else None
727