Backup_Repos/beets
1
0
Fork 0
mirror of https://github.com/beetbox/beets.git synced 2025-12-28 03:22:39 +01:00
Code Issues Projects Releases Wiki Activity

Make musicbrainz plugin talk to musicbrainz directly (#6052)

Browse source 
This PR refactors the MusicBrainz plugin implementation by replacing the
`musicbrainzngs` library with direct HTTP API calls using `requests` and
`requests-ratelimiter`.

**Key Changes:**

- **New utilities module**: Added `beetsplug/_utils/requests.py` with
`TimeoutSession` class and HTTP error handling (`HTTPNotFoundError`,
`CaptchaError`)
- **MusicBrainz API rewrite**: Replaced `musicbrainzngs` dependency with
custom `MusicBrainzAPI` class using direct HTTP requests
- **Rate limiting**: Integrated `requests-ratelimiter` for API rate
limiting instead of `musicbrainzngs.set_rate_limit()`
- **Data structure updates**: Updated field names to match MusicBrainz
JSON API v2 format (e.g., `medium-list` → `media`, `track-list` →
`tracks`)
- **Dependency management**: 
- Made `musicbrainzngs` optional and added it to plugin-specific extras
(`listenbrainz`, `mbcollection`, `missing`, `parentwork`). Updated
plugin docs accordingly.
- Made `requests` a required dependency to ensure backwards
compatibility (ideally, we would make it an optional dependency under
`musicbrainz` extra).
- **Error handling**: Simplified error handling by removing
`MusicBrainzAPIError` wrapper class

**Benefits:**
- Direct control over HTTP requests
- Consistent rate limiting across all network requests
- Better alignment with modern MusicBrainz API responses

The changes maintain backward compatibility while modernizing the
underlying implementation.

Fixes #5553
Fixes #5095
This commit is contained in:
Šarūnas Nejus 2025-12-21 01:08:10 +00:00 committed by GitHub
commit c1904b1f69
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 3327 additions and 1757 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
.git-blame-ignore-revs
View file

@ -49,6 +49,10 @@ f36bc497c8c8f89004f3f6879908d3f0b25123e1
c490ac5810b70f3cf5fd8649669838e8fdb19f4d
# Importer restructure
9147577b2b19f43ca827e9650261a86fb0450cef
# Move functionality under MusicBrainz plugin
529aaac7dced71266c6d69866748a7d044ec20ff
# musicbrainz: reorder methods
5dc6f45110b99f0cc8dbb94251f9b1f6d69583fa
# Copy paste query, types from library to dbcore
1a045c91668c771686f4c871c84f1680af2e944b
# Library restructure (split library.py into multiple modules)

4
.github/workflows/ci.yaml vendored
View file

@ -66,7 +66,7 @@ jobs:
- if: ${{ env.IS_MAIN_PYTHON != 'true' }}
name: Test without coverage
run: |
poetry install --without=lint --extras=autobpm --extras=lyrics --extras=replaygain --extras=reflink --extras=fetchart --extras=chroma --extras=sonosupdate
poetry install --without=lint --extras=autobpm --extras=lyrics --extras=replaygain --extras=reflink --extras=fetchart --extras=chroma --extras=sonosupdate --extras=parentwork
poe test
- if: ${{ env.IS_MAIN_PYTHON == 'true' }}
@ -74,7 +74,7 @@ jobs:
env:
LYRICS_UPDATED: ${{ steps.lyrics-update.outputs.any_changed }}
run: |
poetry install --extras=autobpm --extras=lyrics --extras=docs --extras=replaygain --extras=reflink --extras=fetchart --extras=chroma --extras=sonosupdate
poetry install --extras=autobpm --extras=lyrics --extras=docs --extras=replaygain --extras=reflink --extras=fetchart --extras=chroma --extras=sonosupdate --extras=parentwork
poe docs
poe test-with-coverage

175
beetsplug/_utils/requests.py Normal file
View file

@ -0,0 +1,175 @@
from __future__ import annotations
import atexit
import threading
from contextlib import contextmanager
from functools import cached_property
from http import HTTPStatus
from typing import TYPE_CHECKING, Any, ClassVar, Generic, Protocol, TypeVar
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from beets import __version__
if TYPE_CHECKING:
from collections.abc import Iterator
class BeetsHTTPError(requests.exceptions.HTTPError):
STATUS: ClassVar[HTTPStatus]
def __init__(self, *args, **kwargs) -> None:
super().__init__(
f"HTTP Error: {self.STATUS.value} {self.STATUS.phrase}",
*args,
**kwargs,
)
class HTTPNotFoundError(BeetsHTTPError):
STATUS = HTTPStatus.NOT_FOUND
class Closeable(Protocol):
"""Protocol for objects that have a close method."""
def close(self) -> None: ...
C = TypeVar("C", bound=Closeable)
class SingletonMeta(type, Generic[C]):
"""Metaclass ensuring a single shared instance per class.
Creates one instance per class type on first instantiation, reusing it
for all subsequent calls. Automatically registers cleanup on program exit
for proper resource management.
"""
_instances: ClassVar[dict[type[Any], Any]] = {}
_lock: ClassVar[threading.Lock] = threading.Lock()
def __call__(cls, *args: Any, **kwargs: Any) -> C:
if cls not in cls._instances:
with cls._lock:
if cls not in SingletonMeta._instances:
instance = super().__call__(*args, **kwargs)
SingletonMeta._instances[cls] = instance
atexit.register(instance.close)
return SingletonMeta._instances[cls]
class TimeoutAndRetrySession(requests.Session, metaclass=SingletonMeta):
"""HTTP session with sensible defaults.
* default beets User-Agent header
* default request timeout
* automatic retries on transient connection errors
* raises exceptions for HTTP error status codes
"""
def __init__(self, *args, **kwargs) -> None:
super().__init__(*args, **kwargs)
self.headers["User-Agent"] = f"beets/{__version__} https://beets.io/"
retry = Retry(connect=2, total=2, backoff_factor=1)
adapter = HTTPAdapter(max_retries=retry)
self.mount("https://", adapter)
self.mount("http://", adapter)
def request(self, *args, **kwargs):
"""Execute HTTP request with automatic timeout and status validation.
Ensures all requests have a timeout (defaults to 10 seconds) and raises
an exception for HTTP error status codes.
"""
kwargs.setdefault("timeout", 10)
r = super().request(*args, **kwargs)
r.raise_for_status()
return r
class RequestHandler:
"""Manages HTTP requests with custom error handling and session management.
Provides a reusable interface for making HTTP requests with automatic
conversion of standard HTTP errors to beets-specific exceptions. Supports
custom session types and error mappings that can be overridden by
subclasses.
Usage:
Subclass and override :class:`RequestHandler.session_type`,
:class:`RequestHandler.explicit_http_errors` or
:class:`RequestHandler.status_to_error()` to customize behavior.
Use
* :class:`RequestHandler.get_json()` to get JSON response data
* :class:`RequestHandler.get()` to get HTTP response object
* :class:`RequestHandler.request()` to invoke arbitrary HTTP methods
Feel free to define common methods that are used in multiple plugins.
"""
explicit_http_errors: ClassVar[list[type[BeetsHTTPError]]] = [
HTTPNotFoundError
]
def create_session(self) -> TimeoutAndRetrySession:
"""Create a new HTTP session instance.
Can be overridden by subclasses to provide custom session types.
"""
return TimeoutAndRetrySession()
@cached_property
def session(self) -> TimeoutAndRetrySession:
"""Lazily initialize and cache the HTTP session."""
return self.create_session()
def status_to_error(
self, code: int
) -> type[requests.exceptions.HTTPError] | None:
"""Map HTTP status codes to beets-specific exception types.
Searches the configured explicit HTTP errors for a matching status code.
Returns None if no specific error type is registered for the given code.
"""
return next(
(e for e in self.explicit_http_errors if e.STATUS == code), None
)
@contextmanager
def handle_http_error(self) -> Iterator[None]:
"""Convert standard HTTP errors to beets-specific exceptions.
Wraps operations that may raise HTTPError, automatically translating
recognized status codes into their corresponding beets exception types.
Unrecognized errors are re-raised unchanged.
"""
try:
yield
except requests.exceptions.HTTPError as e:
if beets_error := self.status_to_error(e.response.status_code):
raise beets_error(response=e.response) from e
raise
def request(self, *args, **kwargs) -> requests.Response:
"""Perform HTTP request using the session with automatic error handling.
Delegates to the underlying session method while converting recognized
HTTP errors to beets-specific exceptions through the error handler.
"""
with self.handle_http_error():
return self.session.request(*args, **kwargs)
def get(self, *args, **kwargs) -> requests.Response:
"""Perform HTTP GET request with automatic error handling."""
return self.request("get", *args, **kwargs)
def get_json(self, *args, **kwargs):
"""Fetch and parse JSON data from an HTTP endpoint."""
return self.get(*args, **kwargs).json()

91
beetsplug/lyrics.py
View file

@ -16,7 +16,6 @@
from __future__ import annotations
import atexit
import itertools
import math
import re
@ -25,7 +24,6 @@ from contextlib import contextmanager, suppress
from dataclasses import dataclass
from functools import cached_property, partial, total_ordering
from html import unescape
from http import HTTPStatus
from itertools import groupby
from pathlib import Path
from typing import TYPE_CHECKING, NamedTuple
@ -36,14 +34,17 @@ import requests
from bs4 import BeautifulSoup
from unidecode import unidecode
import beets
from beets import plugins, ui
from beets.autotag.distance import string_dist
from beets.util.config import sanitize_choices
from ._utils.requests import HTTPNotFoundError, RequestHandler
if TYPE_CHECKING:
from collections.abc import Iterable, Iterator
import confuse
from beets.importer import ImportTask
from beets.library import Item, Library
from beets.logging import BeetsLogger as Logger
@ -56,41 +57,12 @@ if TYPE_CHECKING:
TranslatorAPI,
)
USER_AGENT = f"beets/{beets.__version__}"
INSTRUMENTAL_LYRICS = "[Instrumental]"
class NotFoundError(requests.exceptions.HTTPError):
pass
class CaptchaError(requests.exceptions.HTTPError):
pass
class TimeoutSession(requests.Session):
def request(self, *args, **kwargs):
"""Wrap the request method to raise an exception on HTTP errors."""
kwargs.setdefault("timeout", 10)
r = super().request(*args, **kwargs)
if r.status_code == HTTPStatus.NOT_FOUND:
raise NotFoundError("HTTP Error: Not Found", response=r)
if 300 <= r.status_code < 400:
raise CaptchaError("Captcha is required", response=r)
r.raise_for_status()
return r
r_session = TimeoutSession()
r_session.headers.update({"User-Agent": USER_AGENT})
@atexit.register
def close_session():
"""Close the requests session on shut down."""
r_session.close()
def __init__(self, *args, **kwargs) -> None:
super().__init__("Captcha is required", *args, **kwargs)
# Utilities.
@ -186,9 +158,18 @@ def slug(text: str) -> str:
return re.sub(r"\W+", "-", unidecode(text).lower().strip()).strip("-")
class RequestHandler:
class LyricsRequestHandler(RequestHandler):
_log: Logger
def status_to_error(self, code: int) -> type[requests.HTTPError] | None:
if err := super().status_to_error(code):
return err
if 300 <= code < 400:
return CaptchaError
return None
def debug(self, message: str, *args) -> None:
"""Log a debug message with the class name."""
self._log.debug(f"{self.__class__.__name__}: {message}", *args)
@ -208,7 +189,7 @@ class RequestHandler:
return f"{url}?{urlencode(params)}"
def fetch_text(
def get_text(
self, url: str, params: JSONDict | None = None, **kwargs
) -> str:
"""Return text / HTML data from the given URL.
@ -218,21 +199,21 @@ class RequestHandler:
"""
url = self.format_url(url, params)
self.debug("Fetching HTML from {}", url)
r = r_session.get(url, **kwargs)
r = self.get(url, **kwargs)
r.encoding = None
return r.text
def fetch_json(self, url: str, params: JSONDict | None = None, **kwargs):
def get_json(self, url: str, params: JSONDict | None = None, **kwargs):
"""Return JSON data from the given URL."""
url = self.format_url(url, params)
self.debug("Fetching JSON from {}", url)
return r_session.get(url, **kwargs).json()
return super().get_json(url, **kwargs)
def post_json(self, url: str, params: JSONDict | None = None, **kwargs):
"""Send POST request and return JSON response."""
url = self.format_url(url, params)
self.debug("Posting JSON to {}", url)
return r_session.post(url, **kwargs).json()
return self.request("post", url, **kwargs).json()
@contextmanager
def handle_request(self) -> Iterator[None]:
@ -251,8 +232,10 @@ class BackendClass(type):
return cls.__name__.lower()
class Backend(RequestHandler, metaclass=BackendClass):
def __init__(self, config, log):
class Backend(LyricsRequestHandler, metaclass=BackendClass):
config: confuse.Subview
def __init__(self, config: confuse.Subview, log: Logger) -> None:
self._log = log
self.config = config
@ -356,10 +339,10 @@ class LRCLib(Backend):
if album:
get_params["album_name"] = album
yield self.fetch_json(self.SEARCH_URL, params=base_params)
yield self.get_json(self.SEARCH_URL, params=base_params)
with suppress(NotFoundError):
yield [self.fetch_json(self.GET_URL, params=get_params)]
with suppress(HTTPNotFoundError):
yield [self.get_json(self.GET_URL, params=get_params)]
@classmethod
def pick_best_match(cls, lyrics: Iterable[LRCLyrics]) -> LRCLyrics | None:
@ -407,7 +390,7 @@ class MusiXmatch(Backend):
def fetch(self, artist: str, title: str, *_) -> tuple[str, str] | None:
url = self.build_url(artist, title)
html = self.fetch_text(url)
html = self.get_text(url)
if "We detected that your IP is blocked" in html:
self.warn("Failed: Blocked IP address")
return None
@ -532,7 +515,7 @@ class SearchBackend(SoupMixin, Backend):
def fetch(self, artist: str, title: str, *_) -> tuple[str, str] | None:
"""Fetch lyrics for the given artist and title."""
for result in self.get_results(artist, title):
if (html := self.fetch_text(result.url)) and (
if (html := self.get_text(result.url)) and (
lyrics := self.scrape(html)
):
return lyrics, result.url
@ -562,7 +545,7 @@ class Genius(SearchBackend):
return {"Authorization": f"Bearer {self.config['genius_api_key']}"}
def search(self, artist: str, title: str) -> Iterable[SearchResult]:
search_data: GeniusAPI.Search = self.fetch_json(
search_data: GeniusAPI.Search = self.get_json(
self.SEARCH_URL,
params={"q": f"{artist} {title}"},
headers=self.headers,
@ -591,7 +574,7 @@ class Tekstowo(SearchBackend):
return self.SEARCH_URL.format(quote_plus(unidecode(artistitle)))
def search(self, artist: str, title: str) -> Iterable[SearchResult]:
if html := self.fetch_text(self.build_url(title, artist)):
if html := self.get_text(self.build_url(title, artist)):
soup = self.get_soup(html)
for tag in soup.select("div[class=flex-group] > a[title*=' - ']"):
artist, title = str(tag["title"]).split(" - ", 1)
@ -657,12 +640,12 @@ class Google(SearchBackend):
html = Html.remove_ads(super().pre_process_html(html))
return Html.remove_formatting(Html.merge_paragraphs(html))
def fetch_text(self, *args, **kwargs) -> str:
def get_text(self, *args, **kwargs) -> str:
"""Handle an error so that we can continue with the next URL."""
kwargs.setdefault("allow_redirects", False)
with self.handle_request():
try:
return super().fetch_text(*args, **kwargs)
return super().get_text(*args, **kwargs)
except CaptchaError:
self.ignored_domains.add(urlparse(args[0]).netloc)
raise
@ -718,7 +701,7 @@ class Google(SearchBackend):
"excludeTerms": ", ".join(self.EXCLUDE_PAGES),
}
data: GoogleCustomSearchAPI.Response = self.fetch_json(
data: GoogleCustomSearchAPI.Response = self.get_json(
self.SEARCH_URL, params=params
)
for item in data.get("items", []):
@ -743,7 +726,7 @@ class Google(SearchBackend):
@dataclass
class Translator(RequestHandler):
class Translator(LyricsRequestHandler):
TRANSLATE_URL = "https://api.cognitive.microsofttranslator.com/translate"
LINE_PARTS_RE = re.compile(r"^(\[\d\d:\d\d.\d\d\]|) *(.*)$")
SEPARATOR = " | "
@ -953,7 +936,7 @@ class RestFiles:
ui.print_(textwrap.dedent(text))
class LyricsPlugin(RequestHandler, plugins.BeetsPlugin):
class LyricsPlugin(LyricsRequestHandler, plugins.BeetsPlugin):
BACKEND_BY_NAME = {
b.name: b for b in [LRCLib, Google, Genius, Tekstowo, MusiXmatch]
}

56
beetsplug/mbpseudo.py
View file

@ -17,12 +17,10 @@
from __future__ import annotations
import itertools
import traceback
from copy import deepcopy
from typing import TYPE_CHECKING, Any
import mediafile
import musicbrainzngs
from typing_extensions import override
from beets import config
@ -32,8 +30,6 @@ from beets.autotag.match import assign_items
from beets.plugins import find_plugins
from beets.util.id_extractors import extract_release_id
from beetsplug.musicbrainz import (
RELEASE_INCLUDES,
MusicBrainzAPIError,
MusicBrainzPlugin,
_merge_pseudo_and_actual_album,
_preferred_alias,
@ -53,8 +49,6 @@ class MusicBrainzPseudoReleasePlugin(MusicBrainzPlugin):
def __init__(self) -> None:
super().__init__()
self._release_getter = musicbrainzngs.get_release_by_id
self.config.add(
{
"scripts": [],
@ -143,33 +137,25 @@ class MusicBrainzPseudoReleasePlugin(MusicBrainzPlugin):
if release.get("status") == _STATUS_PSEUDO:
return official_release
elif pseudo_release_ids := self._intercept_mb_release(release):
album_id = self._extract_id(pseudo_release_ids[0])
try:
raw_pseudo_release = self._release_getter(
album_id, RELEASE_INCLUDES
)["release"]
pseudo_release = super().album_info(raw_pseudo_release)
if self.config["custom_tags_only"].get(bool):
self._replace_artist_with_alias(
raw_pseudo_release, pseudo_release
)
self._add_custom_tags(official_release, pseudo_release)
return official_release
else:
return PseudoAlbumInfo(
pseudo_release=_merge_pseudo_and_actual_album(
pseudo_release, official_release
),
official_release=official_release,
)
except musicbrainzngs.MusicBrainzError as exc:
raise MusicBrainzAPIError(
exc,
"get pseudo-release by ID",
album_id,
traceback.format_exc(),
if (ids := self._intercept_mb_release(release)) and (
album_id := self._extract_id(ids[0])
):
raw_pseudo_release = self.api.get_release(album_id)
pseudo_release = super().album_info(raw_pseudo_release)
if self.config["custom_tags_only"].get(bool):
self._replace_artist_with_alias(
raw_pseudo_release, pseudo_release
)
self._add_custom_tags(official_release, pseudo_release)
return official_release
else:
return PseudoAlbumInfo(
pseudo_release=_merge_pseudo_and_actual_album(
pseudo_release, official_release
),
official_release=official_release,
)
else:
return official_release
@ -181,7 +167,7 @@ class MusicBrainzPseudoReleasePlugin(MusicBrainzPlugin):
return [
pr_id
for rel in data.get("release-relation-list", [])
for rel in data.get("release-relations", [])
if (pr_id := self._wanted_pseudo_release_id(album_id, rel))
is not None
]
@ -234,7 +220,7 @@ class MusicBrainzPseudoReleasePlugin(MusicBrainzPlugin):
"artist-credit", []
)
aliases = [
artist_credit.get("artist", {}).get("alias-list", [])
artist_credit.get("artist", {}).get("aliases", [])
for artist_credit in artist_credits
]
@ -247,7 +233,7 @@ class MusicBrainzPseudoReleasePlugin(MusicBrainzPlugin):
aliases_flattened,
)
if alias_dict := _preferred_alias(aliases_flattened, [locale]):
if alias := alias_dict.get("alias"):
if alias := alias_dict.get("name"):
self._log.debug("Got alias '{0}'", alias)
pseudo_release.artist = alias
for track in pseudo_release.tracks:

430
beetsplug/musicbrainz.py
View file

@ -16,16 +16,17 @@
from __future__ import annotations
import traceback
import operator
from collections import Counter
from contextlib import suppress
from functools import cached_property
from itertools import product
from dataclasses import dataclass
from functools import cached_property, singledispatchmethod
from itertools import groupby, product
from typing import TYPE_CHECKING, Any
from urllib.parse import urljoin
import musicbrainzngs
from confuse.exceptions import NotFoundError
from requests_ratelimiter import LimiterMixin
import beets
import beets.autotag.hooks
@ -34,6 +35,12 @@ from beets.metadata_plugins import MetadataSourcePlugin
from beets.util.deprecation import deprecate_for_user
from beets.util.id_extractors import extract_release_id
from ._utils.requests import (
HTTPNotFoundError,
RequestHandler,
TimeoutAndRetrySession,
)
if TYPE_CHECKING:
from collections.abc import Iterable, Sequence
from typing import Literal
@ -57,56 +64,33 @@ FIELDS_TO_MB_KEYS = {
"year": "date",
}
musicbrainzngs.set_useragent("beets", beets.__version__, "https://beets.io/")
RELEASE_INCLUDES = [
"artists",
"media",
"recordings",
"release-groups",
"labels",
"artist-credits",
"aliases",
"recording-level-rels",
"work-rels",
"work-level-rels",
"artist-rels",
"isrcs",
"url-rels",
"release-rels",
"genres",
"tags",
]
class MusicBrainzAPIError(util.HumanReadableError):
"""An error while talking to MusicBrainz. The `query` field is the
parameter to the action and may have any type.
"""
def __init__(self, reason, verb, query, tb=None):
self.query = query
if isinstance(reason, musicbrainzngs.WebServiceError):
reason = "MusicBrainz not reachable"
super().__init__(reason, verb, tb)
def get_message(self):
return f"{self._reasonstr()} in {self.verb} with query {self.query!r}"
RELEASE_INCLUDES = list(
{
"artists",
"media",
"recordings",
"release-groups",
"labels",
"artist-credits",
"aliases",
"recording-level-rels",
"work-rels",
"work-level-rels",
"artist-rels",
"isrcs",
"url-rels",
"release-rels",
"genres",
"tags",
}
& set(musicbrainzngs.VALID_INCLUDES["release"])
)
TRACK_INCLUDES = list(
{
"artists",
"aliases",
"isrcs",
"work-level-rels",
"artist-rels",
}
& set(musicbrainzngs.VALID_INCLUDES["recording"])
)
TRACK_INCLUDES = [
"artists",
"aliases",
"isrcs",
"work-level-rels",
"artist-rels",
]
BROWSE_INCLUDES = [
"artist-credits",
@ -115,18 +99,95 @@ BROWSE_INCLUDES = [
"recording-rels",
"release-rels",
]
if "work-level-rels" in musicbrainzngs.VALID_BROWSE_INCLUDES["recording"]:
BROWSE_INCLUDES.append("work-level-rels")
BROWSE_CHUNKSIZE = 100
BROWSE_MAXTRACKS = 500
class LimiterTimeoutSession(LimiterMixin, TimeoutAndRetrySession):
pass
@dataclass
class MusicBrainzAPI(RequestHandler):
api_host: str
rate_limit: float
def create_session(self) -> LimiterTimeoutSession:
return LimiterTimeoutSession(per_second=self.rate_limit)
def get_entity(
self, entity: str, inc_list: list[str] | None = None, **kwargs
) -> JSONDict:
if inc_list:
kwargs["inc"] = "+".join(inc_list)
return self._group_relations(
self.get_json(
f"{self.api_host}/ws/2/{entity}",
params={**kwargs, "fmt": "json"},
)
)
def get_release(self, id_: str) -> JSONDict:
return self.get_entity(f"release/{id_}", inc_list=RELEASE_INCLUDES)
def get_recording(self, id_: str) -> JSONDict:
return self.get_entity(f"recording/{id_}", inc_list=TRACK_INCLUDES)
def browse_recordings(self, **kwargs) -> list[JSONDict]:
kwargs.setdefault("limit", BROWSE_CHUNKSIZE)
kwargs.setdefault("inc_list", BROWSE_INCLUDES)
return self.get_entity("recording", **kwargs)["recordings"]
@singledispatchmethod
@classmethod
def _group_relations(cls, data: Any) -> Any:
"""Normalize MusicBrainz 'relations' into type-keyed fields recursively.
This helper rewrites payloads that use a generic 'relations' list into
a structure that is easier to consume downstream. When a mapping
contains 'relations', those entries are regrouped by their 'target-type'
and stored under keys like '<target-type>-relations'. The original
'relations' key is removed to avoid ambiguous access patterns.
The transformation is applied recursively so that nested objects and
sequences are normalized consistently, while non-container values are
left unchanged.
"""
return data
@_group_relations.register(list)
@classmethod
def _(cls, data: list[Any]) -> list[Any]:
return [cls._group_relations(i) for i in data]
@_group_relations.register(dict)
@classmethod
def _(cls, data: JSONDict) -> JSONDict:
for k, v in list(data.items()):
if k == "relations":
get_target_type = operator.methodcaller("get", "target-type")
for target_type, group in groupby(
sorted(v, key=get_target_type), get_target_type
):
relations = [
{k: v for k, v in item.items() if k != "target-type"}
for item in group
]
data[f"{target_type}-relations"] = cls._group_relations(
relations
)
data.pop("relations")
else:
data[k] = cls._group_relations(v)
return data
def _preferred_alias(
aliases: list[JSONDict], languages: list[str] | None = None
) -> JSONDict | None:
"""Given a list of alias structures for an artist credit, select
and return the user's preferred alias or None if no matching
alias is found.
"""
if not aliases:
return None
@ -149,8 +210,8 @@ def _preferred_alias(
for alias in valid_aliases:
if (
alias["locale"] == locale
and "primary" in alias
and alias.get("type", "").lower() not in ignored_alias_types
and alias.get("primary")
and (alias.get("type") or "").lower() not in ignored_alias_types
):
matches.append(alias)
@ -174,36 +235,33 @@ def _multi_artist_credit(
artist_sort_parts = []
artist_credit_parts = []
for el in credit:
if isinstance(el, str):
# Join phrase.
if include_join_phrase:
artist_parts.append(el)
artist_credit_parts.append(el)
artist_sort_parts.append(el)
alias = _preferred_alias(el["artist"].get("aliases", ()))
# An artist.
if alias:
cur_artist_name = alias["name"]
else:
alias = _preferred_alias(el["artist"].get("alias-list", ()))
cur_artist_name = el["artist"]["name"]
artist_parts.append(cur_artist_name)
# An artist.
if alias:
cur_artist_name = alias["alias"]
else:
cur_artist_name = el["artist"]["name"]
artist_parts.append(cur_artist_name)
# Artist sort name.
if alias:
artist_sort_parts.append(alias["sort-name"])
elif "sort-name" in el["artist"]:
artist_sort_parts.append(el["artist"]["sort-name"])
else:
artist_sort_parts.append(cur_artist_name)
# Artist sort name.
if alias:
artist_sort_parts.append(alias["sort-name"])
elif "sort-name" in el["artist"]:
artist_sort_parts.append(el["artist"]["sort-name"])
else:
artist_sort_parts.append(cur_artist_name)
# Artist credit.
if "name" in el:
artist_credit_parts.append(el["name"])
else:
artist_credit_parts.append(cur_artist_name)
# Artist credit.
if "name" in el:
artist_credit_parts.append(el["name"])
else:
artist_credit_parts.append(cur_artist_name)
if include_join_phrase and (joinphrase := el.get("joinphrase")):
artist_parts.append(joinphrase)
artist_sort_parts.append(joinphrase)
artist_credit_parts.append(joinphrase)
return (
artist_parts,
@ -273,9 +331,9 @@ def _preferred_release_event(
].as_str_seq()
for country in preferred_countries:
for event in release.get("release-event-list", {}):
for event in release.get("release-events", {}):
try:
if country in event["area"]["iso-3166-1-code-list"]:
if country in event["area"]["iso-3166-1-codes"]:
return country, event["date"]
except KeyError:
pass
@ -307,30 +365,6 @@ def _set_date_str(
setattr(info, key, date_num)
def _is_translation(r):
_trans_key = "transl-tracklisting"
return r["type"] == _trans_key and r["direction"] == "backward"
def _find_actual_release_from_pseudo_release(
pseudo_rel: JSONDict,
) -> JSONDict | None:
try:
relations = pseudo_rel["release"]["release-relation-list"]
except KeyError:
return None
# currently we only support trans(liter)ation's
translations = [r for r in relations if _is_translation(r)]
if not translations:
return None
actual_id = translations[0]["target"]
return musicbrainzngs.get_release_by_id(actual_id, RELEASE_INCLUDES)
def _merge_pseudo_and_actual_album(
pseudo: beets.autotag.hooks.AlbumInfo, actual: beets.autotag.hooks.AlbumInfo
) -> beets.autotag.hooks.AlbumInfo:
@ -374,7 +408,21 @@ def _merge_pseudo_and_actual_album(
class MusicBrainzPlugin(MetadataSourcePlugin):
@cached_property
def genres_field(self) -> str:
return f"{self.config['genres_tag'].as_choice(['genre', 'tag'])}-list"
return f"{self.config['genres_tag'].as_choice(['genre', 'tag'])}s"
@cached_property
def api(self) -> MusicBrainzAPI:
hostname = self.config["host"].as_str()
if hostname == "musicbrainz.org":
hostname, rate_limit = "https://musicbrainz.org", 1.0
else:
https = self.config["https"].get(bool)
hostname = f"http{'s' if https else ''}://{hostname}"
rate_limit = (
self.config["ratelimit"].get(int)
/ self.config["ratelimit_interval"].as_number()
)
return MusicBrainzAPI(hostname, rate_limit)
def __init__(self):
"""Set up the python-musicbrainz-ngs module according to settings
@ -409,16 +457,6 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
"'musicbrainz.searchlimit' configuration option",
"'musicbrainz.search_limit'",
)
hostname = self.config["host"].as_str()
https = self.config["https"].get(bool)
# Only call set_hostname when a custom server is configured. Since
# musicbrainz-ngs connects to musicbrainz.org with HTTPS by default
if hostname != "musicbrainz.org":
musicbrainzngs.set_hostname(hostname, https)
musicbrainzngs.set_rate_limit(
self.config["ratelimit_interval"].as_number(),
self.config["ratelimit"].get(int),
)
def track_info(
self,
@ -465,9 +503,9 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
info.artists_ids = _artist_ids(recording["artist-credit"])
info.artist_id = info.artists_ids[0]
if recording.get("artist-relation-list"):
if recording.get("artist-relations"):
info.remixer = _get_related_artist_names(
recording["artist-relation-list"], relation_type="remixer"
recording["artist-relations"], relation_type="remixer"
)
if recording.get("length"):
@ -475,13 +513,13 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
info.trackdisambig = recording.get("disambiguation")
if recording.get("isrc-list"):
info.isrc = ";".join(recording["isrc-list"])
if recording.get("isrcs"):
info.isrc = ";".join(recording["isrcs"])
lyricist = []
composer = []
composer_sort = []
for work_relation in recording.get("work-relation-list", ()):
for work_relation in recording.get("work-relations", ()):
if work_relation["type"] != "performance":
continue
info.work = work_relation["work"]["title"]
@ -490,7 +528,7 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
info.work_disambig = work_relation["work"]["disambiguation"]
for artist_relation in work_relation["work"].get(
"artist-relation-list", ()
"artist-relations", ()
):
if "type" in artist_relation:
type = artist_relation["type"]
@ -508,7 +546,7 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
info.composer_sort = ", ".join(composer_sort)
arranger = []
for artist_relation in recording.get("artist-relation-list", ()):
for artist_relation in recording.get("artist-relations", ()):
if "type" in artist_relation:
type = artist_relation["type"]
if type == "arranger":
@ -540,9 +578,9 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
release["artist-credit"], include_join_phrase=False
)
ntracks = sum(len(m["track-list"]) for m in release["medium-list"])
ntracks = sum(len(m["tracks"]) for m in release["media"])
# The MusicBrainz API omits 'artist-relation-list' and 'work-relation-list'
# The MusicBrainz API omits 'relations'
# when the release has more than 500 tracks. So we use browse_recordings
# on chunks of tracks to recover the same information in this case.
if ntracks > BROWSE_MAXTRACKS:
@ -551,35 +589,30 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
for i in range(0, ntracks, BROWSE_CHUNKSIZE):
self._log.debug("Retrieving tracks starting at {}", i)
recording_list.extend(
musicbrainzngs.browse_recordings(
release=release["id"],
limit=BROWSE_CHUNKSIZE,
includes=BROWSE_INCLUDES,
offset=i,
)["recording-list"]
self.api.browse_recordings(release=release["id"], offset=i)
)
track_map = {r["id"]: r for r in recording_list}
for medium in release["medium-list"]:
for recording in medium["track-list"]:
for medium in release["media"]:
for recording in medium["tracks"]:
recording_info = track_map[recording["recording"]["id"]]
recording["recording"] = recording_info
# Basic info.
track_infos = []
index = 0
for medium in release["medium-list"]:
for medium in release["media"]:
disctitle = medium.get("title")
format = medium.get("format")
if format in config["match"]["ignored_media"].as_str_seq():
continue
all_tracks = medium["track-list"]
all_tracks = medium["tracks"]
if (
"data-track-list" in medium
"data-tracks" in medium
and not config["match"]["ignore_data_tracks"]
):
all_tracks += medium["data-track-list"]
all_tracks += medium["data-tracks"]
track_count = len(all_tracks)
if "pregap" in medium:
@ -594,7 +627,7 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
if (
"video" in track["recording"]
and track["recording"]["video"] == "true"
and track["recording"]["video"]
and config["match"]["ignore_video_tracks"]
):
continue
@ -648,7 +681,7 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
artists=artists_names,
artists_ids=album_artist_ids,
tracks=track_infos,
mediums=len(release["medium-list"]),
mediums=len(release["media"]),
artist_sort=artist_sort_name,
artists_sort=artists_sort_names,
artist_credit=artist_credit_name,
@ -688,9 +721,9 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
rel_primarytype = release["release-group"]["primary-type"]
if rel_primarytype:
albumtypes.append(rel_primarytype.lower())
if "secondary-type-list" in release["release-group"]:
if release["release-group"]["secondary-type-list"]:
for sec_type in release["release-group"]["secondary-type-list"]:
if "secondary-types" in release["release-group"]:
if release["release-group"]["secondary-types"]:
for sec_type in release["release-group"]["secondary-types"]:
albumtypes.append(sec_type.lower())
info.albumtypes = albumtypes
@ -706,8 +739,8 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
_set_date_str(info, release_group_date, True)
# Label name.
if release.get("label-info-list"):
label_info = release["label-info-list"][0]
if release.get("label-info"):
label_info = release["label-info"][0]
if label_info.get("label"):
label = label_info["label"]["name"]
if label != "[no label]":
@ -721,10 +754,10 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
info.language = rep.get("language")
# Media (format).
if release["medium-list"]:
if release["media"]:
# If all media are the same, use that medium name
if len({m.get("format") for m in release["medium-list"]}) == 1:
info.media = release["medium-list"][0].get("format")
if len({m.get("format") for m in release["media"]}) == 1:
info.media = release["media"][0].get("format")
# Otherwise, let's just call it "Media"
else:
info.media = "Media"
@ -748,11 +781,11 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
wanted_sources = {
site for site, wanted in external_ids.items() if wanted
}
if wanted_sources and (url_rels := release.get("url-relation-list")):
if wanted_sources and (url_rels := release.get("url-relations")):
urls = {}
for source, url in product(wanted_sources, url_rels):
if f"{source}.com" in (target := url["target"]):
if f"{source}.com" in (target := url["url"]["resource"]):
urls[source] = target
self._log.debug(
"Found link to {} release via MusicBrainz",
@ -790,17 +823,20 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
def get_album_criteria(
self, items: Sequence[Item], artist: str, album: str, va_likely: bool
) -> dict[str, str]:
criteria = {
"release": album,
"alias": album,
"tracks": str(len(items)),
} | ({"arid": VARIOUS_ARTISTS_ID} if va_likely else {"artist": artist})
criteria = {"release": album} | (
{"arid": VARIOUS_ARTISTS_ID} if va_likely else {"artist": artist}
)
for tag, mb_field in self.extra_mb_field_by_tag.items():
most_common, _ = util.plurality(i.get(tag) for i in items)
value = str(most_common)
if tag == "catalognum":
value = value.replace(" ", "")
if tag == "tracks":
value = str(len(items))
elif tag == "alias":
value = album
else:
most_common, _ = util.plurality(i.get(tag) for i in items)
value = str(most_common)
if tag == "catalognum":
value = value.replace(" ", "")
criteria[mb_field] = value
@ -817,20 +853,17 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
using the provided criteria. Handles API errors by converting them into
MusicBrainzAPIError exceptions with contextual information.
"""
filters = {
k: _v for k, v in filters.items() if (_v := v.lower().strip())
}
self._log.debug(
"Searching for MusicBrainz {}s with: {!r}", query_type, filters
query = " AND ".join(
f'{k}:"{_v}"'
for k, v in filters.items()
if (_v := v.lower().strip())
)
try:
method = getattr(musicbrainzngs, f"search_{query_type}s")
res = method(limit=self.config["search_limit"].get(), **filters)
except musicbrainzngs.MusicBrainzError as exc:
raise MusicBrainzAPIError(
exc, f"{query_type} search", filters, traceback.format_exc()
)
return res[f"{query_type}-list"]
self._log.debug(
"Searching for MusicBrainz {}s with: {!r}", query_type, query
)
return self.api.get_entity(
query_type, query=query, limit=self.config["search_limit"].get()
)[f"{query_type}s"]
def candidates(
self,
@ -842,7 +875,10 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
criteria = self.get_album_criteria(items, artist, album, va_likely)
release_ids = (r["id"] for r in self._search_api("release", criteria))
yield from filter(None, map(self.album_for_id, release_ids))
for id_ in release_ids:
with suppress(HTTPNotFoundError):
if album_info := self.album_for_id(id_):
yield album_info
def item_candidates(
self, item: Item, artist: str, title: str
@ -865,29 +901,27 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
self._log.debug("Invalid MBID ({}).", album_id)
return None
try:
res = musicbrainzngs.get_release_by_id(albumid, RELEASE_INCLUDES)
res = self.api.get_release(albumid)
# resolve linked release relations
actual_res = None
# resolve linked release relations
actual_res = None
if res["release"].get("status") == "Pseudo-Release":
actual_res = _find_actual_release_from_pseudo_release(res)
except musicbrainzngs.ResponseError:
self._log.debug("Album ID match failed.")
return None
except musicbrainzngs.MusicBrainzError as exc:
raise MusicBrainzAPIError(
exc, "get release by ID", albumid, traceback.format_exc()
)
if res.get("status") == "Pseudo-Release" and (
relations := res.get("release-relations")
):
for rel in relations:
if (
rel["type"] == "transl-tracklisting"
and rel["direction"] == "backward"
):
actual_res = self.api.get_release(rel["target"])
# release is potentially a pseudo release
release = self.album_info(res["release"])
release = self.album_info(res)
# should be None unless we're dealing with a pseudo release
if actual_res is not None:
actual_release = self.album_info(actual_res["release"])
actual_release = self.album_info(actual_res)
return _merge_pseudo_and_actual_album(release, actual_release)
else:
return release
@ -902,13 +936,7 @@ class MusicBrainzPlugin(MetadataSourcePlugin):
self._log.debug("Invalid MBID ({}).", track_id)
return None
try:
res = musicbrainzngs.get_recording_by_id(trackid, TRACK_INCLUDES)
except musicbrainzngs.ResponseError:
self._log.debug("Track ID match failed.")
return None
except musicbrainzngs.MusicBrainzError as exc:
raise MusicBrainzAPIError(
exc, "get recording by ID", trackid, traceback.format_exc()
)
return self.track_info(res["recording"])
with suppress(HTTPNotFoundError):
return self.track_info(self.api.get_recording(trackid))
return None

9
docs/plugins/listenbrainz.rst
View file

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

15
docs/plugins/mbcollection.rst
View file

@ -6,9 +6,18 @@ maintain your `music collection`_ list there.
.. _music collection: https://musicbrainz.org/doc/Collections
To begin, just enable the ``mbcollection`` plugin in your configuration (see
:ref:`using-plugins`). Then, add your MusicBrainz username and password to your
:doc:`configuration file </reference/config>` under a ``musicbrainz`` section:
Installation
------------
To use the ``mbcollection`` plugin, first enable it in your configuration (see
:ref:`using-plugins`). Then, install ``beets`` with ``mbcollection`` extra
.. code-block:: bash
pip install "beets[mbcollection]"
Then, add your MusicBrainz username and password to your :doc:`configuration
file </reference/config>` under a ``musicbrainz`` section:
::

15
docs/plugins/missing.rst
View file

@ -5,12 +5,21 @@ This plugin adds a new command, ``missing`` or ``miss``, which finds and lists
missing tracks for albums in your collection. Each album requires one network
call to album data source.
Installation
------------
To use the ``missing`` plugin, first enable it in your configuration (see
:ref:`using-plugins`). Then, install ``beets`` with ``missing`` extra
.. code-block:: bash
pip install "beets[missing]"
Usage
-----
Add the ``missing`` plugin to your configuration (see :ref:`using-plugins`). The
``beet missing`` command fetches album information from the origin data source
and lists names of the **tracks** that are missing from your library.
The ``beet missing`` command fetches album information from the origin data
source and lists names of the **tracks** that are missing from your library.
It can also list the names of missing **albums** for each artist, although this
is limited to albums from the MusicBrainz data source only.

9
docs/plugins/musicbrainz.rst
View file

@ -69,15 +69,14 @@ Default
.. conf:: ratelimit
:default: 1
Controls the number of Web service requests per second.
**Do not change the rate limit setting** if you're using the main MusicBrainz
server---on this public server, you're limited_ to one request per second.
Controls the number of Web service requests per second. This setting applies only
to custom servers. The official MusicBrainz server enforces a rate limit of 1
request per second.
.. conf:: ratelimit_interval
:default: 1.0
The time interval (in seconds) for the rate limit.
The time interval (in seconds) for the rate limit. Only applies to custom servers.
.. conf:: enabled
:default: yes

11
docs/plugins/parentwork.rst
View file

@ -38,8 +38,15 @@ This plugin adds seven tags:
to keep track of recordings whose works have changed.
- **parentwork_date**: The composition date of the parent work.
To use the ``parentwork`` plugin, enable it in your configuration (see
:ref:`using-plugins`).
Installation
------------
To use the ``parentwork`` plugin, first enable it in your configuration (see
:ref:`using-plugins`). Then, install ``beets`` with ``parentwork`` extra
.. code-block:: bash
pip install "beets[parentwork]"
Configuration
-------------

41
poetry.lock generated
View file

@ -1838,7 +1838,7 @@ type = ["mypy", "mypy-extensions"]
name = "musicbrainzngs"
version = "0.7.1"
description = "Python bindings for the MusicBrainz NGS and the Cover Art Archive webservices"
optional = false
optional = true
python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*"
files = [
{file = "musicbrainzngs-0.7.1-py2.py3-none-any.whl", hash = "sha256:e841a8f975104c0a72290b09f59326050194081a5ae62ee512f41915090e1a10"},
@ -2683,6 +2683,21 @@ docs = ["sphinx", "sphinx_rtd_theme"]
fuzzer = ["atheris", "hypothesis"]
test = ["coverage[toml] (>=5.2)", "hypothesis", "pytest (>=6.0)", "pytest-benchmark", "pytest-cov", "pytest-timeout"]
[[package]]
name = "pyrate-limiter"
version = "2.10.0"
description = "Python Rate-Limiter using Leaky-Bucket Algorithm"
optional = false
python-versions = ">=3.7,<4.0"
files = [
{file = "pyrate_limiter-2.10.0-py3-none-any.whl", hash = "sha256:a99e52159f5ed5eb58118bed8c645e30818e7c0e0d127a0585c8277c776b0f7f"},
{file = "pyrate_limiter-2.10.0.tar.gz", hash = "sha256:98cc52cdbe058458e945ae87d4fd5a73186497ffa545ee6e98372f8599a5bd34"},
]
[package.extras]
all = ["filelock (>=3.0)", "redis (>=3.3,<4.0)", "redis-py-cluster (>=2.1.3,<3.0.0)"]
docs = ["furo (>=2022.3.4,<2023.0.0)", "myst-parser (>=0.17)", "sphinx (>=4.3.0,<5.0.0)", "sphinx-autodoc-typehints (>=1.17,<2.0)", "sphinx-copybutton (>=0.5)", "sphinxcontrib-apidoc (>=0.3,<0.4)"]
[[package]]
name = "pytest"
version = "8.4.2"
@ -3236,6 +3251,24 @@ requests = ">=2.0.0"
[package.extras]
rsa = ["oauthlib[signedtoken] (>=3.0.0)"]
[[package]]
name = "requests-ratelimiter"
version = "0.7.0"
description = "Rate-limiting for the requests library"
optional = false
python-versions = "<4.0,>=3.7"
files = [
{file = "requests_ratelimiter-0.7.0-py3-none-any.whl", hash = "sha256:1a7ef2faaa790272722db8539728690046237766fcc479f85b9591e5356a8185"},
{file = "requests_ratelimiter-0.7.0.tar.gz", hash = "sha256:a070c8a359a6f3a001b0ccb08f17228b7ae0a6e21d8df5b6f6bd58389cddde45"},
]
[package.dependencies]
pyrate-limiter = "<3.0"
requests = ">=2.20"
[package.extras]
docs = ["furo (>=2023.3,<2024.0)", "myst-parser (>=1.0)", "sphinx (>=5.2,<6.0)", "sphinx-autodoc-typehints (>=1.22,<2.0)", "sphinx-copybutton (>=0.5)"]
[[package]]
name = "resampy"
version = "0.4.3"
@ -4174,9 +4207,13 @@ import = ["py7zr", "rarfile"]
kodiupdate = ["requests"]
lastgenre = ["pylast"]
lastimport = ["pylast"]
listenbrainz = ["musicbrainzngs"]
lyrics = ["beautifulsoup4", "langdetect", "requests"]
mbcollection = ["musicbrainzngs"]
metasync = ["dbus-python"]
missing = ["musicbrainzngs"]
mpdstats = ["python-mpd2"]
parentwork = ["musicbrainzngs"]
plexupdate = ["requests"]
reflink = ["reflink"]
replaygain = ["PyGObject"]
@ -4189,4 +4226,4 @@ web = ["flask", "flask-cors"]
[metadata]
lock-version = "2.0"
python-versions = ">=3.10,<4"
content-hash = "9e154214b2f404415ef17df83f926a326ffb62a83b3901a404946110354d4067"
content-hash = "8cf2ad0e6a842511e1215720a63bfdf9d5f49345410644cbb0b5fd8fb74f50d2"

11
pyproject.toml
View file

@ -48,13 +48,14 @@ confuse = ">=2.1.0"
jellyfish = "*"
lap = ">=0.5.12"
mediafile = ">=0.12.0"
musicbrainzngs = ">=0.4"
numpy = [
{ python = "<3.13", version = ">=2.0.2" },
{ python = ">=3.13", version = ">=2.3.4" },
]
platformdirs = ">=3.5.0"
pyyaml = "*"
requests = ">=2.32.5"
requests-ratelimiter = ">=0.7.0"
typing_extensions = "*"
unidecode = ">=1.3.6"
@ -68,6 +69,7 @@ scipy = [ # for librosa
{ python = "<3.13", version = ">=1.13.1", optional = true },
{ python = ">=3.13", version = ">=1.16.1", optional = true },
]
musicbrainzngs = { version = ">=0.4", optional = true }
numba = [ # for librosa
{ python = "<3.13", version = ">=0.60", optional = true },
{ python = ">=3.13", version = ">=0.62.1", optional = true },
@ -83,7 +85,6 @@ python3-discogs-client = { version = ">=2.3.15", optional = true }
pyxdg = { version = "*", optional = true }
rarfile = { version = "*", optional = true }
reflink = { version = "*", optional = true }
requests = { version = "*", optional = true }
resampy = { version = ">=0.4.3", optional = true }
requests-oauthlib = { version = ">=0.6.1", optional = true }
soco = { version = "*", optional = true }
@ -93,7 +94,7 @@ pydata-sphinx-theme = { version = "*", optional = true }
sphinx = { version = "*", optional = true }
sphinx-design = { version = ">=0.6.1", optional = true }
sphinx-copybutton = { version = ">=0.5.2", optional = true }
titlecase = {version = "^2.4.1", optional = true}
titlecase = { version = "^2.4.1", optional = true }
[tool.poetry.group.test.dependencies]
beautifulsoup4 = "*"
@ -164,9 +165,13 @@ import = ["py7zr", "rarfile"]
kodiupdate = ["requests"]
lastgenre = ["pylast"]
lastimport = ["pylast"]
listenbrainz = ["musicbrainzngs"]
lyrics = ["beautifulsoup4", "langdetect", "requests"]
mbcollection = ["musicbrainzngs"]
metasync = ["dbus-python"]
missing = ["musicbrainzngs"]
mpdstats = ["python-mpd2"]
parentwork = ["musicbrainzngs"]
plexupdate = ["requests"]
reflink = ["reflink"]
replaygain = [

124
test/plugins/test_mbpseudo.py
View file

@ -1,14 +1,14 @@
import json
import pathlib
from copy import deepcopy
import pytest
from beets import config
from beets.autotag import AlbumMatch
from beets.autotag.distance import Distance
from beets.autotag.hooks import AlbumInfo, TrackInfo
from beets.library import Item
from beets.test.helper import PluginMixin
from beets.test.helper import ConfigMixin, PluginMixin
from beetsplug._typing import JSONDict
from beetsplug.mbpseudo import (
_STATUS_PSEUDO,
@ -18,6 +18,23 @@ from beetsplug.mbpseudo import (
@pytest.fixture(scope="module")
def rsrc_dir(pytestconfig: pytest.Config):
return pytestconfig.rootpath / "test" / "rsrc" / "mbpseudo"
@pytest.fixture
def official_release(rsrc_dir: pathlib.Path) -> JSONDict:
info_json = (rsrc_dir / "official_release.json").read_text(encoding="utf-8")
return json.loads(info_json)
@pytest.fixture
def pseudo_release(rsrc_dir: pathlib.Path) -> JSONDict:
info_json = (rsrc_dir / "pseudo_release.json").read_text(encoding="utf-8")
return json.loads(info_json)
@pytest.fixture
def official_release_info() -> AlbumInfo:
return AlbumInfo(
tracks=[TrackInfo(title="百花繚乱")],
@ -26,7 +43,7 @@ def official_release_info() -> AlbumInfo:
)
@pytest.fixture(scope="module")
@pytest.fixture
def pseudo_release_info() -> AlbumInfo:
return AlbumInfo(
tracks=[TrackInfo(title="In Bloom")],
@ -35,6 +52,14 @@ def pseudo_release_info() -> AlbumInfo:
)
@pytest.fixture(scope="module", autouse=True)
def config():
config = ConfigMixin().config
with pytest.MonkeyPatch.context() as m:
m.setattr("beetsplug.mbpseudo.config", config)
yield config
class TestPseudoAlbumInfo:
def test_album_id_always_from_pseudo(
self, official_release_info: AlbumInfo, pseudo_release_info: AlbumInfo
@ -62,8 +87,7 @@ class TestPseudoAlbumInfo:
info = PseudoAlbumInfo(
pseudo_release_info, official_release_info, data_source="test"
)
item = Item()
item["title"] = "百花繚乱"
item = Item(title="百花繚乱")
assert info.determine_best_ref([item]) == "official"
@ -71,37 +95,29 @@ class TestPseudoAlbumInfo:
assert info.data_source == "test"
@pytest.fixture(scope="module")
def rsrc_dir(pytestconfig: pytest.Config):
return pytestconfig.rootpath / "test" / "rsrc" / "mbpseudo"
class TestMBPseudoPlugin(PluginMixin):
class TestMBPseudoMixin(PluginMixin):
plugin = "mbpseudo"
@pytest.fixture(autouse=True)
def patch_get_release(self, monkeypatch, pseudo_release: JSONDict):
monkeypatch.setattr(
"beetsplug.musicbrainz.MusicBrainzAPI.get_release",
lambda _, album_id: deepcopy(
{pseudo_release["id"]: pseudo_release}[album_id]
),
)
@pytest.fixture(scope="class")
def plugin_config(self):
return {"scripts": ["Latn", "Dummy"]}
@pytest.fixture(scope="class")
@pytest.fixture
def mbpseudo_plugin(self, plugin_config) -> MusicBrainzPseudoReleasePlugin:
self.config[self.plugin].set(plugin_config)
return MusicBrainzPseudoReleasePlugin()
@pytest.fixture
def official_release(self, rsrc_dir: pathlib.Path) -> JSONDict:
info_json = (rsrc_dir / "official_release.json").read_text(
encoding="utf-8"
)
return json.loads(info_json)
@pytest.fixture
def pseudo_release(self, rsrc_dir: pathlib.Path) -> JSONDict:
info_json = (rsrc_dir / "pseudo_release.json").read_text(
encoding="utf-8"
)
return json.loads(info_json)
class TestMBPseudoPlugin(TestMBPseudoMixin):
def test_scripts_init(
self, mbpseudo_plugin: MusicBrainzPseudoReleasePlugin
):
@ -129,7 +145,7 @@ class TestMBPseudoPlugin(PluginMixin):
mbpseudo_plugin: MusicBrainzPseudoReleasePlugin,
pseudo_release: JSONDict,
):
album_info = mbpseudo_plugin.album_info(pseudo_release["release"])
album_info = mbpseudo_plugin.album_info(pseudo_release)
assert not isinstance(album_info, PseudoAlbumInfo)
assert album_info.data_source == "MusicBrainzPseudoRelease"
assert album_info.albumstatus == _STATUS_PSEUDO
@ -148,9 +164,9 @@ class TestMBPseudoPlugin(PluginMixin):
official_release: JSONDict,
json_key: str,
):
del official_release["release"]["release-relation-list"][0][json_key]
del official_release["release-relations"][0][json_key]
album_info = mbpseudo_plugin.album_info(official_release["release"])
album_info = mbpseudo_plugin.album_info(official_release)
assert not isinstance(album_info, PseudoAlbumInfo)
assert album_info.data_source == "MusicBrainzPseudoRelease"
@ -159,11 +175,11 @@ class TestMBPseudoPlugin(PluginMixin):
mbpseudo_plugin: MusicBrainzPseudoReleasePlugin,
official_release: JSONDict,
):
official_release["release"]["release-relation-list"][0]["release"][
official_release["release-relations"][0]["release"][
"text-representation"
]["script"] = "Null"
album_info = mbpseudo_plugin.album_info(official_release["release"])
album_info = mbpseudo_plugin.album_info(official_release)
assert not isinstance(album_info, PseudoAlbumInfo)
assert album_info.data_source == "MusicBrainzPseudoRelease"
@ -171,12 +187,8 @@ class TestMBPseudoPlugin(PluginMixin):
self,
mbpseudo_plugin: MusicBrainzPseudoReleasePlugin,
official_release: JSONDict,
pseudo_release: JSONDict,
):
mbpseudo_plugin._release_getter = (
lambda album_id, includes: pseudo_release
)
album_info = mbpseudo_plugin.album_info(official_release["release"])
album_info = mbpseudo_plugin.album_info(official_release)
assert isinstance(album_info, PseudoAlbumInfo)
assert album_info.data_source == "MusicBrainzPseudoRelease"
@ -226,40 +238,19 @@ class TestMBPseudoPlugin(PluginMixin):
assert match.info.album == "In Bloom"
class TestMBPseudoPluginCustomTagsOnly(PluginMixin):
plugin = "mbpseudo"
class TestMBPseudoPluginCustomTagsOnly(TestMBPseudoMixin):
@pytest.fixture(scope="class")
def mbpseudo_plugin(self) -> MusicBrainzPseudoReleasePlugin:
self.config[self.plugin]["scripts"] = ["Latn"]
self.config[self.plugin]["custom_tags_only"] = True
return MusicBrainzPseudoReleasePlugin()
@pytest.fixture(scope="class")
def official_release(self, rsrc_dir: pathlib.Path) -> JSONDict:
info_json = (rsrc_dir / "official_release.json").read_text(
encoding="utf-8"
)
return json.loads(info_json)
@pytest.fixture(scope="class")
def pseudo_release(self, rsrc_dir: pathlib.Path) -> JSONDict:
info_json = (rsrc_dir / "pseudo_release.json").read_text(
encoding="utf-8"
)
return json.loads(info_json)
def plugin_config(self):
return {"scripts": ["Latn", "Dummy"], "custom_tags_only": True}
def test_custom_tags(
self,
config,
mbpseudo_plugin: MusicBrainzPseudoReleasePlugin,
official_release: JSONDict,
pseudo_release: JSONDict,
):
config["import"]["languages"] = []
mbpseudo_plugin._release_getter = (
lambda album_id, includes: pseudo_release
)
album_info = mbpseudo_plugin.album_info(official_release["release"])
config["import"]["languages"] = ["en", "jp"]
album_info = mbpseudo_plugin.album_info(official_release)
assert not isinstance(album_info, PseudoAlbumInfo)
assert album_info.data_source == "MusicBrainzPseudoRelease"
assert album_info["album_transl"] == "In Bloom"
@ -269,15 +260,12 @@ class TestMBPseudoPluginCustomTagsOnly(PluginMixin):
def test_custom_tags_with_import_languages(
self,
config,
mbpseudo_plugin: MusicBrainzPseudoReleasePlugin,
official_release: JSONDict,
pseudo_release: JSONDict,
):
config["import"]["languages"] = ["en", "jp"]
mbpseudo_plugin._release_getter = (
lambda album_id, includes: pseudo_release
)
album_info = mbpseudo_plugin.album_info(official_release["release"])
config["import"]["languages"] = []
album_info = mbpseudo_plugin.album_info(official_release)
assert not isinstance(album_info, PseudoAlbumInfo)
assert album_info.data_source == "MusicBrainzPseudoRelease"
assert album_info["album_transl"] == "In Bloom"

549
test/plugins/test_musicbrainz.py
View file

@ -64,10 +64,10 @@ class MBAlbumInfoTest(MusicBrainzTestCase):
}
],
"date": "3001",
"medium-list": [],
"genre-list": [{"count": 1, "name": "GENRE"}],
"tag-list": [{"count": 1, "name": "TAG"}],
"label-info-list": [
"media": [],
"genres": [{"count": 1, "name": "GENRE"}],
"tags": [{"count": 1, "name": "TAG"}],
"label-info": [
{
"catalog-number": "CATALOG NUMBER",
"label": {"name": "LABEL NAME"},
@ -83,7 +83,7 @@ class MBAlbumInfoTest(MusicBrainzTestCase):
}
if multi_artist_credit:
release["artist-credit"].append(" & ") # add join phase
release["artist-credit"][0]["joinphrase"] = " & "
release["artist-credit"].append(
{
"artist": {
@ -124,7 +124,7 @@ class MBAlbumInfoTest(MusicBrainzTestCase):
]
if multi_artist_credit:
track["artist-credit"].append(" & ") # add join phase
track["artist-credit"][0]["joinphrase"] = " & "
track["artist-credit"].append(
{
"artist": {
@ -148,11 +148,11 @@ class MBAlbumInfoTest(MusicBrainzTestCase):
"number": "A1",
}
data_track_list.append(data_track)
release["medium-list"].append(
release["media"].append(
{
"position": "1",
"track-list": track_list,
"data-track-list": data_track_list,
"tracks": track_list,
"data-tracks": data_track_list,
"format": medium_format,
"title": "MEDIUM TITLE",
}
@ -188,7 +188,7 @@ class MBAlbumInfoTest(MusicBrainzTestCase):
}
]
if multi_artist_credit:
track["artist-credit"].append(" & ") # add join phase
track["artist-credit"][0]["joinphrase"] = " & "
track["artist-credit"].append(
{
"artist": {
@ -200,7 +200,7 @@ class MBAlbumInfoTest(MusicBrainzTestCase):
}
)
if remixer:
track["artist-relation-list"] = [
track["artist-relations"] = [
{
"type": "remixer",
"type-id": "RELATION TYPE ID",
@ -215,7 +215,7 @@ class MBAlbumInfoTest(MusicBrainzTestCase):
}
]
if video:
track["video"] = "true"
track["video"] = True
if disambiguation:
track["disambiguation"] = disambiguation
return track
@ -301,10 +301,10 @@ class MBAlbumInfoTest(MusicBrainzTestCase):
"number": "A1",
}
]
release["medium-list"].append(
release["media"].append(
{
"position": "2",
"track-list": second_track_list,
"tracks": second_track_list,
}
)
@ -700,15 +700,15 @@ class ArtistFlatteningTest(unittest.TestCase):
def _add_alias(self, credit_dict, suffix="", locale="", primary=False):
alias = {
"alias": f"ALIAS{suffix}",
"name": f"ALIAS{suffix}",
"locale": locale,
"sort-name": f"ALIASSORT{suffix}",
}
if primary:
alias["primary"] = "primary"
if "alias-list" not in credit_dict["artist"]:
credit_dict["artist"]["alias-list"] = []
credit_dict["artist"]["alias-list"].append(alias)
if "aliases" not in credit_dict["artist"]:
credit_dict["artist"]["aliases"] = []
credit_dict["artist"]["aliases"].append(alias)
def test_single_artist(self):
credit = [self._credit_dict()]
@ -725,7 +725,10 @@ class ArtistFlatteningTest(unittest.TestCase):
assert c == ["CREDIT"]
def test_two_artists(self):
credit = [self._credit_dict("a"), " AND ", self._credit_dict("b")]
credit = [
{**self._credit_dict("a"), "joinphrase": " AND "},
self._credit_dict("b"),
]
a, s, c = musicbrainz._flatten_artist_credit(credit)
assert a == "NAMEa AND NAMEb"
assert s == "SORTa AND SORTb"
@ -783,86 +786,84 @@ class MBLibraryTest(MusicBrainzTestCase):
def test_follow_pseudo_releases(self):
side_effect = [
{
"release": {
"title": "pseudo",
"id": "d2a6f856-b553-40a0-ac54-a321e8e2da02",
"status": "Pseudo-Release",
"medium-list": [
{
"track-list": [
{
"id": "baz",
"recording": {
"title": "translated title",
"id": "bar",
"length": 42,
},
"position": 9,
"number": "A1",
}
],
"position": 5,
}
],
"artist-credit": [
{
"artist": {
"name": "some-artist",
"id": "some-id",
},
}
],
"release-group": {
"id": "another-id",
},
"release-relation-list": [
{
"type": "transl-tracklisting",
"target": "d2a6f856-b553-40a0-ac54-a321e8e2da01",
"direction": "backward",
}
],
}
"title": "pseudo",
"id": "d2a6f856-b553-40a0-ac54-a321e8e2da02",
"status": "Pseudo-Release",
"media": [
{
"tracks": [
{
"id": "baz",
"recording": {
"title": "translated title",
"id": "bar",
"length": 42,
},
"position": 9,
"number": "A1",
}
],
"position": 5,
}
],
"artist-credit": [
{
"artist": {
"name": "some-artist",
"id": "some-id",
},
}
],
"release-group": {
"id": "another-id",
},
"release-relations": [
{
"type": "transl-tracklisting",
"target": "d2a6f856-b553-40a0-ac54-a321e8e2da01",
"direction": "backward",
}
],
},
{
"release": {
"title": "actual",
"id": "d2a6f856-b553-40a0-ac54-a321e8e2da01",
"status": "Official",
"medium-list": [
{
"track-list": [
{
"id": "baz",
"recording": {
"title": "original title",
"id": "bar",
"length": 42,
},
"position": 9,
"number": "A1",
}
],
"position": 5,
}
],
"artist-credit": [
{
"artist": {
"name": "some-artist",
"id": "some-id",
},
}
],
"release-group": {
"id": "another-id",
},
"country": "COUNTRY",
}
"title": "actual",
"id": "d2a6f856-b553-40a0-ac54-a321e8e2da01",
"status": "Official",
"media": [
{
"tracks": [
{
"id": "baz",
"recording": {
"title": "original title",
"id": "bar",
"length": 42,
},
"position": 9,
"number": "A1",
}
],
"position": 5,
}
],
"artist-credit": [
{
"artist": {
"name": "some-artist",
"id": "some-id",
},
}
],
"release-group": {
"id": "another-id",
},
"country": "COUNTRY",
},
]
with mock.patch("musicbrainzngs.get_release_by_id") as gp:
with mock.patch(
"beetsplug.musicbrainz.MusicBrainzAPI.get_release"
) as gp:
gp.side_effect = side_effect
album = self.mb.album_for_id("d2a6f856-b553-40a0-ac54-a321e8e2da02")
assert album.country == "COUNTRY"
@ -870,44 +871,43 @@ class MBLibraryTest(MusicBrainzTestCase):
def test_pseudo_releases_with_empty_links(self):
side_effect = [
{
"release": {
"title": "pseudo",
"id": "d2a6f856-b553-40a0-ac54-a321e8e2da02",
"status": "Pseudo-Release",
"medium-list": [
{
"track-list": [
{
"id": "baz",
"recording": {
"title": "translated title",
"id": "bar",
"length": 42,
},
"position": 9,
"number": "A1",
}
],
"position": 5,
}
],
"artist-credit": [
{
"artist": {
"name": "some-artist",
"id": "some-id",
},
}
],
"release-group": {
"id": "another-id",
},
"release-relation-list": [],
}
},
"title": "pseudo",
"id": "d2a6f856-b553-40a0-ac54-a321e8e2da02",
"status": "Pseudo-Release",
"media": [
{
"tracks": [
{
"id": "baz",
"recording": {
"title": "translated title",
"id": "bar",
"length": 42,
},
"position": 9,
"number": "A1",
}
],
"position": 5,
}
],
"artist-credit": [
{
"artist": {
"name": "some-artist",
"id": "some-id",
},
}
],
"release-group": {
"id": "another-id",
},
}
]
with mock.patch("musicbrainzngs.get_release_by_id") as gp:
with mock.patch(
"beetsplug.musicbrainz.MusicBrainzAPI.get_release"
) as gp:
gp.side_effect = side_effect
album = self.mb.album_for_id("d2a6f856-b553-40a0-ac54-a321e8e2da02")
assert album.country is None
@ -915,43 +915,43 @@ class MBLibraryTest(MusicBrainzTestCase):
def test_pseudo_releases_without_links(self):
side_effect = [
{
"release": {
"title": "pseudo",
"id": "d2a6f856-b553-40a0-ac54-a321e8e2da02",
"status": "Pseudo-Release",
"medium-list": [
{
"track-list": [
{
"id": "baz",
"recording": {
"title": "translated title",
"id": "bar",
"length": 42,
},
"position": 9,
"number": "A1",
}
],
"position": 5,
}
],
"artist-credit": [
{
"artist": {
"name": "some-artist",
"id": "some-id",
},
}
],
"release-group": {
"id": "another-id",
},
}
},
"title": "pseudo",
"id": "d2a6f856-b553-40a0-ac54-a321e8e2da02",
"status": "Pseudo-Release",
"media": [
{
"tracks": [
{
"id": "baz",
"recording": {
"title": "translated title",
"id": "bar",
"length": 42,
},
"position": 9,
"number": "A1",
}
],
"position": 5,
}
],
"artist-credit": [
{
"artist": {
"name": "some-artist",
"id": "some-id",
},
}
],
"release-group": {
"id": "another-id",
},
}
]
with mock.patch("musicbrainzngs.get_release_by_id") as gp:
with mock.patch(
"beetsplug.musicbrainz.MusicBrainzAPI.get_release"
) as gp:
gp.side_effect = side_effect
album = self.mb.album_for_id("d2a6f856-b553-40a0-ac54-a321e8e2da02")
assert album.country is None
@ -959,50 +959,50 @@ class MBLibraryTest(MusicBrainzTestCase):
def test_pseudo_releases_with_unsupported_links(self):
side_effect = [
{
"release": {
"title": "pseudo",
"id": "d2a6f856-b553-40a0-ac54-a321e8e2da02",
"status": "Pseudo-Release",
"medium-list": [
{
"track-list": [
{
"id": "baz",
"recording": {
"title": "translated title",
"id": "bar",
"length": 42,
},
"position": 9,
"number": "A1",
}
],
"position": 5,
}
],
"artist-credit": [
{
"artist": {
"name": "some-artist",
"id": "some-id",
},
}
],
"release-group": {
"id": "another-id",
},
"release-relation-list": [
{
"type": "remaster",
"target": "d2a6f856-b553-40a0-ac54-a321e8e2da01",
"direction": "backward",
}
],
}
},
"title": "pseudo",
"id": "d2a6f856-b553-40a0-ac54-a321e8e2da02",
"status": "Pseudo-Release",
"media": [
{
"tracks": [
{
"id": "baz",
"recording": {
"title": "translated title",
"id": "bar",
"length": 42,
},
"position": 9,
"number": "A1",
}
],
"position": 5,
}
],
"artist-credit": [
{
"artist": {
"name": "some-artist",
"id": "some-id",
},
}
],
"release-group": {
"id": "another-id",
},
"release-relations": [
{
"type": "remaster",
"target": "d2a6f856-b553-40a0-ac54-a321e8e2da01",
"direction": "backward",
}
],
}
]
with mock.patch("musicbrainzngs.get_release_by_id") as gp:
with mock.patch(
"beetsplug.musicbrainz.MusicBrainzAPI.get_release"
) as gp:
gp.side_effect = side_effect
album = self.mb.album_for_id("d2a6f856-b553-40a0-ac54-a321e8e2da02")
assert album.country is None
@ -1047,15 +1047,13 @@ class TestMusicBrainzPlugin(PluginMixin):
assert mb.get_album_criteria(items, "Artist ", " Album", va_likely) == {
"release": " Album",
"alias": " Album",
"tracks": str(len(items)),
**expected_additional_criteria,
}
def test_item_candidates(self, monkeypatch, mb):
monkeypatch.setattr(
"musicbrainzngs.search_recordings",
lambda *_, **__: {"recording-list": [self.RECORDING]},
"beetsplug.musicbrainz.MusicBrainzAPI.get_json",
lambda *_, **__: {"recordings": [self.RECORDING]},
)
candidates = list(mb.item_candidates(Item(), "hello", "there"))
@ -1065,34 +1063,32 @@ class TestMusicBrainzPlugin(PluginMixin):
def test_candidates(self, monkeypatch, mb):
monkeypatch.setattr(
"musicbrainzngs.search_releases",
lambda *_, **__: {"release-list": [{"id": self.mbid}]},
"beetsplug.musicbrainz.MusicBrainzAPI.get_json",
lambda *_, **__: {"releases": [{"id": self.mbid}]},
)
monkeypatch.setattr(
"musicbrainzngs.get_release_by_id",
"beetsplug.musicbrainz.MusicBrainzAPI.get_release",
lambda *_, **__: {
"release": {
"title": "hi",
"id": self.mbid,
"status": "status",
"medium-list": [
{
"track-list": [
{
"id": "baz",
"recording": self.RECORDING,
"position": 9,
"number": "A1",
}
],
"position": 5,
}
],
"artist-credit": [
{"artist": {"name": "some-artist", "id": "some-id"}}
],
"release-group": {"id": "another-id"},
}
"title": "hi",
"id": self.mbid,
"status": "status",
"media": [
{
"tracks": [
{
"id": "baz",
"recording": self.RECORDING,
"position": 9,
"number": "A1",
}
],
"position": 5,
}
],
"artist-credit": [
{"artist": {"name": "some-artist", "id": "some-id"}}
],
"release-group": {"id": "another-id"},
},
)
candidates = list(mb.candidates([], "hello", "there", False))
@ -1100,3 +1096,84 @@ class TestMusicBrainzPlugin(PluginMixin):
assert len(candidates) == 1
assert candidates[0].tracks[0].track_id == self.RECORDING["id"]
assert candidates[0].album == "hi"
def test_group_relations():
raw_release = {
"id": "r1",
"relations": [
{"target-type": "artist", "type": "vocal", "name": "A"},
{"target-type": "url", "type": "streaming", "url": "http://s"},
{"target-type": "url", "type": "purchase", "url": "http://p"},
{
"target-type": "work",
"type": "performance",
"work": {
"relations": [
{
"artist": {"name": "幾田りら"},
"target-type": "artist",
"type": "composer",
},
{
"target-type": "url",
"type": "lyrics",
"url": {
"resource": "https://utaten.com/lyric/tt24121002/"
},
},
{
"artist": {"name": "幾田りら"},
"target-type": "artist",
"type": "lyricist",
},
{
"target-type": "url",
"type": "lyrics",
"url": {
"resource": "https://www.uta-net.com/song/366579/"
},
},
],
"title": "百花繚乱",
"type": "Song",
},
},
],
}
assert musicbrainz.MusicBrainzAPI._group_relations(raw_release) == {
"id": "r1",
"artist-relations": [{"type": "vocal", "name": "A"}],
"url-relations": [
{"type": "streaming", "url": "http://s"},
{"type": "purchase", "url": "http://p"},
],
"work-relations": [
{
"type": "performance",
"work": {
"artist-relations": [
{"type": "composer", "artist": {"name": "幾田りら"}},
{"type": "lyricist", "artist": {"name": "幾田りら"}},
],
"url-relations": [
{
"type": "lyrics",
"url": {
"resource": "https://utaten.com/lyric/tt24121002/"
},
},
{
"type": "lyrics",
"url": {
"resource": "https://www.uta-net.com/song/366579/"
},
},
],
"title": "百花繚乱",
"type": "Song",
},
},
],
}

58
test/plugins/utils/test_request_handler.py Normal file
View file

@ -0,0 +1,58 @@
import io
from http import HTTPStatus
from unittest.mock import Mock
from urllib.error import URLError
import pytest
import requests
from urllib3 import HTTPResponse
from urllib3.exceptions import NewConnectionError
from beetsplug._utils.requests import RequestHandler
class TestRequestHandlerRetry:
@pytest.fixture(autouse=True)
def patch_connection(self, monkeypatch, last_response):
monkeypatch.setattr(
"urllib3.connectionpool.HTTPConnectionPool._make_request",
Mock(
side_effect=[
NewConnectionError(None, "Connection failed"),
URLError("bad"),
last_response,
]
),
)
@pytest.fixture
def request_handler(self):
return RequestHandler()
@pytest.mark.parametrize(
"last_response",
[
HTTPResponse(
body=io.BytesIO(b"success"),
status=HTTPStatus.OK,
preload_content=False,
),
],
ids=["success"],
)
def test_retry_on_connection_error(self, request_handler):
"""Verify that the handler retries on connection errors."""
response = request_handler.get("http://example.com/api")
assert response.text == "success"
assert response.status_code == HTTPStatus.OK
@pytest.mark.parametrize(
"last_response", [ConnectionResetError], ids=["conn_error"]
)
def test_retry_exhaustion(self, request_handler):
"""Verify that the handler raises an error after exhausting retries."""
with pytest.raises(
requests.exceptions.ConnectionError, match="Max retries exceeded"
):
request_handler.get("http://example.com/api")

2663
test/rsrc/mbpseudo/official_release.json
View file

File diff suppressed because it is too large Load diff

819
test/rsrc/mbpseudo/pseudo_release.json
View file

@ -1,346 +1,515 @@
{
"release": {
"id": "dc3ee2df-0bc1-49eb-b8c4-34473d279a43",
"title": "In Bloom",
"status": "Pseudo-Release",
"quality": "normal",
"text-representation": {
"language": "eng",
"script": "Latn"
},
"artist-credit": [
{
"name": "Lilas Ikuta",
"artist": {
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"type": "Person",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"country": "JP",
"alias-list": [
"aliases": [],
"artist-credit": [
{
"artist": {
"aliases": [
{
"begin": null,
"end": null,
"ended": false,
"locale": "en",
"name": "Lilas Ikuta",
"primary": true,
"sort-name": "Ikuta, Lilas",
"type": "Artist name",
"type-id": "894afba6-2816-3c24-8072-eadb66bd04bc"
}
],
"country": "JP",
"disambiguation": "",
"genres": [
{
"count": 1,
"disambiguation": "",
"id": "eba7715e-ee26-4989-8d49-9db382955419",
"name": "j-pop"
},
{
"count": 1,
"disambiguation": "",
"id": "455f264b-db00-4716-991d-fbd32dc24523",
"name": "singer-songwriter"
}
],
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"tags": [
{
"count": 1,
"name": "j-pop"
},
{
"count": 1,
"name": "singer-songwriter"
}
],
"type": "Person",
"type-id": "b6e035f4-3ce9-331c-97df-83397230b0df"
},
"joinphrase": "",
"name": "Lilas Ikuta"
}
],
"asin": null,
"barcode": null,
"cover-art-archive": {
"artwork": false,
"back": false,
"count": 0,
"darkened": false,
"front": false
},
"disambiguation": "",
"genres": [],
"id": "dc3ee2df-0bc1-49eb-b8c4-34473d279a43",
"label-info": [],
"media": [
{
"format": "Digital Media",
"format-id": "907a28d9-b3b2-3ef6-89a8-7b18d91d4794",
"id": "606faab7-60fa-3a8b-a40f-2c66150cce81",
"position": 1,
"title": "",
"track-count": 1,
"track-offset": 0,
"tracks": [
{
"artist-credit": [
{
"locale": "en",
"sort-name": "Ikuta, Lilas",
"type": "Artist name",
"primary": "primary",
"alias": "Lilas Ikuta"
"artist": {
"aliases": [
{
"begin": null,
"end": null,
"ended": false,
"locale": "en",
"name": "Lilas Ikuta",
"primary": true,
"sort-name": "Ikuta, Lilas",
"type": "Artist name",
"type-id": "894afba6-2816-3c24-8072-eadb66bd04bc"
}
],
"country": "JP",
"disambiguation": "",
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"type": "Person",
"type-id": "b6e035f4-3ce9-331c-97df-83397230b0df"
},
"joinphrase": "",
"name": "Lilas Ikuta"
}
],
"alias-count": 1,
"tag-list": [
{
"count": "1",
"name": "j-pop"
},
{
"count": "1",
"name": "singer-songwriter"
}
]
}
}
],
"release-group": {
"id": "da0d6bbb-f44b-4fff-8739-9d72db0402a1",
"type": "Single",
"title": "百花繚乱",
"first-release-date": "2025-01-10",
"primary-type": "Single",
"artist-credit": [
{
"artist": {
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"type": "Person",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"country": "JP",
"alias-list": [
{
"locale": "en",
"sort-name": "Ikuta, Lilas",
"type": "Artist name",
"primary": "primary",
"alias": "Lilas Ikuta"
}
],
"alias-count": 1,
"tag-list": [
{
"count": "1",
"name": "j-pop"
},
{
"count": "1",
"name": "singer-songwriter"
}
]
}
}
],
"artist-credit-phrase": "幾田りら"
},
"cover-art-archive": {
"artwork": "false",
"count": "0",
"front": "false",
"back": "false"
},
"label-info-list": [],
"label-info-count": 0,
"medium-list": [
{
"position": "1",
"format": "Digital Media",
"track-list": [
{
"id": "2018b012-a184-49a2-a464-fb4628a89588",
"position": "1",
"number": "1",
"title": "In Bloom",
"length": "179239",
"id": "2018b012-a184-49a2-a464-fb4628a89588",
"length": 179239,
"number": "1",
"position": 1,
"recording": {
"aliases": [],
"artist-credit": [
{
"name": "Lilas Ikuta",
"artist": {
"country": "JP",
"disambiguation": "",
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"type": "Person",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"type": "Person",
"type-id": "b6e035f4-3ce9-331c-97df-83397230b0df"
},
"joinphrase": "",
"name": "幾田りら"
}
],
"artist-relations": [
{
"artist": {
"country": "JP",
"alias-list": [
"disambiguation": "Japanese composer/arranger/guitarist, agehasprings",
"id": "f24241fb-4d89-4bf2-8336-3f2a7d2c0025",
"name": "KOHD",
"sort-name": "KOHD",
"type": "Person",
"type-id": "b6e035f4-3ce9-331c-97df-83397230b0df"
},
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": null,
"direction": "backward",
"end": null,
"ended": false,
"source-credit": "",
"target-credit": "",
"type": "arranger",
"type-id": "22661fb8-cdb7-4f67-8385-b2a8be6c9f0d"
},
{
"artist": {
"country": "JP",
"disambiguation": "",
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"type": "Person",
"type-id": "b6e035f4-3ce9-331c-97df-83397230b0df"
},
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": "2025",
"direction": "backward",
"end": "2025",
"ended": true,
"source-credit": "",
"target-credit": "Lilas Ikuta",
"type": "phonographic copyright",
"type-id": "7fd5fbc0-fbf4-4d04-be23-417d50a4dc30"
},
{
"artist": {
"country": "JP",
"disambiguation": "",
"id": "1d27ab8a-a0df-47cf-b4cc-d2d7a0712a05",
"name": "山本秀哉",
"sort-name": "Yamamoto, Shuya",
"type": "Person",
"type-id": "b6e035f4-3ce9-331c-97df-83397230b0df"
},
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": null,
"direction": "backward",
"end": null,
"ended": false,
"source-credit": "",
"target-credit": "",
"type": "producer",
"type-id": "5c0ceac3-feb4-41f0-868d-dc06f6e27fc0"
},
{
"artist": {
"country": "JP",
"disambiguation": "",
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"type": "Person",
"type-id": "b6e035f4-3ce9-331c-97df-83397230b0df"
},
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": null,
"direction": "backward",
"end": null,
"ended": false,
"source-credit": "",
"target-credit": "",
"type": "vocal",
"type-id": "0fdbe3c6-7700-4a31-ae54-b53f06ae1cfa"
}
],
"disambiguation": "",
"first-release-date": "2025-01-10",
"genres": [],
"id": "781724c1-a039-41e6-bd9b-770c3b9d5b8e",
"isrcs": [
"JPP302400868"
],
"length": 179546,
"tags": [],
"title": "百花繚乱",
"url-relations": [
{
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": null,
"direction": "forward",
"end": null,
"ended": false,
"source-credit": "",
"target-credit": "",
"type": "free streaming",
"type-id": "7e41ef12-a124-4324-afdb-fdbae687a89c",
"url": {
"id": "d076eaf9-5fde-4f6e-a946-cde16b67aa3b",
"resource": "https://open.spotify.com/track/782PTXsbAWB70ySDZ5NHmP"
}
},
{
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": null,
"direction": "forward",
"end": null,
"ended": false,
"source-credit": "",
"target-credit": "",
"type": "purchase for download",
"type-id": "92777657-504c-4acb-bd33-51a201bd57e1",
"url": {
"id": "64879627-6eca-4755-98b5-b2234a8dbc61",
"resource": "https://music.apple.com/jp/song/1857886416"
}
},
{
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": null,
"direction": "forward",
"end": null,
"ended": false,
"source-credit": "",
"target-credit": "",
"type": "streaming",
"type-id": "b5f3058a-666c-406f-aafb-f9249fc7b122",
"url": {
"id": "64879627-6eca-4755-98b5-b2234a8dbc61",
"resource": "https://music.apple.com/jp/song/1857886416"
}
}
],
"video": false,
"work-relations": [
{
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": null,
"direction": "forward",
"end": null,
"ended": false,
"source-credit": "",
"target-credit": "",
"type": "performance",
"type-id": "a3005666-a872-32c3-ad06-98af558e99b0",
"work": {
"artist-relations": [
{
"locale": "en",
"sort-name": "Ikuta, Lilas",
"type": "Artist name",
"primary": "primary",
"alias": "Lilas Ikuta"
}
],
"alias-count": 1,
"tag-list": [
{
"count": "1",
"name": "j-pop"
"artist": {
"country": "JP",
"disambiguation": "",
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"type": "Person",
"type-id": "b6e035f4-3ce9-331c-97df-83397230b0df"
},
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": null,
"direction": "backward",
"end": null,
"ended": false,
"source-credit": "",
"target-credit": "",
"type": "composer",
"type-id": "d59d99ea-23d4-4a80-b066-edca32ee158f"
},
{
"count": "1",
"name": "singer-songwriter"
"artist": {
"country": "JP",
"disambiguation": "",
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"type": "Person",
"type-id": "b6e035f4-3ce9-331c-97df-83397230b0df"
},
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": null,
"direction": "backward",
"end": null,
"ended": false,
"source-credit": "",
"target-credit": "",
"type": "lyricist",
"type-id": "3e48faba-ec01-47fd-8e89-30e81161661c"
}
],
"attributes": [],
"disambiguation": "",
"id": "9e14d6b2-ac7d-43e9-82a9-561bc76ce2ed",
"iswcs": [],
"language": "jpn",
"languages": [
"jpn"
],
"title": "百花繚乱",
"type": "Song",
"type-id": "f061270a-2fd6-32f1-a641-f0f8676d14e6",
"url-relations": [
{
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": null,
"direction": "backward",
"end": null,
"ended": false,
"source-credit": "",
"target-credit": "",
"type": "lyrics",
"type-id": "e38e65aa-75e0-42ba-ace0-072aeb91a538",
"url": {
"id": "dfac3640-6b23-4991-a59c-7cb80e8eb950",
"resource": "https://utaten.com/lyric/tt24121002/"
}
},
{
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": null,
"direction": "backward",
"end": null,
"ended": false,
"source-credit": "",
"target-credit": "",
"type": "lyrics",
"type-id": "e38e65aa-75e0-42ba-ace0-072aeb91a538",
"url": {
"id": "b1b5d5df-e79d-4cda-bb2a-8014e5505415",
"resource": "https://www.uta-net.com/song/366579/"
}
}
]
}
}
],
"recording": {
"id": "781724c1-a039-41e6-bd9b-770c3b9d5b8e",
"title": "百花繚乱",
"length": "179546",
"artist-credit": [
{
"artist": {
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"type": "Person",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"country": "JP",
"alias-list": [
{
"locale": "en",
"sort-name": "Ikuta, Lilas",
"type": "Artist name",
"primary": "primary",
"alias": "Lilas Ikuta"
}
],
"alias-count": 1,
"tag-list": [
{
"count": "1",
"name": "j-pop"
},
{
"count": "1",
"name": "singer-songwriter"
}
]
}
}
],
"isrc-list": [
"JPP302400868"
],
"isrc-count": 1,
"artist-relation-list": [
{
"type": "arranger",
"type-id": "22661fb8-cdb7-4f67-8385-b2a8be6c9f0d",
"target": "f24241fb-4d89-4bf2-8336-3f2a7d2c0025",
"direction": "backward",
"artist": {
"id": "f24241fb-4d89-4bf2-8336-3f2a7d2c0025",
"type": "Person",
"name": "KOHD",
"sort-name": "KOHD",
"country": "JP",
"disambiguation": "Japanese composer/arranger/guitarist, agehasprings"
}
},
{
"type": "phonographic copyright",
"type-id": "7fd5fbc0-fbf4-4d04-be23-417d50a4dc30",
"target": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"direction": "backward",
"begin": "2025",
"end": "2025",
"ended": "true",
"artist": {
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"type": "Person",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"country": "JP"
},
"target-credit": "Lilas Ikuta"
},
{
"type": "producer",
"type-id": "5c0ceac3-feb4-41f0-868d-dc06f6e27fc0",
"target": "1d27ab8a-a0df-47cf-b4cc-d2d7a0712a05",
"direction": "backward",
"artist": {
"id": "1d27ab8a-a0df-47cf-b4cc-d2d7a0712a05",
"type": "Person",
"name": "山本秀哉",
"sort-name": "Yamamoto, Shuya",
"country": "JP"
}
},
{
"type": "vocal",
"type-id": "0fdbe3c6-7700-4a31-ae54-b53f06ae1cfa",
"target": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"direction": "backward",
"artist": {
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"type": "Person",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"country": "JP"
}
}
],
"work-relation-list": [
{
"type": "performance",
"type-id": "a3005666-a872-32c3-ad06-98af558e99b0",
"target": "9e14d6b2-ac7d-43e9-82a9-561bc76ce2ed",
"direction": "forward",
"work": {
"id": "9e14d6b2-ac7d-43e9-82a9-561bc76ce2ed",
"type": "Song",
"title": "百花繚乱",
"language": "jpn",
"artist-relation-list": [
{
"type": "composer",
"type-id": "d59d99ea-23d4-4a80-b066-edca32ee158f",
"target": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"direction": "backward",
"artist": {
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"type": "Person",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"country": "JP"
}
},
{
"type": "lyricist",
"type-id": "3e48faba-ec01-47fd-8e89-30e81161661c",
"target": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"direction": "backward",
"artist": {
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"type": "Person",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"country": "JP"
}
}
],
"url-relation-list": [
{
"type": "lyrics",
"type-id": "e38e65aa-75e0-42ba-ace0-072aeb91a538",
"target": "https://utaten.com/lyric/tt24121002/",
"direction": "backward"
},
{
"type": "lyrics",
"type-id": "e38e65aa-75e0-42ba-ace0-072aeb91a538",
"target": "https://www.uta-net.com/song/366579/",
"direction": "backward"
}
]
}
}
],
"artist-credit-phrase": "幾田りら"
]
},
"title": "In Bloom"
}
]
}
],
"packaging": null,
"packaging-id": null,
"quality": "normal",
"release-group": {
"aliases": [],
"artist-credit": [
{
"artist": {
"aliases": [
{
"begin": null,
"end": null,
"ended": false,
"locale": "en",
"name": "Lilas Ikuta",
"primary": true,
"sort-name": "Ikuta, Lilas",
"type": "Artist name",
"type-id": "894afba6-2816-3c24-8072-eadb66bd04bc"
}
],
"country": "JP",
"disambiguation": "",
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"type": "Person",
"type-id": "b6e035f4-3ce9-331c-97df-83397230b0df"
},
"joinphrase": "",
"name": "幾田りら"
}
],
"disambiguation": "",
"first-release-date": "2025-01-10",
"genres": [],
"id": "da0d6bbb-f44b-4fff-8739-9d72db0402a1",
"primary-type": "Single",
"primary-type-id": "d6038452-8ee0-3f68-affc-2de9a1ede0b9",
"secondary-type-ids": [],
"secondary-types": [],
"tags": [],
"title": "百花繚乱"
},
"release-relations": [
{
"attribute-ids": {},
"attribute-values": {},
"attributes": [],
"begin": null,
"direction": "backward",
"end": null,
"ended": false,
"release": {
"artist-credit": [
{
"artist": {
"country": "JP",
"disambiguation": "",
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"type": null,
"type-id": null
},
"artist-credit-phrase": "Lilas Ikuta",
"track_or_recording_length": "179239"
"joinphrase": "",
"name": "幾田りら"
}
],
"track-count": 1
}
],
"medium-count": 1,
"release-relation-list": [
{
"type": "transl-tracklisting",
"type-id": "fc399d47-23a7-4c28-bfcf-0607a562b644",
"target": "a5ce1d11-2e32-45a4-b37f-c1589d46b103",
"direction": "backward",
"release": {
"id": "a5ce1d11-2e32-45a4-b37f-c1589d46b103",
"title": "百花繚乱",
"quality": "normal",
"text-representation": {
"language": "jpn",
"script": "Jpan"
},
"artist-credit": [
{
"artist": {
"id": "55e42264-ef27-49d8-93fd-29f930dc96e4",
"name": "幾田りら",
"sort-name": "Ikuta, Lilas",
"country": "JP"
}
}
],
"date": "2025-01-10",
"country": "XW",
"release-event-list": [
{
"date": "2025-01-10",
"area": {
"id": "525d4e18-3d00-31b9-a58b-a146a916de8f",
"name": "[Worldwide]",
"sort-name": "[Worldwide]",
"iso-3166-1-code-list": [
"XW"
]
}
}
],
"release-event-count": 1,
"barcode": "199066336168",
"medium-list": [],
"medium-count": 0,
"artist-credit-phrase": "幾田りら"
}
}
],
"artist-credit-phrase": "Lilas Ikuta"
}
}
"barcode": "199066336168",
"country": "XW",
"date": "2025-01-10",
"disambiguation": "",
"id": "a5ce1d11-2e32-45a4-b37f-c1589d46b103",
"media": [],
"packaging": null,
"packaging-id": null,
"quality": "normal",
"release-events": [
{
"area": {
"disambiguation": "",
"id": "525d4e18-3d00-31b9-a58b-a146a916de8f",
"iso-3166-1-codes": [
"XW"
],
"name": "[Worldwide]",
"sort-name": "[Worldwide]",
"type": null,
"type-id": null
},
"date": "2025-01-10"
}
],
"release-group": null,
"status": null,
"status-id": null,
"text-representation": {
"language": "jpn",
"script": "Jpan"
},
"title": "百花繚乱"
},
"source-credit": "",
"target-credit": "",
"type": "transl-tracklisting",
"type-id": "fc399d47-23a7-4c28-bfcf-0607a562b644"
}
],
"status": "Pseudo-Release",
"status-id": "41121bb9-3413-3818-8a9a-9742318349aa",
"tags": [],
"text-representation": {
"language": "eng",
"script": "Latn"
},
"title": "In Bloom"
}