Migrate listenbrainz plugin to use our MusicBrainzAPI implementation

beetsplug/_utils/musicbrainz.py

@@ -8,13 +8,15 @@ from typing import TYPE_CHECKING, Any
 
 from requests_ratelimiter import LimiterMixin
 
-from beets import config
+from beets import config, logging
 
 from .requests import RequestHandler, TimeoutAndRetrySession
 
 if TYPE_CHECKING:
     from .._typing import JSONDict
 
+log = logging.getLogger(__name__)
+
 
 class LimiterTimeoutSession(LimiterMixin, TimeoutAndRetrySession):
     pass
@@ -63,6 +65,26 @@ class MusicBrainzAPI(RequestHandler):
             )
         )
 
+    def search_entity(
+        self, entity: str, filters: dict[str, str], **kwargs
+    ) -> list[JSONDict]:
+        """Search for MusicBrainz entities matching the given filters.
+
+        * Query is constructed by combining the provided filters using AND logic
+        * Each filter key-value pair is formatted as 'key:"value"' unless
+          - 'key' is empty, in which case only the value is used, '"value"'
+          - 'value' is empty, in which case the filter is ignored
+        * Values are lowercased and stripped of whitespace.
+        """
+        query = " AND ".join(
+            ":".join(filter(None, (k, f'"{_v}"')))
+            for k, v in filters.items()
+            if (_v := v.lower().strip())
+        )
+        log.debug("Searching for MusicBrainz {}s with: {!r}", entity, query)
+        kwargs["query"] = query
+        return self.get_entity(entity, **kwargs)[f"{entity}s"]
+
     def get_release(self, id_: str, **kwargs) -> JSONDict:
         return self.get_entity(f"release/{id_}", **kwargs)
 

beetsplug/listenbrainz.py

@@ -2,17 +2,16 @@
 
 import datetime
 
-import musicbrainzngs
 import requests
 
-from beets import __version__, config, ui
+from beets import config, ui
 from beets.plugins import BeetsPlugin
 from beetsplug.lastimport import process_tracks
 
-musicbrainzngs.set_useragent("beets", __version__, "https://beets.io/")
+from ._utils.musicbrainz import MusicBrainzAPIMixin
 
 
-class ListenBrainzPlugin(BeetsPlugin):
+class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin):
     """A Beets plugin for interacting with ListenBrainz."""
 
     ROOT = "http://api.listenbrainz.org/1/"
@@ -131,17 +130,16 @@ class ListenBrainzPlugin(BeetsPlugin):
             )
         return tracks
 
-    def get_mb_recording_id(self, track):
+    def get_mb_recording_id(self, track) -> str | None:
         """Returns the MusicBrainz recording ID for a track."""
