From 4e1fdf0bf3db0c0c6d2aa06dab5fd1769cfb9fda Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=A0ar=C5=ABnas=20Nejus?= Date: Wed, 19 Aug 2026 03:14:39 +0100 Subject: [PATCH] typing: fix annotations in metadata source plugins --- beetsplug/_utils/musicbrainz.py | 39 +++++++++++++++++++++---------- beetsplug/acousticbrainz.py | 4 ++-- beetsplug/beatport.py | 4 ++-- beetsplug/chroma.py | 21 +++++++++-------- beetsplug/deezer.py | 22 ++++++++--------- beetsplug/listenbrainz.py | 38 ++++++++++++++++-------------- beetsplug/mbsubmit.py | 10 +++++--- beetsplug/spotify.py | 10 ++++---- test/plugins/test_listenbrainz.py | 2 +- 9 files changed, 87 insertions(+), 63 deletions(-) diff --git a/beetsplug/_utils/musicbrainz.py b/beetsplug/_utils/musicbrainz.py index 4d973dc09..8aeb9595a 100644 --- a/beetsplug/_utils/musicbrainz.py +++ b/beetsplug/_utils/musicbrainz.py @@ -454,6 +454,10 @@ class Recording(TypedDict): work_relations: NotRequired[list[WorkRelation]] +class RecordingWithReleases(Recording): + releases: list[BaseRelease] + + class Track(TypedDict): artist_credit: list[ArtistCredit] id: str @@ -501,30 +505,33 @@ class ReleaseRelation(RelationBase): release: ReleaseRelationRelease -class Release(TypedDict): - aliases: list[Alias] +class BaseRelease(TypedDict): artist_credit: list[ArtistCredit] - asin: str | None barcode: str | None - cover_art_archive: CoverArtArchive disambiguation: str - genres: list[Genre] id: str - label_info: list[LabelInfo] - media: list[Medium] packaging: ReleasePackaging | None packaging_id: str | None - quality: ReleaseQuality - release_group: ReleaseGroup status: ReleaseStatus | None status_id: str | None - tags: list[Tag] + quality: ReleaseQuality text_representation: TextRepresentation - title: str - artist_relations: NotRequired[list[ArtistRelation]] country: NotRequired[str | None] date: NotRequired[str] + title: str release_events: NotRequired[list[ReleaseEvent]] + + +class Release(BaseRelease): + aliases: list[Alias] + asin: str | None + cover_art_archive: CoverArtArchive + genres: list[Genre] + label_info: list[LabelInfo] + media: list[Medium] + release_group: ReleaseGroup + tags: list[Tag] + artist_relations: NotRequired[list[ArtistRelation]] release_relations: NotRequired[list[ReleaseRelation]] url_relations: NotRequired[list[UrlRelation]] @@ -696,6 +703,14 @@ class MusicBrainzAPI(RequestHandler): kwargs.setdefault("includes", RECORDING_INCLUDES) return self._lookup("recording", id_, **kwargs) + def get_base_recording_with_releases( + self, id_: str, **kwargs: Unpack[LookupKwargs] + ) -> RecordingWithReleases: + """Retrieve a recording by its MusicBrainz ID.""" + kwargs.setdefault("includes", []) + kwargs["includes"].append("releases") + return self._lookup("recording", id_, **kwargs) + def get_work(self, id_: str, **kwargs: Unpack[LookupKwargs]) -> Work: """Retrieve a work by its MusicBrainz ID.""" return self._lookup("work", id_, **kwargs) diff --git a/beetsplug/acousticbrainz.py b/beetsplug/acousticbrainz.py index 8f043ba81..62f04a3ef 100644 --- a/beetsplug/acousticbrainz.py +++ b/beetsplug/acousticbrainz.py @@ -125,7 +125,7 @@ class AcousticPlugin(plugins.BeetsPlugin): self._fetch_info( items, ui.should_write(), - opts.force_refetch or self.config["force"], + opts.force_refetch or self.config["force"].get(bool), ) cmd.func = func @@ -262,7 +262,7 @@ class AcousticPlugin(plugins.BeetsPlugin): # `composites = {'initial_key': ['B', 'minor']}`. # The recursive traversal. - composites = defaultdict(list) + composites = defaultdict[str, list[str]](list) yield from self._data_to_scheme_child(data, scheme, composites) # When composites has been populated, yield the composite attributes diff --git a/beetsplug/beatport.py b/beetsplug/beatport.py index 337d69361..d8761565b 100644 --- a/beetsplug/beatport.py +++ b/beetsplug/beatport.py @@ -493,12 +493,12 @@ class BeatportPlugin(MetadataSourcePlugin): ) def _get_artist( - self, artists: list[tuple[str, str]] | None + self, artists: Iterable[tuple[str, str]] | None ) -> tuple[str, str | None]: """Returns an artist string (all artists) and an artist_id (the main artist) for a list of Beatport release or track artists. """ - return self.get_artist(artists=artists, id_key=0, name_key=1) + return self.get_artist(artists or [], id_key=0, name_key=1) # type: ignore[arg-type] def _get_tracks(self, query: str) -> list[TrackInfo]: """Returns a list of TrackInfo objects for a Beatport query.""" diff --git a/beetsplug/chroma.py b/beetsplug/chroma.py index 4083bfd43..d149b79d1 100644 --- a/beetsplug/chroma.py +++ b/beetsplug/chroma.py @@ -148,7 +148,7 @@ def acoustid_match(log: Logger, path: bytes) -> None: # 'countries' to then sort preferred countries first. country_patterns = config["match"]["preferred"]["countries"].as_str_seq() countries = [re.compile(pat, re.I) for pat in country_patterns] - original_year = config["match"]["preferred"]["original_year"] + original_year = config["match"]["preferred"]["original_year"].as_str() releases.sort( key=partial( releases_key, countries=countries, original_year=original_year @@ -170,7 +170,7 @@ def _all_releases(items: Sequence[Item]) -> Iterator[str]: which releases the items have in common. Generates release IDs. """ # Count the number of "hits" for each release. - relcounts = defaultdict(int) + relcounts = defaultdict[str, int](int) for item in items: if item.path not in _matches: continue @@ -236,11 +236,11 @@ class AcoustidPlugin(MetadataSourcePlugin): if self.mb is None: return [] - albums = [] - for relid in prefix(_all_releases(items), MAX_RELEASES): - album = self.mb.album_for_id(relid) - if album: - albums.append(album) + albums = [ + a + for relid in prefix(_all_releases(items), MAX_RELEASES) + if (a := self.mb.album_for_id(relid)) + ] self._log.debug("acoustid album candidates: {}", len(albums)) return albums @@ -409,7 +409,8 @@ def submit_items( log: Logger, userkey: str, items: Sequence[Item], chunksize: int = 64 ) -> None: """Submit fingerprints for the items to the Acoustid server.""" - data = [] # The running list of dictionaries to submit. + # The running list of dictionaries to submit. + data: list[JSONDict] = [] def submit_chunk() -> None: """Submit the current accumulated fingerprint data.""" @@ -496,10 +497,10 @@ class ScoredItem: self.score = score def __lt__(self, other: object) -> bool: - return self.score < other.score + return type(self) is type(other) and self.score < other.score def __gt__(self, other: object) -> bool: - return self.score > other.score + return type(self) is type(other) and self.score > other.score def __str__(self) -> str: percent = f"{round(self.score * 100, 2)}%".rjust(6) diff --git a/beetsplug/deezer.py b/beetsplug/deezer.py index bdee75744..c4100b22e 100644 --- a/beetsplug/deezer.py +++ b/beetsplug/deezer.py @@ -264,20 +264,20 @@ class DeezerPlugin(SearchApiMetadataSourcePlugin[IDResponse]): self._log.debug("No deezer_track_id present for: {}", item) continue try: - rank = self.fetch_data( - f"{self.track_url}{deezer_track_id}" - ).get("rank") - self._log.debug( - "Deezer track: {} has {} rank", deezer_track_id, rank - ) + track = self.fetch_data(f"{self.track_url}{deezer_track_id}") except Exception as e: self._log.debug("Invalid Deezer track_id: {}", e) continue - item.deezer_track_rank = int(rank) - item.store() - item.deezer_updated = time.time() - if write: - item.try_write() + else: + if track and (rank := track.get("rank") is not None): + self._log.debug( + "Deezer track: {} has {} rank", deezer_track_id, rank + ) + item.deezer_track_rank = int(rank) + item.store() + item.deezer_updated = time.time() + if write: + item.try_write() def fetch_data(self, url: str) -> JSONDict | None: try: diff --git a/beetsplug/listenbrainz.py b/beetsplug/listenbrainz.py index 40bdcf9d6..5c572dfd7 100644 --- a/beetsplug/listenbrainz.py +++ b/beetsplug/listenbrainz.py @@ -21,6 +21,7 @@ from ._utils.playcount import update_play_counts from ._utils.requests import TimeoutAndRetrySession if TYPE_CHECKING: + from collections.abc import Iterable from pathlib import Path from beets.library import Library @@ -146,7 +147,7 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin): self._log.info("{} play-counts imported", found) @staticmethod - def _aggregate_listens(tracks: list[Track]) -> list[Track]: + def _aggregate_listens(tracks: Iterable[Track]) -> list[Track]: """Aggregate individual listen events into per-track play counts. ListenBrainz returns individual listen events (each with playcount=1). @@ -304,7 +305,7 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin): return all_listens - def get_tracks_from_listens(self, listens: list[Listen]) -> list[Track]: + def get_tracks_from_listens(self, listens: Iterable[Listen]) -> list[Track]: """Returns a list of tracks from a list of listens.""" tracks: list[Track] = [] for track in listens: @@ -346,7 +347,9 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin): def get_listenbrainz_playlists(self) -> list[JSONDict]: resp = self.get_playlists_createdfor(self.username) - playlists = resp.get("playlists") + if not resp: + return [] + playlists = resp.get("playlists", []) listenbrainz_playlists = [] for playlist in playlists: @@ -387,7 +390,7 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin): def get_tracks_from_playlist(self, playlist: JSONDict) -> list[JSONDict]: """This function returns a list of tracks in the playlist.""" tracks = [] - for track in playlist.get("playlist").get("track"): + for track in playlist.get("playlist", {}).get("track"): identifier = track.get("identifier") if isinstance(identifier, list): identifier = identifier[0] @@ -401,23 +404,19 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin): ) return self.get_track_info(tracks) - def get_track_info(self, tracks: list[JSONDict]) -> list[JSONDict]: + def get_track_info(self, tracks: Iterable[JSONDict]) -> list[JSONDict]: track_info = [] for track in tracks: - identifier = track.get("identifier") - recording = self.mb_api.get_recording( - identifier, includes=["releases", "artist-credits"] - ) + identifier = track["identifier"] + recording = self.mb_api.get_base_recording_with_releases(identifier) title = recording.get("title") - artist_credit = recording.get("artist_credit", []) - if artist_credit: - artist = artist_credit[0].get("artist", {}).get("name") + if artist_credit := next(iter(recording["artist_credit"]), None): + artist = artist_credit.get("artist", {}).get("name") else: artist = None - releases = recording.get("releases", []) - if releases: - album = releases[0].get("title") - date = releases[0].get("date") + if release := next(iter(recording["releases"]), None): + album = release["title"] + date = release.get("date") year = date.split("-")[0] if date else None else: album = None @@ -455,5 +454,8 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin): f"- {selected_playlist['date']}" ) # Fetch and return tracks from the selected playlist - playlist = self.get_playlist(selected_playlist.get("identifier")) - return self.get_tracks_from_playlist(playlist) + if (identifier := selected_playlist.get("identifier")) and ( + playlist := self.get_playlist(identifier) + ): + return self.get_tracks_from_playlist(playlist) + return [] diff --git a/beetsplug/mbsubmit.py b/beetsplug/mbsubmit.py index a2ba0a572..d0e1b5e76 100644 --- a/beetsplug/mbsubmit.py +++ b/beetsplug/mbsubmit.py @@ -10,13 +10,13 @@ implemented by MusicBrainz yet. from __future__ import annotations import subprocess +from functools import cached_property from typing import TYPE_CHECKING from beets import ui from beets.autotag import Recommendation from beets.plugins import BeetsPlugin from beets.util import PromptChoice, displayable_path -from beetsplug.info import print_data if TYPE_CHECKING: import optparse @@ -73,9 +73,13 @@ class MBSubmitPlugin(BeetsPlugin): except OSError as exc: self._log.error("Could not open picard, got error:\n{}", exc) + @cached_property + def fmt(self) -> str: + return self.config["format"].as_str() + def print_tracks(self, session: ImportSession, task: ImportTask) -> None: for i in sorted(task.items, key=lambda i: i.track): - print_data(None, i, self.config["format"].as_str()) + ui.print_(format(i, self.fmt)) def commands(self) -> list[ui.Subcommand]: """Add beet UI commands for mbsubmit.""" @@ -94,4 +98,4 @@ class MBSubmitPlugin(BeetsPlugin): def _mbsubmit(self, items: Sequence[Item]) -> None: """Print track information to be submitted to MusicBrainz.""" for i in sorted(items, key=lambda i: i.track): - print_data(None, i, self.config["format"].as_str()) + ui.print_(format(i, self.fmt)) diff --git a/beetsplug/spotify.py b/beetsplug/spotify.py index 7f9a79d2d..e88070336 100644 --- a/beetsplug/spotify.py +++ b/beetsplug/spotify.py @@ -27,7 +27,7 @@ from beets.metadata_plugins import IDResponse, SearchApiMetadataSourcePlugin from beets.util import chunks if TYPE_CHECKING: - from collections.abc import Mapping, Sequence + from collections.abc import Iterable, Mapping, Sequence from beets.library import Item, Library from beets.metadata_plugins import QueryType, SearchParams @@ -319,7 +319,7 @@ class SpotifyPlugin( raise APIError("Request failed.") def _multi_artist_credit( - self, artists: list[dict[str | int, str]] + self, artists: Iterable[dict[str | int, str]] ) -> tuple[list[str], list[str]]: """Given a list of artist dictionaries, accumulate data into a pair of lists: the first being the artist names, and the second being the @@ -607,7 +607,7 @@ class SpotifyPlugin( return True def _match_library_tracks( - self, library: Library, keywords: list[str] + self, library: Library, keywords: Sequence[str] ) -> list[SearchResponseAlbums | SearchResponseTracks] | None: """Get simplified track object dicts for library tracks. @@ -883,7 +883,9 @@ class SpotifyPlugin( for item, _ in items_to_update: item.store() - def track_info(self, track_id: str) -> tuple[Any, Any, Any, Any]: + def track_info( + self, track_id: str + ) -> tuple[int | None, str | None, str | None, str | None]: """Fetch a track's popularity and external IDs using its Spotify ID.""" track_data = self._handle_response( "get", f"{self.track_url}/{track_id}" diff --git a/test/plugins/test_listenbrainz.py b/test/plugins/test_listenbrainz.py index 8fa3a3cd2..92f7209f1 100644 --- a/test/plugins/test_listenbrainz.py +++ b/test/plugins/test_listenbrainz.py @@ -43,7 +43,7 @@ class TestListenBrainzPlugin(ConfigMixin): def test_get_track_info(self, plugin, requests_mock): requests_mock.get( - "/ws/2/recording/id1?inc=releases%2Bartist-credits", + "/ws/2/recording/id1?inc=releases", json={ "title": "T", "artist-credit": [],