feat: auto-discover similar tracks during autoplay via Last.fm/MusicBrainz

Every Nth autoplay pick (configurable via discover_ratio), query Last.fm
for similar tracks. When Last.fm has no key or returns nothing, fall back
to MusicBrainz tag-based recording search (no API key needed). Discovered
tracks are resolved via yt-dlp and deduplicated within the session. If
discovery fails, the kept-deck shuffle continues as before.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
user
2026-02-23 21:19:41 +01:00
parent 56f6b9822f
commit da9ed51c74
6 changed files with 933 additions and 17 deletions

View File

@@ -5,7 +5,7 @@ import importlib.util
import json
import sys
from dataclasses import dataclass
from unittest.mock import MagicMock, patch
from unittest.mock import AsyncMock, MagicMock, patch
# -- Load plugin module directly ---------------------------------------------
@@ -521,3 +521,142 @@ class TestCmdTags:
with patch.object(_mod, "_get_top_tags", return_value=tags):
asyncio.run(_mod.cmd_tags(bot, msg))
assert any("Lateralus:" in r for r in bot.replied)
# ---------------------------------------------------------------------------
# TestParseTitle
# ---------------------------------------------------------------------------
class TestParseTitle:
def test_dash_separator(self):
assert _mod._parse_title("Tool - Lateralus") == ("Tool", "Lateralus")
def test_double_dash(self):
assert _mod._parse_title("Tool -- Lateralus") == ("Tool", "Lateralus")
def test_pipe_separator(self):
assert _mod._parse_title("Tool | Lateralus") == ("Tool", "Lateralus")
def test_tilde_separator(self):
assert _mod._parse_title("Tool ~ Lateralus") == ("Tool", "Lateralus")
def test_no_separator(self):
assert _mod._parse_title("Lateralus") == ("", "Lateralus")
def test_empty_string(self):
assert _mod._parse_title("") == ("", "")
def test_strips_whitespace(self):
assert _mod._parse_title(" Tool - Lateralus ") == ("Tool", "Lateralus")
def test_first_separator_wins(self):
"""Only the first matching separator is used."""
assert _mod._parse_title("A - B - C") == ("A", "B - C")
def test_dash_priority_over_pipe(self):
"""Dash separator is tried before pipe."""
assert _mod._parse_title("A - B | C") == ("A", "B | C")
# ---------------------------------------------------------------------------
# TestDiscoverSimilar
# ---------------------------------------------------------------------------
class TestDiscoverSimilar:
def test_lastfm_path(self):
"""Returns Last.fm result when API key + results available."""
bot = _FakeBot(api_key="test-key")
tracks = [{"name": "Found", "artist": {"name": "Band"}, "match": "0.9"}]
with patch.object(_mod, "_get_similar_tracks", return_value=tracks):
result = asyncio.run(
_mod.discover_similar(bot, "Tool - Lateralus"),
)
assert result == ("Band", "Found")
def test_lastfm_empty_falls_to_musicbrainz(self):
"""Falls back to MusicBrainz when Last.fm returns nothing."""
bot = _FakeBot(api_key="test-key")
mb_picks = [{"artist": "MB Artist", "title": "MB Song"}]
with patch.object(_mod, "_get_similar_tracks", return_value=[]), \
patch("plugins._musicbrainz.mb_search_artist", return_value="mbid"), \
patch("plugins._musicbrainz.mb_artist_tags", return_value=["rock"]), \
patch("plugins._musicbrainz.mb_find_similar_recordings",
return_value=mb_picks):
result = asyncio.run(
_mod.discover_similar(bot, "Tool - Lateralus"),
)
assert result == ("MB Artist", "MB Song")
def test_no_api_key_uses_musicbrainz(self):
"""Skips Last.fm when no API key, goes straight to MusicBrainz."""
bot = _FakeBot(api_key="")
mb_picks = [{"artist": "MB Band", "title": "MB Track"}]
with patch.dict("os.environ", {}, clear=True), \
patch("plugins._musicbrainz.mb_search_artist", return_value="mbid"), \
patch("plugins._musicbrainz.mb_artist_tags", return_value=["rock"]), \
patch("plugins._musicbrainz.mb_find_similar_recordings",
return_value=mb_picks):
result = asyncio.run(
_mod.discover_similar(bot, "Tool - Lateralus"),
)
assert result == ("MB Band", "MB Track")
def test_both_fail_returns_none(self):
"""Returns None when both Last.fm and MusicBrainz fail."""
bot = _FakeBot(api_key="test-key")
with patch.object(_mod, "_get_similar_tracks", return_value=[]), \
patch("plugins._musicbrainz.mb_search_artist", return_value=None):
result = asyncio.run(
_mod.discover_similar(bot, "Tool - Lateralus"),
)
assert result is None
def test_no_artist_returns_none(self):
"""Returns None when title has no artist component."""
bot = _FakeBot(api_key="test-key")
result = asyncio.run(
_mod.discover_similar(bot, "Lateralus"),
)
assert result is None
def test_musicbrainz_import_error_handled(self):
"""Gracefully handles import error for _musicbrainz module."""
bot = _FakeBot(api_key="test-key")
with patch.object(_mod, "_get_similar_tracks", return_value=[]), \
patch.dict("sys.modules", {"plugins._musicbrainz": None}):
result = asyncio.run(
_mod.discover_similar(bot, "Tool - Lateralus"),
)
assert result is None
def test_lastfm_exception_falls_to_musicbrainz(self):
"""Last.fm exception triggers MusicBrainz fallback."""
bot = _FakeBot(api_key="test-key")
mb_picks = [{"artist": "Fallback", "title": "Song"}]
with patch.object(_mod, "_get_similar_tracks",
side_effect=Exception("API down")), \
patch("plugins._musicbrainz.mb_search_artist", return_value="mbid"), \
patch("plugins._musicbrainz.mb_artist_tags", return_value=["rock"]), \
patch("plugins._musicbrainz.mb_find_similar_recordings",
return_value=mb_picks):
result = asyncio.run(
_mod.discover_similar(bot, "Tool - Lateralus"),
)
assert result == ("Fallback", "Song")
def test_lastfm_pick_missing_name_falls_to_musicbrainz(self):
"""Falls to MB when Last.fm result has empty artist/title."""
bot = _FakeBot(api_key="test-key")
tracks = [{"name": "", "artist": {"name": ""}, "match": "0.9"}]
mb_picks = [{"artist": "MB", "title": "Track"}]
with patch.object(_mod, "_get_similar_tracks", return_value=tracks), \
patch("plugins._musicbrainz.mb_search_artist", return_value="mbid"), \
patch("plugins._musicbrainz.mb_artist_tags", return_value=["rock"]), \
patch("plugins._musicbrainz.mb_find_similar_recordings",
return_value=mb_picks):
result = asyncio.run(
_mod.discover_similar(bot, "Tool - Lateralus"),
)
assert result == ("MB", "Track")

