diff --git a/beetsplug/_utils/musicbrainz.py b/beetsplug/_utils/musicbrainz.py index d78a3edd5b..8aeb9595ad 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]] @@ -618,7 +625,7 @@ def request(self, *args, **kwargs) -> Response: kwargs["params"]["fmt"] = "json" return super().request(*args, **kwargs) - def get_json(self, *args, **kwargs): + def get_json(self, *args, **kwargs) -> JSONDict: """Fetch JSON data from MusicBrainz and normalize its field names.""" return self._normalize_data(super().get_json(*args, **kwargs)) @@ -696,6 +703,14 @@ def get_recording( 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/_utils/requests.py b/beetsplug/_utils/requests.py index 33356ff8a4..7c742695a6 100644 --- a/beetsplug/_utils/requests.py +++ b/beetsplug/_utils/requests.py @@ -90,7 +90,7 @@ def __init__(self, *args, **kwargs) -> None: self.mount("https://", adapter) self.mount("http://", adapter) - def request(self, *args, **kwargs): + def request(self, *args, **kwargs) -> requests.Response: """Execute HTTP request with automatic timeout and status validation. Ensures all requests have a timeout (defaults to 10 seconds) and raises @@ -115,7 +115,7 @@ class RateLimitAdapter(HTTPAdapter): Override `_wait_time()` for custom strategies (token bucket, burst, etc.). """ - def __init__(self, rate_limit: float = 0.25, **kwargs): + def __init__(self, rate_limit: float = 0.25, **kwargs) -> None: super().__init__(**kwargs) self.rate_limit = rate_limit self._last_request_time = 0.0 @@ -125,7 +125,9 @@ def _wait_time(self, elapsed: float) -> float: """Return seconds to wait. Override for custom rate limiting.""" return max(0, self.rate_limit - elapsed) - def send(self, request: requests.PreparedRequest, *args, **kwargs): + def send( + self, request: requests.PreparedRequest, *args, **kwargs + ) -> requests.Response: with self._lock: elapsed = time.monotonic() - self._last_request_time wait = self._wait_time(elapsed) @@ -222,6 +224,6 @@ def delete(self, *args, **kwargs) -> requests.Response: """Perform HTTP DELETE request with automatic error handling.""" return self.request("delete", *args, **kwargs) - def get_json(self, *args, **kwargs): + def get_json(self, *args, **kwargs) -> Any: """Fetch and parse JSON data from an HTTP endpoint.""" return self.get(*args, **kwargs).json() diff --git a/beetsplug/acousticbrainz.py b/beetsplug/acousticbrainz.py index c70014d8ff..62f04a3ef4 100644 --- a/beetsplug/acousticbrainz.py +++ b/beetsplug/acousticbrainz.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections import defaultdict -from typing import TYPE_CHECKING, ClassVar, Protocol +from typing import TYPE_CHECKING, Any, ClassVar, Protocol import requests @@ -12,8 +12,12 @@ from beets.exceptions import UserError if TYPE_CHECKING: + from collections.abc import Iterator, Sequence + from beets.importer import ImportSession, ImportTask - from beets.library import Library + from beets.library import Item, Library + + from ._typing import JSONDict class AcousticBrainzCLIOpts(Protocol): @@ -101,7 +105,7 @@ def __init__(self) -> None: if self.config["auto"]: self.register_listener("import_task_files", self.import_task_files) - def commands(self): + def commands(self) -> list[ui.Subcommand]: cmd = ui.Subcommand( "acousticbrainz", help="fetch metadata from AcousticBrainz" ) @@ -121,7 +125,7 @@ def func( 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 @@ -133,7 +137,7 @@ def import_task_files( """Function is called upon beet import.""" self._fetch_info(task.imported_items(), False, True) - def _get_data(self, mbid): + def _get_data(self, mbid: str) -> JSONDict: if not self.base_url: raise UserError( "This plugin is deprecated since AcousticBrainz has shut " @@ -161,7 +165,9 @@ def _get_data(self, mbid): return data - def _fetch_info(self, items, write, force): + def _fetch_info( + self, items: Sequence[Item], write: bool, force: bool + ) -> None: """Fetch additional information from AcousticBrainz for the `item`s.""" tags = self.config["tags"].as_str_seq() for item in items: @@ -200,7 +206,9 @@ def _fetch_info(self, items, write, force): if write: item.try_write() - def _map_data_to_scheme(self, data, scheme): + def _map_data_to_scheme( + self, data: JSONDict, scheme: JSONDict + ) -> Iterator[tuple[str, Any]]: """Given `data` as a structure of nested dictionaries, and `scheme` as a structure of nested dictionaries , `yield` tuples `(attr, val)` where `attr` and `val` are corresponding leaf @@ -254,7 +262,7 @@ def _map_data_to_scheme(self, data, scheme): # `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 @@ -262,7 +270,12 @@ def _map_data_to_scheme(self, data, scheme): for composite_attr, value_parts in composites.items(): yield composite_attr, " ".join(value_parts) - def _data_to_scheme_child(self, subdata, subscheme, composites): + def _data_to_scheme_child( + self, + subdata: JSONDict, + subscheme: JSONDict, + composites: dict[str, list[str]], + ) -> Iterator[tuple[str, Any]]: """The recursive business logic of :meth:`_map_data_to_scheme`: Traverse two structures of nested dictionaries in parallel and `yield` tuples of corresponding leaf nodes. @@ -300,7 +313,7 @@ def _data_to_scheme_child(self, subdata, subscheme, composites): ) -def _generate_urls(base_url, mbid): +def _generate_urls(base_url: str, mbid: str) -> Iterator[str]: """Generates AcousticBrainz end point urls for given `mbid`.""" for level in LEVELS: yield f"{base_url}{mbid}{level}" diff --git a/beetsplug/beatport.py b/beetsplug/beatport.py index e5334bef65..d8761565b1 100644 --- a/beetsplug/beatport.py +++ b/beetsplug/beatport.py @@ -44,7 +44,11 @@ class BeatportClient: _api_base = "https://oauth-api.beatport.com" def __init__( - self, c_key, c_secret, auth_key=None, auth_secret=None + self, + c_key: str, + c_secret: str, + auth_key: str | None = None, + auth_secret: str | None = None, ) -> None: """Initiate the client with OAuth information. @@ -118,7 +122,7 @@ def search( self, query: str, release_type: Literal["release", "track"], - details=True, + details: bool = True, ) -> Iterator[BeatportRelease | BeatportTrack]: """Perform a search of the Beatport catalogue. @@ -389,7 +393,7 @@ def item_candidates( self._log.debug("API Error: {} (query: {})", e, query) return [] - def album_for_id(self, album_id: str): + def album_for_id(self, album_id: str) -> AlbumInfo | None: """Fetches a release by its Beatport ID and returns an AlbumInfo object or None if the query is not a valid ID or release is not found. """ @@ -404,7 +408,7 @@ def album_for_id(self, album_id: str): return self._get_album_info(release) return None - def track_for_id(self, track_id: str): + def track_for_id(self, track_id: str) -> TrackInfo | None: """Fetches a track by its Beatport ID and returns a TrackInfo object or None if the track is not a valid Beatport ID or track is not found. """ @@ -488,13 +492,15 @@ def _get_track_info(self, track: BeatportTrack) -> TrackInfo: genres=track.genres, ) - def _get_artist(self, artists): + def _get_artist( + 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): + def _get_tracks(self, query: str) -> list[TrackInfo]: """Returns a list of TrackInfo objects for a Beatport query.""" bp_tracks = self.client.search(query, release_type="track") return [self._get_track_info(x) for x in bp_tracks] diff --git a/beetsplug/chroma.py b/beetsplug/chroma.py index 4c57a2475a..d149b79d17 100644 --- a/beetsplug/chroma.py +++ b/beetsplug/chroma.py @@ -8,7 +8,7 @@ import re from collections import defaultdict from functools import cached_property, partial -from typing import TYPE_CHECKING, Protocol +from typing import TYPE_CHECKING, Any, Protocol import acoustid import confuse @@ -21,14 +21,16 @@ if TYPE_CHECKING: import optparse - from collections.abc import Iterable, Iterator + from collections.abc import Iterable, Iterator, Sequence - from beets.autotag import TrackInfo + from beets.autotag import AlbumInfo, TrackInfo from beets.importer import ImportSession, ImportTask - from beets.library import Library - from beets.library.models import Item + from beets.library import Item, Library + from beets.logging import BeetsLogger as Logger from beetsplug.musicbrainz import MusicBrainzPlugin + from ._typing import JSONDict + class ChromaSearchCLIOpts(Protocol): count: int @@ -58,7 +60,7 @@ class ChromaSearchCLIOpts(Protocol): _acoustids: dict[bytes, str] = {} -def prefix(it, count): +def prefix(it: Iterable[Any], count: int) -> Iterator[Any]: """Truncate an iterable to at most `count` items.""" for i, v in enumerate(it): if i >= count: @@ -66,7 +68,9 @@ def prefix(it, count): yield v -def releases_key(release, countries, original_year): +def releases_key( + release: JSONDict, countries: Sequence[re.Pattern[str]], original_year: str +) -> tuple[int, int, int, int]: """Used as a key to sort releases by date then preferred country""" date = release.get("date") if date and original_year: @@ -89,7 +93,7 @@ def releases_key(release, countries, original_year): return (year, month, day, country_key) -def acoustid_match(log, path): +def acoustid_match(log: Logger, path: bytes) -> None: """Gets metadata for a file from Acoustid and populates the _matches, _fingerprints, and _acoustids dictionaries accordingly. """ @@ -144,7 +148,7 @@ def acoustid_match(log, path): # '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 @@ -161,12 +165,12 @@ def acoustid_match(log, path): # Plugin structure and autotagging logic. -def _all_releases(items): +def _all_releases(items: Sequence[Item]) -> Iterator[str]: """Given an iterable of Items, determines (according to Acoustid) 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 @@ -216,7 +220,7 @@ def fingerprint_task( ) -> None: return fingerprint_task(self._log, task, session) - def track_distance(self, item, info): + def track_distance(self, item: Item, info: TrackInfo) -> Distance: dist = Distance() if item.path not in _matches or not info.track_id: # Match failed or no track ID. @@ -226,20 +230,24 @@ def track_distance(self, item, info): dist.add_expr("track_id", info.track_id not in recording_ids) return dist - def candidates(self, items, artist, album, va_likely): + def candidates( + self, items: Sequence[Item], artist: str, album: str, va_likely: bool + ) -> list[AlbumInfo]: 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 - def item_candidates(self, item, artist, title) -> Iterable[TrackInfo]: + def item_candidates( + self, item: Item, artist: str, title: str + ) -> Iterable[TrackInfo]: if item.path not in _matches: return [] @@ -255,15 +263,15 @@ def item_candidates(self, item, artist, title) -> Iterable[TrackInfo]: self._log.debug("acoustid item candidates: {}", len(tracks)) return tracks - def album_for_id(self, *args, **kwargs): + def album_for_id(self, *args, **kwargs) -> None: # Lookup by fingerprint ID does not make too much sense. return None - def track_for_id(self, *args, **kwargs): + def track_for_id(self, *args, **kwargs) -> None: # Lookup by fingerprint ID does not make too much sense. return None - def commands(self): + def commands(self) -> list[ui.Subcommand]: submit_cmd = ui.Subcommand( "submit", help="submit Acoustid fingerprints" ) @@ -293,7 +301,7 @@ def fingerprint_cmd_func( return [submit_cmd, fingerprint_cmd, self.chromasearch_cmd()] - def chromasearch_cmd(self): + def chromasearch_cmd(self) -> ui.Subcommand: cmd = ui.Subcommand( "chromasearch", help="search local database by chroma fingerprint" ) @@ -375,7 +383,9 @@ def search_cmd_func( # Hooks into import process. -def fingerprint_task(log, task: ImportTask, session: ImportSession) -> None: +def fingerprint_task( + log: Logger, task: ImportTask, session: ImportSession +) -> None: """Fingerprint each item in the task for later use during the autotagging candidate search. """ @@ -395,11 +405,14 @@ def apply_acoustid_metadata(task: ImportTask, session: ImportSession) -> None: # UI commands. -def submit_items(log, userkey, items, chunksize=64): +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(): + def submit_chunk() -> None: """Submit the current accumulated fingerprint data.""" log.info("submitting {} fingerprints", len(data)) try: @@ -440,7 +453,9 @@ def submit_chunk(): submit_chunk() -def fingerprint_item(log, item, write=False, quiet=False): +def fingerprint_item( + log: Logger, item: Item, write: bool = False, quiet: bool = False +) -> str | None: """Get the fingerprint for an Item. If the item already has a fingerprint, it is not regenerated. If fingerprint generation fails, return None. If the items are associated with a library, they are @@ -481,13 +496,13 @@ def __init__(self, item: Item, score: float) -> None: self.item = item self.score = score - def __lt__(self, other): - return self.score < other.score + def __lt__(self, other: object) -> bool: + return type(self) is type(other) and self.score < other.score - def __gt__(self, other): - return self.score > other.score + def __gt__(self, other: object) -> bool: + return type(self) is type(other) and self.score > other.score - def __str__(self): + def __str__(self) -> str: percent = f"{round(self.score * 100, 2)}%".rjust(6) if self.score >= 0.95: percent = colorize("text_success", percent) @@ -504,7 +519,7 @@ def __init__(self, n: int) -> None: self.n = n self.heap: list[ScoredItem] = [] - def add(self, value: ScoredItem): + def add(self, value: ScoredItem) -> None: if len(self.heap) < self.n: heapq.heappush(self.heap, value) else: diff --git a/beetsplug/deezer.py b/beetsplug/deezer.py index fdc69ed0ca..c4100b22ee 100644 --- a/beetsplug/deezer.py +++ b/beetsplug/deezer.py @@ -41,7 +41,7 @@ class DeezerPlugin(SearchApiMetadataSourcePlugin[IDResponse]): def __init__(self) -> None: super().__init__() - def commands(self): + def commands(self) -> list[ui.Subcommand]: """Add beet UI commands to interact with Deezer.""" deezer_update_cmd = ui.Subcommand( "deezerupdate", help=f"Update {self.data_source} rank" @@ -252,7 +252,7 @@ def get_search_response(self, params: SearchParams) -> list[IDResponse]: response.raise_for_status() return response.json()["data"] - def deezerupdate(self, items: Sequence[Item], write: bool): + def deezerupdate(self, items: Sequence[Item], write: bool) -> None: """Obtain rank information from Deezer.""" for index, item in enumerate(items, start=1): self._log.info( @@ -264,22 +264,22 @@ def deezerupdate(self, items: Sequence[Item], write: bool): 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() - - def fetch_data(self, url: str): + 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: response = requests.get(url, timeout=10) response.raise_for_status() diff --git a/beetsplug/discogs/__init__.py b/beetsplug/discogs/__init__.py index bb62a0a960..a24e28f7a9 100644 --- a/beetsplug/discogs/__init__.py +++ b/beetsplug/discogs/__init__.py @@ -33,6 +33,7 @@ if TYPE_CHECKING: from collections.abc import Callable, Iterator, Sequence + from beets.importer import ImportSession from beets.library import Item from beets.metadata_plugins import QueryType, SearchParams @@ -78,7 +79,7 @@ class DiscogsPlugin(SearchApiMetadataSourcePlugin[IDResponse]): - def __init__(self): + def __init__(self) -> None: super().__init__() self.config.add( { @@ -123,7 +124,7 @@ def extra_discogs_field_by_tag(self) -> dict[str, str]: return field_by_tag - def setup(self, session=None) -> None: + def setup(self, session: ImportSession | None = None) -> None: """Create the `discogs_client` field. Authenticate if necessary.""" c_key = self.config["apikey"].as_str() c_secret = self.config["apisecret"].as_str() diff --git a/beetsplug/listenbrainz.py b/beetsplug/listenbrainz.py index 3ab2f7fa8b..5c572dfd78 100644 --- a/beetsplug/listenbrainz.py +++ b/beetsplug/listenbrainz.py @@ -21,10 +21,12 @@ from ._utils.requests import TimeoutAndRetrySession if TYPE_CHECKING: + from collections.abc import Iterable from pathlib import Path from beets.library import Library + from ._typing import JSONDict from ._utils.playcount import Track @@ -63,7 +65,7 @@ class ListenBrainzPlugin(MusicBrainzAPIMixin, BeetsPlugin): "listenbrainz_play_count": types.INTEGER } - def __init__(self): + def __init__(self) -> None: """Initialize the plugin.""" super().__init__() self.token = self.config["token"].get() @@ -72,7 +74,7 @@ def __init__(self): self.AUTH_HEADER = {"Authorization": f"Token {self.token}"} config["listenbrainz"]["token"].redact = True - def commands(self): + def commands(self) -> list[ui.Subcommand]: """Add beet UI commands to interact with ListenBrainz.""" lbupdate_cmd = ui.Subcommand( "lbimport", help="Import ListenBrainz history" @@ -110,10 +112,10 @@ def func(lib: Library, opts: LBImportCLIOpts, args: list[str]) -> None: def _lbupdate( self, - lib, + lib: Library, export_file: str | None = None, max_listens: int | None = None, - ): + ) -> None: """Update play counts from ListenBrainz listening history.""" listens: list[Listen] | None if export_file is not None: @@ -145,7 +147,7 @@ def _lbupdate( 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). @@ -171,7 +173,9 @@ def _aggregate_listens(tracks: list[Track]) -> list[Track]: for key, info in track_info.items() ] - def _make_request(self, url, params=None): + def _make_request( + self, url: str, params: JSONDict | None = None + ) -> JSONDict | None: """Makes a request to the ListenBrainz API. Respects the X-RateLimit-* headers returned by the server: if the @@ -232,7 +236,11 @@ def import_listenbrainz_data_export( return all_listens def get_listens( - self, min_ts=None, max_ts=None, count=None, max_total=None + self, + min_ts: int | None = None, + max_ts: int | None = None, + count: int | None = None, + max_total: int | None = None, ) -> list[Listen] | None: """Gets the listening history of a given user from the ListenBrainz API. @@ -297,7 +305,7 @@ def get_listens( 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: @@ -321,7 +329,7 @@ def get_tracks_from_listens(self, listens: list[Listen]) -> list[Track]: ) return tracks - def get_mb_recording_id(self, track) -> str | None: + def get_mb_recording_id(self, track: JSONDict) -> str | None: """Returns the MusicBrainz recording ID for a track.""" results = self.mb_api.search( "recording", @@ -332,14 +340,16 @@ def get_mb_recording_id(self, track) -> str | None: ) return next((r["id"] for r in results), None) - def get_playlists_createdfor(self, username): + def get_playlists_createdfor(self, username: str) -> JSONDict | None: """Returns a list of playlists created by a user.""" url = f"{self.ROOT}/user/{username}/playlists/createdfor" return self._make_request(url) - def get_listenbrainz_playlists(self): + 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: @@ -372,15 +382,15 @@ def get_listenbrainz_playlists(self): self._log.debug("Playlist: {0[type]} - {0[date]}", playlist) return listenbrainz_playlists - def get_playlist(self, identifier): + def get_playlist(self, identifier: str) -> JSONDict | None: """Returns a playlist.""" url = f"{self.ROOT}/playlist/{identifier}" return self._make_request(url) - def get_tracks_from_playlist(self, playlist): + 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] @@ -394,23 +404,19 @@ def get_tracks_from_playlist(self, playlist): ) return self.get_track_info(tracks) - def get_track_info(self, tracks): + 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 @@ -426,7 +432,9 @@ def get_track_info(self, tracks): ) return track_info - def get_weekly_playlist(self, playlist_type, most_recent=True): + def get_weekly_playlist( + self, playlist_type: str, most_recent: bool = True + ) -> list[JSONDict]: # Fetch all playlists playlists = self.get_listenbrainz_playlists() # Filter playlists by type @@ -446,5 +454,8 @@ def get_weekly_playlist(self, playlist_type, most_recent=True): 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 4c7b61bfaa..d0e1b5e764 100644 --- a/beetsplug/mbsubmit.py +++ b/beetsplug/mbsubmit.py @@ -10,19 +10,20 @@ 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 + from collections.abc import Sequence from beets.importer import ImportSession, ImportTask - from beets.library import Library + from beets.library import Item, Library class MBSubmitPlugin(BeetsPlugin): @@ -61,7 +62,7 @@ def before_choose_candidate_event( ] return [] - def picard(self, session, task): + def picard(self, session: ImportSession, task: ImportTask) -> None: paths = [] for p in task.paths: paths.append(displayable_path(p)) @@ -72,11 +73,15 @@ def picard(self, session, task): except OSError as exc: self._log.error("Could not open picard, got error:\n{}", exc) - def print_tracks(self, session, task): + @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): + def commands(self) -> list[ui.Subcommand]: """Add beet UI commands for mbsubmit.""" mbsubmit_cmd = ui.Subcommand( "mbsubmit", help="Submit Tracks to MusicBrainz" @@ -90,7 +95,7 @@ def func(lib: Library, opts: optparse.Values, args: list[str]) -> None: return [mbsubmit_cmd] - def _mbsubmit(self, items): + 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 4fbeb8da5a..e880703364 100644 --- a/beetsplug/spotify.py +++ b/beetsplug/spotify.py @@ -27,7 +27,7 @@ from beets.util import chunks if TYPE_CHECKING: - from collections.abc import Sequence + from collections.abc import Iterable, Mapping, Sequence from beets.library import Item, Library from beets.metadata_plugins import QueryType, SearchParams @@ -151,7 +151,7 @@ class SpotifyPlugin( "valence": "spotify_valence", } - def __init__(self): + def __init__(self) -> None: super().__init__() self.config.add( { @@ -176,7 +176,7 @@ def __init__(self): ) # Protects audio_features_available self.setup() - def setup(self): + def setup(self) -> None: """Retrieve previously saved OAuth token or generate a new one.""" try: @@ -319,7 +319,7 @@ def _handle_response( 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 @@ -590,7 +590,7 @@ def func( sync_cmd.func = func return [spotify_cmd, sync_cmd] - def _parse_opts(self, opts): + def _parse_opts(self, opts: SpotifyCLIOpts) -> bool: if opts.mode: self.config["mode"].set(opts.mode) @@ -606,7 +606,9 @@ def _parse_opts(self, opts): self.opts = opts return True - def _match_library_tracks(self, library: Library, keywords: list[str]): + def _match_library_tracks( + self, library: Library, keywords: Sequence[str] + ) -> list[SearchResponseAlbums | SearchResponseTracks] | None: """Get simplified track object dicts for library tracks. Matches tracks based on the specified ``keywords``. @@ -718,7 +720,9 @@ def _match_library_tracks(self, library: Library, keywords: list[str]): return results - def _output_match_results(self, results): + def _output_match_results( + self, results: Sequence[Mapping[str, Any]] | None + ) -> None: """Open a playlist or print Spotify URLs. Uses the provided track object dicts. @@ -879,7 +883,9 @@ def _fetch_info( for item, _ in items_to_update: item.store() - def track_info(self, track_id: str): + 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}" @@ -898,7 +904,7 @@ def track_info(self, track_id: str): external_ids.get("upc"), ) - def track_audio_features(self, track_id: str): + def track_audio_features(self, track_id: str) -> JSONDict | None: """Fetch track audio features by its Spotify ID. Thread-safe: avoids redundant API calls and logs the 403 warning only diff --git a/test/plugins/test_listenbrainz.py b/test/plugins/test_listenbrainz.py index 8fa3a3cd2b..92f7209f19 100644 --- a/test/plugins/test_listenbrainz.py +++ b/test/plugins/test_listenbrainz.py @@ -43,7 +43,7 @@ def test_get_mb_recording_id( 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": [],