Merge branch 'master' into feature/add-artist-to-item-entry-template

Author: matlads

.github/workflows/ci.yaml

@@ -33,7 +33,7 @@ jobs:
         if: matrix.platform == 'ubuntu-latest'
         run: |
           sudo apt update
-          sudo apt install ffmpeg gobject-introspection libcairo2-dev libgirepository-2.0-dev pandoc imagemagick
+          sudo apt install --yes --no-install-recommends ffmpeg gobject-introspection gstreamer1.0-plugins-base python3-gst-1.0 libcairo2-dev libgirepository-2.0-dev pandoc imagemagick
 
       - name: Get changed lyrics files
         id: lyrics-update

.github/workflows/lint.yml

@@ -105,7 +105,6 @@ jobs:
 
       - name: Type check code
         uses: liskin/gh-problem-matcher-wrap@v3
-        continue-on-error: true
         with:
           linters: mypy
           run: poe check-types --show-column-numbers --no-error-summary ${{ needs.changed-files.outputs.changed_python_files }}

beets/autotag/distance.py

@@ -8,7 +8,7 @@
 from jellyfish import levenshtein_distance
 from unidecode import unidecode
 
-from beets import config, plugins
+from beets import config, metadata_plugins
 from beets.util import as_string, cached_classproperty, get_most_common_tags
 
 if TYPE_CHECKING:
