danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Library: wire single->album resolution into import detection (gated, fail-safe)

Browse source 
detect_album_info_web gains a last-resort step: when a track matched a SINGLE
with no usable album context, look up the parent ALBUM that contains it (via
get_artist_albums_for_source + get_artist_album_tracks) and promote to it, so it
groups with its album-mates and gets the album's cover instead of the single's.

GATED behind metadata_enhancement.single_to_album (default OFF) — it's a
per-import metadata lookup, so it's opt-in, matching the canonical-version
pattern. Fully fail-safe: flag off, no source, or any client error/miss -> None,
so the track stays exactly as matched (never worse than today). The promoted
album name is forced past get_import_clean_album (which otherwise pins the
single's name) so grouping + tags use the album. 4 glue seam tests added
(promote-when-enabled, disabled-by-default, no-match, client-raises); 462
import-suite tests pass.
This commit is contained in:
BoulderBadgeDad 2026-06-18 09:30:11 -07:00
parent 00b26fc5f1
commit 58363ae510
2 changed files with 159 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

88
core/imports/context.py
View file

@ -8,6 +8,10 @@ from __future__ import annotations
from typing import Any, Dict, Optional
from utils.logging_config import get_logger
logger = get_logger("imports.context")
def _as_dict(value: Any) -> Dict[str, Any]:
return value if isinstance(value, dict) else {}
@ -440,4 +444,86 @@ def detect_album_info_web(context, artist_context=None):
force_album=True,
)
return None
# Last resort: the track matched a SINGLE with no usable album context —
# look up the parent ALBUM that actually contains it (gated, fail-safe).
return _resolve_single_to_parent_album(context, artist_context)
def _resolve_single_to_parent_album(context, artist_context):
"""A single-matched track -> a promoted album_info for its parent album, or
None. GATED by ``metadata_enhancement.single_to_album`` (default OFF — it's a
per-import metadata lookup, so it's opt-in). Fail-safe: any miss/error returns
None so the track stays exactly as it was matched (never worse than today)."""
try:
from core.metadata.common import get_config_manager
if not get_config_manager().get("metadata_enhancement.single_to_album", False):
return None
except Exception:
return None
try:
source = get_import_source(context)
track_info = get_import_track_info(context)
original_search = get_import_original_search(context)
track_title = (track_info.get("name") or original_search.get("title") or "").strip()
artist_name = (extract_artist_name(artist_context)
or get_import_clean_artist(context, default="")).strip()
if not source or not track_title or not artist_name:
return None
artist_id = str(get_import_source_ids(context).get("artist_id") or "")
from core.metadata.album_tracks import (
get_artist_albums_for_source,
get_artist_album_tracks,
)
from core.imports.single_to_album import resolve_single_to_album
def _acc(o, *ks):
for k in ks:
v = o.get(k) if isinstance(o, dict) else getattr(o, k, None)
if v:
return v
return None
def fetch_candidates():
albums = get_artist_albums_for_source(
source, artist_id, artist_name=artist_name,
album_type="album", limit=20) or []
return [{"name": _acc(a, "name", "title"),
"album_type": _acc(a, "album_type") or "album",
"id": _acc(a, "id", "album_id")} for a in albums]
def fetch_tracks(alb):
payload = get_artist_album_tracks(
str(alb.get("id") or ""), artist_name=artist_name,
album_name=alb.get("name") or "") or {}
return [(_acc(t, "title", "name", "track_name") or "")
for t in (payload.get("tracks") or [])]
album = resolve_single_to_album(
track_title,
fetch_album_candidates=fetch_candidates,
fetch_album_tracks=fetch_tracks)
if not album or not album.get("name"):
return None
logger.info("single->album: re-homed '%s' onto parent album '%s'",
track_title, album["name"])
promoted = build_import_album_info(
context,
album_info={
"album_name": album["name"],
"track_number": track_info.get("track_number"),
"disc_number": track_info.get("disc_number"),
"album_image_url": "",
"confidence": 0.5,
},
force_album=True,
)
# build_import_album_info resolves album_name via get_import_clean_album,
# which prefers original_search.album (the SINGLE's name); override it
# with the resolved parent album so grouping + tags use the album.
promoted["album_name"] = album["name"]
return promoted
except Exception as e:
logger.debug("single->album resolution failed: %s", e)
return None

