Merge branch 'beetbox:master' into play-randomize

Author: amiv1

beets/metadata_plugins.py

@@ -11,25 +11,32 @@
 import re
 from contextlib import contextmanager, nullcontext
 from functools import cache, cached_property, wraps
-from typing import TYPE_CHECKING, Generic, Literal, TypedDict, TypeVar
+from typing import (
+    TYPE_CHECKING,
+    Generic,
+    Literal,
+    NamedTuple,
+    TypedDict,
+    TypeVar,
+)
 
 import unidecode
 from confuse import NotFoundError
-from typing_extensions import NotRequired
 
 from beets import config, logging
 from beets.util import cached_classproperty
 from beets.util.id_extractors import extract_release_id
 
 from .plugins import BeetsPlugin, find_plugins, notify_info_yielded
 
+Ret = TypeVar("Ret")
+QueryType = Literal["album", "track"]
+
 if TYPE_CHECKING:
     from collections.abc import Callable, Iterable, Iterator, Sequence
 
     from .autotag.hooks import AlbumInfo, Item, TrackInfo
 
-    Ret = TypeVar("Ret")
-
 # Global logger.
 log = logging.getLogger("beets")
 
@@ -198,7 +205,7 @@ def item_candidates(
         """
         raise NotImplementedError
 
-    def albums_for_ids(self, ids: Sequence[str]) -> Iterable[AlbumInfo | None]:
+    def albums_for_ids(self, ids: Iterable[str]) -> Iterable[AlbumInfo | None]:
         """Batch lookup of album metadata for a list of album IDs.
 
         Given a list of album identifiers, yields corresponding AlbumInfo objects.
@@ -209,7 +216,7 @@ def albums_for_ids(self, ids: Sequence[str]) -> Iterable[AlbumInfo | None]:
 
         return (self.album_for_id(id) for id in ids)
 
-    def tracks_for_ids(self, ids: Sequence[str]) -> Iterable[TrackInfo | None]:
+    def tracks_for_ids(self, ids: Iterable[str]) -> Iterable[TrackInfo | None]:
         """Batch lookup of track metadata for a list of track IDs.
 
         Given a list of track identifiers, yields corresponding TrackInfo objects.
@@ -283,9 +290,17 @@ class IDResponse(TypedDict):
     id: str
 
 
-class SearchFilter(TypedDict):
-    artist: NotRequired[str]
-    album: NotRequired[str]
+class SearchParams(NamedTuple):
+    """Bundle normalized search context passed to provider search hooks.
+
+    Shared search orchestration constructs this value so plugin hooks receive
+    one object describing search intent, query text, and provider filters.
+    """
+
+    query_type: QueryType
+    query: str
+    filters: dict[str, str]
+    limit: int
 
 
 R = TypeVar("R", bound=IDResponse)
@@ -312,77 +327,92 @@ def __init__(self, *args, **kwargs) -> None:
         )
 
     @abc.abstractmethod
-    def _search_api(
+    def get_search_query_with_filters(
         self,
-        query_type: Literal["album", "track"],
-        filters: SearchFilter,
-        query_string: str = "",
-    ) -> Sequence[R]:
-        """Perform a search on the API.
+        query_type: QueryType,
+        items: Sequence[Item],
+        artist: str,
+        name: str,
+        va_likely: bool,
+    ) -> tuple[str, dict[str, str]]:
+        """Build query text and API filters for a provider search.
+
+        Subclasses can override this hook when their API requires a query format
+        or filter set that differs from the default text-based construction.
+
+        :param query_type: The type of query to perform. Either *album* or *track*
+        :param items: List of items the search is being performed for
+        :param artist: Artist name
+        :param name: Album or track name, depending on ``query_type``
+        :param va_likely: Whether the search is likely to be for various artists
+        :return: Tuple of (``query`` text, ``filters`` dict) to use for the
+            search API call
+        """
+
+    @abc.abstractmethod
+    def get_search_response(self, params: SearchParams) -> Sequence[R]:
+        """Fetch raw search results for a provider request.
 
-        :param query_type: The type of query to perform.
-        :param filters: A dictionary of filters to apply to the search.
-        :param query_string: Additional query to include in the search.
+        Implementations should return records containing source IDs so shared
+        candidate resolution can perform ID-based album and track lookups.
 
-        Should return a list of identifiers for the requested type (album or track).
+        :param params: :py:namedtuple:`~SearchParams` named tuple
+        :return: Sequence of IDResponse dicts containing at least an "id" key for each
         """
+
         raise NotImplementedError
 
+    def _search_api(
+        self, query_type: QueryType, query: str, filters: dict[str, str]
+    ) -> Sequence[R]:
+        """Run shared provider search orchestration and return ID-bearing results.
+
+        This path applies optional query normalization and default limits, then
+        delegates API access to provider hooks with consistent logging and
+        failure handling.
+        """
+        if self.config["search_query_ascii"].get():
+            query = unidecode.unidecode(query)
+
+        limit = self.config["search_limit"].get(int)
+        params = SearchParams(query_type, query, filters, limit)
+
+        self._log.debug("Searching for '{}' with {}", query, filters)
+        try:
+            response_data = self.get_search_response(params)
+        except Exception as e:
+            if config["raise_on_error"].get(bool):
+                raise
+            self._log.error(
+                "Error searching {.data_source}: {}", self, e, exc_info=True
+            )
+            return ()
+
+        self._log.debug("Found {} result(s)", len(response_data))
+        return response_data
+
+    def _get_candidates(
+        self, query_type: QueryType, *args, **kwargs
+    ) -> Sequence[R]:
+        """Resolve query hooks and execute one provider search request."""
+
+        return self._search_api(
+            query_type,
+            *self.get_search_query_with_filters(query_type, *args, **kwargs),
+        )
+
     def candidates(
         self,
         items: Sequence[Item],
         artist: str,
         album: str,
         va_likely: bool,
     ) -> Iterable[AlbumInfo]:
