mirror of
https://github.com/beetbox/beets.git
synced 2025-12-28 11:32:30 +01:00
Merge e89d97dfe2 into 2bd77b9895
This commit is contained in:
commit
d3c3e31445
10 changed files with 387 additions and 229 deletions
|
|
@ -17,10 +17,13 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from copy import deepcopy
|
||||
from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING, Any, TypeVar
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
from beets import plugins
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from beets.library import Item
|
||||
|
||||
|
|
@ -54,6 +57,16 @@ class AttrDict(dict[str, V]):
|
|||
class Info(AttrDict[Any]):
|
||||
"""Container for metadata about a musical entity."""
|
||||
|
||||
Identifier = tuple[str | None, str | None]
|
||||
|
||||
@property
|
||||
def id(self) -> str | None:
|
||||
raise NotImplementedError
|
||||
|
||||
@property
|
||||
def identifier(self) -> Identifier:
|
||||
return (self.data_source, self.id)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
album: str | None = None,
|
||||
|
|
@ -95,6 +108,10 @@ class AlbumInfo(Info):
|
|||
user items, and later to drive tagging decisions once selected.
|
||||
"""
|
||||
|
||||
@property
|
||||
def id(self) -> str | None:
|
||||
return self.album_id
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
tracks: list[TrackInfo],
|
||||
|
|
@ -167,6 +184,10 @@ class TrackInfo(Info):
|
|||
stand alone for singleton matching.
|
||||
"""
|
||||
|
||||
@property
|
||||
def id(self) -> str | None:
|
||||
return self.track_id
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
|
|
@ -214,16 +235,23 @@ class TrackInfo(Info):
|
|||
|
||||
|
||||
# Structures that compose all the information for a candidate match.
|
||||
|
||||
|
||||
class AlbumMatch(NamedTuple):
|
||||
@dataclass
|
||||
class Match:
|
||||
distance: Distance
|
||||
info: Info
|
||||
|
||||
|
||||
@dataclass
|
||||
class AlbumMatch(Match):
|
||||
info: AlbumInfo
|
||||
mapping: dict[Item, TrackInfo]
|
||||
extra_items: list[Item]
|
||||
extra_tracks: list[TrackInfo]
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
plugins.send("album_matched", match=self)
|
||||
|
||||
class TrackMatch(NamedTuple):
|
||||
distance: Distance
|
||||
|
||||
@dataclass
|
||||
class TrackMatch(Match):
|
||||
info: TrackInfo
|
||||
|
|
|
|||
|
|
@ -19,12 +19,12 @@ releases and tracks.
|
|||
from __future__ import annotations
|
||||
|
||||
from enum import IntEnum
|
||||
from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar
|
||||
from typing import TYPE_CHECKING, NamedTuple, TypeVar
|
||||
|
||||
import lap
|
||||
import numpy as np
|
||||
|
||||
from beets import config, logging, metadata_plugins, 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
|
||||
|
||||
|
|
@ -35,6 +35,11 @@ if TYPE_CHECKING:
|
|||
|
||||
from beets.library import Item
|
||||
|
||||
from .hooks import Info
|
||||
|
||||
AnyMatch = TypeVar("AnyMatch", TrackMatch, AlbumMatch)
|
||||
Candidates = dict[Info.Identifier, AnyMatch]
|
||||
|
||||
# Global logger.
|
||||
log = logging.getLogger("beets")
|
||||
|
||||
|
|
@ -98,28 +103,21 @@ def assign_items(
|
|||
return mapping, extra_items, extra_tracks
|
||||
|
||||
|
||||
def match_by_id(items: Iterable[Item]) -> AlbumInfo | None:
|
||||
"""If the items are tagged with an external source ID, return an
|
||||
AlbumInfo object for the corresponding album. Otherwise, returns
|
||||
None.
|
||||
def match_by_id(album_id: str | None, consensus: bool) -> Iterable[AlbumInfo]:
|
||||
"""Return album candidates for the given album id.
|
||||
|
||||
Make sure that the ID is present and that there is consensus on it among
|
||||
the items being tagged.
|
||||
"""
|
||||
albumids = (item.mb_albumid for item in items if item.mb_albumid)
|
||||
|
||||
# Did any of the items have an MB album ID?
|
||||
try:
|
||||
first = next(albumids)
|
||||
except StopIteration:
|
||||
if not album_id:
|
||||
log.debug("No album ID found.")
|
||||
return None
|
||||
elif not consensus:
|
||||
log.debug("No album ID consensus.")
|
||||
else:
|
||||
log.debug("Searching for discovered album ID: {}", album_id)
|
||||
return metadata_plugins.albums_for_ids([album_id])
|
||||
|
||||
# Is there a consensus on the MB album ID?
|
||||
for other in albumids:
|
||||
if other != first:
|
||||
log.debug("No album ID consensus.")
|
||||
return None
|
||||
# If all album IDs are equal, look up the album.
|
||||
log.debug("Searching for discovered album ID: {}", first)
|
||||
return metadata_plugins.album_for_id(first)
|
||||
return ()
|
||||
|
||||
|
||||
def _recommendation(
|
||||
|
|
@ -179,9 +177,6 @@ def _recommendation(
|
|||
return rec
|
||||
|
||||
|
||||
AnyMatch = TypeVar("AnyMatch", TrackMatch, AlbumMatch)
|
||||
|
||||
|
||||
def _sort_candidates(candidates: Iterable[AnyMatch]) -> Sequence[AnyMatch]:
|
||||
"""Sort candidates by distance."""
|
||||
return sorted(candidates, key=lambda match: match.distance)
|
||||
|
|
@ -189,7 +184,7 @@ def _sort_candidates(candidates: Iterable[AnyMatch]) -> Sequence[AnyMatch]:
|
|||
|
||||
def _add_candidate(
|
||||
items: Sequence[Item],
|
||||
results: dict[Any, AlbumMatch],
|
||||
results: Candidates[AlbumMatch],
|
||||
info: AlbumInfo,
|
||||
):
|
||||
"""Given a candidate AlbumInfo object, attempt to add the candidate
|
||||
|
|
@ -197,7 +192,10 @@ def _add_candidate(
|
|||
checking the track count, ordering the items, checking for
|
||||
duplicates, and calculating the distance.
|
||||
"""
|
||||
log.debug("Candidate: {0.artist} - {0.album} ({0.album_id})", info)
|
||||
log.debug(
|
||||
"Candidate: {0.artist} - {0.album} ({0.album_id}) from {0.data_source}",
|
||||
info,
|
||||
)
|
||||
|
||||
# Discard albums with zero tracks.
|
||||
if not info.tracks:
|
||||
|
|
@ -205,7 +203,7 @@ def _add_candidate(
|
|||
return
|
||||
|
||||
# Prevent duplicates.
|
||||
if info.album_id and info.album_id in results:
|
||||
if info.identifier in results:
|
||||
log.debug("Duplicate.")
|
||||
return
|
||||
|
||||
|
|
@ -231,7 +229,7 @@ def _add_candidate(
|
|||
return
|
||||
|
||||
log.debug("Success. Distance: {}", dist)
|
||||
results[info.album_id] = hooks.AlbumMatch(
|
||||
results[info.identifier] = hooks.AlbumMatch(
|
||||
dist, info, mapping, extra_items, extra_tracks
|
||||
)
|
||||
|
||||
|
|
@ -266,38 +264,35 @@ def tag_album(
|
|||
log.debug("Tagging {} - {}", cur_artist, cur_album)
|
||||
|
||||
# The output result, keys are the MB album ID.
|
||||
candidates: dict[Any, AlbumMatch] = {}
|
||||
candidates: Candidates[AlbumMatch] = {}
|
||||
|
||||
# Search by explicit ID.
|
||||
if search_ids:
|
||||
for search_id in search_ids:
|
||||
log.debug("Searching for album ID: {}", search_id)
|
||||
if info := metadata_plugins.album_for_id(search_id):
|
||||
_add_candidate(items, candidates, info)
|
||||
if opt_candidate := candidates.get(info.album_id):
|
||||
plugins.send("album_matched", match=opt_candidate)
|
||||
log.debug("Searching for album IDs: {}", search_ids)
|
||||
for _info in metadata_plugins.albums_for_ids(search_ids):
|
||||
_add_candidate(items, candidates, _info)
|
||||
|
||||
# Use existing metadata or text search.
|
||||
else:
|
||||
# Try search based on current ID.
|
||||
if info := match_by_id(items):
|
||||
for info in match_by_id(
|
||||
likelies["mb_albumid"], consensus["mb_albumid"]
|
||||
):
|
||||
_add_candidate(items, candidates, info)
|
||||
for candidate in candidates.values():
|
||||
plugins.send("album_matched", match=candidate)
|
||||
|
||||
rec = _recommendation(list(candidates.values()))
|
||||
log.debug("Album ID match recommendation is {}", rec)
|
||||
if candidates and not config["import"]["timid"]:
|
||||
# If we have a very good MBID match, return immediately.
|
||||
# Otherwise, this match will compete against metadata-based
|
||||
# matches.
|
||||
if rec == Recommendation.strong:
|
||||
log.debug("ID match.")
|
||||
return (
|
||||
cur_artist,
|
||||
cur_album,
|
||||
Proposal(list(candidates.values()), rec),
|
||||
)
|
||||
rec = _recommendation(list(candidates.values()))
|
||||
log.debug("Album ID match recommendation is {}", rec)
|
||||
if candidates and not config["import"]["timid"]:
|
||||
# If we have a very good MBID match, return immediately.
|
||||
# Otherwise, this match will compete against metadata-based
|
||||
# matches.
|
||||
if rec == Recommendation.strong:
|
||||
log.debug("ID match.")
|
||||
return (
|
||||
cur_artist,
|
||||
cur_album,
|
||||
Proposal(list(candidates.values()), rec),
|
||||
)
|
||||
|
||||
# Search terms.
|
||||
if not (search_artist and search_album):
|
||||
|
|
@ -318,8 +313,6 @@ def tag_album(
|
|||
items, search_artist, search_album, va_likely
|
||||
):
|
||||
_add_candidate(items, candidates, matched_candidate)
|
||||
if opt_candidate := candidates.get(matched_candidate.album_id):
|
||||
plugins.send("album_matched", match=opt_candidate)
|
||||
|
||||
log.debug("Evaluating {} candidates.", len(candidates))
|
||||
# Sort and get the recommendation.
|
||||
|
|
@ -343,25 +336,22 @@ def tag_item(
|
|||
"""
|
||||
# Holds candidates found so far: keys are MBIDs; values are
|
||||
# (distance, TrackInfo) pairs.
|
||||
candidates = {}
|
||||
candidates: Candidates[TrackMatch] = {}
|
||||
rec: Recommendation | None = None
|
||||
|
||||
# First, try matching by the external source ID.
|
||||
trackids = search_ids or [t for t in [item.mb_trackid] if t]
|
||||
if trackids:
|
||||
for trackid in trackids:
|
||||
log.debug("Searching for track 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.
|
||||
rec = _recommendation(_sort_candidates(candidates.values()))
|
||||
if (
|
||||
rec == Recommendation.strong
|
||||
and not config["import"]["timid"]
|
||||
):
|
||||
log.debug("Track ID match.")
|
||||
return Proposal(_sort_candidates(candidates.values()), rec)
|
||||
log.debug("Searching for track IDs: {}", trackids)
|
||||
for info in metadata_plugins.tracks_for_ids(trackids):
|
||||
dist = track_distance(item, info, incl_artist=True)
|
||||
candidates[info.identifier] = hooks.TrackMatch(dist, info)
|
||||
|
||||
# If this is a good match, then don't keep searching.
|
||||
rec = _recommendation(_sort_candidates(candidates.values()))
|
||||
if rec == Recommendation.strong and not config["import"]["timid"]:
|
||||
log.debug("Track ID match.")
|
||||
return Proposal(_sort_candidates(candidates.values()), rec)
|
||||
|
||||
# If we're searching by ID, don't proceed.
|
||||
if search_ids:
|
||||
|
|
@ -381,7 +371,7 @@ def tag_item(
|
|||
item, search_artist, search_title
|
||||
):
|
||||
dist = track_distance(item, track_info, incl_artist=True)
|
||||
candidates[track_info.track_id] = hooks.TrackMatch(dist, track_info)
|
||||
candidates[track_info.identifier] = hooks.TrackMatch(dist, track_info)
|
||||
|
||||
# Sort by distance and return with recommendation.
|
||||
log.debug("Found {} candidates.", len(candidates))
|
||||
|
|
|
|||
|
|
@ -34,6 +34,14 @@ def find_metadata_source_plugins() -> list[MetadataSourcePlugin]:
|
|||
return [p for p in find_plugins() if hasattr(p, "data_source")] # type: ignore[misc]
|
||||
|
||||
|
||||
@cache
|
||||
def get_metadata_source(name: str) -> MetadataSourcePlugin | None:
|
||||
"""Get metadata source plugin by name."""
|
||||
name = name.lower()
|
||||
plugins = find_metadata_source_plugins()
|
||||
return next((p for p in plugins if p.data_source.lower() == name), None)
|
||||
|
||||
|
||||
@notify_info_yielded("albuminfo_received")
|
||||
def candidates(*args, **kwargs) -> Iterable[AlbumInfo]:
|
||||
"""Return matching album candidates from all metadata source plugins."""
|
||||
|
|
@ -48,28 +56,38 @@ def item_candidates(*args, **kwargs) -> Iterable[TrackInfo]:
|
|||
yield from plugin.item_candidates(*args, **kwargs)
|
||||
|
||||
|
||||
def album_for_id(_id: str) -> AlbumInfo | None:
|
||||
"""Get AlbumInfo object for the given ID string.
|
||||
|
||||
A single ID can yield just a single album, so we return the first match.
|
||||
"""
|
||||
@notify_info_yielded("albuminfo_received")
|
||||
def albums_for_ids(ids: Sequence[str]) -> Iterable[AlbumInfo]:
|
||||
"""Return matching albums from all metadata sources for the given ID."""
|
||||
for plugin in find_metadata_source_plugins():
|
||||
if info := plugin.album_for_id(album_id=_id):
|
||||
send("albuminfo_received", info=info)
|
||||
return info
|
||||
yield from plugin.albums_for_ids(ids)
|
||||
|
||||
|
||||
@notify_info_yielded("trackinfo_received")
|
||||
def tracks_for_ids(ids: Sequence[str]) -> Iterable[TrackInfo]:
|
||||
"""Return matching tracks from all metadata sources for the given ID."""
|
||||
for plugin in find_metadata_source_plugins():
|
||||
yield from plugin.tracks_for_ids(ids)
|
||||
|
||||
|
||||
def album_for_id(_id: str, data_source: str) -> AlbumInfo | None:
|
||||
"""Get AlbumInfo object for the given ID and data source."""
|
||||
if (plugin := get_metadata_source(data_source)) and (
|
||||
info := plugin.album_for_id(_id)
|
||||
):
|
||||
send("albuminfo_received", info=info)
|
||||
return info
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def track_for_id(_id: str) -> TrackInfo | None:
|
||||
"""Get TrackInfo object for the given ID string.
|
||||
|
||||
A single ID can yield just a single track, so we return the first match.
|
||||
"""
|
||||
for plugin in find_metadata_source_plugins():
|
||||
if info := plugin.track_for_id(_id):
|
||||
send("trackinfo_received", info=info)
|
||||
return info
|
||||
def track_for_id(_id: str, data_source: str) -> TrackInfo | None:
|
||||
"""Get TrackInfo object for the given ID and data source."""
|
||||
if (plugin := get_metadata_source(data_source)) and (
|
||||
info := plugin.track_for_id(_id)
|
||||
):
|
||||
send("trackinfo_received", info=info)
|
||||
return info
|
||||
|
||||
return None
|
||||
|
||||
|
|
@ -169,7 +187,7 @@ class MetadataSourcePlugin(BeetsPlugin, metaclass=abc.ABCMeta):
|
|||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
def albums_for_ids(self, ids: Sequence[str]) -> Iterable[AlbumInfo | None]:
|
||||
def albums_for_ids(self, ids: Sequence[str]) -> Iterable[AlbumInfo]:
|
||||
"""Batch lookup of album metadata for a list of album IDs.
|
||||
|
||||
Given a list of album identifiers, yields corresponding AlbumInfo objects.
|
||||
|
|
@ -178,9 +196,9 @@ class MetadataSourcePlugin(BeetsPlugin, metaclass=abc.ABCMeta):
|
|||
single calls to album_for_id.
|
||||
"""
|
||||
|
||||
return (self.album_for_id(id) for id in ids)
|
||||
return filter(None, (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: Sequence[str]) -> Iterable[TrackInfo]:
|
||||
"""Batch lookup of track metadata for a list of track IDs.
|
||||
|
||||
Given a list of track identifiers, yields corresponding TrackInfo objects.
|
||||
|
|
@ -189,7 +207,7 @@ class MetadataSourcePlugin(BeetsPlugin, metaclass=abc.ABCMeta):
|
|||
single calls to track_for_id.
|
||||
"""
|
||||
|
||||
return (self.track_for_id(id) for id in ids)
|
||||
return filter(None, (self.track_for_id(id) for id in ids))
|
||||
|
||||
def _extract_id(self, url: str) -> str | None:
|
||||
"""Extract an ID from a URL for this metadata source plugin.
|
||||
|
|
|
|||
|
|
@ -72,17 +72,25 @@ class MBSyncPlugin(BeetsPlugin):
|
|||
query.
|
||||
"""
|
||||
for item in lib.items(query + ["singleton:true"]):
|
||||
if not item.mb_trackid:
|
||||
if not (track_id := item.mb_trackid):
|
||||
self._log.info(
|
||||
"Skipping singleton with no mb_trackid: {}", item
|
||||
)
|
||||
continue
|
||||
|
||||
if not (data_source := item.get("data_source")):
|
||||
self._log.info(
|
||||
"Skipping singleton without data source: {}", item
|
||||
)
|
||||
continue
|
||||
|
||||
if not (
|
||||
track_info := metadata_plugins.track_for_id(item.mb_trackid)
|
||||
track_info := metadata_plugins.track_for_id(
|
||||
track_id, data_source
|
||||
)
|
||||
):
|
||||
self._log.info(
|
||||
"Recording ID not found: {0.mb_trackid} for track {0}", item
|
||||
"Recording ID not found: {} for track {}", track_id, item
|
||||
)
|
||||
continue
|
||||
|
||||
|
|
@ -97,15 +105,24 @@ class MBSyncPlugin(BeetsPlugin):
|
|||
"""
|
||||
# Process matching albums.
|
||||
for album in lib.albums(query):
|
||||
if not album.mb_albumid:
|
||||
if not (album_id := album.mb_albumid):
|
||||
self._log.info("Skipping album with no mb_albumid: {}", album)
|
||||
continue
|
||||
|
||||
if not (
|
||||
album_info := metadata_plugins.album_for_id(album.mb_albumid)
|
||||
data_source := album.get("data_source")
|
||||
or album.items()[0].get("data_source")
|
||||
):
|
||||
self._log.info("Skipping album without data source: {}", album)
|
||||
continue
|
||||
|
||||
if not (
|
||||
album_info := metadata_plugins.album_for_id(
|
||||
album_id, data_source
|
||||
)
|
||||
):
|
||||
self._log.info(
|
||||
"Release ID {0.mb_albumid} not found for album {0}", album
|
||||
"Release ID {} not found for album {}", album_id, album
|
||||
)
|
||||
continue
|
||||
|
||||
|
|
|
|||
|
|
@ -219,10 +219,17 @@ class MissingPlugin(BeetsPlugin):
|
|||
if len(album.items()) == album.albumtotal:
|
||||
return
|
||||
|
||||
item_mbids = {x.mb_trackid for x in album.items()}
|
||||
# fetch missing items
|
||||
# TODO: Implement caching that without breaking other stuff
|
||||
if album_info := metadata_plugins.album_for_id(album.mb_albumid):
|
||||
if (
|
||||
data_source := album.get("data_source")
|
||||
or album.items()[0].get("data_source")
|
||||
) and (
|
||||
album_info := metadata_plugins.album_for_id(
|
||||
album.mb_albumid, data_source
|
||||
)
|
||||
):
|
||||
item_mbids = {x.mb_trackid for x in album.items()}
|
||||
for track_info in album_info.tracks:
|
||||
if track_info.track_id not in item_mbids:
|
||||
self._log.debug(
|
||||
|
|
|
|||
0
test/autotag/__init__.py
Normal file
0
test/autotag/__init__.py
Normal file
177
test/autotag/test_match.py
Normal file
177
test/autotag/test_match.py
Normal file
|
|
@ -0,0 +1,177 @@
|
|||
import pytest
|
||||
|
||||
from beets import metadata_plugins
|
||||
from beets.autotag import AlbumInfo, TrackInfo, match
|
||||
from beets.library import Item
|
||||
from beets.test.helper import ConfigMixin
|
||||
|
||||
|
||||
class TestAssignment(ConfigMixin):
|
||||
A = "one"
|
||||
B = "two"
|
||||
C = "three"
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _setup_config(self):
|
||||
self.config["match"]["track_length_grace"] = 10
|
||||
self.config["match"]["track_length_max"] = 30
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
# 'expected' is a tuple of expected (mapping, extra_items, extra_tracks)
|
||||
"item_titles, track_titles, expected",
|
||||
[
|
||||
# items ordering gets corrected
|
||||
([A, C, B], [A, B, C], ({A: A, B: B, C: C}, [], [])),
|
||||
# unmatched tracks are returned as 'extra_tracks'
|
||||
# the first track is unmatched
|
||||
([B, C], [A, B, C], ({B: B, C: C}, [], [A])),
|
||||
# the middle track is unmatched
|
||||
([A, C], [A, B, C], ({A: A, C: C}, [], [B])),
|
||||
# the last track is unmatched
|
||||
([A, B], [A, B, C], ({A: A, B: B}, [], [C])),
|
||||
# unmatched items are returned as 'extra_items'
|
||||
([A, C, B], [A, C], ({A: A, C: C}, [B], [])),
|
||||
],
|
||||
)
|
||||
def test_assign_tracks(self, item_titles, track_titles, expected):
|
||||
expected_mapping, expected_extra_items, expected_extra_tracks = expected
|
||||
|
||||
items = [Item(title=title) for title in item_titles]
|
||||
tracks = [TrackInfo(title=title) for title in track_titles]
|
||||
|
||||
mapping, extra_items, extra_tracks = match.assign_items(items, tracks)
|
||||
|
||||
assert (
|
||||
{i.title: t.title for i, t in mapping.items()},
|
||||
[i.title for i in extra_items],
|
||||
[t.title for t in extra_tracks],
|
||||
) == (expected_mapping, expected_extra_items, expected_extra_tracks)
|
||||
|
||||
def test_order_works_when_track_names_are_entirely_wrong(self):
|
||||
# A real-world test case contributed by a user.
|
||||
def item(i, length):
|
||||
return Item(
|
||||
artist="ben harper",
|
||||
album="burn to shine",
|
||||
title=f"ben harper - Burn to Shine {i}",
|
||||
track=i,
|
||||
length=length,
|
||||
)
|
||||
|
||||
items = []
|
||||
items.append(item(1, 241.37243007106997))
|
||||
items.append(item(2, 342.27781704375036))
|
||||
items.append(item(3, 245.95070222338137))
|
||||
items.append(item(4, 472.87662515485437))
|
||||
items.append(item(5, 279.1759535763187))
|
||||
items.append(item(6, 270.33333768012))
|
||||
items.append(item(7, 247.83435613222923))
|
||||
items.append(item(8, 216.54504531525072))
|
||||
items.append(item(9, 225.72775379800484))
|
||||
items.append(item(10, 317.7643606963552))
|
||||
items.append(item(11, 243.57001238834192))
|
||||
items.append(item(12, 186.45916150485752))
|
||||
|
||||
def info(index, title, length):
|
||||
return TrackInfo(title=title, length=length, index=index)
|
||||
|
||||
trackinfo = []
|
||||
trackinfo.append(info(1, "Alone", 238.893))
|
||||
trackinfo.append(info(2, "The Woman in You", 341.44))
|
||||
trackinfo.append(info(3, "Less", 245.59999999999999))
|
||||
trackinfo.append(info(4, "Two Hands of a Prayer", 470.49299999999999))
|
||||
trackinfo.append(info(5, "Please Bleed", 277.86599999999999))
|
||||
trackinfo.append(info(6, "Suzie Blue", 269.30599999999998))
|
||||
trackinfo.append(info(7, "Steal My Kisses", 245.36000000000001))
|
||||
trackinfo.append(info(8, "Burn to Shine", 214.90600000000001))
|
||||
trackinfo.append(info(9, "Show Me a Little Shame", 224.0929999999999))
|
||||
trackinfo.append(info(10, "Forgiven", 317.19999999999999))
|
||||
trackinfo.append(info(11, "Beloved One", 243.733))
|
||||
trackinfo.append(info(12, "In the Lord's Arms", 186.13300000000001))
|
||||
|
||||
expected = dict(zip(items, trackinfo)), [], []
|
||||
|
||||
assert match.assign_items(items, trackinfo) == expected
|
||||
|
||||
|
||||
class TestTagMultipleDataSources:
|
||||
@pytest.fixture
|
||||
def shared_track_id(self):
|
||||
return "track-12345"
|
||||
|
||||
@pytest.fixture
|
||||
def shared_album_id(self):
|
||||
return "album-12345"
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _setup_plugins(self, monkeypatch, shared_album_id, shared_track_id):
|
||||
class StubPlugin(metadata_plugins.MetadataSourcePlugin):
|
||||
@property
|
||||
def track(self):
|
||||
return TrackInfo(
|
||||
artist="Artist",
|
||||
title="Title",
|
||||
track_id=shared_track_id,
|
||||
data_source=self.data_source,
|
||||
)
|
||||
|
||||
@property
|
||||
def album(self):
|
||||
return AlbumInfo(
|
||||
[self.track],
|
||||
artist="Albumartist",
|
||||
album="Album",
|
||||
album_id=shared_album_id,
|
||||
data_source=self.data_source,
|
||||
)
|
||||
|
||||
def album_for_id(self, album_id):
|
||||
return self.album if album_id == shared_album_id else None
|
||||
|
||||
def track_for_id(self, track_id):
|
||||
return self.track if track_id == shared_track_id else None
|
||||
|
||||
def candidates(self, *_, **__):
|
||||
yield self.album
|
||||
|
||||
def item_candidates(self, *_, **__):
|
||||
yield self.track
|
||||
|
||||
class DeezerPlugin(StubPlugin):
|
||||
data_source = "Deezer"
|
||||
|
||||
class DiscogsPlugin(StubPlugin):
|
||||
data_source = "Discogs"
|
||||
|
||||
monkeypatch.setattr(
|
||||
metadata_plugins,
|
||||
"find_metadata_source_plugins",
|
||||
lambda: [DeezerPlugin(), DiscogsPlugin()],
|
||||
)
|
||||
|
||||
def check_proposal(self, proposal):
|
||||
sources = [
|
||||
candidate.info.data_source for candidate in proposal.candidates
|
||||
]
|
||||
assert len(sources) == 2
|
||||
assert set(sources) == {"Discogs", "Deezer"}
|
||||
|
||||
def test_search_album_ids(self, shared_album_id):
|
||||
_, _, proposal = match.tag_album([Item()], search_ids=[shared_album_id])
|
||||
|
||||
self.check_proposal(proposal)
|
||||
|
||||
def test_search_album_current_id(self, shared_album_id):
|
||||
_, _, proposal = match.tag_album([Item(mb_albumid=shared_album_id)])
|
||||
|
||||
self.check_proposal(proposal)
|
||||
|
||||
def test_search_track_ids(self, shared_track_id):
|
||||
proposal = match.tag_item(Item(), search_ids=[shared_track_id])
|
||||
|
||||
self.check_proposal(proposal)
|
||||
|
||||
def test_search_track_current_id(self, shared_track_id):
|
||||
proposal = match.tag_item(Item(mb_trackid=shared_track_id))
|
||||
|
||||
self.check_proposal(proposal)
|
||||
|
|
@ -45,10 +45,15 @@ class MbsyncCliTest(PluginTestCase):
|
|||
album="old album",
|
||||
mb_albumid="album id",
|
||||
mb_trackid="track id",
|
||||
data_source="data_source",
|
||||
)
|
||||
self.lib.add_album([album_item])
|
||||
|
||||
singleton = Item(title="old title", mb_trackid="singleton id")
|
||||
singleton = Item(
|
||||
title="old title",
|
||||
mb_trackid="singleton id",
|
||||
data_source="data_source",
|
||||
)
|
||||
self.lib.add(singleton)
|
||||
|
||||
self.run_command("mbsync")
|
||||
|
|
|
|||
|
|
@ -17,97 +17,9 @@
|
|||
import pytest
|
||||
|
||||
from beets import autotag, config
|
||||
from beets.autotag import AlbumInfo, TrackInfo, correct_list_fields, match
|
||||
from beets.autotag import AlbumInfo, TrackInfo, correct_list_fields
|
||||
from beets.library import Item
|
||||
from beets.test.helper import BeetsTestCase, ConfigMixin
|
||||
|
||||
|
||||
class TestAssignment(ConfigMixin):
|
||||
A = "one"
|
||||
B = "two"
|
||||
C = "three"
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _setup_config(self):
|
||||
self.config["match"]["track_length_grace"] = 10
|
||||
self.config["match"]["track_length_max"] = 30
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
# 'expected' is a tuple of expected (mapping, extra_items, extra_tracks)
|
||||
"item_titles, track_titles, expected",
|
||||
[
|
||||
# items ordering gets corrected
|
||||
([A, C, B], [A, B, C], ({A: A, B: B, C: C}, [], [])),
|
||||
# unmatched tracks are returned as 'extra_tracks'
|
||||
# the first track is unmatched
|
||||
([B, C], [A, B, C], ({B: B, C: C}, [], [A])),
|
||||
# the middle track is unmatched
|
||||
([A, C], [A, B, C], ({A: A, C: C}, [], [B])),
|
||||
# the last track is unmatched
|
||||
([A, B], [A, B, C], ({A: A, B: B}, [], [C])),
|
||||
# unmatched items are returned as 'extra_items'
|
||||
([A, C, B], [A, C], ({A: A, C: C}, [B], [])),
|
||||
],
|
||||
)
|
||||
def test_assign_tracks(self, item_titles, track_titles, expected):
|
||||
expected_mapping, expected_extra_items, expected_extra_tracks = expected
|
||||
|
||||
items = [Item(title=title) for title in item_titles]
|
||||
tracks = [TrackInfo(title=title) for title in track_titles]
|
||||
|
||||
mapping, extra_items, extra_tracks = match.assign_items(items, tracks)
|
||||
|
||||
assert (
|
||||
{i.title: t.title for i, t in mapping.items()},
|
||||
[i.title for i in extra_items],
|
||||
[t.title for t in extra_tracks],
|
||||
) == (expected_mapping, expected_extra_items, expected_extra_tracks)
|
||||
|
||||
def test_order_works_when_track_names_are_entirely_wrong(self):
|
||||
# A real-world test case contributed by a user.
|
||||
def item(i, length):
|
||||
return Item(
|
||||
artist="ben harper",
|
||||
album="burn to shine",
|
||||
title=f"ben harper - Burn to Shine {i}",
|
||||
track=i,
|
||||
length=length,
|
||||
)
|
||||
|
||||
items = []
|
||||
items.append(item(1, 241.37243007106997))
|
||||
items.append(item(2, 342.27781704375036))
|
||||
items.append(item(3, 245.95070222338137))
|
||||
items.append(item(4, 472.87662515485437))
|
||||
items.append(item(5, 279.1759535763187))
|
||||
items.append(item(6, 270.33333768012))
|
||||
items.append(item(7, 247.83435613222923))
|
||||
items.append(item(8, 216.54504531525072))
|
||||
items.append(item(9, 225.72775379800484))
|
||||
items.append(item(10, 317.7643606963552))
|
||||
items.append(item(11, 243.57001238834192))
|
||||
items.append(item(12, 186.45916150485752))
|
||||
|
||||
def info(index, title, length):
|
||||
return TrackInfo(title=title, length=length, index=index)
|
||||
|
||||
trackinfo = []
|
||||
trackinfo.append(info(1, "Alone", 238.893))
|
||||
trackinfo.append(info(2, "The Woman in You", 341.44))
|
||||
trackinfo.append(info(3, "Less", 245.59999999999999))
|
||||
trackinfo.append(info(4, "Two Hands of a Prayer", 470.49299999999999))
|
||||
trackinfo.append(info(5, "Please Bleed", 277.86599999999999))
|
||||
trackinfo.append(info(6, "Suzie Blue", 269.30599999999998))
|
||||
trackinfo.append(info(7, "Steal My Kisses", 245.36000000000001))
|
||||
trackinfo.append(info(8, "Burn to Shine", 214.90600000000001))
|
||||
trackinfo.append(info(9, "Show Me a Little Shame", 224.0929999999999))
|
||||
trackinfo.append(info(10, "Forgiven", 317.19999999999999))
|
||||
trackinfo.append(info(11, "Beloved One", 243.733))
|
||||
trackinfo.append(info(12, "In the Lord's Arms", 186.13300000000001))
|
||||
|
||||
expected = dict(zip(items, trackinfo)), [], []
|
||||
|
||||
assert match.assign_items(items, trackinfo) == expected
|
||||
from beets.test.helper import BeetsTestCase
|
||||
|
||||
|
||||
class ApplyTestUtil:
|
||||
|
|
|
|||
|
|
@ -1517,7 +1517,7 @@ class ImportPretendTest(IOMixin, AutotagImportTestCase):
|
|||
assert self.__run(importer) == [f"No files imported from {empty_path}"]
|
||||
|
||||
|
||||
def mocked_get_album_by_id(id_):
|
||||
def mocked_get_albums_by_ids(ids):
|
||||
"""Return album candidate for the given id.
|
||||
|
||||
The two albums differ only in the release title and artist name, so that
|
||||
|
|
@ -1525,32 +1525,34 @@ def mocked_get_album_by_id(id_):
|
|||
ImportHelper.prepare_album_for_import().
|
||||
"""
|
||||
# Map IDs to (release title, artist), so the distances are different.
|
||||
album, artist = {
|
||||
album_artist_map = {
|
||||
ImportIdTest.ID_RELEASE_0: ("VALID_RELEASE_0", "TAG ARTIST"),
|
||||
ImportIdTest.ID_RELEASE_1: ("VALID_RELEASE_1", "DISTANT_MATCH"),
|
||||
}[id_]
|
||||
}
|
||||
|
||||
return AlbumInfo(
|
||||
album_id=id_,
|
||||
album=album,
|
||||
artist_id="some-id",
|
||||
artist=artist,
|
||||
albumstatus="Official",
|
||||
tracks=[
|
||||
TrackInfo(
|
||||
track_id="bar",
|
||||
title="foo",
|
||||
artist_id="some-id",
|
||||
artist=artist,
|
||||
length=59,
|
||||
index=9,
|
||||
track_allt="A2",
|
||||
)
|
||||
],
|
||||
)
|
||||
for id_ in ids:
|
||||
album, artist = album_artist_map[id_]
|
||||
yield AlbumInfo(
|
||||
album_id=id_,
|
||||
album=album,
|
||||
artist_id="some-id",
|
||||
artist=artist,
|
||||
albumstatus="Official",
|
||||
tracks=[
|
||||
TrackInfo(
|
||||
track_id="bar",
|
||||
title="foo",
|
||||
artist_id="some-id",
|
||||
artist=artist,
|
||||
length=59,
|
||||
index=9,
|
||||
track_allt="A2",
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def mocked_get_track_by_id(id_):
|
||||
def mocked_get_tracks_by_ids(ids):
|
||||
"""Return track candidate for the given id.
|
||||
|
||||
The two tracks differ only in the release title and artist name, so that
|
||||
|
|
@ -1558,27 +1560,29 @@ def mocked_get_track_by_id(id_):
|
|||
ImportHelper.prepare_album_for_import().
|
||||
"""
|
||||
# Map IDs to (recording title, artist), so the distances are different.
|
||||
title, artist = {
|
||||
title_artist_map = {
|
||||
ImportIdTest.ID_RECORDING_0: ("VALID_RECORDING_0", "TAG ARTIST"),
|
||||
ImportIdTest.ID_RECORDING_1: ("VALID_RECORDING_1", "DISTANT_MATCH"),
|
||||
}[id_]
|
||||
}
|
||||
|
||||
return TrackInfo(
|
||||
track_id=id_,
|
||||
title=title,
|
||||
artist_id="some-id",
|
||||
artist=artist,
|
||||
length=59,
|
||||
)
|
||||
for id_ in ids:
|
||||
title, artist = title_artist_map[id_]
|
||||
yield TrackInfo(
|
||||
track_id=id_,
|
||||
title=title,
|
||||
artist_id="some-id",
|
||||
artist=artist,
|
||||
length=59,
|
||||
)
|
||||
|
||||
|
||||
@patch(
|
||||
"beets.metadata_plugins.track_for_id",
|
||||
Mock(side_effect=mocked_get_track_by_id),
|
||||
"beets.metadata_plugins.tracks_for_ids",
|
||||
Mock(side_effect=mocked_get_tracks_by_ids),
|
||||
)
|
||||
@patch(
|
||||
"beets.metadata_plugins.album_for_id",
|
||||
Mock(side_effect=mocked_get_album_by_id),
|
||||
"beets.metadata_plugins.albums_for_ids",
|
||||
Mock(side_effect=mocked_get_albums_by_ids),
|
||||
)
|
||||
class ImportIdTest(ImportTestCase):
|
||||
ID_RELEASE_0 = "00000000-0000-0000-0000-000000000000"
|
||||
|
|
|
|||
Loading…
Reference in a new issue