@@ -409,7 +409,7 @@ def track_distance(
         dist.add_expr("medium", item.disc != track_info.medium)
 
     # Plugins.
-    dist.update(plugins.track_distance(item, track_info))
+    dist.update(metadata_plugins.track_distance(item, track_info))
 
     return dist
 
@@ -526,6 +526,6 @@ def distance(
         dist.add("unmatched_tracks", 1.0)
 
     # Plugins.
-    dist.update(plugins.album_distance(items, album_info, mapping))
+    dist.update(metadata_plugins.album_distance(items, album_info, mapping))
 
     return dist

beets/autotag/match.py

@@ -24,7 +24,7 @@
 import lap
 import numpy as np
 
-from beets import config, logging, plugins
+from beets import config, logging, metadata_plugins
 from beets.autotag import AlbumInfo, AlbumMatch, TrackInfo, TrackMatch, hooks
 from beets.util import get_most_common_tags
 
@@ -119,7 +119,7 @@ def match_by_id(items: Iterable[Item]) -> AlbumInfo | None:
             return None
     # If all album IDs are equal, look up the album.
     log.debug("Searching for discovered album ID: {0}", first)
-    return plugins.album_for_id(first)
+    return metadata_plugins.album_for_id(first)
 
 
 def _recommendation(
@@ -274,7 +274,7 @@ def tag_album(
     if search_ids:
         for search_id in search_ids:
             log.debug("Searching for album ID: {0}", search_id)
-            if info := plugins.album_for_id(search_id):
+            if info := metadata_plugins.album_for_id(search_id):
                 _add_candidate(items, candidates, info)
 
     # Use existing metadata or text search.
@@ -311,7 +311,7 @@ def tag_album(
         log.debug("Album might be VA: {0}", va_likely)
 
         # Get the results from the data sources.
-        for matched_candidate in plugins.candidates(
+        for matched_candidate in metadata_plugins.candidates(
             items, search_artist, search_album, va_likely
         ):
             _add_candidate(items, candidates, matched_candidate)
@@ -346,7 +346,7 @@ def tag_item(
     if trackids:
         for trackid in trackids:
             log.debug("Searching for track ID: {0}", trackid)
-            if info := plugins.track_for_id(trackid):
+            if info := metadata_plugins.track_for_id(trackid):
                 dist = track_distance(item, info, incl_artist=True)
                 candidates[info.track_id] = hooks.TrackMatch(dist, info)
                 # If this is a good match, then don't keep searching.
@@ -372,7 +372,7 @@ def tag_item(
     log.debug("Item search terms: {0} - {1}", search_artist, search_title)
 
     # Get and evaluate candidate metadata.
-    for track_info in plugins.item_candidates(
+    for track_info in metadata_plugins.item_candidates(
         item, search_artist, search_title
     ):
         dist = track_distance(item, track_info, incl_artist=True)

beets/dbcore/db.py

@@ -289,19 +289,22 @@ class Model(ABC, Generic[D]):
     terms.
     """
 
-    _types: dict[str, types.Type] = {}
-    """Optional Types for non-fixed (i.e., flexible and computed) fields.
-    """
+    @cached_classproperty
+    def _types(cls) -> dict[str, types.Type]:
+        """Optional types for non-fixed (flexible and computed) fields."""
+        return {}
 
     _sorts: dict[str, type[FieldSort]] = {}
     """Optional named sort criteria. The keys are strings and the values
     are subclasses of `Sort`.
     """
 
-    _queries: dict[str, FieldQueryType] = {}
-    """Named queries that use a field-like `name:value` syntax but which
-    do not relate to any specific field.
-    """
+    @cached_classproperty
+    def _queries(cls) -> dict[str, FieldQueryType]:
+        """Named queries that use a field-like `name:value` syntax but which
+        do not relate to any specific field.
+        """
+        return {}
 
     _always_dirty = False
     """By default, fields only become "dirty" when their value actually

beets/dbcore/query.py

@@ -28,6 +28,7 @@
 from typing import TYPE_CHECKING, Any, Generic, TypeVar, Union
 
 from beets import util
+from beets.util.units import raw_seconds_short
 
 if TYPE_CHECKING:
     from beets.dbcore.db import AnyModel, Model
@@ -892,7 +893,7 @@ def _convert(self, s: str) -> float | None:
         if not s:
             return None
         try:
-            return util.raw_seconds_short(s)
+            return raw_seconds_short(s)
         except ValueError:
             try:
                 return float(s)

beets/dbcore/types.py

@@ -292,7 +292,7 @@ class DelimitedString(BaseString[list[str], list[str]]):
     containing delimiter-separated values.
     """
 
-    model_type = list
+    model_type = list[str]
 
     def __init__(self, delimiter: str):
         self.delimiter = delimiter

beets/importer/stages.py

@@ -70,6 +70,7 @@ def query_tasks(session: ImportSession):
     Instead of finding files from the filesystem, a query is used to
     match items from the library.
     """
+    task: ImportTask
     if session.config["singletons"]:
         # Search for items.
         for item in session.lib.items(session.query):
@@ -143,9 +144,7 @@ def lookup_candidates(session: ImportSession, task: ImportTask):
 
     # Restrict the initial lookup to IDs specified by the user via the -m
     # option. Currently all the IDs are passed onto the tasks directly.
-    task.search_ids = session.config["search_ids"].as_str_seq()
-
-    task.lookup_candidates()
+    task.lookup_candidates(session.config["search_ids"].as_str_seq())
 
 
 @pipeline.stage

beets/importer/tasks.py

@@ -22,7 +22,7 @@
 from collections import defaultdict
 from enum import Enum
 from tempfile import mkdtemp
-from typing import TYPE_CHECKING, Callable, Iterable, Sequence
+from typing import TYPE_CHECKING, Any, Callable, Iterable, Sequence
 
 import mediafile
 
@@ -32,6 +32,8 @@
 from .state import ImportState
 
 if TYPE_CHECKING:
+    from beets.autotag.match import Recommendation
+
     from .session import ImportSession
 
 # Global logger.
@@ -159,6 +161,7 @@ class ImportTask(BaseImportTask):
     cur_album: str | None = None
     cur_artist: str | None = None
     candidates: Sequence[autotag.AlbumMatch | autotag.TrackMatch] = []
+    rec: Recommendation | None = None
 
     def __init__(
         self,
@@ -167,11 +170,9 @@ def __init__(
         items: Iterable[library.Item] | None,
     ):
         super().__init__(toppath, paths, items)
-        self.rec = None
         self.should_remove_duplicates = False
         self.should_merge_duplicates = False
         self.is_album = True
-        self.search_ids = []  # user-supplied candidate IDs.
 
     def set_choice(
         self, choice: Action | autotag.AlbumMatch | autotag.TrackMatch
@@ -356,20 +357,17 @@ def handle_created(self, session: ImportSession):
             tasks = [t for inner in tasks for t in inner]
         return tasks
 
-    def lookup_candidates(self):
-        """Retrieve and store candidates for this album. User-specified
-        candidate IDs are stored in self.search_ids: if present, the
-        initial lookup is restricted to only those IDs.
+    def lookup_candidates(self, search_ids: list[str]) -> None:
+        """Retrieve and store candidates for this album.
+
+        If User-specified ``search_ids`` list is not empty, the lookup is
+        restricted to only those IDs.
         """
-        artist, album, prop = autotag.tag_album(
-            self.items, search_ids=self.search_ids
+        self.cur_artist, self.cur_album, (self.candidates, self.rec) = (
+            autotag.tag_album(self.items, search_ids=search_ids)
         )
-        self.cur_artist = artist
-        self.cur_album = album
-        self.candidates = prop.candidates
-        self.rec = prop.recommendation
 
-    def find_duplicates(self, lib: library.Library):
+    def find_duplicates(self, lib: library.Library) -> list[library.Album]:
         """Return a list of albums from `lib` with the same artist and
         album name as the task.
         """
@@ -695,12 +693,12 @@ def _emit_imported(self, lib):
         for item in self.imported_items():
             plugins.send("item_imported", lib=lib, item=item)
 
-    def lookup_candidates(self):
-        prop = autotag.tag_item(self.item, search_ids=self.search_ids)
-        self.candidates = prop.candidates
-        self.rec = prop.recommendation
+    def lookup_candidates(self, search_ids: list[str]) -> None:
+        self.candidates, self.rec = autotag.tag_item(
+            self.item, search_ids=search_ids
+        )
 
-    def find_duplicates(self, lib):
+    def find_duplicates(self, lib: library.Library) -> list[library.Item]:  # type: ignore[override] # Need splitting Singleton and Album tasks into separate classes
         """Return a list of items from `lib` that have the same artist
         and title as the task.
         """
@@ -802,6 +800,11 @@ def _emit_imported(self, lib):
         pass
 
 
+ArchiveHandler = tuple[
+    Callable[[util.StrPath], bool], Callable[[util.StrPath], Any]
+]
+
+
 class ArchiveImportTask(SentinelImportTask):
     """An import task that represents the processing of an archive.
 
@@ -827,42 +830,41 @@ def is_archive(cls, path):
         if not os.path.isfile(path):
             return False
 
-        for path_test, _ in cls.handlers():
+        for path_test, _ in cls.handlers:
             if path_test(os.fsdecode(path)):
                 return True
         return False
 
-    @classmethod
-    def handlers(cls):
+    @util.cached_classproperty
+    def handlers(cls) -> list[ArchiveHandler]:
         """Returns a list of archive handlers.
 
         Each handler is a `(path_test, ArchiveClass)` tuple. `path_test`
         is a function that returns `True` if the given path can be
         handled by `ArchiveClass`. `ArchiveClass` is a class that
         implements the same interface as `tarfile.TarFile`.
         """
-        if not hasattr(cls, "_handlers"):
-            cls._handlers: list[tuple[Callable, ...]] = []
-            from zipfile import ZipFile, is_zipfile
+        _handlers: list[ArchiveHandler] = []
+        from zipfile import ZipFile, is_zipfile
 
-            cls._handlers.append((is_zipfile, ZipFile))
-            import tarfile
+        _handlers.append((is_zipfile, ZipFile))
+        import tarfile
 
-            cls._handlers.append((tarfile.is_tarfile, tarfile.open))
-            try:
-                from rarfile import RarFile, is_rarfile
-            except ImportError:
-                pass
-            else:
-                cls._handlers.append((is_rarfile, RarFile))
-            try:
-                from py7zr import SevenZipFile, is_7zfile
-            except ImportError:
-                pass
-            else:
-                cls._handlers.append((is_7zfile, SevenZipFile))
+        _handlers.append((tarfile.is_tarfile, tarfile.open))
+        try:
+            from rarfile import RarFile, is_rarfile
+        except ImportError:
+            pass
+        else:
+            _handlers.append((is_rarfile, RarFile))
+        try:
+            from py7zr import SevenZipFile, is_7zfile
+        except ImportError:
+            pass
+        else:
+            _handlers.append((is_7zfile, SevenZipFile))
 
-        return cls._handlers
+        return _handlers
 
     def cleanup(self, copy=False, delete=False, move=False):
         """Removes the temporary directory the archive was extracted to."""
@@ -879,7 +881,7 @@ def extract(self):
         """
         assert self.toppath is not None, "toppath must be set"
 
-        for path_test, handler_class in self.handlers():
+        for path_test, handler_class in self.handlers:
             if path_test(os.fsdecode(self.toppath)):
                 break
         else:
@@ -925,7 +927,7 @@ def __init__(self, toppath: util.PathBytes, session: ImportSession):
         self.imported = 0  # "Real" tasks created.
         self.is_archive = ArchiveImportTask.is_archive(util.syspath(toppath))
 
-    def tasks(self):
+    def tasks(self) -> Iterable[ImportTask]:
         """Yield all import tasks for music found in the user-specified
         path `self.toppath`. Any necessary sentinel tasks are also
         produced.
@@ -1114,7 +1116,10 @@ def albums_in_dir(path: util.PathBytes):
     a list of Items that is probably an album. Specifically, any folder
     containing any media files is an album.
     """
-    collapse_pat = collapse_paths = collapse_items = None
+    collapse_paths: list[util.PathBytes] = []
+    collapse_items: list[util.PathBytes] = []
+    collapse_pat = None
+
     ignore: list[str] = config["ignore"].as_str_seq()
     ignore_hidden: bool = config["ignore_hidden"].get(bool)
 
@@ -1139,7 +1144,7 @@ def albums_in_dir(path: util.PathBytes):
                 # proceed to process the current one.
                 if collapse_items:
                     yield collapse_paths, collapse_items
-                collapse_pat = collapse_paths = collapse_items = None
+                collapse_pat, collapse_paths, collapse_items = None, [], []
 
         # Check whether this directory looks like the *first* directory
         # in a multi-disc sequence. There are two indicators: the file