View File

@@ -2362,3 +2362,245 @@ class TestKeptRepair:
stored = json.loads(raw)
assert stored["filename"] == "song.webm"
assert (music_dir / "song.webm").is_file()
# ---------------------------------------------------------------------------
# TestAutoplayDiscovery
# ---------------------------------------------------------------------------
class TestAutoplayDiscovery:
"""Tests for the discovery integration in _play_loop autoplay."""
def test_config_defaults(self):
"""Default discover/discover_ratio values are set."""
bot = _FakeBot()
ps = _mod._ps(bot)
assert ps["discover"] is True
assert ps["discover_ratio"] == 3
def test_config_from_toml(self):
"""Config values are read from bot config."""
bot = _FakeBot()
bot.config = {"music": {"discover": False, "discover_ratio": 5}}
# Reset pstate so _ps re-reads config
bot._pstate.clear()
ps = _mod._ps(bot)
assert ps["discover"] is False
assert ps["discover_ratio"] == 5
def test_discovery_triggers_on_ratio(self, tmp_path):
"""Discovery is attempted when autoplay_count is a multiple of ratio."""
bot = _FakeBot()
ps = _mod._ps(bot)
ps["autoplay"] = True
ps["discover"] = True
ps["discover_ratio"] = 1 # trigger every pick
ps["autoplay_cooldown"] = 0
ps["duck_silence"] = 0
# Seed history so discovery has something to reference
ps["history"] = [
_mod._Track(url="x", title="Tool - Lateralus", requester="a"),
]
# Set up kept tracks for fallback pool
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "a.opus").write_bytes(b"audio")
bot.state.set("music", "keep:1", json.dumps({
"url": "https://example.com/a", "title": "Kept Track",
"filename": "a.opus", "id": 1,
}))
discover_called = []
async def fake_discover(b, title):
discover_called.append(title)
return ("Deftones", "Change")
lastfm_mod = MagicMock()
lastfm_mod.discover_similar = fake_discover
bot.registry._modules = {"lastfm": lastfm_mod}
resolved = [("https://youtube.com/watch?v=x", "Deftones - Change")]
async def _run():
with patch.object(_mod, "_MUSIC_DIR", music_dir), \
patch.object(_mod, "_resolve_tracks", return_value=resolved), \
patch.object(_mod, "_download_track", return_value=None):
task = asyncio.create_task(
_mod._play_loop(bot, seek=0.0, fade_in=False),
)
# Let it pick a track, then cancel
await asyncio.sleep(0.5)
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_run())
assert len(discover_called) >= 1
assert discover_called[0] == "Tool - Lateralus"
def test_discovery_disabled(self, tmp_path):
"""Discovery is skipped when discover=False."""
bot = _FakeBot()
ps = _mod._ps(bot)
ps["autoplay"] = True
ps["discover"] = False
ps["discover_ratio"] = 1
ps["autoplay_cooldown"] = 0
ps["duck_silence"] = 0
ps["history"] = [
_mod._Track(url="x", title="Tool - Lateralus", requester="a"),
]
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "a.opus").write_bytes(b"audio")
bot.state.set("music", "keep:1", json.dumps({
"url": "https://example.com/a", "title": "Kept Track",
"filename": "a.opus", "id": 1,
}))
discover_called = []
async def fake_discover(b, title):
discover_called.append(title)
return ("X", "Y")
lastfm_mod = MagicMock()
lastfm_mod.discover_similar = fake_discover
bot.registry._modules = {"lastfm": lastfm_mod}
async def _run():
with patch.object(_mod, "_MUSIC_DIR", music_dir), \
patch.object(_mod, "_download_track", return_value=None):
task = asyncio.create_task(
_mod._play_loop(bot, seek=0.0, fade_in=False),
)
await asyncio.sleep(0.5)
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_run())
assert discover_called == []
def test_discovery_dedup(self):
"""Same discovered track is not resolved twice (dedup by seen set)."""
# Unit-test the dedup logic directly: simulate the set-based
# deduplication that _play_loop uses with _discover_seen.
_discover_seen: set[str] = set()
def _would_resolve(artist: str, title: str) -> bool:
key = f"{artist.lower()}:{title.lower()}"
if key in _discover_seen:
return False
_discover_seen.add(key)
return True
assert _would_resolve("Deftones", "Change") is True
assert _would_resolve("Deftones", "Change") is False
assert _would_resolve("deftones", "change") is False
assert _would_resolve("Tool", "Sober") is True
def test_discovery_fallback_to_kept(self, tmp_path):
"""Falls back to kept deck when discovery returns None."""
bot = _FakeBot()
ps = _mod._ps(bot)
ps["autoplay"] = True
ps["discover"] = True
ps["discover_ratio"] = 1
ps["autoplay_cooldown"] = 0
ps["duck_silence"] = 0
ps["history"] = [
_mod._Track(url="x", title="Tool - Lateralus", requester="a"),
]
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "a.opus").write_bytes(b"audio")
bot.state.set("music", "keep:1", json.dumps({
"url": "https://example.com/a", "title": "Kept Track",
"filename": "a.opus", "id": 1,
}))
async def fake_discover(b, title):
return None
lastfm_mod = MagicMock()
lastfm_mod.discover_similar = fake_discover
bot.registry._modules = {"lastfm": lastfm_mod}
queued_titles = []
async def _run():
with patch.object(_mod, "_MUSIC_DIR", music_dir), \
patch.object(_mod, "_download_track", return_value=None):
task = asyncio.create_task(
_mod._play_loop(bot, seek=0.0, fade_in=False),
)
await asyncio.sleep(0.5)
# Check what was queued -- should be kept track, not discovered
if ps.get("current"):
queued_titles.append(ps["current"].title)
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_run())
# The kept track should have been used as fallback
if queued_titles:
assert queued_titles[0] == "Kept Track"
def test_no_history_skips_discovery(self, tmp_path):
"""Discovery is skipped when history is empty."""
bot = _FakeBot()
ps = _mod._ps(bot)
ps["autoplay"] = True
ps["discover"] = True
ps["discover_ratio"] = 1
ps["autoplay_cooldown"] = 0
ps["duck_silence"] = 0
ps["history"] = []
music_dir = tmp_path / "music"
music_dir.mkdir()
(music_dir / "a.opus").write_bytes(b"audio")
bot.state.set("music", "keep:1", json.dumps({
"url": "https://example.com/a", "title": "Kept Track",
"filename": "a.opus", "id": 1,
}))
discover_called = []
async def fake_discover(b, title):
discover_called.append(title)
return ("X", "Y")
lastfm_mod = MagicMock()
lastfm_mod.discover_similar = fake_discover
bot.registry._modules = {"lastfm": lastfm_mod}
async def _run():
with patch.object(_mod, "_MUSIC_DIR", music_dir), \
patch.object(_mod, "_download_track", return_value=None):
task = asyncio.create_task(
_mod._play_loop(bot, seek=0.0, fade_in=False),
)
await asyncio.sleep(0.5)
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
asyncio.run(_run())
assert discover_called == []