-        resp = musicbrainzngs.search_recordings(
-            query=track["track_metadata"].get("track_name"),
-            release=track["track_metadata"].get("release_name"),
-            strict=True,
+        results = self.mb_api.search_entity(
+            "recording",
+            {
+                "": track["track_metadata"].get("track_name"),
+                "release": track["track_metadata"].get("release_name"),
+            },
         )
-        if resp.get("recording-count") == "1":
-            return resp.get("recording-list")[0].get("id")
-        else:
-            return None
+        return next((r["id"] for r in results), None)
 
     def get_playlists_createdfor(self, username):
         """Returns a list of playlists created by a user."""
@@ -209,17 +207,16 @@ class ListenBrainzPlugin(BeetsPlugin):
         track_info = []
         for track in tracks:
             identifier = track.get("identifier")
-            resp = musicbrainzngs.get_recording_by_id(
+            recording = self.mb_api.get_recording(
                 identifier, includes=["releases", "artist-credits"]
             )
-            recording = resp.get("recording")
             title = recording.get("title")
             artist_credit = recording.get("artist-credit", [])
             if artist_credit:
                 artist = artist_credit[0].get("artist", {}).get("name")
             else:
                 artist = None
-            releases = recording.get("release-list", [])
+            releases = recording.get("releases", [])
             if releases:
                 album = releases[0].get("title")
                 date = releases[0].get("date")

beetsplug/musicbrainz.py

@@ -751,17 +751,9 @@ class MusicBrainzPlugin(MusicBrainzAPIMixin, MetadataSourcePlugin):
         using the provided criteria. Handles API errors by converting them into
         MusicBrainzAPIError exceptions with contextual information.
         """
-        query = " AND ".join(
-            f'{k}:"{_v}"'
-            for k, v in filters.items()
-            if (_v := v.lower().strip())
+        return self.mb_api.search_entity(
+            query_type, filters, limit=self.config["search_limit"].get()
         )
-        self._log.debug(
-            "Searching for MusicBrainz {}s with: {!r}", query_type, query
-        )
-        return self.mb_api.get_entity(
-            query_type, query=query, limit=self.config["search_limit"].get()
-        )[f"{query_type}s"]
 
     def candidates(
         self,

docs/plugins/listenbrainz.rst

@@ -6,15 +6,16 @@ ListenBrainz Plugin
 The ListenBrainz plugin for beets allows you to interact with the ListenBrainz
 service.
 
-Installation
-------------
+Configuration
+-------------
 
-To use the ``listenbrainz`` plugin, first enable it in your configuration (see
-:ref:`using-plugins`). Then, install ``beets`` with ``listenbrainz`` extra
+To enable the ListenBrainz plugin, add the following to your beets configuration
+file (config.yaml_):
 
-.. code-block:: bash
+.. code-block:: yaml
 
-    pip install "beets[listenbrainz]"
+    plugins:
+        - listenbrainz
 
 You can then configure the plugin by providing your Listenbrainz token (see
 intructions here_) and username:

poetry.lock

@@ -4180,7 +4180,6 @@ import = ["py7zr", "rarfile"]
 kodiupdate = ["requests"]
 lastgenre = ["pylast"]
 lastimport = ["pylast"]
-listenbrainz = ["musicbrainzngs"]
 lyrics = ["beautifulsoup4", "langdetect", "requests"]
 mbcollection = ["musicbrainzngs"]
 metasync = ["dbus-python"]
@@ -4199,4 +4198,4 @@ web = ["flask", "flask-cors"]
 [metadata]
 lock-version = "2.0"
 python-versions = ">=3.10,<4"
-content-hash = "45c7dc4ec30f4460a09554d0ec0ebcafebff097386e005e29e12830d16d223dd"
+content-hash = "d9141a482e4990a4466a121a59deaeaf46e5613ff0af315f277110935e391e63"

pyproject.toml

@@ -163,7 +163,6 @@ import = ["py7zr", "rarfile"]
 kodiupdate = ["requests"]
 lastgenre = ["pylast"]
 lastimport = ["pylast"]
-listenbrainz = ["musicbrainzngs"]
 lyrics = ["beautifulsoup4", "langdetect", "requests"]
 mbcollection = ["musicbrainzngs"]
 metasync = ["dbus-python"]

test/plugins/test_listenbrainz.py

@@ -6,41 +6,33 @@ from beetsplug.listenbrainz import ListenBrainzPlugin
 
 class TestListenBrainzPlugin(ConfigMixin):
     @pytest.fixture(scope="class")
-    def plugin(self):
+    def plugin(self) -> ListenBrainzPlugin:
         self.config["listenbrainz"]["token"] = "test_token"
         self.config["listenbrainz"]["username"] = "test_user"
         return ListenBrainzPlugin()
 
     @pytest.mark.parametrize(
         "search_response, expected_id",
-        [
-            (
-                {"recording-count": "1", "recording-list": [{"id": "id1"}]},
-                "id1",
-            ),
-            ({"recording-count": "0"}, None),
-        ],
+        [([{"id": "id1"}], "id1"), ([], None)],
         ids=["found", "not_found"],
     )
     def test_get_mb_recording_id(
-        self, monkeypatch, plugin, search_response, expected_id
+        self, plugin, requests_mock, search_response, expected_id
     ):
-        monkeypatch.setattr(
-            "musicbrainzngs.search_recordings", lambda *_, **__: search_response
+        requests_mock.get(
+            "/ws/2/recording", json={"recordings": search_response}
         )
         track = {"track_metadata": {"track_name": "S", "release_name": "A"}}
 
         assert plugin.get_mb_recording_id(track) == expected_id
 
-    def test_get_track_info(self, monkeypatch, plugin):
-        monkeypatch.setattr(
-            "musicbrainzngs.get_recording_by_id",
-            lambda *_, **__: {
-                "recording": {
-                    "title": "T",
-                    "artist-credit": [],
-                    "release-list": [{"title": "Al", "date": "2023-01"}],
-                }
+    def test_get_track_info(self, plugin, requests_mock):
+        requests_mock.get(
+            "/ws/2/recording/id1?inc=releases%2Bartist-credits",
+            json={
+                "title": "T",
+                "artist-credit": [],
+                "releases": [{"title": "Al", "date": "2023-01"}],
             },
         )