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, soundtracks and
444        singles named after the song, so an ambiguous or unknown name yields no year
445        rather than a guess.
446        Costs up to two MusicBrainz requests.
447
448        :param artist_name: Name of the track's primary artist.
449        :param track_name: Name of the track.
450        :return: The earliest known release year, or None if MusicBrainz does not know it.
451        """
452        result = await self._search_release_groups_by_track_name(artist_name, track_name)
453        if not result or not (release_groups := result[1]):
454            return None
455        # the release groups are sorted oldest first, and undated ones sort last
456        release_year = _release_year(release_groups[0][1])
457        # the release found already dates the song, so a lookup that fails costs this song
458        # precision rather than the year the search already supplied
459        first_release_year: int | None = None
460        with suppress(Exception):
461            first_release_year = await self._earliest_first_release_year(
462                [release_group.id for release_group, _ in release_groups]
463            )
464        if first_release_year is None:
465            return release_year
466        if release_year is None:
467            return first_release_year
468        if release_year - first_release_year > MIN_FIRST_RELEASE_CORRECTION_YEARS:
469            return first_release_year
470        return release_year
471
472    @staticmethod
473    def _link_type_for_relation(relation: MusicBrainzRelation) -> LinkType | None:
474        if link_type := URL_RELATION_TYPE_MAPPING.get(relation.type):
475            return link_type
476        if relation.type == "social network" and relation.url:
477            url_lower = relation.url.resource.lower()
478            for host, link_type in SOCIAL_HOST_MAPPING:
479                if host in url_lower:
480                    return link_type
481        return None
482
483    async def _search_release_groups_by_track_name(
484        self, artist_name: str, track_name: str
485    ) -> tuple[dict[str, Any], list[tuple[MusicBrainzReleaseGroup, str]]] | None:
486        """
487        Search recordings by artist and track name and aggregate their release groups.
488
489        :param artist_name: Artist name to search for.
490        :param track_name: Track name to search for.
491        :return: Tuple of (raw artist, release groups with their earliest release date sorted
492            oldest first), or None when no recording matched. The release groups are empty
493            when the matched recordings carry no studio album, soundtrack or same-named single.
494        """
495        search_artist = re.sub(LUCENE_SPECIAL, r"\\\1", artist_name)
496        search_track = re.sub(LUCENE_SPECIAL, r"\\\1", track_name)
497        result = await self._api_client.get_data(
498            "recording",
499            query=f'"{search_track}" AND artist:"{search_artist}"',
500            limit="100",
501        )
502        if not result or "recordings" not in result:
503            return None
504
505        # Collect all matching recordings with their artist and first-release-date
506        matches: list[tuple[dict[str, Any], dict[str, Any], str]] = []
507        for strict in (True, False):
508            for item in result["recordings"]:
509                if not compare_strings(item["title"], track_name, strict):
510                    continue
511                for artist_credit in item.get("artist-credit", []):
512                    artist = artist_credit.get("artist", {})
513                    artist_matches = compare_strings(artist.get("name", ""), artist_name, strict)
514                    if not artist_matches:
515                        for alias in artist.get("aliases", []):
516                            if compare_strings(alias.get("name", ""), artist_name, strict):
517                                artist_matches = True
518                                break
519                    if artist_matches:
520                        first_release = item.get("first-release-date", "") or ""
521                        matches.append((item, artist, first_release))
522                        break
523            if matches:
524                break
525
526        if not matches:
527            return None
528
529        # Sort by first-release-date to find the earliest (likely original studio recording)
530        matches.sort(key=lambda x: x[2] if x[2] else "9999")
531
532        # Aggregate release groups from ALL matching recordings
533        # This ensures we find albums even if the first recording only has singles
534        all_release_groups: dict[str, tuple[MusicBrainzReleaseGroup, str]] = {}
535        for recording, _, _ in matches:
536            for rg, release_date in self._get_release_groups_with_dates(recording, track_name):
537                rg_id = rg.id
538                if rg_id in all_release_groups:
539                    existing_rg, existing_date = all_release_groups[rg_id]
540                    if release_date and (not existing_date or release_date < existing_date):
541                        if not rg.barcode:
542                            rg.barcode = existing_rg.barcode
543                        all_release_groups[rg_id] = (rg, release_date)
544                    elif rg.barcode and not existing_rg.barcode:
545                        existing_rg.barcode = rg.barcode
546                else:
547                    all_release_groups[rg_id] = (rg, release_date)
548
549        if not all_release_groups:
550            # Fall back to the earliest recording (for artist lookup at least)
551            return (matches[0][1], [])
552        # Sort by release date
553        sorted_groups = sorted(all_release_groups.values(), key=lambda x: x[1] if x[1] else "9999")
554        return (matches[0][1], sorted_groups)
555
556    def _get_release_groups_with_dates(
557        self, recording: dict[str, Any], track_name: str
558    ) -> list[tuple[MusicBrainzReleaseGroup, str]]:
559        """
560        Collect release groups for a recording with their release dates.
561
562        Filters out compilations, live and other rereleases, including the compilations
563        credited to Various Artists rather than tagged as such. Soundtracks are kept,
564        since a song written for a film is first released on one.
565        For singles, only includes those where the title matches the track name.
566        Returns list of (release_group, release_date) tuples for singles and studio albums.
567
568        :param recording: MusicBrainz recording dict.
569        :param track_name: Track name to match against single titles.
570        """
571        releases = recording.get("releases", [])
572        if not releases:
573            return []
574
575        # Collect release groups with their earliest release date, deduplicating by ID
576        seen: dict[str, tuple[MusicBrainzReleaseGroup, str]] = {}
577
578        for release in releases:
579            # Skip bootleg and pseudo-releases
580            release_status = release.get("status", "")
581            if release_status in ("Bootleg", "Pseudo-Release"):
582                continue
583
584            # Plenty of hits compilations carry no Compilation secondary type, so they pass
585            # for studio albums. Their releases are credited to Various Artists, which an
586            # album or single of one artist never is.
587            if _is_various_artists_release(release):
588                continue
589
590            rg = release.get("release-group", {})
591            rg_id = rg.get("id")
592            if not rg_id:
593                continue
594
595            primary_type = rg.get("primary-type")
596            secondary_types = rg.get("secondary-types", [])
597
598            # Only include singles and studio albums (no compilations, live, etc.)
599            if primary_type not in ("Album", "Single"):
600                continue
601            # A song written for a film or musical is first released on its soundtrack, which
602            # MusicBrainz types as an album with a Soundtrack secondary type. Any other
603            # secondary type, alongside Soundtrack or not, means the release group is not
604            # where the song came out.
605            if secondary_types and secondary_types != ["Soundtrack"]:
606                continue
607
608            # For singles, only include if the title matches the track name
609            # (avoid B-sides and bonus tracks on unrelated singles)
610            if primary_type == "Single":
611                if not compare_strings(rg.get("title", ""), track_name, strict=False):
612                    continue
613
614            release_date = release.get("date", "") or ""
615            barcode = release.get("barcode") or None
616
617            # Keep the earliest release date per release group
618            if rg_id in seen:
619                existing_rg, existing_date = seen[rg_id]
620                if release_date and (not existing_date or release_date < existing_date):
621                    mb_rg = MusicBrainzReleaseGroup.from_raw(rg)
622                    mb_rg.barcode = barcode or existing_rg.barcode
623                    seen[rg_id] = (mb_rg, release_date)
624                elif barcode and not existing_rg.barcode:
625                    existing_rg.barcode = barcode
626            else:
627                mb_rg = MusicBrainzReleaseGroup.from_raw(rg)
628                mb_rg.barcode = barcode
629                seen[rg_id] = (mb_rg, release_date)
630
631        return list(seen.values())
632
633    async def _earliest_first_release_year(self, release_group_ids: list[str]) -> int | None:
634        """
635        Get the year the oldest of the given release groups was first released.
636
637        :param release_group_ids: MusicBrainz release group IDs to look up.
638        :return: The earliest known first release year, or None if MusicBrainz knows none.
639        """
640        # a recording search never yields more than a handful of release groups, so one
641        # query covers them all
642        safe_ids = (re.sub(LUCENE_SPECIAL, r"\\\1", rg_id) for rg_id in release_group_ids)
643        result = await self._api_client.get_data(
644            "release-group",
645            query=f"rgid:({' OR '.join(safe_ids)})",
646            limit="100",
647        )
648        if not result:
649            return None
650        years = [
651            year
652            for release_group in result.get("release-groups", [])
653            if (year := _release_year(release_group.get("first-release-date") or "")) is not None
654        ]
655        return min(years, default=None)
656
657
658def _is_various_artists_release(release: dict[str, Any]) -> bool:
659    """
660    Return whether a MusicBrainz release is credited to Various Artists.
661
662    :param release: MusicBrainz release dict from a recording search.
663    """
664    # MusicBrainz always credits the Various Artists entity by id, while its display name is
665    # localized and other artists are named after it, so only the id identifies it.
666    return any(
667        (credit.get("artist") or {}).get("id") == VARIOUS_ARTISTS_MBID
668        for credit in release.get("artist-credit") or ()
669    )
670
671
672def _release_year(release_date: str) -> int | None:
673    """
674    Read the year off a MusicBrainz date of any precision.
675
676    :param release_date: MusicBrainz date, as a year, year-month or full date.
677    :return: The year, or None if the date is absent or unparsable.
678    """
679    return int(year) if (year := release_date[:4]).isdigit() else None
680