diff --git a/.git-blame-ignore-revs b/.git-blame-ignore-revs index 8866dd6039..a9b43923d3 100644 --- a/.git-blame-ignore-revs +++ b/.git-blame-ignore-revs @@ -299,3 +299,5 @@ f37306b0cc17f17291ea5010178f190a95b29fd5 27900f58fbeba06483b77c677045ad1df8e9af45 # typing: rename AnyLibModel -> AlbumOrItem 33a01fe0b37942a38c8b88d0fc8167edd02ca81b +# typing: add types to command handlers +14eabc60600ff3c8ef1908100ae7abd3aa998bc4 diff --git a/beets/dbcore/sort.py b/beets/dbcore/sort.py index 159696c167..16a5e7c90e 100644 --- a/beets/dbcore/sort.py +++ b/beets/dbcore/sort.py @@ -5,6 +5,8 @@ from typing import TYPE_CHECKING, Any if TYPE_CHECKING: + from collections.abc import Sequence + from beets.dbcore.db import AnyModel, Model @@ -19,8 +21,8 @@ def order_clause(self) -> str | None: """ return None - def sort(self, items: list[AnyModel]) -> list[AnyModel]: - """Sort the list of objects and return a list.""" + def sort(self, items: Sequence[AnyModel]) -> Sequence[AnyModel]: + """Sort the given sequence of model objects.""" return sorted(items) def is_slow(self) -> bool: @@ -72,7 +74,7 @@ def is_slow(self) -> bool: return True return False - def sort(self, items: list[AnyModel]) -> list[AnyModel]: + def sort(self, items: Sequence[AnyModel]) -> Sequence[AnyModel]: slow_sorts = [] switch_slow = False for sort in reversed(self.sorts): @@ -114,7 +116,7 @@ def __init__( self.ascending = ascending self.case_insensitive = case_insensitive - def sort(self, objs: list[AnyModel]) -> list[AnyModel]: + def sort(self, objs: Sequence[AnyModel]) -> Sequence[AnyModel]: # TODO: Support flexible attributes with different types (e.g. a mix # of strings and numbers) without falling over. @@ -186,7 +188,7 @@ def is_slow(self) -> bool: class NullSort(Sort): """No sorting. Leave results unsorted.""" - def sort(self, items: list[AnyModel]) -> list[AnyModel]: + def sort(self, items: Sequence[AnyModel]) -> Sequence[AnyModel]: return items def __nonzero__(self) -> bool: @@ -214,7 +216,7 @@ def order_clause(self) -> str: return f"COALESCE(NULLIF({field}_sort, ''), {field}) {collate} {order}" - def sort(self, objs: list[AnyModel]) -> list[AnyModel]: + def sort(self, objs: Sequence[AnyModel]) -> Sequence[AnyModel]: def key(obj: Model) -> str | bytes: val = obj[f"{self.field}_sort"] or obj[self.field] return val.lower() if self.case_insensitive else val diff --git a/beets/ui/__init__.py b/beets/ui/__init__.py index 4f604b4e3f..a89598187a 100644 --- a/beets/ui/__init__.py +++ b/beets/ui/__init__.py @@ -468,7 +468,10 @@ def add_album_option(self, flags=("-a", "--album")): Sets the album property on the options extracted from the CLI. """ album = optparse.Option( - *flags, action="store_true", help="match albums instead of tracks" + *flags, + action="store_true", + default=False, + help="match albums instead of tracks", ) self.add_option(album) self._album_flags = set(flags) @@ -581,7 +584,7 @@ class Subcommand: invoked by a SubcommandOptionParser. """ - func: Callable[[library.Library, optparse.Values, list[str]], Any] + func: Callable[[library.Library, Any, list[str]], Any] def __init__(self, name, parser=None, help="", aliases=(), hide=False): # noqa: A002 """Creates a new subcommand. name is the primary way to invoke diff --git a/beets/ui/commands/completion.py b/beets/ui/commands/completion.py index 776c389b4c..783f59e498 100644 --- a/beets/ui/commands/completion.py +++ b/beets/ui/commands/completion.py @@ -1,16 +1,27 @@ """The 'completion' command: print shell script for command line completion.""" +from __future__ import annotations + import os import re +from typing import TYPE_CHECKING from beets import library, logging, plugins, ui from beets.util import syspath +if TYPE_CHECKING: + import optparse + + from beets.library import Library + + # Global logger. log = logging.getLogger("beets") -def print_completion(*args): +def print_completion( + lib: Library, opts: optparse.Values, args: list[str] +) -> None: from beets.ui.commands import default_commands for line in completion_script(default_commands + plugins.commands()): diff --git a/beets/ui/commands/config.py b/beets/ui/commands/config.py index 31a5487fdd..449b237a1b 100644 --- a/beets/ui/commands/config.py +++ b/beets/ui/commands/config.py @@ -1,13 +1,27 @@ """The 'config' command: show and edit user configuration.""" +from __future__ import annotations + import os +from typing import TYPE_CHECKING, Protocol from beets import config, ui from beets.exceptions import UserError from beets.util import displayable_path, editor_command, interactive_open +if TYPE_CHECKING: + from beets.library import Library + + +class ConfigCLIOpts(Protocol): + paths: bool | None + defaults: bool + edit: bool | None + redact: bool + config: str | None + -def config_func(lib, opts, args): +def config_func(lib: Library, opts: ConfigCLIOpts, args: list[str]) -> None: # Make sure lazy configuration is loaded config.resolve() @@ -84,6 +98,7 @@ def config_edit(cli_options): "-d", "--defaults", action="store_true", + default=False, help="include the default configuration", ) config_cmd.parser.add_option( diff --git a/beets/ui/commands/fields.py b/beets/ui/commands/fields.py index fc309b9d1a..792f794bda 100644 --- a/beets/ui/commands/fields.py +++ b/beets/ui/commands/fields.py @@ -1,9 +1,17 @@ """The `fields` command: show available fields for queries and format strings.""" +from __future__ import annotations + import textwrap +from typing import TYPE_CHECKING from beets import library, ui +if TYPE_CHECKING: + import optparse + + from beets.library import Library + def _print_keys(query): """Given a SQLite query result, print the `key` field of each @@ -13,7 +21,7 @@ def _print_keys(query): ui.print_(f" {row['key']}") -def fields_func(lib, opts, args): +def fields_func(lib: Library, opts: optparse.Values, args: list[str]) -> None: def _print_rows(names): ui.print_(textwrap.indent("\n".join(sorted(names)), " ")) diff --git a/beets/ui/commands/help.py b/beets/ui/commands/help.py index e50b8ef2de..76dbcb66f0 100644 --- a/beets/ui/commands/help.py +++ b/beets/ui/commands/help.py @@ -1,8 +1,17 @@ """The 'help' command: show help information for commands.""" +from __future__ import annotations + +from typing import TYPE_CHECKING + from beets import ui from beets.exceptions import UserError +if TYPE_CHECKING: + import optparse + + from beets.library import Library + class HelpCommand(ui.Subcommand): def __init__(self): @@ -12,7 +21,9 @@ def __init__(self): help="give detailed help on a specific sub-command", ) - def func(self, lib, opts, args): + def func( + self, lib: Library, opts: optparse.Values, args: list[str] + ) -> None: if args: cmdname = args[0] helpcommand = self.root_parser._subcommand_for_name(cmdname) diff --git a/beets/ui/commands/import_/__init__.py b/beets/ui/commands/import_/__init__.py index 0828d695a5..1068998753 100644 --- a/beets/ui/commands/import_/__init__.py +++ b/beets/ui/commands/import_/__init__.py @@ -3,7 +3,7 @@ from __future__ import annotations import os -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Protocol from beets import config, logging, plugins, ui from beets.exceptions import UserError @@ -22,6 +22,12 @@ log = logging.getLogger("beets") +class ImportCLIOpts(Protocol): + copy: bool | None + library: bool | None + from_logfiles: list[str] | None + + def paths_from_logfile(path: str) -> Iterator[str]: """Parse the logfile and yield skipped paths to pass to the `import` command. @@ -91,8 +97,8 @@ def import_files( plugins.send("import", lib=lib, paths=paths) -def import_func(lib: Library, opts: optparse.Values, args: list[str]) -> None: - config["import"].set_args(opts) +def import_func(lib: Library, opts: ImportCLIOpts, args: list[str]) -> None: + config["import"].set_args(vars(opts)) # Special case: --copy flag suppresses import_move (which would # otherwise take precedence). diff --git a/beets/ui/commands/list.py b/beets/ui/commands/list.py index cb92b9b790..8511d9820d 100644 --- a/beets/ui/commands/list.py +++ b/beets/ui/commands/list.py @@ -1,7 +1,18 @@ """The 'list' command: query and show library contents.""" +from __future__ import annotations + +from typing import TYPE_CHECKING, Protocol + from beets import ui +if TYPE_CHECKING: + from beets.library import Library + + +class ListCLIOpts(Protocol): + album: bool + def list_items(lib, query, album, fmt=""): """Print out items in lib matching query. If album, then search for @@ -15,7 +26,7 @@ def list_items(lib, query, album, fmt=""): ui.print_(format(item, fmt)) -def list_func(lib, opts, args): +def list_func(lib: Library, opts: ListCLIOpts, args: list[str]) -> None: list_items(lib, args, opts.album) diff --git a/beets/ui/commands/modify.py b/beets/ui/commands/modify.py index e03b2de2ff..8891e1c643 100644 --- a/beets/ui/commands/modify.py +++ b/beets/ui/commands/modify.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING, NamedTuple +from typing import TYPE_CHECKING, NamedTuple, Protocol from beets import library, ui from beets.dbcore import types @@ -12,7 +12,15 @@ from .utils import do_query if TYPE_CHECKING: - from beets.library import LibModel + from beets.library import LibModel, Library + + +class ModifyCLIOpts(Protocol): + album: bool + inherit: bool + move: bool | None + write: bool | None + yes: bool | None class ModifyOperation(NamedTuple): @@ -150,7 +158,7 @@ def modify_parse_args(args, is_album: bool): return query, mods, dels -def modify_func(lib, opts, args): +def modify_func(lib: Library, opts: ModifyCLIOpts, args: list[str]) -> None: query, mods, dels = modify_parse_args(args, is_album=opts.album) if not mods and not dels: raise UserError("no modifications specified") diff --git a/beets/ui/commands/move.py b/beets/ui/commands/move.py index 907f44ca47..1d136be2eb 100644 --- a/beets/ui/commands/move.py +++ b/beets/ui/commands/move.py @@ -3,7 +3,7 @@ from __future__ import annotations import os -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Protocol from beets import logging, ui from beets.exceptions import UserError @@ -13,12 +13,21 @@ from .utils import do_query if TYPE_CHECKING: - from beets.util import PathLike + from beets.library import Library # Global logger. log = logging.getLogger("beets") +class MoveCLIOpts(Protocol): + album: bool + copy: bool + dest: str | None + export: bool + pretend: bool + timid: bool + + def show_path_changes(path_changes): """Given a list of tuples (source, destination) that indicate the path changes, log the changes as INFO-level output to the beets log. @@ -62,7 +71,7 @@ def show_path_changes(path_changes): def move_items( lib, - dest_path: PathLike, + dest: bytes | None, query, copy, album, @@ -74,7 +83,6 @@ def move_items( dest is None, then the library's base directory is used, making the command "consolidate" files. """ - dest = os.fsencode(dest_path) if dest_path else dest_path items, albums = do_query(lib, query, album, False) objs = albums if album else items num_objs = len(objs) @@ -147,10 +155,9 @@ def isalbummoved(album): obj.move(operation=MoveOperation.MOVE, basedir=dest) -def move_func(lib, opts, args): - dest = opts.dest +def move_func(lib: Library, opts: MoveCLIOpts, args: list[str]) -> None: + dest = normpath(opts.dest) if opts.dest else None if dest is not None: - dest = normpath(dest) if not os.path.isdir(syspath(dest)): raise UserError(f"no such directory: {displayable_path(dest)}") @@ -188,14 +195,15 @@ def move_func(lib, opts, args): "-t", "--timid", dest="timid", + default=False, action="store_true", help="always confirm all actions", ) move_cmd.parser.add_option( "-e", "--export", - default=False, action="store_true", + default=False, help="copy without changing the database path", ) move_cmd.parser.add_album_option() diff --git a/beets/ui/commands/remove.py b/beets/ui/commands/remove.py index 997a4b48cd..053d685960 100644 --- a/beets/ui/commands/remove.py +++ b/beets/ui/commands/remove.py @@ -1,9 +1,22 @@ """The `remove` command: remove items from the library (and optionally delete files).""" +from __future__ import annotations + +from typing import TYPE_CHECKING, Protocol + from beets import ui from .utils import do_query +if TYPE_CHECKING: + from beets.library import Library + + +class RemoveCLIOpts(Protocol): + album: bool + delete: bool | None + force: bool | None + def remove_items(lib, query, album, delete, force): """Remove items matching query from lib. If album, then match and @@ -67,7 +80,7 @@ def fmt_album(a): obj.remove(delete) -def remove_func(lib, opts, args): +def remove_func(lib: Library, opts: RemoveCLIOpts, args: list[str]) -> None: remove_items(lib, args, opts.album, opts.delete, opts.force) diff --git a/beets/ui/commands/stats.py b/beets/ui/commands/stats.py index d51d4d8ae9..4ff976f80f 100644 --- a/beets/ui/commands/stats.py +++ b/beets/ui/commands/stats.py @@ -1,15 +1,26 @@ """The 'stats' command: show library statistics.""" +from __future__ import annotations + import os +from typing import TYPE_CHECKING, Protocol from beets import logging, ui from beets.util import syspath from beets.util.units import human_bytes, human_seconds +if TYPE_CHECKING: + from beets.library import Library + + # Global logger. log = logging.getLogger("beets") +class StatsCLIOpts(Protocol): + exact: bool + + def show_stats(lib, query, exact): """Shows some statistics about the matched items.""" items = lib.items(query) @@ -49,7 +60,7 @@ def show_stats(lib, query, exact): Album artists: {len(album_artists)}""") -def stats_func(lib, opts, args): +def stats_func(lib: Library, opts: StatsCLIOpts, args: list[str]) -> None: show_stats(lib, args, opts.exact) @@ -57,6 +68,10 @@ def stats_func(lib, opts, args): "stats", help="show statistics about the library or a query" ) stats_cmd.parser.add_option( - "-e", "--exact", action="store_true", help="exact size and time" + "-e", + "--exact", + action="store_true", + default=False, + help="exact size and time", ) stats_cmd.func = stats_func diff --git a/beets/ui/commands/update.py b/beets/ui/commands/update.py index c8251c4222..01900e9370 100644 --- a/beets/ui/commands/update.py +++ b/beets/ui/commands/update.py @@ -1,6 +1,9 @@ """The `update` command: Update library contents according to on-disk tags.""" +from __future__ import annotations + import os +from typing import TYPE_CHECKING, Protocol from beets import library, logging, ui from beets.util import ancestry, syspath @@ -8,10 +11,22 @@ from .utils import do_query +if TYPE_CHECKING: + from beets.library import Library + + # Global logger. log = logging.getLogger("beets") +class UpdateCLIOpts(Protocol): + album: bool + exclude_fields: list[str] | None + fields: list[str] | None + move: bool | None + pretend: bool | None + + def update_items(lib, query, album, move, pretend, fields, exclude_fields=None): """For all the items matched by the query, update the library to reflect the item's embedded tags. @@ -38,10 +53,14 @@ def update_items(lib, query, album, move, pretend, fields, exclude_fields=None): item_fields = fields # get all the album fields to update album_fields = fields or library.Album._fields.keys() - if exclude_fields: + if exclude_fields_set := set(exclude_fields or []): # remove any excluded fields from the item and album sets - item_fields = [f for f in item_fields if f not in exclude_fields] - album_fields = [f for f in album_fields if f not in exclude_fields] + item_fields = [ + f for f in item_fields if f not in exclude_fields_set + ] + album_fields = [ + f for f in album_fields if f not in exclude_fields_set + ] # Walk through the items and pick up their changes. affected_albums = set() @@ -73,8 +92,7 @@ def update_items(lib, query, album, move, pretend, fields, exclude_fields=None): # Special-case album artist when it matches track artist. (Hacky # but necessary for preserving album-level metadata for non- # autotagged imports.) - if not item.albumartist: - old_item = lib.get_item(item.id) + if not item.albumartist and (old_item := lib.get_item(item.id)): if old_item.albumartist == old_item.artist == item.artist: item.albumartist = old_item.albumartist item._dirty.discard("albumartist") @@ -130,11 +148,11 @@ def update_items(lib, query, album, move, pretend, fields, exclude_fields=None): album.store(fields=album_fields) -def update_func(lib, opts, args): +def update_func(lib: Library, opts: UpdateCLIOpts, args: list[str]) -> None: # Verify that the library folder exists to prevent accidental wipes. if not os.path.isdir(syspath(lib.directory)): ui.print_("Library path is unavailable or does not exist.") - ui.print_(lib.directory) + ui.print_(os.fsdecode(lib.directory)) if not ui.input_yn("Are you sure you want to continue (y/n)?", True): return update_items( diff --git a/beets/ui/commands/version.py b/beets/ui/commands/version.py index a93c373a44..e1de09c964 100644 --- a/beets/ui/commands/version.py +++ b/beets/ui/commands/version.py @@ -1,12 +1,20 @@ """The 'version' command: show version information.""" +from __future__ import annotations + from platform import python_version +from typing import TYPE_CHECKING import beets from beets import plugins, ui +if TYPE_CHECKING: + import optparse + + from beets.library import Library + -def show_version(*args): +def show_version(lib: Library, opts: optparse.Values, args: list[str]) -> None: ui.print_(f"beets version {beets.__version__}") ui.print_(f"Python version {python_version()}") # Show plugins. diff --git a/beets/ui/commands/write.py b/beets/ui/commands/write.py index 87fba8236e..d4f3d88347 100644 --- a/beets/ui/commands/write.py +++ b/beets/ui/commands/write.py @@ -1,16 +1,28 @@ """The `write` command: write tag information to files.""" +from __future__ import annotations + import os +from typing import TYPE_CHECKING, Protocol from beets import library, logging, ui from beets.util import syspath from .utils import do_query +if TYPE_CHECKING: + from beets.library import Library + + # Global logger. log = logging.getLogger("beets") +class WriteCLIOpts(Protocol): + force: bool + pretend: bool + + def write_items(lib, query, pretend, force): """Write tag information from the database to the respective files in the filesystem. @@ -40,7 +52,7 @@ def write_items(lib, query, pretend, force): item.try_sync(True, False) -def write_func(lib, opts, args): +def write_func(lib: Library, opts: WriteCLIOpts, args: list[str]) -> None: write_items(lib, args, opts.pretend, opts.force) @@ -49,12 +61,14 @@ def write_func(lib, opts, args): "-p", "--pretend", action="store_true", + default=False, help="show all changes but do nothing", ) write_cmd.parser.add_option( "-f", "--force", action="store_true", + default=False, help="write tags even if the existing tags match the database", ) write_cmd.func = write_func diff --git a/beetsplug/absubmit.py b/beetsplug/absubmit.py index 1897922093..b043d8ac17 100644 --- a/beetsplug/absubmit.py +++ b/beetsplug/absubmit.py @@ -1,5 +1,7 @@ """Calculate acoustic information and submit to AcousticBrainz.""" +from __future__ import annotations + import errno import hashlib import json @@ -7,12 +9,22 @@ import shutil import subprocess import tempfile +from typing import TYPE_CHECKING, Protocol import requests from beets import plugins, ui, util from beets.exceptions import UserError +if TYPE_CHECKING: + from beets.library import Library + + +class ABSubmitCLIOpts(Protocol): + force_refetch: bool + pretend_fetch: bool + + # We use this field to check whether AcousticBrainz info is present. PROBE_FIELD = "mood_acoustic" @@ -113,7 +125,9 @@ def commands(self): cmd.func = self.command return [cmd] - def command(self, lib, opts, args): + def command( + self, lib: Library, opts: ABSubmitCLIOpts, args: list[str] + ) -> None: if not self.url: raise UserError( "This plugin is deprecated since AcousticBrainz no longer " diff --git a/beetsplug/acousticbrainz.py b/beetsplug/acousticbrainz.py index 2c3c7c353e..c70014d8ff 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 +from typing import TYPE_CHECKING, ClassVar, Protocol import requests @@ -13,6 +13,11 @@ if TYPE_CHECKING: from beets.importer import ImportSession, ImportTask + from beets.library import Library + + +class AcousticBrainzCLIOpts(Protocol): + force_refetch: bool LEVELS = ["/low-level", "/high-level"] @@ -109,7 +114,9 @@ def commands(self): help="re-download data when already present", ) - def func(lib, opts, args): + def func( + lib: Library, opts: AcousticBrainzCLIOpts, args: list[str] + ) -> None: items = lib.items(args) self._fetch_info( items, diff --git a/beetsplug/aura.py b/beetsplug/aura.py index 990a3e8123..912c7a8628 100644 --- a/beetsplug/aura.py +++ b/beetsplug/aura.py @@ -6,7 +6,7 @@ import re from dataclasses import dataclass from mimetypes import guess_type -from typing import TYPE_CHECKING, ClassVar +from typing import TYPE_CHECKING, ClassVar, Protocol from flask import ( Blueprint, @@ -32,6 +32,11 @@ from beets.dbcore.query import SQLiteType from beets.library import LibModel, Library + +class AuraCLIOpts(Protocol): + debug: bool + + # Constants # AURA server information @@ -931,7 +936,7 @@ def __init__(self): def commands(self): """Add subcommand used to run the AURA server.""" - def run_aura(lib, opts, args): + def run_aura(lib: Library, opts: AuraCLIOpts, args: list[str]) -> None: """Run the application using Flask's built in-server. Args: diff --git a/beetsplug/autobpm.py b/beetsplug/autobpm.py index 0e0e3a88b5..8df5610a5e 100644 --- a/beetsplug/autobpm.py +++ b/beetsplug/autobpm.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Protocol import librosa import numpy as np @@ -16,6 +16,11 @@ from beets.library import Item, Library +class AutoBPMCLIOpts(Protocol): + force: bool + quiet: bool + + class AutoBPMPlugin(BeetsPlugin): def __init__(self) -> None: super().__init__() @@ -62,7 +67,9 @@ def commands(self) -> list[Subcommand]: cmd.func = self.command return [cmd] - def command(self, lib: Library, opts, args: list[str]) -> None: + def command( + self, lib: Library, opts: AutoBPMCLIOpts, args: list[str] + ) -> None: force = self.config["force"].get(bool) or opts.force quiet = self.config["quiet"].get(bool) or opts.quiet self.calculate_bpm( diff --git a/beetsplug/badfiles.py b/beetsplug/badfiles.py index 3c7a1d6777..b2f9886d7a 100644 --- a/beetsplug/badfiles.py +++ b/beetsplug/badfiles.py @@ -7,7 +7,7 @@ import shlex import sys from subprocess import STDOUT, CalledProcessError, check_output, list2cmdline -from typing import TYPE_CHECKING, Literal +from typing import TYPE_CHECKING, Literal, Protocol import confuse @@ -19,10 +19,15 @@ if TYPE_CHECKING: from beets.importer import ImportSession, ImportTask + from beets.library import Library ImportAction = Literal["abort", "skip", "continue"] +class BadCLIOpts(Protocol): + verbose: bool + + class CheckerCommandError(Exception): """Raised when running a checker failed. @@ -249,7 +254,7 @@ def on_import_task_before_choice( raise Exception(f"Unexpected selection: {sel}") return None - def command(self, lib, opts, args): + def command(self, lib: Library, opts: BadCLIOpts, args: list[str]) -> None: # Get items from arguments items = lib.items(args) self.verbose = opts.verbose diff --git a/beetsplug/bareasc.py b/beetsplug/bareasc.py index 9d6810f779..9b40e0ff5a 100644 --- a/beetsplug/bareasc.py +++ b/beetsplug/bareasc.py @@ -4,6 +4,10 @@ """Provides a bare-ASCII matching query.""" +from __future__ import annotations + +from typing import TYPE_CHECKING, Protocol + from unidecode import unidecode from beets import ui @@ -11,6 +15,13 @@ from beets.plugins import BeetsPlugin from beets.ui import print_ +if TYPE_CHECKING: + from beets.library import Library + + +class BareascCLIOpts(Protocol): + album: bool + class BareascQuery(StringFieldQuery[str]): """Compare items using bare ASCII, without accents etc.""" @@ -63,13 +74,15 @@ def commands(self): cmd.func = self.unidecode_list return [cmd] - def unidecode_list(self, lib, opts, args): + def unidecode_list( + self, lib: Library, opts: BareascCLIOpts, args: list[str] + ) -> None: """Emulate normal 'list' command but with unidecode output.""" album = opts.album # Copied from commands.py - list_items if album: - for album in lib.albums(args): - bare = unidecode(str(album)) + for album_obj in lib.albums(args): + bare = unidecode(str(album_obj)) print_(bare) else: for item in lib.items(args): diff --git a/beetsplug/bench.py b/beetsplug/bench.py index 8553357a5e..f058084777 100644 --- a/beetsplug/bench.py +++ b/beetsplug/bench.py @@ -1,7 +1,10 @@ """Some simple performance benchmarks for beets.""" +from __future__ import annotations + import cProfile import timeit +from typing import TYPE_CHECKING, Protocol from beets import importer, plugins, ui from beets.autotag import Source, tag_album @@ -9,8 +12,24 @@ from beets.util.pathformats import PF_KEY_DEFAULT from beetsplug._utils import vfs +if TYPE_CHECKING: + from collections.abc import Sequence + + from beets.library import Item, Library + + +class BenchAunique(Protocol): + profile: bool + -def aunique_benchmark(lib, prof): +class BenchMatch(Protocol): + profile: bool + id: str | None + + +def aunique_benchmark( + lib: Library, opts: BenchAunique, args: list[str] +) -> None: def _build_tree(): vfs.libtree(lib) @@ -18,7 +37,7 @@ def _build_tree(): lib.path_formats = [ (PF_KEY_DEFAULT, "$albumartist/$album%aunique{}/$track $title") ] - if prof: + if opts.profile: cProfile.runctx( "_build_tree()", {}, @@ -33,7 +52,7 @@ def _build_tree(): lib.path_formats = [ (PF_KEY_DEFAULT, "$albumartist/$album%lower{}/$track $title") ] - if prof: + if opts.profile: cProfile.runctx( "_build_tree()", {}, @@ -45,14 +64,13 @@ def _build_tree(): print("Without %aunique:", interval) -def match_benchmark(lib, prof, query=None, album_id=None): +def match_benchmark(lib: Library, opts: BenchMatch, args: list[str]) -> None: # If no album ID is provided, we'll match against a suitably huge # album. - if not album_id: - album_id = "9c5c043e-bc69-4edb-81a4-1aaf9c81e6dc" + id_ = opts.id or "9c5c043e-bc69-4edb-81a4-1aaf9c81e6dc" # Get an album from the library to use as the source for the match. - items = lib.albums(query).get().items() + items: Sequence[Item] = i.items() if (i := lib.albums(args).get()) else [] # Ensure fingerprinting is invoked (if enabled). plugins.send( @@ -64,9 +82,9 @@ def match_benchmark(lib, prof, query=None, album_id=None): # Run the match. def _run_match(): source = Source.from_items(items) - tag_album(source, search_ids=[album_id]) + tag_album(source, search_ids=[id_]) - if prof: + if opts.profile: cProfile.runctx( "_run_match()", {}, {"_run_match": _run_match}, "match.prof" ) @@ -89,9 +107,7 @@ def commands(self): default=False, help="performance profiling", ) - aunique_bench_cmd.func = lambda lib, opts, args: aunique_benchmark( - lib, opts.profile - ) + aunique_bench_cmd.func = aunique_benchmark match_bench_cmd = ui.Subcommand( "bench_match", help="benchmark for track matching" @@ -106,8 +122,6 @@ def commands(self): match_bench_cmd.parser.add_option( "-i", "--id", default=None, help="album ID to match against" ) - match_bench_cmd.func = lambda lib, opts, args: match_benchmark( - lib, opts.profile, args, opts.id - ) + match_bench_cmd.func = match_benchmark return [aunique_bench_cmd, match_bench_cmd] diff --git a/beetsplug/bpd/__init__.py b/beetsplug/bpd/__init__.py index bd61083622..73db57b68a 100644 --- a/beetsplug/bpd/__init__.py +++ b/beetsplug/bpd/__init__.py @@ -27,7 +27,10 @@ from beetsplug._utils import vfs if TYPE_CHECKING: + import optparse + from beets.dbcore.query import Query + from beets.library import Library try: @@ -1621,21 +1624,19 @@ def commands(self): "bpd", help="run an MPD-compatible music player server" ) - def func(lib, opts, args): + def func(lib: Library, opts: optparse.Values, args: list[str]) -> None: host = self.config["host"].as_str() host = args.pop(0) if args else host port = args.pop(0) if args else self.config["port"].get(int) if args: - ctrl_port = args.pop(0) + ctrl_port = int(args.pop(0)) else: ctrl_port = self.config["control_port"].get(int) if args: raise UserError("too many arguments") password = self.config["password"].as_str() volume = self.config["volume"].get(int) - self.start_bpd( - lib, host, int(port), password, volume, int(ctrl_port) - ) + self.start_bpd(lib, host, int(port), password, volume, ctrl_port) cmd.func = func return [cmd] diff --git a/beetsplug/bpm.py b/beetsplug/bpm.py index 8604df4400..f9f6a461e8 100644 --- a/beetsplug/bpm.py +++ b/beetsplug/bpm.py @@ -1,10 +1,18 @@ """Determine BPM by pressing a key to the rhythm.""" +from __future__ import annotations + import time +from typing import TYPE_CHECKING from beets import ui from beets.plugins import BeetsPlugin +if TYPE_CHECKING: + import optparse + + from beets.library import Library + def bpm(max_strokes): """Returns average BPM (possibly of a playing song) @@ -42,7 +50,9 @@ def commands(self): cmd.func = self.command return [cmd] - def command(self, lib, opts, args): + def command( + self, lib: Library, opts: optparse.Values, args: list[str] + ) -> None: write = ui.should_write() self.get_bpm(lib.items(args), write) diff --git a/beetsplug/bpsync.py b/beetsplug/bpsync.py index 828c97b01f..956294bb2c 100644 --- a/beetsplug/bpsync.py +++ b/beetsplug/bpsync.py @@ -1,5 +1,9 @@ """Update library's tags using Beatport.""" +from __future__ import annotations + +from typing import TYPE_CHECKING, Protocol + from beets import library, ui, util from beets.autotag import AlbumMatch, Distance, TrackMatch from beets.plugins import BeetsPlugin, apply_item_changes @@ -7,6 +11,15 @@ from .beatport import BeatportPlugin +if TYPE_CHECKING: + from beets.library import Library + + +class BPSyncCLIOpts(Protocol): + move: bool | None + pretend: bool | None + write: bool | None + class BPSyncPlugin(BeetsPlugin): def __init__(self): @@ -49,7 +62,7 @@ def commands(self): cmd.func = self.func return [cmd] - def func(self, lib, opts, args): + def func(self, lib: Library, opts: BPSyncCLIOpts, args: list[str]) -> None: """Command handler for the bpsync function.""" move = ui.should_move(opts.move) pretend = opts.pretend diff --git a/beetsplug/chroma.py b/beetsplug/chroma.py index 01154b115b..4c57a2475a 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 +from typing import TYPE_CHECKING, Protocol import acoustid import confuse @@ -20,13 +20,23 @@ from beets.util.color import colorize if TYPE_CHECKING: + import optparse from collections.abc import Iterable, Iterator from beets.autotag import TrackInfo from beets.importer import ImportSession, ImportTask + from beets.library import Library from beets.library.models import Item from beetsplug.musicbrainz import MusicBrainzPlugin + +class ChromaSearchCLIOpts(Protocol): + count: int + full: bool | None + search: str | None + write: bool | None + + API_KEY = "1vOwZtEn" SCORE_THRESH = 0.5 TRACK_ID_WEIGHT = 10.0 @@ -258,7 +268,9 @@ def commands(self): "submit", help="submit Acoustid fingerprints" ) - def submit_cmd_func(lib, opts, args): + def submit_cmd_func( + lib: Library, opts: optparse.Values, args: list[str] + ) -> None: try: apikey = config["acoustid"]["apikey"].as_str() except confuse.NotFoundError: @@ -271,7 +283,9 @@ def submit_cmd_func(lib, opts, args): "fingerprint", help="generate fingerprints for items without them" ) - def fingerprint_cmd_func(lib, opts, args): + def fingerprint_cmd_func( + lib: Library, opts: optparse.Values, args: list[str] + ) -> None: for item in lib.items(args): fingerprint_item(self._log, item, write=ui.should_write()) @@ -315,7 +329,9 @@ def chromasearch_cmd(self): help="Write computed fingerprints to files", ) - def search_cmd_func(lib, opts, args): + def search_cmd_func( + lib: Library, opts: ChromaSearchCLIOpts, args: list[str] + ) -> None: if not opts.search: raise UserError("no --search provided") if opts.count <= 0: @@ -348,8 +364,8 @@ def search_cmd_func(lib, opts, args): if score > 0: top.add(ScoredItem(item, score)) - for item in top: - ui.print_(str(item)) + for scored_item in top: + ui.print_(str(scored_item)) cmd.func = search_cmd_func diff --git a/beetsplug/convert.py b/beetsplug/convert.py index 92e4f2f8f4..6a20d46012 100644 --- a/beetsplug/convert.py +++ b/beetsplug/convert.py @@ -10,7 +10,7 @@ import threading from functools import cached_property from string import Template -from typing import TYPE_CHECKING, Literal, NamedTuple +from typing import TYPE_CHECKING, Literal, NamedTuple, Protocol import mediafile from confuse import ConfigTypeError, Optional @@ -26,8 +26,6 @@ from beetsplug._utils import art if TYPE_CHECKING: - import optparse - from beets.importer import ImportSession, ImportTask from beets.library import Album, Library from beets.util.pathformats import PathFormat @@ -36,6 +34,13 @@ # Keep track of temporary transcoded files for deletion. _temp_files: list[bytes] = [] + +class ConvertCLIOpts(Protocol): + album: bool + keep_new: bool + yes: bool | None + + # Some convenient alternate names for formats. ALIASES = {"windows media": "wma", "vorbis": "ogg"} @@ -651,7 +656,7 @@ def copy_album_art(self, album: Album) -> None: util.copy(album.artpath, dest) def convert_func( - self, lib: Library, opts: optparse.Values, args: list[str] + self, lib: Library, opts: ConvertCLIOpts, args: list[str] ) -> None: self.config.set(vars(opts)) pretend = self.pretend diff --git a/beetsplug/deezer.py b/beetsplug/deezer.py index 629c87bd48..fdc69ed0ca 100644 --- a/beetsplug/deezer.py +++ b/beetsplug/deezer.py @@ -17,6 +17,7 @@ VARIOUS_ARTISTS_ID = 5080 if TYPE_CHECKING: + import optparse from collections.abc import Sequence from beets.library import Item, Library @@ -46,7 +47,7 @@ def commands(self): "deezerupdate", help=f"Update {self.data_source} rank" ) - def func(lib: Library, opts, args): + def func(lib: Library, opts: optparse.Values, args: list[str]) -> None: items = lib.items(args) self.deezerupdate(list(items), ui.should_write()) diff --git a/beetsplug/duplicates.py b/beetsplug/duplicates.py index 6ce02703d0..8bfeffd467 100644 --- a/beetsplug/duplicates.py +++ b/beetsplug/duplicates.py @@ -1,7 +1,10 @@ """List duplicate tracks or albums.""" +from __future__ import annotations + import os import shlex +from typing import TYPE_CHECKING from beets.library import Album, Item from beets.plugins import BeetsPlugin @@ -14,6 +17,13 @@ subprocess, ) +if TYPE_CHECKING: + import optparse + from collections.abc import Sequence + + from beets.library import LibModel, Library + + PLUGIN = "duplicates" @@ -128,7 +138,7 @@ def __init__(self): self._command.parser.add_all_common_options() def commands(self): - def _dup(lib, opts, args): + def _dup(lib: Library, opts: optparse.Values, args: list[str]) -> None: self.config.set_args(opts) album = self.config["album"].get(bool) checksum = self.config["checksum"].get(str) @@ -146,6 +156,7 @@ def _dup(lib, opts, args): strict = self.config["strict"].get(bool) tag = self.config["tag"].get(str) + items: Sequence[LibModel] if album: if not keys: keys = ["mb_albumid"] diff --git a/beetsplug/edit.py b/beetsplug/edit.py index d944bfca0b..2fb85e97d4 100644 --- a/beetsplug/edit.py +++ b/beetsplug/edit.py @@ -8,7 +8,7 @@ import subprocess from collections import Counter from tempfile import NamedTemporaryFile -from typing import TYPE_CHECKING, Any, cast +from typing import TYPE_CHECKING, Any, Protocol, cast import yaml @@ -22,6 +22,7 @@ if TYPE_CHECKING: from beets.importer import ImportSession, ImportTask + from beets.library import Library # These "safe" types can avoid the format/parse cycle that most fields go # through: they are safe to edit with native YAML types. @@ -39,6 +40,12 @@ ITEM_ONLY_FIELDS = Item._field_names - Album._field_names +class EditCLIOpts(Protocol): + album: bool + all: bool | None + field: list[str] | None + + class ParseError(Exception): """The modified file is unreadable. The user should be offered a chance to fix the error. @@ -174,7 +181,9 @@ def commands(self): edit_command.func = self._edit_command return [edit_command] - def _edit_command(self, lib, opts, args): + def _edit_command( + self, lib: Library, opts: EditCLIOpts, args: list[str] + ) -> None: """The CLI command function for the `beet edit` command.""" # Get the objects to edit. items, albums = do_query(lib, args, opts.album, False) diff --git a/beetsplug/embedart.py b/beetsplug/embedart.py index abd03fbcaa..3c4156ecf5 100644 --- a/beetsplug/embedart.py +++ b/beetsplug/embedart.py @@ -5,7 +5,7 @@ import os import tempfile from mimetypes import guess_extension -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Protocol import requests @@ -18,13 +18,28 @@ from beetsplug._utils import art if TYPE_CHECKING: - import optparse from collections.abc import Sequence from beets.importer import ImportSession, ImportTask from beets.library import Album, LibModel, Library +class EmbedArtCLIOpts(Protocol): + file: str | None + url: str | None + yes: bool | None + + +class ExtractArtCLIOpts(Protocol): + associate: bool | None + filename: str | None + outpath: str | None + + +class ClearArtCLIOpts(Protocol): + yes: bool | None + + def _confirm(objs: Sequence[LibModel], album: bool) -> bool: """Show the list of affected objects (items or albums) and confirm that the user wants to modify their artwork. @@ -109,7 +124,7 @@ def commands(self) -> list[ui.Subcommand]: ifempty = self.config["ifempty"].get(bool) def embed_func( - lib: Library, opts: optparse.Values, args: list[str] + lib: Library, opts: EmbedArtCLIOpts, args: list[str] ) -> None: if opts.file: imagepath = normpath(opts.file) @@ -210,7 +225,7 @@ def embed_func( ) def extract_func( - lib: Library, opts: optparse.Values, args: list[str] + lib: Library, opts: ExtractArtCLIOpts, args: list[str] ) -> None: if opts.outpath: art.extract_first( @@ -247,7 +262,7 @@ def extract_func( ) def clear_func( - lib: Library, opts: optparse.Values, args: list[str] + lib: Library, opts: ClearArtCLIOpts, args: list[str] ) -> None: items = lib.items(args) # Confirm with user. diff --git a/beetsplug/export.py b/beetsplug/export.py index 67f58d510e..4e6250ca94 100644 --- a/beetsplug/export.py +++ b/beetsplug/export.py @@ -1,10 +1,13 @@ """Exports data from beets""" +from __future__ import annotations + import codecs import csv import json import sys from datetime import date, datetime +from typing import TYPE_CHECKING, Literal, Protocol, get_args from xml.etree import ElementTree import mediafile @@ -13,6 +16,20 @@ from beets.plugins import BeetsPlugin from beetsplug.info import library_data, tag_data +if TYPE_CHECKING: + from beets.library import Library + +Format = Literal["json", "jsonlines", "csv", "xml"] + + +class ExportCLIOpts(Protocol): + library: bool | None + album: bool + append: bool + included_keys: list[str] + output: str | None + format: Format | None + class ExportEncoder(json.JSONEncoder): """Deals with dates because JSON doesn't have a standard""" @@ -68,18 +85,13 @@ def __init__(self): def commands(self): cmd = ui.Subcommand("export", help="export data from beets") cmd.func = self.run + cmd.parser.add_album_option() cmd.parser.add_option( "-l", "--library", action="store_true", help="show library fields instead of tags", ) - cmd.parser.add_option( - "-a", - "--album", - action="store_true", - help='show album fields instead of tracks (implies "--library")', - ) cmd.parser.add_option( "--append", action="store_true", @@ -107,10 +119,13 @@ def commands(self): ) return [cmd] - def run(self, lib, opts, args): + def run(self, lib: Library, opts: ExportCLIOpts, args: list[str]) -> None: file_path = opts.output file_mode = "a" if opts.append else "w" - file_format = opts.format or self.config["default_format"].get(str) + default_format: Format = self.config["default_format"].as_choice( + get_args(Format) + ) + file_format = opts.format or default_format file_format_is_line_based = file_format == "jsonlines" format_options = self.config[file_format]["formatting"].get(dict) diff --git a/beetsplug/fetchart.py b/beetsplug/fetchart.py index b13b8b6770..02f4c84eb4 100644 --- a/beetsplug/fetchart.py +++ b/beetsplug/fetchart.py @@ -9,7 +9,7 @@ from contextlib import closing from enum import Enum from functools import cached_property -from typing import TYPE_CHECKING, Any, AnyStr, ClassVar, Literal +from typing import TYPE_CHECKING, Any, AnyStr, ClassVar, Literal, Protocol import confuse import requests @@ -29,6 +29,12 @@ from beets.library import Album, Library from beets.logging import BeetsLogger as Logger + +class FetchArtCLIOpts(Protocol): + force: bool + quiet: bool + + try: from bs4 import BeautifulSoup, Tag @@ -1548,7 +1554,7 @@ def commands(self) -> list[ui.Subcommand]: help="quiet mode: do not output albums that already have artwork", ) - def func(lib: Library, opts, args) -> None: + def func(lib: Library, opts: FetchArtCLIOpts, args: list[str]) -> None: self.batch_fetch_art(lib, lib.albums(args), opts.force, opts.quiet) cmd.func = func diff --git a/beetsplug/fish.py b/beetsplug/fish.py index 8175bced26..ee10f5def4 100644 --- a/beetsplug/fish.py +++ b/beetsplug/fish.py @@ -7,13 +7,26 @@ `beet fish -e genres -e albumartist` """ +from __future__ import annotations + import os from operator import attrgetter +from typing import TYPE_CHECKING, Protocol from beets import library, plugins, ui from beets.plugins import BeetsPlugin from beets.ui import commands +if TYPE_CHECKING: + from beets.library import Library + + +class FishCLIOpts(Protocol): + extravalues: list[str] | None + noFields: bool # noqa: N815 + output: str + + BL_NEED2 = """complete -c beet -n '__fish_beet_needs_command' {} {}\n""" BL_USE3 = """complete -c beet -n '__fish_beet_using_command {}' {} {}\n""" BL_SUBS = """complete -c beet -n '__fish_at_level {} ""' {} {}\n""" @@ -80,7 +93,7 @@ def commands(self): ) return [cmd] - def run(self, lib, opts, args): + def run(self, lib: Library, opts: FishCLIOpts, args: list[str]) -> None: # Gather the commands from Beets core and its plugins. # Collect the album and track fields. # If specified, also collect the values for these fields. diff --git a/beetsplug/freedesktop.py b/beetsplug/freedesktop.py index a84bb45ff8..816e03c189 100644 --- a/beetsplug/freedesktop.py +++ b/beetsplug/freedesktop.py @@ -1,8 +1,17 @@ """Creates freedesktop.org-compliant .directory files on an album level.""" +from __future__ import annotations + +from typing import TYPE_CHECKING + from beets import ui from beets.plugins import BeetsPlugin +if TYPE_CHECKING: + import optparse + + from beets.library import Library + class FreedesktopPlugin(BeetsPlugin): def commands(self): @@ -13,7 +22,9 @@ def commands(self): deprecated.func = self.deprecation_message return [deprecated] - def deprecation_message(self, lib, opts, args): + def deprecation_message( + self, lib: Library, opts: optparse.Values, args: list[str] + ) -> None: ui.print_( "This plugin is deprecated. Its functionality is " "superseded by the 'thumbnails' plugin" diff --git a/beetsplug/ftintitle.py b/beetsplug/ftintitle.py index 74d9d7ed0c..200e3faf2c 100644 --- a/beetsplug/ftintitle.py +++ b/beetsplug/ftintitle.py @@ -9,9 +9,11 @@ from beets import config, plugins, ui if TYPE_CHECKING: + import optparse + from beets.autotag import AlbumInfo, Info, TrackInfo from beets.importer import ImportSession, ImportTask - from beets.library import Album, Item + from beets.library import Album, Item, Library DEFAULT_BRACKET_KEYWORDS: tuple[str, ...] = ( "abridged", @@ -256,7 +258,7 @@ def __init__(self) -> None: ) def commands(self) -> list[ui.Subcommand]: - def func(lib, opts, args): + def func(lib: Library, opts: optparse.Values, args: list[str]) -> None: self.config.set_args(opts) write = ui.should_write() diff --git a/beetsplug/info.py b/beetsplug/info.py index b3ddf8c51a..3c9559e72f 100644 --- a/beetsplug/info.py +++ b/beetsplug/info.py @@ -1,6 +1,9 @@ """Shows file metadata.""" +from __future__ import annotations + import os +from typing import TYPE_CHECKING, Any, Protocol import mediafile @@ -9,6 +12,18 @@ from beets.plugins import BeetsPlugin from beets.util import displayable_path, normpath, syspath +if TYPE_CHECKING: + from beets.library import Library + + +class InfoCLIOpts(Protocol): + album: bool + format: str | None + included_keys: list[str] + keys_only: bool | None + library: bool | None + summarize: bool | None + def tag_data(lib, args, album=False): query = [] @@ -135,18 +150,13 @@ class InfoPlugin(BeetsPlugin): def commands(self): cmd = ui.Subcommand("info", help="show file metadata") cmd.func = self.run + cmd.parser.add_album_option() cmd.parser.add_option( "-l", "--library", action="store_true", help="show library fields instead of tags", ) - cmd.parser.add_option( - "-a", - "--album", - action="store_true", - help='show album fields instead of tracks (implies "--library")', - ) cmd.parser.add_option( "-s", "--summarize", @@ -167,7 +177,7 @@ def commands(self): cmd.parser.add_format_option(target="item") return [cmd] - def run(self, lib, opts, args): + def run(self, lib: Library, opts: InfoCLIOpts, args: list[str]) -> None: """Print tag info or library data for each file referenced by args. Main entry point for the `beet info ARGS...` command. @@ -193,7 +203,7 @@ def run(self, lib, opts, args): included_keys = [k for k in included_keys if k != "path"] first = True - summary = {} + summary: dict[str, Any] = {} for data_emitter in data_collector(lib, args, album=opts.album): try: data, item = data_emitter(included_keys or "*") diff --git a/beetsplug/ipfs.py b/beetsplug/ipfs.py index eddcf59404..bb80d9af34 100644 --- a/beetsplug/ipfs.py +++ b/beetsplug/ipfs.py @@ -1,13 +1,32 @@ """Adds support for ipfs. Requires go-ipfs and a running ipfs daemon""" +from __future__ import annotations + import os import shutil import subprocess import tempfile +from contextlib import contextmanager +from types import SimpleNamespace +from typing import TYPE_CHECKING, Protocol from beets import config, library, ui, util from beets.plugins import BeetsPlugin +if TYPE_CHECKING: + from collections.abc import Iterator + + from beets.library import Library + + +class IPFSCLIOpts(Protocol): + _import: bool | None + _list: bool | None + add: bool | None + get: bool | None + play: bool | None + publish: bool | None + class IPFSPlugin(BeetsPlugin): def __init__(self): @@ -54,7 +73,7 @@ def commands(self): help="Play music from remote libraries", ) - def func(lib, opts, args): + def func(lib: Library, opts: IPFSCLIOpts, args: list[str]) -> None: if opts.add: for album in lib.albums(args): if len(album.items()) == 0: @@ -88,14 +107,19 @@ def auto_add(self, session, task): if self.ipfs_add(task.album): task.album.store() - def ipfs_play(self, lib, opts, args): + def ipfs_play( + self, lib: Library, opts: IPFSCLIOpts, args: list[str] + ) -> None: from beetsplug.play import PlayPlugin - jlib = self.get_remote_lib(lib) player = PlayPlugin() config["play"]["relative_to"] = None - player.album = True - player.play_music(jlib, player, args) + # set opts that `_play_command` expects + play_opts = SimpleNamespace( + album=True, randomize=None, args=None, yes=None + ) + with self.remote_lib(lib) as jlib: + player._play_command(jlib, play_opts, args) def ipfs_add(self, album): try: @@ -248,19 +272,26 @@ def ipfs_list(self, lib, args): ui.print_(format(album, fmt), " : ", album.ipfs.decode()) def query(self, lib, args): - rlib = self.get_remote_lib(lib) - return rlib.albums(args) + with self.remote_lib(lib) as rlib: + return rlib.albums(args) def _remote_libs_path(self, lib): lib_root = os.path.dirname(os.fsencode(lib.path)) return os.path.join(lib_root, b"remotes") - def get_remote_lib(self, lib): + @contextmanager + def remote_lib(self, lib: Library) -> Iterator[Library]: remote_libs = self._remote_libs_path(lib) path = os.path.join(remote_libs, b"joined.db") if not os.path.isfile(path): raise OSError - return library.Library(path) + + remote_lib = library.Library(path) + + try: + yield remote_lib + finally: + remote_lib._close() def ipfs_added_albums(self, rlib, tmpname): """Returns a new library with only albums/items added to ipfs""" diff --git a/beetsplug/keyfinder.py b/beetsplug/keyfinder.py index b86b4e2a41..bf4bad2bd0 100644 --- a/beetsplug/keyfinder.py +++ b/beetsplug/keyfinder.py @@ -1,11 +1,19 @@ """Uses the `KeyFinder` program to add the `initial_key` field.""" +from __future__ import annotations + import os.path import subprocess +from typing import TYPE_CHECKING from beets import ui, util from beets.plugins import BeetsPlugin +if TYPE_CHECKING: + import optparse + + from beets.library import Library + class KeyFinderPlugin(BeetsPlugin): def __init__(self): @@ -22,7 +30,9 @@ def commands(self): cmd.func = self.command return [cmd] - def command(self, lib, opts, args): + def command( + self, lib: Library, opts: optparse.Values, args: list[str] + ) -> None: self.find_key(lib.items(args), write=ui.should_write()) def imported(self, session, task): diff --git a/beetsplug/lastgenre/__init__.py b/beetsplug/lastgenre/__init__.py index 1d1c55eeec..d1afd161e1 100644 --- a/beetsplug/lastgenre/__init__.py +++ b/beetsplug/lastgenre/__init__.py @@ -14,7 +14,7 @@ from collections import defaultdict from functools import cached_property, singledispatchmethod from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Protocol import confuse import yaml @@ -27,7 +27,6 @@ from .client import LastFmClient if TYPE_CHECKING: - import optparse from collections.abc import Iterable from beets.importer import ImportSession, ImportTask @@ -47,6 +46,10 @@ #: The label is used for logging and describes the source and filtering applied. +class LastGenreCLIOpts(Protocol): + album: bool + + # Canonicalization tree processing. @@ -740,9 +743,9 @@ def commands(self) -> list[ui.Subcommand]: lastgenre_cmd.parser.set_defaults(album=True) def lastgenre_func( - lib: library.Library, opts: optparse.Values, args: list[str] + lib: library.Library, opts: LastGenreCLIOpts, args: list[str] ) -> None: - self.config.set_args(opts) + self.config.set_args(vars(opts)) method = lib.albums if opts.album else lib.items for obj in method(args): diff --git a/beetsplug/lastimport.py b/beetsplug/lastimport.py index b07c7c6b1f..d996fe33e6 100644 --- a/beetsplug/lastimport.py +++ b/beetsplug/lastimport.py @@ -12,6 +12,10 @@ from ._utils.playcount import update_play_counts if TYPE_CHECKING: + import optparse + + from beets.library import Library + from ._utils.playcount import Track API_URL = "https://ws.audioscrobbler.com/2.0/" @@ -29,7 +33,7 @@ def __init__(self): def commands(self): cmd = ui.Subcommand("lastimport", help="import last.fm play-count") - def func(lib, opts, args): + def func(lib: Library, opts: optparse.Values, args: list[str]) -> None: import_lastfm(lib, self._log) cmd.func = func diff --git a/beetsplug/limit.py b/beetsplug/limit.py index 52f73aad4a..06ee1ce74a 100644 --- a/beetsplug/limit.py +++ b/beetsplug/limit.py @@ -7,15 +7,29 @@ query language). """ +from __future__ import annotations + from collections import deque from itertools import islice +from typing import TYPE_CHECKING, Protocol from beets.dbcore import FieldQuery from beets.plugins import BeetsPlugin from beets.ui import Subcommand, print_ +if TYPE_CHECKING: + from collections.abc import Iterable + + from beets.library import LibModel, Library + + +class LsLimitCLIOpts(Protocol): + album: bool + head: int | None + tail: int | None + -def lslimit(lib, opts, args): +def lslimit(lib: Library, opts: LsLimitCLIOpts, args: list[str]) -> None: """Query command with head/tail.""" if (opts.head is not None) and (opts.tail is not None): @@ -23,6 +37,7 @@ def lslimit(lib, opts, args): if (opts.head or opts.tail or 0) < 0: raise ValueError("Limit value must be non-negative") + objs: Iterable[LibModel] if opts.album: objs = lib.albums(args) else: diff --git a/beetsplug/listenbrainz.py b/beetsplug/listenbrainz.py index b09b0c73f0..3ab2f7fa8b 100644 --- a/beetsplug/listenbrainz.py +++ b/beetsplug/listenbrainz.py @@ -7,7 +7,7 @@ import time import zipfile from collections import Counter -from typing import TYPE_CHECKING, ClassVar, TypedDict +from typing import TYPE_CHECKING, ClassVar, Protocol, TypedDict import requests @@ -23,9 +23,16 @@ if TYPE_CHECKING: from pathlib import Path + from beets.library import Library + from ._utils.playcount import Track +class LBImportCLIOpts(Protocol): + export_file: str | None + max_listens: int | None + + class Listen(TypedDict): listened_at: int track_metadata: TrackMetadata @@ -93,7 +100,7 @@ def commands(self): ), ) - def func(lib, opts, args): + def func(lib: Library, opts: LBImportCLIOpts, args: list[str]) -> None: self._lbupdate( lib, export_file=opts.export_file, max_listens=opts.max_listens ) diff --git a/beetsplug/lyrics.py b/beetsplug/lyrics.py index c9e5c529d9..5fd6d4ccde 100644 --- a/beetsplug/lyrics.py +++ b/beetsplug/lyrics.py @@ -12,7 +12,7 @@ from html import unescape from itertools import filterfalse, groupby from pathlib import Path -from typing import TYPE_CHECKING, ClassVar, NamedTuple +from typing import TYPE_CHECKING, ClassVar, NamedTuple, Protocol from urllib.parse import quote, quote_plus, urlencode, urlparse import requests @@ -54,6 +54,11 @@ HtmlTransformer = Callable[[str], str] +class LyricsCLIOpts(Protocol): + print: bool + rest_directory: str | None + + class CaptchaError(requests.exceptions.HTTPError): def __init__(self, *args, **kwargs) -> None: super().__init__("Captcha is required", *args, **kwargs) @@ -1142,7 +1147,7 @@ def commands(self): help="do not fetch missing lyrics", ) - def func(lib: Library, opts, args) -> None: + def func(lib: Library, opts: LyricsCLIOpts, args: list[str]) -> None: # The "write to files" option corresponds to the # import_write config value. self.config.set(vars(opts)) diff --git a/beetsplug/mbcollection.py b/beetsplug/mbcollection.py index 1823b0e9a9..e7dc1f7069 100644 --- a/beetsplug/mbcollection.py +++ b/beetsplug/mbcollection.py @@ -16,6 +16,7 @@ from ._utils.requests import BeetsHTTPError if TYPE_CHECKING: + import optparse from collections.abc import Iterable, Iterator from requests import Response @@ -182,7 +183,9 @@ def commands(self): mbupdate.func = self.update_collection return [mbupdate] - def update_collection(self, lib: Library, opts, args) -> None: + def update_collection( + self, lib: Library, opts: optparse.Values, args: list[str] + ) -> None: self.config.set_args(opts) remove_missing = self.config["remove"].get(bool) self.update_album_list(lib, lib.albums(), remove_missing) diff --git a/beetsplug/mbsubmit.py b/beetsplug/mbsubmit.py index dd040c47bd..4c7b61bfaa 100644 --- a/beetsplug/mbsubmit.py +++ b/beetsplug/mbsubmit.py @@ -19,7 +19,10 @@ from beetsplug.info import print_data if TYPE_CHECKING: + import optparse + from beets.importer import ImportSession, ImportTask + from beets.library import Library class MBSubmitPlugin(BeetsPlugin): @@ -79,7 +82,7 @@ def commands(self): "mbsubmit", help="Submit Tracks to MusicBrainz" ) - def func(lib, opts, args): + def func(lib: Library, opts: optparse.Values, args: list[str]) -> None: items = lib.items(args) self._mbsubmit(items) diff --git a/beetsplug/mbsync.py b/beetsplug/mbsync.py index ecdbe0b7d6..0060c0460c 100644 --- a/beetsplug/mbsync.py +++ b/beetsplug/mbsync.py @@ -1,11 +1,23 @@ """Synchronise library metadata with metadata source backends.""" +from __future__ import annotations + from collections import defaultdict +from typing import TYPE_CHECKING, Protocol from beets import library, metadata_plugins, ui, util from beets.autotag import AlbumMatch, Distance, TrackMatch from beets.plugins import BeetsPlugin, apply_item_changes +if TYPE_CHECKING: + from beets.library import Library + + +class MBSyncCLIOpts(Protocol): + move: bool | None + pretend: bool | None + write: bool | None + class MBSyncPlugin(BeetsPlugin): def __init__(self): @@ -45,7 +57,7 @@ def commands(self): cmd.func = self.func return [cmd] - def func(self, lib, opts, args): + def func(self, lib: Library, opts: MBSyncCLIOpts, args: list[str]) -> None: """Command handler for the mbsync function.""" move = ui.should_move(opts.move) pretend = opts.pretend diff --git a/beetsplug/metasync/__init__.py b/beetsplug/metasync/__init__.py index 52d4b54fc6..483e050814 100644 --- a/beetsplug/metasync/__init__.py +++ b/beetsplug/metasync/__init__.py @@ -4,7 +4,7 @@ from abc import ABCMeta, abstractmethod from importlib import import_module -from typing import TYPE_CHECKING, ClassVar +from typing import TYPE_CHECKING, ClassVar, Protocol from confuse import ConfigValueError @@ -13,6 +13,7 @@ if TYPE_CHECKING: from beets.dbcore import types + from beets.library import Library METASYNC_MODULE = "beetsplug.metasync" @@ -20,6 +21,11 @@ SOURCES = {"amarok": "Amarok", "itunes": "Itunes"} +class MetaSyncCLIOpts(Protocol): + pretend: bool | None + sources: list[str] + + class MetaSource(metaclass=ABCMeta): item_types: ClassVar[dict[str, types.Type]] @@ -84,7 +90,9 @@ def commands(self): cmd.func = self.func return [cmd] - def func(self, lib, opts, args): + def func( + self, lib: Library, opts: MetaSyncCLIOpts, args: list[str] + ) -> None: """Command handler for the metasync function.""" pretend = opts.pretend diff --git a/beetsplug/missing.py b/beetsplug/missing.py index 09913140b8..6f8d5555af 100644 --- a/beetsplug/missing.py +++ b/beetsplug/missing.py @@ -16,6 +16,7 @@ from ._utils.musicbrainz import MusicBrainzAPIMixin if TYPE_CHECKING: + import optparse from collections.abc import Iterator from beets.library import Library @@ -151,7 +152,7 @@ def __init__(self): self._command.parser.add_format_option() def commands(self): - def _miss(lib, opts, args): + def _miss(lib: Library, opts: optparse.Values, args: list[str]) -> None: self.config.set_args(opts) albms = self.config["album"].get() diff --git a/beetsplug/mpdstats.py b/beetsplug/mpdstats.py index 3df20659c0..fa0791527c 100644 --- a/beetsplug/mpdstats.py +++ b/beetsplug/mpdstats.py @@ -1,6 +1,8 @@ +from __future__ import annotations + import os import time -from typing import ClassVar +from typing import TYPE_CHECKING, ClassVar import mpd @@ -10,6 +12,12 @@ from beets.exceptions import UserError from beets.util import displayable_path +if TYPE_CHECKING: + import optparse + + from beets.library import Library + + # If we lose the connection, how many times do we want to retry and how # much time should we wait between retries? RETRIES = 10 @@ -351,7 +359,7 @@ def commands(self): help="set the password of the MPD server to connect to", ) - def func(lib, opts, args): + def func(lib: Library, opts: optparse.Values, args: list[str]) -> None: mpd_config.set_args(opts) try: diff --git a/beetsplug/parentwork.py b/beetsplug/parentwork.py index 4d3f6749e2..c59d8c55f6 100644 --- a/beetsplug/parentwork.py +++ b/beetsplug/parentwork.py @@ -14,6 +14,9 @@ from ._utils.musicbrainz import MusicBrainzAPIMixin if TYPE_CHECKING: + import optparse + + from beets.library import Library from beetsplug._utils.musicbrainz import Work @@ -27,7 +30,7 @@ def __init__(self): self.import_stages = [self.imported] def commands(self): - def func(lib, opts, args): + def func(lib: Library, opts: optparse.Values, args: list[str]) -> None: self.config.set_args(opts) force_parent = self.config["force"].get(bool) write = ui.should_write() diff --git a/beetsplug/play.py b/beetsplug/play.py index 8bbce78861..c6ffa8756f 100644 --- a/beetsplug/play.py +++ b/beetsplug/play.py @@ -6,7 +6,7 @@ import shlex import subprocess from os.path import relpath -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Protocol from beets import config, ui, util from beets.exceptions import UserError @@ -16,13 +16,24 @@ from beets.util.color import colorize if TYPE_CHECKING: + from collections.abc import Sequence + from beets.importer import ImportSession, ImportTask + from beets.library import LibModel, Library # Indicate where arguments should be inserted into the command string. # If this is missing, they're placed at the end. ARGS_MARKER = "$args" + +class PlayCLIOpts(Protocol): + album: bool + args: str | None + randomize: bool | None + yes: bool | None + + # Indicate where the playlist file (with absolute path) should be inserted into # the command string. If this is missing, its placed at the end, but before # arguments. @@ -102,7 +113,9 @@ def commands(self): play_command.func = self._play_command return [play_command] - def _play_command(self, lib, opts, args): + def _play_command( + self, lib: Library, opts: PlayCLIOpts, args: list[str] + ) -> None: """The CLI command function for `beet play`. Create a list of paths from query, determine if tracks or albums are to be played. """ @@ -112,6 +125,7 @@ def _play_command(self, lib, opts, args): relative_to = util.normpath(relative_to) # Perform search by album and add folders rather than tracks to # playlist. + selection: Sequence[LibModel] if opts.album: selection = lib.albums(args) paths = [] diff --git a/beetsplug/random.py b/beetsplug/random.py index 20565689f2..cee2cf98cc 100644 --- a/beetsplug/random.py +++ b/beetsplug/random.py @@ -3,19 +3,26 @@ import random from itertools import groupby, islice from operator import methodcaller -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Protocol from beets.plugins import BeetsPlugin from beets.ui import Subcommand, print_ if TYPE_CHECKING: - import optparse from collections.abc import Iterable from beets.library import LibModel, Library -def random_func(lib: Library, opts: optparse.Values, args: list[str]): +class RandomCLIOpts(Protocol): + album: bool + equal_chance: bool + field: str + number: int + time: float | None + + +def random_func(lib: Library, opts: RandomCLIOpts, args: list[str]): """Select some random items or albums and print the results.""" # Fetch all the objects matching the query into a list. objs = lib.albums(args) if opts.album else lib.items(args) @@ -44,6 +51,7 @@ def random_func(lib: Library, opts: optparse.Values, args: list[str]): "-e", "--equal-chance", action="store_true", + default=False, help="each field has the same chance", ) random_cmd.parser.add_option( diff --git a/beetsplug/replaygain.py b/beetsplug/replaygain.py index 29dca42445..c2a660e980 100644 --- a/beetsplug/replaygain.py +++ b/beetsplug/replaygain.py @@ -16,7 +16,7 @@ from multiprocessing.pool import ThreadPool from pathlib import Path from threading import Event, Thread -from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypeVar +from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol, TypeVar from beets import ui from beets.exceptions import UserError @@ -24,7 +24,6 @@ from beets.util import command_output, syspath if TYPE_CHECKING: - import optparse from collections.abc import Callable, Sequence from logging import Logger @@ -33,6 +32,14 @@ from beets.importer import ImportSession, ImportTask from beets.library import Album, Item, Library + +class ReplayGainCLIOpts(Protocol): + album: bool + force: bool + threads: int | None + write: bool | None + + # Utilities. @@ -1567,7 +1574,7 @@ def imported(self, session: ImportSession, task: ImportTask): self.handle_track(task.item, False, self.force_on_import) def command_func( - self, lib: Library, opts: optparse.Values, args: list[str] + self, lib: Library, opts: ReplayGainCLIOpts, args: list[str] ): try: write = ui.should_write(opts.write) diff --git a/beetsplug/scrub.py b/beetsplug/scrub.py index 51d1aeef47..3eacf29892 100644 --- a/beetsplug/scrub.py +++ b/beetsplug/scrub.py @@ -4,7 +4,7 @@ from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Protocol import mediafile import mutagen @@ -14,6 +14,11 @@ if TYPE_CHECKING: from beets.importer import ImportSession, ImportTask + from beets.library import Library + + +class ScrubCLIOpts(Protocol): + write: bool _MUTAGEN_FORMATS = { @@ -46,7 +51,9 @@ def __init__(self) -> None: self.register_listener("import_task_files", self.import_task_files) def commands(self): - def scrub_func(lib, opts, args): + def scrub_func( + lib: Library, opts: ScrubCLIOpts, args: list[str] + ) -> None: # Walk through matching files and remove tags. for item in lib.items(args): self._log.info("scrubbing: {.filepath}", item) diff --git a/beetsplug/smartplaylist.py b/beetsplug/smartplaylist.py index addedac1b3..1f69955dfb 100644 --- a/beetsplug/smartplaylist.py +++ b/beetsplug/smartplaylist.py @@ -6,7 +6,7 @@ from collections import defaultdict from functools import cached_property from shlex import quote as shell_quote -from typing import TYPE_CHECKING, Any, TypeAlias +from typing import TYPE_CHECKING, Any, Literal, Protocol, TypeAlias from urllib.parse import quote from urllib.request import pathname2url @@ -40,6 +40,19 @@ ] +class SmartPlaylistCLIOpts(Protocol): + pretend: bool | None + format: str + playlist_dir: str + dest_regen: bool + relative_to: str | None + prefix: str + forward_slash: bool + urlencode: bool + uri_format: str | None + output: Literal["m3u", "extm3u"] + + class SmartPlaylistPlugin(plugins.BeetsPlugin): def __init__(self) -> None: super().__init__() @@ -177,7 +190,9 @@ def commands(self) -> list[ui.Subcommand]: spl_update.func = self.update_cmd return [spl_update] - def update_cmd(self, lib: Library, opts: Any, args: list[str]) -> None: + def update_cmd( + self, lib: Library, opts: SmartPlaylistCLIOpts, args: list[str] + ) -> None: self.build_queries() if args: args_set = set(args) diff --git a/beetsplug/spotify.py b/beetsplug/spotify.py index c8ab4e0ab4..4fbeb8da5a 100644 --- a/beetsplug/spotify.py +++ b/beetsplug/spotify.py @@ -13,7 +13,7 @@ import time import webbrowser from http import HTTPStatus -from typing import TYPE_CHECKING, Any, ClassVar, Literal, TypedDict +from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol, TypedDict import confuse import requests @@ -36,6 +36,15 @@ DEFAULT_WAITING_TIME = 5 +class SpotifyCLIOpts(Protocol): + mode: str | None + show_failures: bool | None + + +class SpotifySyncCLIOpts(Protocol): + force_refetch: bool + + class TrackDetails(TypedDict): """Popularity and external IDs returned by the /v1/tracks batch endpoint.""" @@ -530,7 +539,9 @@ def get_search_response( def commands(self) -> list[ui.Subcommand]: # autotagger import command - def queries(lib, opts, args): + def queries( + lib: Library, opts: SpotifyCLIOpts, args: list[str] + ) -> None: success = self._parse_opts(opts) if success: results = self._match_library_tracks(lib, args) @@ -570,7 +581,9 @@ def queries(lib, opts, args): help="re-download data when already present", ) - def func(lib, opts, args): + def func( + lib: Library, opts: SpotifySyncCLIOpts, args: list[str] + ) -> None: items = lib.items(args) self._fetch_info(lib, items, ui.should_write(), opts.force_refetch) @@ -593,7 +606,7 @@ def _parse_opts(self, opts): self.opts = opts return True - def _match_library_tracks(self, library: Library, keywords: str): + def _match_library_tracks(self, library: Library, keywords: list[str]): """Get simplified track object dicts for library tracks. Matches tracks based on the specified ``keywords``. diff --git a/beetsplug/subsonicplaylist.py b/beetsplug/subsonicplaylist.py index 976cd5b2c3..69d149b724 100644 --- a/beetsplug/subsonicplaylist.py +++ b/beetsplug/subsonicplaylist.py @@ -1,6 +1,9 @@ +from __future__ import annotations + import random import string from hashlib import md5 +from typing import TYPE_CHECKING from urllib.parse import urlencode from xml.etree import ElementTree @@ -11,6 +14,12 @@ from beets.plugins import BeetsPlugin from beets.ui import Subcommand +if TYPE_CHECKING: + import optparse + + from beets.library import Library + + __author__ = "https://github.com/MrNuggelz" @@ -90,7 +99,9 @@ def get_playlist(self, playlist_id): return name, tracks def commands(self): - def build_playlist(lib, opts, args): + def build_playlist( + lib: Library, opts: optparse.Values, args: list[str] + ) -> None: self.config.set_args(opts) ids = self.config["playlist_ids"].as_str_seq() if self.config["playlist_names"].as_str_seq(): diff --git a/beetsplug/thumbnails.py b/beetsplug/thumbnails.py index 8bb5622497..f25d8ab3ba 100644 --- a/beetsplug/thumbnails.py +++ b/beetsplug/thumbnails.py @@ -23,7 +23,9 @@ BASE_DIR = os.path.join(BaseDirectory.xdg_cache_home, "thumbnails") if TYPE_CHECKING: - from beets.library import Album + import optparse + + from beets.library import Album, Library NORMAL_DIR = bytestring_path(os.path.join(BASE_DIR, "normal")) @@ -62,7 +64,9 @@ def commands(self): return [thumbnails_command] - def process_query(self, lib, opts, args): + def process_query( + self, lib: Library, opts: optparse.Values, args: list[str] + ) -> None: self.config.set_args(opts) if self._check_local_ok(): for album in lib.albums(args): diff --git a/beetsplug/tidal/__init__.py b/beetsplug/tidal/__init__.py index ea21920665..44e4481d3a 100644 --- a/beetsplug/tidal/__init__.py +++ b/beetsplug/tidal/__init__.py @@ -5,7 +5,7 @@ import re import time from functools import cached_property -from typing import TYPE_CHECKING, ClassVar, Literal, overload +from typing import TYPE_CHECKING, ClassVar, Literal, Protocol, overload import confuse @@ -19,7 +19,6 @@ from .api import TidalAPI if TYPE_CHECKING: - import optparse from collections.abc import Callable, Iterable, Sequence from beets.autotag import Info @@ -37,6 +36,16 @@ ) +class TidalCLIOpts(Protocol): + auth: bool + + +class TidalSyncCLIOpts(Protocol): + album: bool + force: bool + write: bool + + log = getLogger("beets.tidal") _normalize_label_re = re.compile( @@ -612,7 +621,7 @@ def commands(self) -> list[ui.Subcommand]: ) def auth_func( - lib: Library, opts: optparse.Values, args: list[str] + lib: Library, opts: TidalCLIOpts, args: list[str] ) -> None: if opts.auth: self.api.ui_authenticate_flow() @@ -652,7 +661,7 @@ def auth_func( ) def sync_func( - lib: Library, opts: optparse.Values, args: list[str] + lib: Library, opts: TidalSyncCLIOpts, args: list[str] ) -> None: query = ["data_source:tidal", *args] diff --git a/beetsplug/titlecase.py b/beetsplug/titlecase.py index 0d61afa809..536617e6e6 100644 --- a/beetsplug/titlecase.py +++ b/beetsplug/titlecase.py @@ -15,9 +15,11 @@ from beets.plugins import BeetsPlugin if TYPE_CHECKING: + import optparse + from beets.autotag import Info from beets.importer import ImportSession, ImportTask - from beets.library import Item + from beets.library import Item, Library __author__ = "henryoberholtzer@gmail.com" __version__ = "1.0" @@ -155,7 +157,7 @@ def received_info_handler(self, info: Info) -> None: self.titlecase_fields(track) def commands(self) -> list[ui.Subcommand]: - def func(lib, opts, args): + def func(lib: Library, opts: optparse.Values, args: list[str]) -> None: write = ui.should_write() for item in lib.items(args): self._log.info(f"titlecasing {item.title}:") diff --git a/beetsplug/unimported.py b/beetsplug/unimported.py index 83476ec11c..6d5524cb23 100644 --- a/beetsplug/unimported.py +++ b/beetsplug/unimported.py @@ -3,12 +3,21 @@ beets library database, including art files """ +from __future__ import annotations + import os +from typing import TYPE_CHECKING from beets import util from beets.plugins import BeetsPlugin from beets.ui import Subcommand, print_ +if TYPE_CHECKING: + import optparse + + from beets.library import Library + + __author__ = "https://github.com/MrNuggelz" @@ -18,7 +27,9 @@ def __init__(self): self.config.add({"ignore_extensions": [], "ignore_subdirectories": []}) def commands(self): - def print_unimported(lib, opts, args): + def print_unimported( + lib: Library, opts: optparse.Values, args: list[str] + ) -> None: ignore_exts = [ f".{x}".encode() for x in self.config["ignore_extensions"].as_str_seq() diff --git a/beetsplug/web/__init__.py b/beetsplug/web/__init__.py index fcd089e245..86f6c93353 100644 --- a/beetsplug/web/__init__.py +++ b/beetsplug/web/__init__.py @@ -1,9 +1,12 @@ """A Web interface to beets.""" +from __future__ import annotations + import base64 import json import os import typing as t +from typing import TYPE_CHECKING, Protocol import flask from flask import jsonify @@ -15,6 +18,14 @@ from beets.dbcore.query import PathQuery from beets.plugins import BeetsPlugin +if TYPE_CHECKING: + from beets.library import Library + + +class WebCLIOpts(Protocol): + debug: bool + + # Type checking hacks if t.TYPE_CHECKING: @@ -445,7 +456,7 @@ def commands(self): help="debug mode", ) - def func(lib, opts, args): + def func(lib: Library, opts: WebCLIOpts, args: list[str]) -> None: args = args if args: self.config["host"] = args.pop(0) @@ -479,7 +490,7 @@ def func(lib, opts, args): # Allow serving behind a reverse proxy if self.config["reverse_proxy"]: - app.wsgi_app = ReverseProxied(app.wsgi_app) + app.wsgi_app = ReverseProxied(app.wsgi_app) # type: ignore[method-assign] # Start the web application. app.run( diff --git a/beetsplug/zero.py b/beetsplug/zero.py index 1462aa43e7..6c5314c8bc 100644 --- a/beetsplug/zero.py +++ b/beetsplug/zero.py @@ -13,8 +13,10 @@ from beets.ui import Subcommand, input_yn if TYPE_CHECKING: + import optparse + from beets.importer import ImportSession, ImportTask - from beets.library import Item + from beets.library import Item, Library __author__ = "baobab@heresiarch.info" @@ -78,7 +80,9 @@ def __init__(self) -> None: def commands(self): zero_command = Subcommand("zero", help="set fields to null") - def zero_fields(lib, opts, args): + def zero_fields( + lib: Library, opts: optparse.Values, args: list[str] + ) -> None: if not args and not input_yn( "Remove fields for all items? (Y/n)", True ): diff --git a/docs/changelog.rst b/docs/changelog.rst index 170708fdc3..5fa420a1d8 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -67,6 +67,8 @@ Bug fixes (:doc:`plugins/musicbrainz`, :doc:`plugins/spotify`, :doc:`plugins/deezer` and :doc:`plugins/discogs`) no longer send a search request when both the query text and the filters are empty. :bug:`6862` +- :doc:`plugins/ipfs`: Fix ``beet ipfs --play`` option to invoke the Play plugin + through its command interface. .. For plugin developers diff --git a/test/plugins/lyrics_pages.py b/test/plugins/lyrics_pages.py index 09760ba568..480717d1d1 100644 --- a/test/plugins/lyrics_pages.py +++ b/test/plugins/lyrics_pages.py @@ -278,7 +278,7 @@ def backend(self) -> str: url_title="Lady Madonna - The Beatles - LETRAS.MUS.BR", ), LyricsPage.make( - "https://lrclib.net/api/get/19648857", + "https://lrclib.net/api/get/23863037", """ [00:08.35] Lady Madonna, children at your feet [00:12.85] Wonder how you manage to make ends meet @@ -310,6 +310,7 @@ def backend(self) -> str: LyricsPage.make( "https://api.lrcmux.dev/get?artist=The+Beatles&title=Lady+Madonna&duration=186&format=lrc&level=line&sources=ytmusic", """ + [00:00.00] [00:08.71] Lady Madonna, children at your feet [00:13.08] Wonder how you manage to make ends meet [00:17.43] Who finds the money when you pay the rent? diff --git a/test/plugins/test_ipfs.py b/test/plugins/test_ipfs.py index c0cfb0c0ab..934d83621a 100644 --- a/test/plugins/test_ipfs.py +++ b/test/plugins/test_ipfs.py @@ -1,9 +1,10 @@ +import os from pathlib import Path from unittest.mock import Mock, patch -from beets import library +from beets import library, util from beets.test import _common -from beets.test.helper import PluginTestCase +from beets.test.helper import PluginTestCase, PluginTestHelper from beetsplug.ipfs import IPFSPlugin @@ -44,11 +45,8 @@ def test_get_remote_lib_accepts_library_path(self): remote_lib._close() ipfs = IPFSPlugin() - added_lib = ipfs.get_remote_lib(self.lib) - try: + with ipfs.remote_lib(self.lib) as added_lib: assert added_lib.path == remote_dir / "joined.db" - finally: - added_lib._close() def mk_test_album(self): items = [_common.item() for _ in range(3)] @@ -77,3 +75,25 @@ def mk_test_album(self): album.store(inherit=False) return album + + +class TestIPFSPlay(PluginTestHelper): + plugin = "ipfs" + db_on_disk = True + + def test_ipfs_play(self, monkeypatch): + """Test that ipfs successfully calls PlayPlugin's play method.""" + # do not attempt to actually play the music + monkeypatch.setattr("beetsplug.play.play", lambda *_: None) + + # we need some music to play + self.add_album_fixture() + + # create remote lib in the expected place, + # see IPFSPlugin._remote_libs_path + remote_dir = Path(os.fsdecode(self.lib.path)).parent / "remotes" + remote_dir.mkdir() + util.copy(self.lib.path, remote_dir / "joined.db") + + # check that we can play without any errors + IPFSPlugin().ipfs_play(self.lib, None, []) diff --git a/test/ui/commands/test_move.py b/test/ui/commands/test_move.py index fbbb69a38f..b9ec6ef792 100644 --- a/test/ui/commands/test_move.py +++ b/test/ui/commands/test_move.py @@ -1,3 +1,4 @@ +import os import shutil from beets import library @@ -29,7 +30,15 @@ def _move( pretend=False, export=False, ): - move_items(self.lib, dest, query, copy, album, pretend, export=export) + move_items( + self.lib, + os.fsencode(dest) if dest else None, + query, + copy, + album, + pretend, + export=export, + ) def test_move_item(self): self._move() diff --git a/test/ui/test_ui.py b/test/ui/test_ui.py index 3b45b1a1e3..0d1c628298 100644 --- a/test/ui/test_ui.py +++ b/test/ui/test_ui.py @@ -375,7 +375,7 @@ def test_album_option(self): parser.add_album_option() assert bool(parser._album_flags) - assert parser.parse_args([]) == ({"album": None}, []) + assert parser.parse_args([]) == ({"album": False}, []) assert parser.parse_args(["-a"]) == ({"album": True}, []) assert parser.parse_args(["--album"]) == ({"album": True}, []) @@ -456,7 +456,7 @@ def test_add_all_common_options(self): parser = ui.CommonOptionsParser() parser.add_all_common_options() assert parser.parse_args([]) == ( - {"album": None, "path": None, "format": None}, + {"album": False, "path": None, "format": None}, [], )