Bring back album_for_id, track_for_id

Restore album_for_id and track_for_id functions in metadata_plugins to
support data source-specific lookups. These functions accept both an ID
and data_source parameter, enabling plugins like mbsync and missing to
retrieve metadata from the correct source.

Update mbsync and missing plugins to use the restored functions with
explicit data_source parameters. Add data_source validation to prevent
lookups when the source is not specified.

Add get_metadata_source helper function to retrieve plugins by their
data_source name, cached for performance.

beets/metadata_plugins.py

@@ -19,7 +19,7 @@ from typing_extensions import NotRequired
 from beets.util import cached_classproperty
 from beets.util.id_extractors import extract_release_id
 
-from .plugins import BeetsPlugin, find_plugins, notify_info_yielded
+from .plugins import BeetsPlugin, find_plugins, notify_info_yielded, send
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Sequence
@@ -34,6 +34,14 @@ def find_metadata_source_plugins() -> list[MetadataSourcePlugin]:
     return [p for p in find_plugins() if hasattr(p, "data_source")]  # type: ignore[misc]
 
 
+@cache
+def get_metadata_source(name: str) -> MetadataSourcePlugin | None:
+    """Get metadata source plugin by name."""
+    name = name.lower()
+    plugins = find_metadata_source_plugins()
+    return next((p for p in plugins if p.data_source.lower() == name), None)
+
+
 @notify_info_yielded("albuminfo_received")
 def candidates(*args, **kwargs) -> Iterable[AlbumInfo]:
     """Return matching album candidates from all metadata source plugins."""
@@ -62,6 +70,28 @@ def tracks_for_ids(ids: Sequence[str]) -> Iterable[TrackInfo]:
         yield from plugin.tracks_for_ids(ids)
 
 
+def album_for_id(_id: str, data_source: str) -> AlbumInfo | None:
+    """Get AlbumInfo object for the given ID and data source."""
+    if (plugin := get_metadata_source(data_source)) and (
+        info := plugin.album_for_id(_id)
+    ):
+        send("albuminfo_received", info=info)
+        return info
+
+    return None
+
+
+def track_for_id(_id: str, data_source: str) -> TrackInfo | None:
+    """Get TrackInfo object for the given ID and data source."""
+    if (plugin := get_metadata_source(data_source)) and (
+        info := plugin.track_for_id(_id)
+    ):
+        send("trackinfo_received", info=info)
+        return info
+
+    return None
+
+
 @cache
 def get_penalty(data_source: str | None) -> float:
     """Get the penalty value for the given data source."""

beetsplug/mbsync.py

@@ -72,17 +72,25 @@ class MBSyncPlugin(BeetsPlugin):
         query.
         """
         for item in lib.items(query + ["singleton:true"]):
-            if not item.mb_trackid:
+            if not (track_id := item.mb_trackid):
                 self._log.info(
                     "Skipping singleton with no mb_trackid: {}", item
                 )
                 continue
 
+            if not (data_source := item.get("data_source")):
+                self._log.info(
+                    "Skipping singleton without data source: {}", item
+                )
+                continue
+
             if not (
-                track_info := metadata_plugins.track_for_id(item.mb_trackid)
+                track_info := metadata_plugins.track_for_id(
+                    track_id, data_source
+                )
             ):
                 self._log.info(
-                    "Recording ID not found: {0.mb_trackid} for track {0}", item
+                    "Recording ID not found: {} for track {}", track_id, item
                 )
                 continue
 
@@ -97,15 +105,24 @@ class MBSyncPlugin(BeetsPlugin):
         """
         # Process matching albums.
         for album in lib.albums(query):
-            if not album.mb_albumid:
+            if not (album_id := album.mb_albumid):
                 self._log.info("Skipping album with no mb_albumid: {}", album)
                 continue
 
             if not (
-                album_info := metadata_plugins.album_for_id(album.mb_albumid)
+                data_source := album.get("data_source")
+                or album.items()[0].get("data_source")
+            ):
+                self._log.info("Skipping album without data source: {}", album)
+                continue
+
+            if not (
+                album_info := metadata_plugins.album_for_id(
+                    album_id, data_source
+                )
             ):
                 self._log.info(
-                    "Release ID {0.mb_albumid} not found for album {0}", album
+                    "Release ID {} not found for album {}", album_id, album
                 )
                 continue
 

beetsplug/missing.py

@@ -219,10 +219,17 @@ class MissingPlugin(BeetsPlugin):
         if len(album.items()) == album.albumtotal:
             return
 
-        item_mbids = {x.mb_trackid for x in album.items()}
         # fetch missing items
         # TODO: Implement caching that without breaking other stuff
-        if album_info := metadata_plugins.album_for_id(album.mb_albumid):
+        if (
+            data_source := album.get("data_source")
+            or album.items()[0].get("data_source")
+        ) and (
+            album_info := metadata_plugins.album_for_id(
+                album.mb_albumid, data_source
+            )
+        ):
+            item_mbids = {x.mb_trackid for x in album.items()}
             for track_info in album_info.tracks:
                 if track_info.track_id not in item_mbids:
                     self._log.debug(

test/plugins/test_mbsync.py

@@ -45,10 +45,15 @@ class MbsyncCliTest(PluginTestCase):
             album="old album",
             mb_albumid="album id",
             mb_trackid="track id",
+            data_source="data_source",
         )
         self.lib.add_album([album_item])
 
-        singleton = Item(title="old title", mb_trackid="singleton id")
+        singleton = Item(
+            title="old title",
+            mb_trackid="singleton id",
+            data_source="data_source",
+        )
         self.lib.add(singleton)
 
         self.run_command("mbsync")