mbsync / missing plugins: consider all metadata backends

beets/autotag/hooks.py

@@ -56,7 +56,7 @@
         return id(self)
 
 
 class AlbumInfo(AttrDict):
     """Describes a canonical release that may be used to match a release
     in the library. Consists of these data members:
 
@@ -166,7 +166,7 @@
         return dupe
 
 
 class TrackInfo(AttrDict):
     """Describes a canonical track present on a release. Appears as part
     of an AlbumInfo's ``tracks`` list. Consists of these data members:
 
@@ -602,8 +602,7 @@
     if the ID is not found.
     """
     try:
-        album = mb.album_for_id(release_id)
-        if album:
+        if album := mb.album_for_id(release_id):
             plugins.send("albuminfo_received", info=album)
         return album
     except mb.MusicBrainzAPIError as exc:
@@ -616,38 +615,25 @@
     if the ID is not found.
     """
     try:
-        track = mb.track_for_id(recording_id)
-        if track:
+        if track := mb.track_for_id(recording_id):
             plugins.send("trackinfo_received", info=track)
         return track
     except mb.MusicBrainzAPIError as exc:
         exc.log(log)
         return None
 
 
-def albums_for_id(album_id: str) -> Iterable[AlbumInfo]:
-    """Get a list of albums for an ID."""
-    a = album_for_mbid(album_id)
-    if a:
-        yield a
-    for a in plugins.album_for_id(album_id):
-        if a:
-            plugins.send("albuminfo_received", info=a)
-            yield a
-
-
-def tracks_for_id(track_id: str) -> Iterable[TrackInfo]:
-    """Get a list of tracks for an ID."""
-    t = track_for_mbid(track_id)
-    if t:
-        yield t
-    for t in plugins.track_for_id(track_id):
-        if t:
-            plugins.send("trackinfo_received", info=t)
-            yield t
+def album_for_id(_id: str) -> AlbumInfo | None:
+    """Get AlbumInfo object for the given ID string."""
+    return album_for_mbid(_id) or plugins.album_for_id(_id)
+
+
+def track_for_id(_id: str) -> TrackInfo | None:
+    """Get TrackInfo object for the given ID string."""
+    return track_for_mbid(_id) or plugins.track_for_id(_id)
 
 
 def invoke_mb(call_func: Callable, *args):
     try:
         return call_func(*args)
     except mb.MusicBrainzAPIError as exc:
@@ -661,8 +647,8 @@
     artist: str,
     album: str,
     va_likely: bool,
     extra_tags: dict,
 ) -> Iterable[tuple]:
     """Search for album matches. ``items`` is a list of Item objects
     that make up the album. ``artist`` and ``album`` are the respective
     names (strings), which may be derived from the item list or may be
@@ -690,7 +676,7 @@
 
 
 @plugins.notify_info_yielded("trackinfo_received")
 def item_candidates(item: Item, artist: str, title: str) -> Iterable[tuple]:
     """Search for item matches. ``item`` is the Item to be matched.
     ``artist`` and ``title`` are strings and either reflect the item or
     are specified by the user.

beets/autotag/match.py