72
tests/imports/test_single_to_album.py
View file

@ -5,10 +5,13 @@ takes injected fetchers, so neither needs a live metadata client.
from __future__ import annotations
import pytest
from core.imports.single_to_album import (
select_parent_album,
resolve_single_to_album,
)
from core.imports.context import detect_album_info_web
# ── pure selector ─────────────────────────────────────────────────────────────
@ -132,3 +135,72 @@ def test_resolver_caps_albums_probed():
fetch_album_tracks=fetch_tracks,
max_albums=5)
assert probed["n"] == 5 # never probes more than the cap
# ── gated wiring through detect_album_info_web (config gate + client shapes) ───
class _Cfg:
def __init__(self, on):
self._on = on
def get(self, key, default=None):
if key == "metadata_enhancement.single_to_album":
return self._on
return default
_SINGLE_CTX = {
"source": "spotify",
"artist": {"id": "art1", "name": "Coldplay"},
# album_type unset + name == track + total_tracks 1 -> is_album False, and the
# existing best-effort skips (album name == track), so the glue is reached.
"album": {"id": "s1", "name": "Yellow", "total_tracks": 1},
"track_info": {"id": "t1", "name": "Yellow", "track_number": 7},
"original_search_result": {"title": "Yellow", "album": "Yellow"},
}
def _patch_clients(monkeypatch, albums, tracks_by_id):
monkeypatch.setattr("core.metadata.album_tracks.get_artist_albums_for_source",
lambda *a, **k: albums)
monkeypatch.setattr("core.metadata.album_tracks.get_artist_album_tracks",
lambda album_id, **k: {"tracks": tracks_by_id.get(album_id, [])})
def test_glue_promotes_single_to_parent_album_when_enabled(monkeypatch):
monkeypatch.setattr("core.metadata.common.get_config_manager", lambda: _Cfg(True))
_patch_clients(
monkeypatch,
albums=[{"name": "Yellow", "album_type": "single", "id": "s1"},
{"name": "Parachutes", "album_type": "album", "id": "a2"}],
tracks_by_id={"a2": [{"title": "Shiver"}, {"title": "Yellow"}]},
)
out = detect_album_info_web(dict(_SINGLE_CTX), {"id": "art1", "name": "Coldplay"})
assert out and out["is_album"] is True
assert out["album_name"] == "Parachutes"
assert out["track_number"] == 7 # preserved
def test_glue_disabled_by_default_returns_none(monkeypatch):
monkeypatch.setattr("core.metadata.common.get_config_manager", lambda: _Cfg(False))
# Even with clients that WOULD match, the flag off => no promotion.
_patch_clients(monkeypatch,
albums=[{"name": "Parachutes", "album_type": "album", "id": "a2"}],
tracks_by_id={"a2": [{"title": "Yellow"}]})
assert detect_album_info_web(dict(_SINGLE_CTX), {"id": "art1", "name": "Coldplay"}) is None
def test_glue_no_match_returns_none(monkeypatch):
monkeypatch.setattr("core.metadata.common.get_config_manager", lambda: _Cfg(True))
_patch_clients(monkeypatch,
albums=[{"name": "Other Album", "album_type": "album", "id": "a9"}],
tracks_by_id={"a9": [{"title": "Different Song"}]})
assert detect_album_info_web(dict(_SINGLE_CTX), {"id": "art1", "name": "Coldplay"}) is None
def test_glue_failsafe_when_client_raises(monkeypatch):
monkeypatch.setattr("core.metadata.common.get_config_manager", lambda: _Cfg(True))
def boom(*a, **k):
raise RuntimeError("spotify down")
monkeypatch.setattr("core.metadata.album_tracks.get_artist_albums_for_source", boom)
assert detect_album_info_web(dict(_SINGLE_CTX), {"id": "art1", "name": "Coldplay"}) is None