-        query_filters: SearchFilter = {}
-        if album:
-            query_filters["album"] = album
-        if not va_likely:
-            query_filters["artist"] = artist
-
-        results = self._search_api("album", query_filters)
-        if not results:
-            return []
-
-        return filter(
-            None, self.albums_for_ids([result["id"] for result in results])
-        )
+        results = self._get_candidates("album", items, artist, album, va_likely)
+        return filter(None, self.albums_for_ids(r["id"] for r in results))
 
     def item_candidates(
         self, item: Item, artist: str, title: str
     ) -> Iterable[TrackInfo]:
-        results = self._search_api(
-            "track", {"artist": artist}, query_string=title
-        )
-        if not results:
-            return []
-
-        return filter(
-            None,
-            self.tracks_for_ids([result["id"] for result in results if result]),
-        )
-
-    def _construct_search_query(
-        self, filters: SearchFilter, query_string: str
-    ) -> str:
-        """Construct a query string with the specified filters and keywords to
-        be provided to the spotify (or similar) search API.
-
-        The returned format was initially designed for spotify's search API but
-        we found is also useful with other APIs that support similar query structures.
-        see `spotify <https://developer.spotify.com/documentation/web-api/reference/search>`_
-        and `deezer <https://developers.deezer.com/api/search>`_.
-
-        :param filters: Field filters to apply.
-        :param query_string: Query keywords to use.
-        :return: Query string to be provided to the search API.
-        """
-
-        components = [query_string, *(f"{k}:'{v}'" for k, v in filters.items())]
-        query = " ".join(filter(None, components))
-
-        if self.config["search_query_ascii"].get():
-            query = unidecode.unidecode(query)
-
-        return query
+        results = self._get_candidates("track", [item], artist, title, False)
+        return filter(None, self.tracks_for_ids(r["id"] for r in results))

beetsplug/_utils/musicbrainz.py

@@ -29,6 +29,8 @@
 
     from requests import Response
 
+    from beets.metadata_plugins import IDResponse
+
     from .._typing import JSONDict
 
 log = logging.getLogger("beets")
@@ -232,7 +234,7 @@ def search(
         entity: Entity,
         filters: dict[str, str],
         **kwargs: Unpack[SearchKwargs],
-    ) -> list[JSONDict]:
+    ) -> list[IDResponse]:
         """Search for MusicBrainz entities matching the given filters.
 
         * Query is constructed by combining the provided filters using AND logic

beetsplug/deezer.py

@@ -18,7 +18,7 @@
 
 import collections
 import time
-from typing import TYPE_CHECKING, ClassVar, Literal
+from typing import TYPE_CHECKING, ClassVar
 
 import requests
 
@@ -31,7 +31,7 @@
     from collections.abc import Sequence
 
     from beets.library import Item, Library
-    from beets.metadata_plugins import SearchFilter
+    from beets.metadata_plugins import QueryType, SearchParams
 
     from ._typing import JSONDict
 
@@ -217,58 +217,34 @@ def _get_track(self, track_data: JSONDict) -> TrackInfo:
             deezer_updated=time.time(),
         )
 
-    def _search_api(
+    def get_search_query_with_filters(
         self,
-        query_type: Literal[
-            "album",
-            "track",
-            "artist",
-            "history",
-            "playlist",
-            "podcast",
-            "radio",
-            "user",
-        ],
-        filters: SearchFilter,
-        query_string: str = "",
-    ) -> Sequence[IDResponse]:
-        """Query the Deezer Search API for the specified ``query_string``, applying
-        the provided ``filters``.
-
-        :param filters: Field filters to apply.
-        :param query_string: Additional query to include in the search.
-        :return: JSON data for the class:`Response <Response>` object or None
-            if no search results are returned.
-        """
-        query = self._construct_search_query(
-            query_string=query_string, filters=filters
-        )
-        self._log.debug("Searching {.data_source} for '{}'", self, query)
-        try:
-            response = requests.get(
-                f"{self.search_url}{query_type}",
-                params={
-                    "q": query,
-                    "limit": self.config["search_limit"].get(),
-                },
-                timeout=10,
-            )
-            response.raise_for_status()
-        except requests.exceptions.RequestException as e:
-            self._log.error(
-                "Error fetching data from {.data_source} API\n Error: {}",
-                self,
-                e,
-            )
-            return ()
-        response_data: Sequence[IDResponse] = response.json().get("data", [])
-        self._log.debug(
-            "Found {} result(s) from {.data_source} for '{}'",
-            len(response_data),
-            self,
-            query,
+        query_type: QueryType,
+        items: Sequence[Item],
+        artist: str,
+        name: str,
+        va_likely: bool,
+    ) -> tuple[str, dict[str, str]]:
+        query = f'album:"{name}"' if query_type == "album" else name
+        if query_type == "track" or not va_likely:
+            query += f' artist:"{artist}"'
+
+        return query, {}
+
+    def get_search_response(self, params: SearchParams) -> list[IDResponse]:
+        """Search Deezer and return the raw result payload entries."""
+
+        response = requests.get(
+            f"{self.search_url}{params.query_type}",
+            params={
+                **params.filters,
+                "q": params.query,
+                "limit": str(params.limit),
+            },
+            timeout=10,
         )
-        return response_data
+        response.raise_for_status()
+        return response.json()["data"]
 
     def deezerupdate(self, items: Sequence[Item], write: bool):
         """Obtain rank information from Deezer."""