@@ -510,8 +510,8 @@ def tag_album(
     if search_ids:
         for search_id in search_ids:
             log.debug("Searching for album ID: {0}", search_id)
-            for album_info_for_id in hooks.albums_for_id(search_id):
-                _add_candidate(items, candidates, album_info_for_id)
+            if info := hooks.album_for_id(search_id):
+                _add_candidate(items, candidates, info)
 
     # Use existing metadata or text search.
     else:
@@ -590,11 +590,9 @@ def tag_item(
     if trackids:
         for trackid in trackids:
             log.debug("Searching for track ID: {0}", trackid)
-            for track_info in hooks.tracks_for_id(trackid):
-                dist = track_distance(item, track_info, incl_artist=True)
-                candidates[track_info.track_id] = hooks.TrackMatch(
-                    dist, track_info
-                )
+            if info := hooks.track_for_id(trackid):
+                dist = track_distance(item, info, incl_artist=True)
+                candidates[info.track_id] = hooks.TrackMatch(dist, info)
                 # If this is a good match, then don't keep searching.
                 rec = _recommendation(_sort_candidates(candidates.values()))
                 if (

beets/plugins.py

@@ -14,18 +14,25 @@
 
 """Support for beets plugins."""
 
+from __future__ import annotations
+
 import abc
 import inspect
 import re
 import traceback
 from collections import defaultdict
 from functools import wraps
+from typing import TYPE_CHECKING
 
 import mediafile
 
 import beets
 from beets import logging
 
+if TYPE_CHECKING:
+    from beets.autotag.hooks import AlbumInfo, TrackInfo
+
+
 PLUGIN_NAMESPACE = "beetsplug"
 
 # Plugins using the Last.fm API can share the same API key.
@@ -290,7 +297,7 @@ def load_plugins(names=()):
             )
 
 
-_instances = {}
+_instances: dict[type[BeetsPlugin], BeetsPlugin] = {}
 
 
 def find_plugins():
@@ -397,20 +404,30 @@ def item_candidates(item, artist, title):
         yield from plugin.item_candidates(item, artist, title)
 
 
-def album_for_id(album_id):
-    """Get AlbumInfo objects for a given ID string."""
+def album_for_id(_id: str) -> AlbumInfo | None:
+    """Get AlbumInfo object for the given ID string.
+
+    A single ID can yield just a single album, so we return the first match.
+    """
     for plugin in find_plugins():
-        album = plugin.album_for_id(album_id)
-        if album:
-            yield album
+        if info := plugin.album_for_id(_id):
+            send("albuminfo_received", info=info)
+            return info
 
+    return None
 
-def track_for_id(track_id):
-    """Get TrackInfo objects for a given ID string."""
+
+def track_for_id(_id: str) -> TrackInfo | None:
+    """Get TrackInfo object for the given ID string.
+
+    A single ID can yield just a single track, so we return the first match.
+    """
     for plugin in find_plugins():
-        track = plugin.track_for_id(track_id)
-        if track:
-            yield track
+        if info := plugin.track_for_id(_id):
+            send("trackinfo_received", info=info)
+            return info
+
+    return None
 
 
 def template_funcs():

beetsplug/deezer.py

@@ -71,7 +71,7 @@ def fetch_data(self, url):
             self._log.error("Error fetching data from {}\n Error: {}", url, e)
             return None
         if "error" in data:
-            self._log.error("Deezer API error: {}", data["error"]["message"])
+            self._log.debug("Deezer API error: {}", data["error"]["message"])
             return None
         return data
 

beetsplug/discogs.py

@@ -29,7 +29,6 @@
 
 import confuse
 from discogs_client import Client, Master, Release
-from discogs_client import __version__ as dc_string
 from discogs_client.exceptions import DiscogsAPIError
 from requests.exceptions import ConnectionError
 from typing_extensions import TypedDict
@@ -64,7 +63,6 @@ class ReleaseFormat(TypedDict):
 class DiscogsPlugin(BeetsPlugin):
     def __init__(self):
         super().__init__()
-        self.check_discogs_client()
         self.config.add(
             {
                 "apikey": API_KEY,
@@ -80,24 +78,7 @@ def __init__(self):
         self.config["apikey"].redact = True
         self.config["apisecret"].redact = True
         self.config["user_token"].redact = True
-        self.discogs_client = None
-        self.register_listener("import_begin", self.setup)
-
-    def check_discogs_client(self):
-        """Ensure python3-discogs-client version >= 2.3.15"""
-        dc_min_version = [2, 3, 15]
-        dc_version = [int(elem) for elem in dc_string.split(".")]
-        min_len = min(len(dc_version), len(dc_min_version))
-        gt_min = [
-            (elem > elem_min)
-            for elem, elem_min in zip(
-                dc_version[:min_len], dc_min_version[:min_len]
-            )
-        ]
-        if True not in gt_min:
-            self._log.warning(
-                "python3-discogs-client version should be >= 2.3.15"
-            )
+        self.setup()
 
     def setup(self, session=None):
         """Create the `discogs_client` field. Authenticate if necessary."""
@@ -184,8 +165,7 @@ def candidates(self, items, artist, album, va_likely, extra_tags=None):
 
         if not album and not artist:
             self._log.debug(
-                "Skipping Discogs query. Files missing album and "
-                "artist tags."
+                "Skipping Discogs query. Files missing album and artist tags."
             )
             return []
 
@@ -259,7 +239,7 @@ def item_candidates(self, item, artist, title):
 
         if not artist and not title:
             self._log.debug(
-                "Skipping Discogs query. File missing artist and " "title tags."
+                "Skipping Discogs query. File missing artist and title tags."
             )
             return []
 

beetsplug/mbsync.py

@@ -12,17 +12,14 @@
 # The above copyright notice and this permission notice shall be
 # included in all copies or substantial portions of the Software.
 
-"""Update library's tags using MusicBrainz."""
+"""Synchronise library metadata with metadata source backends."""
 
-import re
 from collections import defaultdict
 
 from beets import autotag, library, ui, util
 from beets.autotag import hooks
 from beets.plugins import BeetsPlugin, apply_item_changes
 
-MBID_REGEX = r"(\d|\w){8}-(\d|\w){4}-(\d|\w){4}-(\d|\w){4}-(\d|\w){12}"
-
 
 class MBSyncPlugin(BeetsPlugin):
     def __init__(self):
@@ -84,17 +81,7 @@ def singletons(self, lib, query, move, pretend, write):
                 )
                 continue
 
-            # Do we have a valid MusicBrainz track ID?
-            if not re.match(MBID_REGEX, item.mb_trackid):
-                self._log.info(
-                    "Skipping singleton with invalid mb_trackid:" + " {0}",
-                    item_formatted,
-                )
-                continue
-
-            # Get the MusicBrainz recording info.
-            track_info = hooks.track_for_mbid(item.mb_trackid)
-            if not track_info:
+            if not (track_info := hooks.track_for_id(item.mb_trackid)):
                 self._log.info(
                     "Recording ID not found: {0} for track {0}",
                     item.mb_trackid,
@@ -121,18 +108,7 @@ def albums(self, lib, query, move, pretend, write):
                 continue
 
             items = list(a.items())
-
-            # Do we have a valid MusicBrainz album ID?
-            if not re.match(MBID_REGEX, a.mb_albumid):
-                self._log.info(
-                    "Skipping album with invalid mb_albumid: {0}",
-                    album_formatted,
-                )
-                continue
-
-            # Get the MusicBrainz album information.
-            album_info = hooks.album_for_mbid(a.mb_albumid)
-            if not album_info:
+            if not (album_info := hooks.album_for_id(a.mb_albumid)):
                 self._log.info(
                     "Release ID {0} not found for album {1}",
                     a.mb_albumid,
@@ -179,7 +155,7 @@ def albums(self, lib, query, move, pretend, write):
             with lib.transaction():
                 autotag.apply_metadata(album_info, mapping)
                 changed = False
-                # Find any changed item to apply MusicBrainz changes to album.
+                # Find any changed item to apply changes to album.
                 any_changed_item = items[0]
                 for item in items:
                     item_changed = ui.show_model_changes(item)