310
tests/test_musicbrainz.py Normal file
View File

@@ -0,0 +1,310 @@
"""Tests for the MusicBrainz API helper module."""
import importlib.util
import json
import sys
import time
from io import BytesIO
from unittest.mock import MagicMock, patch
# -- Load module directly ----------------------------------------------------
_spec = importlib.util.spec_from_file_location(
"_musicbrainz", "plugins/_musicbrainz.py",
)
_mod = importlib.util.module_from_spec(_spec)
sys.modules["_musicbrainz"] = _mod
_spec.loader.exec_module(_mod)
# -- Helpers -----------------------------------------------------------------
def _make_resp(data: dict) -> MagicMock:
"""Create a fake HTTP response with JSON body."""
resp = MagicMock()
resp.read.return_value = json.dumps(data).encode()
return resp
# ---------------------------------------------------------------------------
# TestMbRequest
# ---------------------------------------------------------------------------
class TestMbRequest:
def setup_method(self):
_mod._last_request = 0.0
def test_returns_parsed_json(self):
resp = _make_resp({"status": "ok"})
with patch("derp.http.urlopen", return_value=resp):
result = _mod._mb_request("artist", {"query": "Tool"})
assert result == {"status": "ok"}
def test_rate_delay_enforced(self):
"""Second call within rate interval triggers sleep."""
_mod._last_request = time.monotonic()
resp = _make_resp({})
slept = []
with patch("derp.http.urlopen", return_value=resp), \
patch.object(_mod.time, "sleep", side_effect=slept.append), \
patch.object(_mod.time, "monotonic", return_value=_mod._last_request + 0.2):
_mod._mb_request("artist", {"query": "X"})
assert len(slept) == 1
assert slept[0] > 0
def test_no_delay_when_interval_elapsed(self):
"""No sleep when enough time has passed since last request."""
_mod._last_request = time.monotonic() - 5.0
resp = _make_resp({})
with patch("derp.http.urlopen", return_value=resp), \
patch.object(_mod.time, "sleep") as mock_sleep:
_mod._mb_request("artist", {"query": "X"})
mock_sleep.assert_not_called()
def test_returns_empty_on_error(self):
with patch("derp.http.urlopen", side_effect=ConnectionError("fail")):
result = _mod._mb_request("artist", {"query": "X"})
assert result == {}
def test_updates_last_request_on_success(self):
_mod._last_request = 0.0
resp = _make_resp({})
with patch("derp.http.urlopen", return_value=resp):
_mod._mb_request("test")
assert _mod._last_request > 0
def test_updates_last_request_on_error(self):
_mod._last_request = 0.0
with patch("derp.http.urlopen", side_effect=Exception("boom")):
_mod._mb_request("test")
assert _mod._last_request > 0
def test_none_params(self):
"""Handles None params without error."""
resp = _make_resp({"ok": True})
with patch("derp.http.urlopen", return_value=resp):
result = _mod._mb_request("test", None)
assert result == {"ok": True}
# ---------------------------------------------------------------------------
# TestMbSearchArtist
# ---------------------------------------------------------------------------
class TestMbSearchArtist:
def setup_method(self):
_mod._last_request = 0.0
def test_returns_mbid(self):
data = {"artists": [{"id": "abc-123", "name": "Tool", "score": 100}]}
with patch.object(_mod, "_mb_request", return_value=data):
result = _mod.mb_search_artist("Tool")
assert result == "abc-123"
def test_returns_none_no_results(self):
with patch.object(_mod, "_mb_request", return_value={"artists": []}):
assert _mod.mb_search_artist("Unknown") is None
def test_returns_none_on_empty_response(self):
with patch.object(_mod, "_mb_request", return_value={}):
assert _mod.mb_search_artist("X") is None
def test_returns_none_low_score(self):
"""Rejects matches with score below 50."""
data = {"artists": [{"id": "low", "name": "Mismatch", "score": 30}]}
with patch.object(_mod, "_mb_request", return_value=data):
assert _mod.mb_search_artist("Tool") is None
def test_returns_none_on_error(self):
with patch.object(_mod, "_mb_request", return_value={}):
assert _mod.mb_search_artist("Error") is None
def test_accepts_high_score(self):
data = {"artists": [{"id": "abc", "name": "Tool", "score": 85}]}
with patch.object(_mod, "_mb_request", return_value=data):
assert _mod.mb_search_artist("Tool") == "abc"
# ---------------------------------------------------------------------------
# TestMbArtistTags
# ---------------------------------------------------------------------------
class TestMbArtistTags:
def setup_method(self):
_mod._last_request = 0.0
def test_returns_sorted_top_5(self):
data = {"tags": [
{"name": "rock", "count": 50},
{"name": "metal", "count": 100},
{"name": "prog", "count": 80},
{"name": "alternative", "count": 60},
{"name": "hard rock", "count": 40},
{"name": "grunge", "count": 30},
]}
with patch.object(_mod, "_mb_request", return_value=data):
result = _mod.mb_artist_tags("mbid-123")
assert len(result) == 5
assert result[0] == "metal"
assert result[1] == "prog"
assert result[2] == "alternative"
assert result[3] == "rock"
assert result[4] == "hard rock"
def test_empty_tags(self):
with patch.object(_mod, "_mb_request", return_value={"tags": []}):
assert _mod.mb_artist_tags("mbid") == []
def test_no_tags_key(self):
with patch.object(_mod, "_mb_request", return_value={}):
assert _mod.mb_artist_tags("mbid") == []
def test_skips_nameless_tags(self):
data = {"tags": [
{"name": "rock", "count": 50},
{"count": 100}, # no name
{"name": "", "count": 80}, # empty name
]}
with patch.object(_mod, "_mb_request", return_value=data):
result = _mod.mb_artist_tags("mbid")
assert result == ["rock"]
def test_fewer_than_5_tags(self):
data = {"tags": [{"name": "jazz", "count": 10}]}
with patch.object(_mod, "_mb_request", return_value=data):
result = _mod.mb_artist_tags("mbid")
assert result == ["jazz"]
# ---------------------------------------------------------------------------
# TestMbFindSimilarRecordings
# ---------------------------------------------------------------------------
class TestMbFindSimilarRecordings:
def setup_method(self):
_mod._last_request = 0.0
def test_returns_dicts(self):
data = {"recordings": [
{
"title": "Song A",
"artist-credit": [{"name": "Other Artist"}],
},
{
"title": "Song B",
"artist-credit": [{"name": "Another Band"}],
},
]}
with patch.object(_mod, "_mb_request", return_value=data):
result = _mod.mb_find_similar_recordings(
"Tool", ["rock", "metal"],
)
assert len(result) == 2
assert result[0] == {"artist": "Other Artist", "title": "Song A"}
assert result[1] == {"artist": "Another Band", "title": "Song B"}
def test_excludes_original_artist(self):
data = {"recordings": [
{
"title": "Own Song",
"artist-credit": [{"name": "Tool"}],
},
{
"title": "Other Song",
"artist-credit": [{"name": "Deftones"}],
},
]}
with patch.object(_mod, "_mb_request", return_value=data):
result = _mod.mb_find_similar_recordings(
"Tool", ["rock"],
)
assert len(result) == 1
assert result[0]["artist"] == "Deftones"
def test_excludes_original_artist_case_insensitive(self):
data = {"recordings": [
{
"title": "Song",
"artist-credit": [{"name": "TOOL"}],
},
]}
with patch.object(_mod, "_mb_request", return_value=data):
result = _mod.mb_find_similar_recordings(
"Tool", ["rock"],
)
assert result == []
def test_deduplicates(self):
data = {"recordings": [
{
"title": "Song A",
"artist-credit": [{"name": "Band X"}],
},
{
"title": "Song A",
"artist-credit": [{"name": "Band X"}],
},
]}
with patch.object(_mod, "_mb_request", return_value=data):
result = _mod.mb_find_similar_recordings(
"Other", ["rock"],
)
assert len(result) == 1
def test_empty_tags(self):
result = _mod.mb_find_similar_recordings("Tool", [])
assert result == []
def test_no_recordings(self):
with patch.object(_mod, "_mb_request", return_value={"recordings": []}):
result = _mod.mb_find_similar_recordings(
"Tool", ["rock"],
)
assert result == []
def test_empty_response(self):
with patch.object(_mod, "_mb_request", return_value={}):
result = _mod.mb_find_similar_recordings(
"Tool", ["rock"],
)
assert result == []
def test_skips_missing_title(self):
data = {"recordings": [
{
"title": "",
"artist-credit": [{"name": "Band"}],
},
]}
with patch.object(_mod, "_mb_request", return_value=data):
result = _mod.mb_find_similar_recordings(
"Other", ["rock"],
)
assert result == []
def test_skips_missing_artist_credit(self):
data = {"recordings": [
{"title": "Song", "artist-credit": []},
{"title": "Song2"},
]}
with patch.object(_mod, "_mb_request", return_value=data):
result = _mod.mb_find_similar_recordings(
"Other", ["rock"],
)
assert result == []
def test_uses_top_two_tags(self):
"""Query should use at most 2 tags."""
with patch.object(_mod, "_mb_request", return_value={}) as mock_req:
_mod.mb_find_similar_recordings(
"Tool", ["rock", "metal", "prog"],
)
call_args = mock_req.call_args
query = call_args[1]["query"] if "query" in (call_args[1] or {}) else call_args[0][1].get("query", "")
# Verify the query contains both tag references
assert "rock" in query or "metal" in query