From 0bdf0f131191c64e1c8f7ef17a89fb2b44e7dcff Mon Sep 17 00:00:00 2001 From: ChrisBeWithYou Date: Sat, 4 Jul 2026 05:22:24 -0500 Subject: [PATCH] feat(enrichment): AcoustID audio-fingerprint identification (opt-in) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Text search can only guess the version; the definitive fix is content-based — fingerprint the actual audio with Chromaprint (fpcalc) and look it up on AcoustID, which maps the fingerprint to the EXACT MusicBrainz recording (the approach Lidarr uses). Sidesteps the studio-vs-live ambiguity entirely. - lib/acoustid_match.py: pure response parsing + config gating (unit-tested); normalizes AcoustID hits into the same candidate shape as mb_match so the review UI + editor Match popup render fingerprint and text hits identically. - server.py: _fpcalc (Chromaprint subprocess), _acoustid_lookup (throttled, offline-guarded HTTP), _identify_by_fingerprint (also available to the library-enrichment pipeline), and POST /api/enrichment/identify (upload the master audio → candidates). - Fully OPT-IN and graceful: absent the fpcalc binary or an ACOUSTID_API_KEY the whole path is a no-op / 503 and the text matcher runs unchanged. Requires (both optional): the `fpcalc` (Chromaprint) binary on PATH/$FPCALC, and a free AcoustID application key in $ACOUSTID_API_KEY. Pure parsing/gating is unit-tested; the fpcalc + live-lookup path needs those two to exercise. Co-Authored-By: Claude Opus 4.8 (1M context) --- lib/acoustid_match.py | 128 +++++++++++++++++++++++++++++++++++ server.py | 126 ++++++++++++++++++++++++++++++++++ tests/test_acoustid_match.py | 94 +++++++++++++++++++++++++ 3 files changed, 348 insertions(+) create mode 100644 lib/acoustid_match.py create mode 100644 tests/test_acoustid_match.py diff --git a/lib/acoustid_match.py b/lib/acoustid_match.py new file mode 100644 index 0000000..2765c13 --- /dev/null +++ b/lib/acoustid_match.py @@ -0,0 +1,128 @@ +"""AcoustID audio-fingerprint identification for MusicBrainz enrichment. + +A flat MusicBrainz *text* search ties every take of a song at the same score — +studio, a dozen live bootlegs, and every compilation — so "AC/DC — Highway to +Hell" returns junk (see lib/mb_match.py's canonical re-ranking, which mitigates +it). The definitive fix is content-based: fingerprint the actual audio with +Chromaprint (`fpcalc`) and look it up on AcoustID, which maps the fingerprint +straight to the *exact* MusicBrainz recording — the same approach Lidarr uses. + +This module is the PURE half (no network, no subprocess): response parsing + +config gating, so it is unit-testable in isolation. server.py owns the `fpcalc` +subprocess and the throttled HTTP GET to api.acoustid.org. + +Operational requirements (both optional — absent ⇒ this path is a graceful +no-op and the text matcher still runs): + * `fpcalc` (Chromaprint) on PATH or at $FPCALC — generates the fingerprint. + * an AcoustID application API key in $ACOUSTID_API_KEY — free from + https://acoustid.org/new-application ; AcoustID etiquette limits to ~3 req/s. +""" + +import os + +ACOUSTID_API_ROOT = "https://api.acoustid.org/v2" + +# The `meta` fields we ask AcoustID to return so a hit resolves to displayable +# metadata without a second MusicBrainz round-trip. +LOOKUP_META = "recordings+releasegroups+compress" + +# Mirror mb_match._SECONDARY_SKIP: release-group secondary types that mark a +# non-canonical (live/comp/remix) release, so we can flag the studio take. +_SECONDARY_SKIP = { + "live", "compilation", "remix", "dj-mix", "mixtape/street", + "demo", "interview", "audiobook", "spokenword", +} + + +def api_key(explicit: str | None = None) -> str: + """The AcoustID application API key: an explicit value (e.g. a host setting) + wins, else $ACOUSTID_API_KEY, else "" (⇒ fingerprinting disabled).""" + return (explicit or os.environ.get("ACOUSTID_API_KEY") or "").strip() + + +def is_configured(explicit_key: str | None = None) -> bool: + """True when an API key is available. `fpcalc` presence is checked by + server.py (it owns the binary lookup); both are required to actually run.""" + return bool(api_key(explicit_key)) + + +def _rg_is_studio(rg: dict) -> bool: + if str(rg.get("type", "")).lower() != "album": + return False + secs = {str(s).lower() for s in (rg.get("secondarytypes") or [])} + return not (secs & _SECONDARY_SKIP) + + +def _best_group(recording: dict) -> dict: + """Prefer a studio Album release-group for the display album, else the first.""" + groups = [g for g in (recording.get("releasegroups") or []) if isinstance(g, dict)] + if not groups: + return {} + groups = sorted(groups, key=lambda g: 0 if _rg_is_studio(g) else 1) + return groups[0] + + +def _first_artist(recording: dict) -> str: + for a in (recording.get("artists") or []): + if isinstance(a, dict) and a.get("name"): + return str(a["name"]) + return "" + + +def parse_lookup_response(body: dict) -> list[dict]: + """Normalize an AcoustID /v2/lookup response into the same flat candidate + shape as mb_match (recording_id / title / artist / album / year / duration / + studio / mb_score / score), so the review UI and the editor's Match popup + render fingerprint hits and text hits identically. `mb_score` carries the + AcoustID confidence (0-100) — a fingerprint hit is high-signal by nature.""" + if not isinstance(body, dict) or body.get("status") != "ok": + return [] + out: list[dict] = [] + seen: set[str] = set() + for result in (body.get("results") or []): + if not isinstance(result, dict): + continue + try: + score = float(result.get("score") or 0.0) + except (TypeError, ValueError): + score = 0.0 + for rec in (result.get("recordings") or []): + if not isinstance(rec, dict) or not rec.get("id"): + continue + rid = str(rec["id"]) + if rid in seen: + continue + seen.add(rid) + rg = _best_group(rec) + year = "" + for rel in (rg.get("releases") or []): + d = (rel or {}).get("date") or {} + y = d.get("year") if isinstance(d, dict) else None + if y: + year = str(y)[:4] + break + dur = rec.get("duration") + try: + duration = int(round(float(dur))) if dur else None + except (TypeError, ValueError): + duration = None + out.append({ + "recording_id": rid, + "title": str(rec.get("title", "") or ""), + "artist": _first_artist(rec), + "album": str(rg.get("title", "") or ""), + "year": year, + "duration": duration, + "isrc": "", + "genres": [], + "studio": _rg_is_studio(rg), + "acoustid_score": round(score, 4), + # Fingerprint hits are content-verified, not text-guessed — carry + # the AcoustID confidence as the display score band. + "mb_score": int(round(score * 100)), + "score": round(score, 4), + "source": "acoustid", + }) + # Best AcoustID confidence first; studio take breaks ties. + out.sort(key=lambda c: (c["acoustid_score"], 1 if c["studio"] else 0), reverse=True) + return out diff --git a/server.py b/server.py index 94f282f..3264c22 100644 --- a/server.py +++ b/server.py @@ -52,6 +52,7 @@ import loosefolder as loosefolder_mod # tier classification + response parsing. No network/DB in there — the # throttled transport and the song_enrichment writes live in this module. import mb_match +import acoustid_match # Metadata extraction lives in a side-effect-free module so ProcessPool # scan workers can import + unpickle _scan_one without re-running this # module's import-time side effects (see lib/scan_worker.py). @@ -6244,6 +6245,97 @@ def _mb_search_recordings(artist, title, limit: int = 8) -> list[dict]: return mb_match.parse_search_response(body or {}) +# ── AcoustID audio fingerprinting (content-based identification) ────────────── +# Optional path: requires the Chromaprint `fpcalc` binary AND an AcoustID API +# key ($ACOUSTID_API_KEY). Both absent ⇒ graceful no-op; the text matcher runs. + +def _fpcalc_bin() -> str | None: + """Locate the Chromaprint `fpcalc` binary: $FPCALC override, else PATH.""" + import shutil + cand = os.environ.get("FPCALC") + if cand and Path(cand).exists(): + return cand + return shutil.which("fpcalc") + + +def _acoustid_available() -> bool: + """True only when the network is on, an API key is set, AND fpcalc exists.""" + return (_enrich_network_enabled() + and acoustid_match.is_configured() + and _fpcalc_bin() is not None) + + +def _fpcalc(path: str) -> "tuple[int, str] | None": + """Fingerprint a local audio file → (duration_seconds, fingerprint). None on + any failure (missing binary/file, decode error, timeout).""" + binp = _fpcalc_bin() + if not binp or not Path(path).exists(): + return None + import subprocess + import json as _json + try: + pr = subprocess.run([binp, "-json", str(path)], + capture_output=True, timeout=30) + except Exception: + return None + if pr.returncode != 0: + return None + try: + data = _json.loads(pr.stdout.decode("utf-8", "replace")) + dur = int(round(float(data.get("duration")))) + fp = str(data.get("fingerprint") or "") + except Exception: + return None + if not fp or dur <= 0: + return None + return dur, fp + + +def _acoustid_lookup(duration: int, fingerprint: str) -> list[dict]: + """Look a fingerprint up on AcoustID → candidate dicts (mb_match shape). + Throttled + offline-guarded like the MusicBrainz path. [] when unavailable + or no hit; raises EnrichTransportError for network-shaped failures.""" + key = acoustid_match.api_key() + if not key or not _enrich_network_enabled(): + return [] + import requests + _enrich_throttle() + try: + resp = requests.get( + f"{acoustid_match.ACOUSTID_API_ROOT}/lookup", + params={ + "client": key, "format": "json", + "meta": acoustid_match.LOOKUP_META, + "duration": duration, "fingerprint": fingerprint, + }, + headers={"User-Agent": _enrich_user_agent()}, + timeout=10, + ) + except requests.RequestException as e: + raise EnrichTransportError(str(e)) from e + if resp.status_code == 429: + raise EnrichTransportError("acoustid 429 (rate limited)") + if resp.status_code != 200: + raise EnrichTransportError(f"acoustid HTTP {resp.status_code}") + try: + body = resp.json() + except ValueError as e: + raise EnrichTransportError("bad JSON from acoustid") from e + return acoustid_match.parse_lookup_response(body) + + +def _identify_by_fingerprint(path: str) -> list[dict]: + """fpcalc + AcoustID lookup for a local audio file. [] if fingerprinting is + unavailable, the file can't be read, or nothing matched. Available to the + library-enrichment pipeline as well as the /identify endpoint.""" + if not _acoustid_available(): + return [] + fp = _fpcalc(path) + if not fp: + return [] + return _acoustid_lookup(fp[0], fp[1]) + + def _mb_lookup_recording(mbid: str) -> dict | None: """Direct lookup for a manifest-carried recording MBID (tier 0).""" body = _mb_http_get( @@ -7436,6 +7528,40 @@ def api_enrichment_search(artist: str = "", title: str = "", limit: int = 8, return {"candidates": mb_match.rank_candidates(ref, cands)} +@app.post("/api/enrichment/identify") +def api_enrichment_identify(file: UploadFile = File(...)): + """Identify a song by AUDIO FINGERPRINT (AcoustID) rather than text — the + reliable way to get the EXACT recording/version (the studio take, not a live + bootleg or an extended cut). Upload the master audio; returns candidates in + the same shape as /search, so the review UI and the editor's Match popup can + render fingerprint hits identically. 503 when fingerprinting isn't configured + (needs the fpcalc Chromaprint binary + an ACOUSTID_API_KEY). Sync route: the + fpcalc subprocess + HTTP run in FastAPI's threadpool.""" + if not _acoustid_available(): + return JSONResponse( + {"error": "audio fingerprinting unavailable", + "detail": "requires the fpcalc (Chromaprint) binary and an " + "ACOUSTID_API_KEY"}, + status_code=503) + content = file.file.read() + if not content: + raise HTTPException(status_code=400, detail="empty upload") + import tempfile + ext = (Path(file.filename or "").suffix or ".bin").lower() + tmpdir = tempfile.mkdtemp(prefix="feedback_acoustid_") + tmp = os.path.join(tmpdir, "audio" + ext) + try: + with open(tmp, "wb") as fh: + fh.write(content) + cands = _identify_by_fingerprint(tmp) + except EnrichTransportError as e: + return JSONResponse({"error": "acoustid unavailable", "detail": str(e)}, + status_code=503) + finally: + shutil.rmtree(tmpdir, ignore_errors=True) + return {"candidates": cands} + + @app.get("/api/startup-status") def startup_status(): return _get_startup_status() diff --git a/tests/test_acoustid_match.py b/tests/test_acoustid_match.py new file mode 100644 index 0000000..d1a6969 --- /dev/null +++ b/tests/test_acoustid_match.py @@ -0,0 +1,94 @@ +"""Pure-function tests for AcoustID fingerprint response parsing + config +gating. No network, no fpcalc binary — server.py owns those seams.""" +import acoustid_match as a + + +def _resp(score=0.97, rec_id="rec-1", title="Highway to Hell", artist="AC/DC", + rg_title="Highway to Hell", rg_type="Album", secondary=None, + year=1979, duration=208.4): + return { + "status": "ok", + "results": [{ + "id": "acoustid-uuid", + "score": score, + "recordings": [{ + "id": rec_id, + "title": title, + "duration": duration, + "artists": [{"id": "a1", "name": artist}], + "releasegroups": [{ + "id": "rg1", "title": rg_title, "type": rg_type, + "secondarytypes": secondary or [], + "releases": [{"date": {"year": year}}], + }], + }], + }], + } + + +def test_parse_maps_the_studio_recording(): + out = a.parse_lookup_response(_resp()) + assert len(out) == 1 + c = out[0] + assert c["recording_id"] == "rec-1" + assert c["title"] == "Highway to Hell" + assert c["artist"] == "AC/DC" + assert c["album"] == "Highway to Hell" + assert c["year"] == "1979" + assert c["duration"] == 208 + assert c["studio"] is True + assert c["source"] == "acoustid" + assert c["mb_score"] == 97 # 0.97 → 0..100 confidence band + assert c["score"] == 0.97 + + +def test_live_release_group_is_not_studio(): + out = a.parse_lookup_response(_resp(rg_type="Album", secondary=["Live"])) + assert out[0]["studio"] is False + + +def test_compilation_is_not_studio(): + out = a.parse_lookup_response(_resp(secondary=["Compilation"])) + assert out[0]["studio"] is False + + +def test_prefers_studio_group_for_album_display(): + resp = _resp() + # Add a comp release-group first; the studio one must win the album pick. + resp["results"][0]["recordings"][0]["releasegroups"].insert(0, { + "id": "rg0", "title": "Greatest Hits", "type": "Album", + "secondarytypes": ["Compilation"], "releases": [{"date": {"year": 2000}}], + }) + c = a.parse_lookup_response(resp)[0] + assert c["album"] == "Highway to Hell" + assert c["studio"] is True + + +def test_dedupes_recording_across_results(): + resp = _resp() + resp["results"].append(dict(resp["results"][0])) # same recording again + assert len(a.parse_lookup_response(resp)) == 1 + + +def test_non_ok_status_and_garbage_return_empty(): + assert a.parse_lookup_response({"status": "error"}) == [] + assert a.parse_lookup_response({}) == [] + assert a.parse_lookup_response(None) == [] + assert a.parse_lookup_response({"status": "ok", "results": []}) == [] + + +def test_higher_acoustid_score_ranks_first(): + resp = _resp(score=0.55, rec_id="low") + resp["results"].append(_resp(score=0.99, rec_id="high")["results"][0]) + out = a.parse_lookup_response(resp) + assert out[0]["recording_id"] == "high" + + +def test_config_gating(monkeypatch): + monkeypatch.delenv("ACOUSTID_API_KEY", raising=False) + assert a.api_key() == "" + assert a.is_configured() is False + assert a.is_configured("explicit-key") is True + monkeypatch.setenv("ACOUSTID_API_KEY", "envkey") + assert a.api_key() == "envkey" + assert a.is_configured() is True