Update mbpseudo implementation for beets 2.5

beets/autotag/match.py

@@ -24,7 +24,7 @@ from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar
 import lap
 import numpy as np
 
-from beets import config, logging, metadata_plugins
+from beets import config, logging, metadata_plugins, plugins
 from beets.autotag import AlbumInfo, AlbumMatch, TrackInfo, TrackMatch, hooks
 from beets.util import get_most_common_tags
 
@@ -274,12 +274,17 @@ def tag_album(
             log.debug("Searching for album ID: {}", search_id)
             if info := metadata_plugins.album_for_id(search_id):
                 _add_candidate(items, candidates, info)
+                if opt_candidate := candidates.get(info.album_id):
+                    plugins.send("album_matched", match=opt_candidate)
 
     # Use existing metadata or text search.
     else:
         # Try search based on current ID.
         if info := match_by_id(items):
             _add_candidate(items, candidates, info)
+            for candidate in candidates.values():
+                plugins.send("album_matched", match=candidate)
+
             rec = _recommendation(list(candidates.values()))
             log.debug("Album ID match recommendation is {}", rec)
             if candidates and not config["import"]["timid"]:
@@ -313,6 +318,8 @@ def tag_album(
             items, search_artist, search_album, va_likely
         ):
             _add_candidate(items, candidates, matched_candidate)
+            if opt_candidate := candidates.get(matched_candidate.album_id):
+                plugins.send("album_matched", match=opt_candidate)
 
     log.debug("Evaluating {} candidates.", len(candidates))
     # Sort and get the recommendation.

beets/plugins.py

@@ -72,6 +72,7 @@ EventType = Literal[
     "album_imported",
     "album_removed",
     "albuminfo_received",
+    "album_matched",
     "before_choose_candidate",
     "before_item_moved",
     "cli_exit",

beetsplug/mbpseudo.py

@@ -16,6 +16,7 @@
 
 from __future__ import annotations
 
+import traceback
 from copy import deepcopy
 from typing import TYPE_CHECKING, Any, Iterable, Sequence
 
@@ -23,12 +24,13 @@ import musicbrainzngs
 from typing_extensions import override
 
 from beets.autotag.distance import Distance, distance
-from beets.autotag.hooks import AlbumInfo, TrackInfo
+from beets.autotag.hooks import AlbumInfo
 from beets.autotag.match import assign_items
 from beets.plugins import find_plugins
 from beets.util.id_extractors import extract_release_id
 from beetsplug.musicbrainz import (
     RELEASE_INCLUDES,
+    MusicBrainzAPIError,
     MusicBrainzPlugin,
     _merge_pseudo_and_actual_album,
 )
@@ -50,6 +52,7 @@ class MusicBrainzPseudoReleasePlugin(MusicBrainzPlugin):
         self._log.debug("Desired scripts: {0}", self._scripts)
 
         self.register_listener("pluginload", self._on_plugins_loaded)
+        self.register_listener("album_matched", self._adjust_final_album_match)
 
     # noinspection PyMethodMayBeStatic
     def _on_plugins_loaded(self):
@@ -77,13 +80,14 @@ class MusicBrainzPseudoReleasePlugin(MusicBrainzPlugin):
                 items, artist, album, va_likely
             ):
                 if isinstance(album_info, PseudoAlbumInfo):
-                    yield album_info.get_official_release()
                     self._log.debug(
                         "Using {0} release for distance calculations for album {1}",
                         album_info.determine_best_ref(items),
                         album_info.album_id,
                     )
-
+                    yield album_info  # first yield pseudo to give it priority
+                    yield album_info.get_official_release()
+                else:
                     yield album_info
 
     @override
@@ -95,16 +99,26 @@ class MusicBrainzPseudoReleasePlugin(MusicBrainzPlugin):
             return official_release
         elif pseudo_release_ids := self._intercept_mb_release(release):
             album_id = self._extract_id(pseudo_release_ids[0])
+            try:
                 raw_pseudo_release = musicbrainzngs.get_release_by_id(
                     album_id, RELEASE_INCLUDES
                 )
-            pseudo_release = super().album_info(raw_pseudo_release["release"])
+                pseudo_release = super().album_info(
+                    raw_pseudo_release["release"]
+                )
                 return PseudoAlbumInfo(
                     pseudo_release=_merge_pseudo_and_actual_album(
                         pseudo_release, official_release
                     ),
                     official_release=official_release,
-                data_source=self.data_source,
+                    data_source="MusicBrainz",
+                )
+            except musicbrainzngs.MusicBrainzError as exc:
+                raise MusicBrainzAPIError(
+                    exc,
+                    "get pseudo-release by ID",
+                    album_id,
+                    traceback.format_exc(),
                 )
         else:
             return official_release
@@ -153,34 +167,20 @@ class MusicBrainzPseudoReleasePlugin(MusicBrainzPlugin):
         else:
             return None
 
-    @override
-    def album_distance(
-        self,
-        items: Sequence[Item],
-        album_info: AlbumInfo,
-        mapping: dict[Item, TrackInfo],
-    ) -> Distance:
-        """We use this function more like a listener for the extra details we are
-        injecting.
-
-        For instances of ``PseudoAlbumInfo`` whose corresponding ``mapping`` is _not_ an
-        instance of ``ImmutableMapping``, we know at this point that all penalties from
-        the normal auto-tagging flow have been applied, so we can switch to the metadata
-        from the pseudo-release for the final proposal.
-        """
-
+    def _adjust_final_album_match(self, match: AlbumMatch):
+        album_info = match.info
         if isinstance(album_info, PseudoAlbumInfo):
-            if not isinstance(mapping, ImmutableMapping):
+            mapping = match.mapping
             self._log.debug(
                 "Switching {0} to pseudo-release source for final proposal",
                 album_info.album_id,
             )
             album_info.use_pseudo_as_ref()
-                new_mappings, _, _ = assign_items(items, album_info.tracks)
+            new_mappings, _, _ = assign_items(
+                list(mapping.keys()), album_info.tracks
+            )
             mapping.update(new_mappings)
 
-        return super().album_distance(items, album_info, mapping)
-
     @override
     def _extract_id(self, url: str) -> str | None:
         return extract_release_id("MusicBrainz", url)
@@ -218,12 +218,17 @@ class PseudoAlbumInfo(AlbumInfo):
         return self.__dict__["_official_release"]
 
     def determine_best_ref(self, items: Sequence[Item]) -> str:
+        ds = self.data_source
+        self.data_source = None
+
         self.use_pseudo_as_ref()
         pseudo_dist = self._compute_distance(items)
 
         self.use_official_as_ref()
         official_dist = self._compute_distance(items)
 
+        self.data_source = ds
+
         if official_dist < pseudo_dist:
             self.use_official_as_ref()
             return "official"
@@ -233,7 +238,7 @@ class PseudoAlbumInfo(AlbumInfo):
 
     def _compute_distance(self, items: Sequence[Item]) -> Distance:
         mapping, _, _ = assign_items(items, self.tracks)
-        return distance(items, self, ImmutableMapping(mapping))
+        return distance(items, self, mapping)
 
     def use_pseudo_as_ref(self):
         self.__dict__["_pseudo_source"] = True
@@ -258,8 +263,3 @@ class PseudoAlbumInfo(AlbumInfo):
             result[k] = deepcopy(v, memo)
 
         return result
-
-
-class ImmutableMapping(dict[Item, TrackInfo]):
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)