diff --git a/beetsplug/_utils/musicbrainz.py b/beetsplug/_utils/musicbrainz.py index 3327269b2..63ffd4aa3 100644 --- a/beetsplug/_utils/musicbrainz.py +++ b/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) diff --git a/beetsplug/listenbrainz.py b/beetsplug/listenbrainz.py index 3729001b1..d054a00cc 100644 --- a/beetsplug/listenbrainz.py +++ b/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") diff --git a/beetsplug/musicbrainz.py b/beetsplug/musicbrainz.py index 38097b2ce..990f21351 100644 --- a/beetsplug/musicbrainz.py +++ b/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, diff --git a/docs/plugins/listenbrainz.rst b/docs/plugins/listenbrainz.rst index 17926e878..ceff0e800 100644 --- a/docs/plugins/listenbrainz.rst +++ b/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: diff --git a/poetry.lock b/poetry.lock index dbd3ecf3d..60cbceebd 100644 --- a/poetry.lock +++ b/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" diff --git a/pyproject.toml b/pyproject.toml index bd46d3026..ed0059610 100644 --- a/pyproject.toml +++ b/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"] diff --git a/test/plugins/test_listenbrainz.py b/test/plugins/test_listenbrainz.py index fa6c4fbab..b94cff219 100644 --- a/test/plugins/test_listenbrainz.py +++ b/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"}], }, )