Take data source into account when deciding duplicate candidates

Author: snejus

beets/autotag/hooks.py

@@ -58,6 +58,16 @@ def __hash__(self) -> int:  # type: ignore[override]
 class Info(AttrDict[Any]):
     """Container for metadata about a musical entity."""
 
+    Identifier = tuple[str | None, str | None]
+
+    @property
+    def id(self) -> str | None:
+        raise NotImplementedError
+
+    @property
+    def identifier(self) -> Identifier:
+        return (self.data_source, self.id)
+
     @cached_property
     def name(self) -> str:
         raise NotImplementedError
@@ -103,6 +113,10 @@ class AlbumInfo(Info):
     user items, and later to drive tagging decisions once selected.
     """
 
+    @property
+    def id(self) -> str | None:
+        return self.album_id
+
     @cached_property
     def name(self) -> str:
         return self.album or ""
@@ -179,6 +193,10 @@ class TrackInfo(Info):
     stand alone for singleton matching.
     """
 
+    @property
+    def id(self) -> str | None:
+        return self.track_id
+
     @cached_property
     def name(self) -> str:
         return self.title or ""

beets/autotag/match.py

@@ -19,7 +19,7 @@
 from __future__ import annotations
 
 from enum import IntEnum
-from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar
+from typing import TYPE_CHECKING, NamedTuple, TypeVar
 
 import lap
 import numpy as np
@@ -35,6 +35,11 @@
 
     from beets.library import Item
 
+    from .hooks import Info
+
+    AnyMatch = TypeVar("AnyMatch", TrackMatch, AlbumMatch)
+    Candidates = dict[Info.Identifier, AnyMatch]
+
 # Global logger.
 log = logging.getLogger("beets")
 
@@ -179,33 +184,33 @@ def _recommendation(
     return rec
 
 
-AnyMatch = TypeVar("AnyMatch", TrackMatch, AlbumMatch)
-
-
 def _sort_candidates(candidates: Iterable[AnyMatch]) -> Sequence[AnyMatch]:
     """Sort candidates by distance."""
     return sorted(candidates, key=lambda match: match.distance)
 
 
 def _add_candidate(
     items: Sequence[Item],
-    results: dict[Any, AlbumMatch],
+    results: Candidates[AlbumMatch],
     info: AlbumInfo,
 ):
     """Given a candidate AlbumInfo object, attempt to add the candidate
     to the output dictionary of AlbumMatch objects. This involves
     checking the track count, ordering the items, checking for
     duplicates, and calculating the distance.
     """
-    log.debug("Candidate: {0.artist} - {0.album} ({0.album_id})", info)
+    log.debug(
+        "Candidate: {0.artist} - {0.album} ({0.album_id}) from {0.data_source}",
+        info,
+    )
 
     # Discard albums with zero tracks.
     if not info.tracks:
         log.debug("No tracks.")
         return
 
     # Prevent duplicates.
-    if info.album_id and info.album_id in results:
+    if info.album_id and info.identifier in results:
         log.debug("Duplicate.")
         return
 
@@ -233,7 +238,7 @@ def _add_candidate(
             return
 
     log.debug("Success. Distance: {}", dist)
-    results[info.album_id] = hooks.AlbumMatch(
+    results[info.identifier] = hooks.AlbumMatch(
         dist, info, dict(item_info_pairs), extra_items, extra_tracks
     )
 
@@ -268,15 +273,15 @@ def tag_album(
     log.debug("Tagging {} - {}", cur_artist, cur_album)
 
     # The output result, keys are the MB album ID.
-    candidates: dict[Any, AlbumMatch] = {}
+    candidates: Candidates[AlbumMatch] = {}
 
     # Search by explicit ID.
     if search_ids:
         for search_id in search_ids:
             log.debug("Searching for album ID: {}", search_id)
             for _info in metadata_plugins.albums_for_ids(search_id):
                 _add_candidate(items, candidates, _info)
-                if opt_candidate := candidates.get(_info.album_id):
+                if opt_candidate := candidates.get(_info.identifier):
                     plugins.send("album_matched", match=opt_candidate)
 
     # Use existing metadata or text search.
@@ -320,7 +325,7 @@ def tag_album(
             items, search_artist, search_name, va_likely
         ):
             _add_candidate(items, candidates, matched_candidate)
-            if opt_candidate := candidates.get(matched_candidate.album_id):
+            if opt_candidate := candidates.get(matched_candidate.identifier):
                 plugins.send("album_matched", match=opt_candidate)
 
     log.debug("Evaluating {} candidates.", len(candidates))
@@ -345,7 +350,7 @@ def tag_item(
     """
     # Holds candidates found so far: keys are MBIDs; values are
     # (distance, TrackInfo) pairs.
-    candidates = {}
+    candidates: Candidates[TrackMatch] = {}
     rec: Recommendation | None = None
 
     # First, try matching by the external source ID.
@@ -355,7 +360,7 @@ def tag_item(
             log.debug("Searching for track ID: {}", trackid)
             for info in metadata_plugins.tracks_for_ids(trackid):
                 dist = track_distance(item, info, incl_artist=True)
-                candidates[info.track_id] = hooks.TrackMatch(dist, info)
+                candidates[info.identifier] = hooks.TrackMatch(dist, info)
 
             # If this is a good match, then don't keep searching.
             rec = _recommendation(_sort_candidates(candidates.values()))
@@ -381,7 +386,7 @@ def tag_item(
         item, search_artist, search_name
     ):
         dist = track_distance(item, track_info, incl_artist=True)
-        candidates[track_info.track_id] = hooks.TrackMatch(dist, track_info)
+        candidates[track_info.identifier] = hooks.TrackMatch(dist, track_info)
 
     # Sort by distance and return with recommendation.
     log.debug("Found {} candidates.", len(candidates))

test/autotag/test_match.py

@@ -162,41 +162,21 @@ def check_proposal(self, proposal):
         assert len(sources) == 2
         assert set(sources) == {"Discogs", "Deezer"}
 
-    @pytest.mark.xfail(
-        reason="Same ID from different sources is considered a duplicate (#6181)",
-        raises=AssertionError,
-        strict=True,
-    )
     def test_search_album_ids(self, shared_album_id):
         _, _, proposal = match.tag_album([Item()], search_ids=[shared_album_id])
 
         self.check_proposal(proposal)
 
-    @pytest.mark.xfail(
-        reason="Same ID from different sources is considered a duplicate (#6181)",
-        raises=AssertionError,
-        strict=True,
-    )
     def test_search_album_current_id(self, shared_album_id):
         _, _, proposal = match.tag_album([Item(mb_albumid=shared_album_id)])
 
         self.check_proposal(proposal)
 
-    @pytest.mark.xfail(
-        reason="The last match wins",
-        raises=AssertionError,
-        strict=True,
-    )
     def test_search_track_ids(self, shared_track_id):
         proposal = match.tag_item(Item(), search_ids=[shared_track_id])
 
         self.check_proposal(proposal)
 
-    @pytest.mark.xfail(
-        reason="The last match wins",
-        raises=AssertionError,
-        strict=True,
-    )
     def test_search_track_current_id(self, shared_track_id):
         proposal = match.tag_item(Item(mb_trackid=shared_track_id))