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