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