feat: add real MusicBrainz resolver and opt-in live test

2026-07-10 23:21:50 +02:00
parent c16e9f1035
commit 85c74ba109
3 changed files with 91 additions and 0 deletions
@@ -0,0 +1,72 @@
+from difflib import SequenceMatcher
+
+from lyra_worker.types import MBTarget
+
+_MIN_TITLE_RATIO = 0.6
+
+
+def _ratio(a: str, b: str) -> float:
+    return SequenceMatcher(None, a.strip().casefold(), b.strip().casefold()).ratio()
+
+
+class MusicBrainzResolver:
+    """Real MbResolver using the MusicBrainz webservice via musicbrainzngs.
+
+    NOT unit-tested offline; see test_musicbrainz_live.py. musicbrainzngs is
+    imported lazily so importing/constructing this class stays offline.
+    """
+
+    def __init__(self, app_name: str = "Lyra", version: str = "0.1", contact: str = "lyra@localhost"):
+        self._app = app_name
+        self._version = version
+        self._contact = contact
+
+    def resolve(self, artist: str, album: str) -> MBTarget | None:
+        import musicbrainzngs
+
+        musicbrainzngs.set_useragent(self._app, self._version, self._contact)
+
+        res = musicbrainzngs.search_release_groups(query=album, artist=artist, limit=5)
+        groups = res.get("release-group-list", [])
+        if not groups:
+            return None
+        rg = max(groups, key=lambda g: _ratio(album, g.get("title", "")))
+        if _ratio(album, rg.get("title", "")) < _MIN_TITLE_RATIO:
+            return None
+
+        canonical_album = rg.get("title", album)
+        canonical_artist = artist
+        credit = rg.get("artist-credit")
+        if isinstance(credit, list) and credit and isinstance(credit[0], dict):
+            canonical_artist = (credit[0].get("artist") or {}).get("name", artist)
+
+        year = None
+        frd = rg.get("first-release-date", "") or ""
+        if frd[:4].isdigit():
+            year = int(frd[:4])
+
+        track_count = None
+        total_duration_s = None
+        rgid = rg.get("id")
+        if rgid:
+            rgfull = musicbrainzngs.get_release_group_by_id(rgid, includes=["releases"])
+            releases = rgfull.get("release-group", {}).get("release-list", [])
+            if releases:
+                rel = musicbrainzngs.get_release_by_id(
+                    releases[0]["id"], includes=["recordings"]
+                ).get("release", {})
+                tracks = []
+                for medium in rel.get("medium-list", []):
+                    tracks.extend(medium.get("track-list", []))
+                if tracks:
+                    track_count = len(tracks)
+                    total_ms = sum(int((t.get("recording") or {}).get("length") or 0) for t in tracks)
+                    total_duration_s = total_ms // 1000 if total_ms else None
+
+        return MBTarget(
+            artist=canonical_artist,
+            album=canonical_album,
+            track_count=track_count,
+            total_duration_s=total_duration_s,
+            year=year,
+        )