music-assistant-server

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