diff --git a/lib/acoustid_match.py b/lib/acoustid_match.py new file mode 100644 index 0000000..f5027b6 --- /dev/null +++ b/lib/acoustid_match.py @@ -0,0 +1,152 @@ +"""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. SPACE-separated, not +# `+`-joined: a literal `+` in the value gets percent-encoded to %2B, which +# AcoustID does NOT split into flags — it then attaches no recording metadata +# and every hit comes back empty (verified: `+` → 0 recordings, space → 28). +# `releases` is what carries the per-release DATE (nested under each +# releasegroup), which we need to pick the earliest original album + fill year. +LOOKUP_META = "recordings releasegroups releases 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 _rg_earliest_year(rg: dict) -> "int | None": + """Earliest release YEAR in a release-group (min over its nested releases' + dates). None when no release carries a date. This is what separates the + original pressing from later reissues/comps sharing the same group.""" + years = [] + for rel in (rg.get("releases") or []): + d = (rel or {}).get("date") + if isinstance(d, dict) and d.get("year"): + try: + years.append(int(d["year"])) + except (TypeError, ValueError): + pass + return min(years) if years else None + + +def _best_group(recording: dict) -> dict: + """Pick the display album: a clean studio Album first, and among those the + EARLIEST-released one — the original, not a later reissue or a compilation + that happens to be typed 'Album' (e.g. a soundtrack). This is what pulls + "Machine Head" ahead of a later comp for "Smoke on the Water". Falls back to + the first group when nothing is a studio album or nothing carries a date.""" + groups = [g for g in (recording.get("releasegroups") or []) if isinstance(g, dict)] + if not groups: + return {} + + def sort_key(g): + yr = _rg_earliest_year(g) + # studio (0) before non-studio (1); then earliest year (undated last). + return (0 if _rg_is_studio(g) else 1, yr if yr is not None else 9999) + + return sorted(groups, key=sort_key)[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) + _yr = _rg_earliest_year(rg) + year = str(_yr) if _yr else "" + 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 c3d4bd3..3118317 100644 --- a/server.py +++ b/server.py @@ -57,6 +57,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). @@ -246,6 +247,11 @@ _DEMO_BLOCKED: list[tuple[str, re.Pattern]] = [ ("POST", re.compile(r"^/api/enrichment/review/.+$")), ("POST", re.compile(r"^/api/enrichment/kick$")), ("GET", re.compile(r"^/api/enrichment/search$")), + # AcoustID audio fingerprinting: both identify endpoints run fpcalc (CPU) + # and spend the shared AcoustID rate budget on the caller's behalf — same + # rule as the search/kick relays above; not for anonymous demo visitors. + ("POST", re.compile(r"^/api/enrichment/identify$")), + ("POST", re.compile(r"^/api/enrichment/identify/.+$")), # Context menus (R2): the per-song re-match mutates the cache + spends # rate limit; Get-info exposes filesystem paths. ("POST", re.compile(r"^/api/enrichment/refresh/.+$")), @@ -6294,6 +6300,179 @@ def _mb_search_recordings(artist, title, limit: int = 12) -> 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. + +_ACOUSTID_MAX_UPLOAD_BYTES = 256 * 1024 * 1024 # 256 MB — an uncompressed master + + +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_settings() -> "tuple[bool, str]": + """(enabled, api_key) for AcoustID, resolved from settings with an env-var + fallback for the key. Opt-in: `acoustid_enabled` defaults off. The key lives + in settings so a user can set it themselves in the UI; $ACOUSTID_API_KEY is a + server-wide fallback for a headless deploy.""" + cfg = _load_config(CONFIG_DIR / "config.json") or {} + enabled = cfg.get("acoustid_enabled", False) is True + key = cfg.get("acoustid_api_key") + if not isinstance(key, str) or not key.strip(): + key = os.environ.get("ACOUSTID_API_KEY", "") + return enabled, (key or "").strip() + + +def _acoustid_available() -> bool: + """True only when the user opted in, a key is set (settings or env), the + network is on, AND fpcalc exists.""" + enabled, key = _acoustid_settings() + return (enabled + and _enrich_network_enabled() + and acoustid_match.is_configured(key) + 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_settings() + if not key or not _enrich_network_enabled(): + return [] + import requests + _enrich_throttle() + try: + # POST, not GET: a fingerprint is multi-KB (a 3.5-min track is ~3.5k + # chars), so a GET crams it into the URL and a long song overflows the + # server's URL limit → a spurious failure. AcoustID accepts the same + # params form-encoded in the body. + resp = requests.post( + f"{acoustid_match.ACOUSTID_API_ROOT}/lookup", + data={ + "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 _acoustid_gate() -> "JSONResponse | None": + """Shared availability gate for the identify endpoints: None when ready, + else a 412 needs_setup (opt-in off / no key → the UI re-prompts) or a 503 + (set up but fpcalc/network missing). Never lets a caller pretend a + fingerprint ran.""" + if _acoustid_available(): + return None + enabled, key = _acoustid_settings() + if not enabled or not key: + return JSONResponse( + {"error": "audio fingerprinting not set up", "needs_setup": True, + "detail": "Turn on AcoustID and add a free API key to identify by audio — " + "it reads the recording itself, far more reliable than text search."}, + status_code=412) + return JSONResponse( + {"error": "audio fingerprinting unavailable", "needs_setup": False, + "detail": "the fpcalc (Chromaprint) binary was not found on the server"}, + status_code=503) + + +def _song_audio_file(filename: str) -> "str | None": + """Resolve a LIBRARY song (by filename/id) to a local master-audio file for + fingerprinting: the full-mix `original_audio` extracted from a sloppak, or a + loose folder's audio. None when the song can't be found or ships no full-mix + audio (some packs carry only stems). Mirrors serve_sloppak_file's containment + guards so a crafted filename can't read outside DLC_DIR / the pack.""" + dlc = _get_dlc_dir() + if not dlc: + return None + resolved = _resolve_dlc_path(dlc, filename) + if resolved is None or not resolved.exists(): + return None + if sloppak_mod.is_sloppak(resolved): + try: + canon = resolved.relative_to(dlc.resolve()).as_posix() + except ValueError: + return None + rel = (sloppak_mod.load_manifest(resolved) or {}).get("original_audio") + if not isinstance(rel, str) or not rel.strip(): + return None + src = sloppak_mod.get_cached_source_dir(canon) + if src is None: + try: + src = sloppak_mod.resolve_source_dir(canon, dlc, SLOPPAK_CACHE_DIR) + except Exception: + return None + target = (src / rel.strip()).resolve() + try: + target.relative_to(src.resolve()) + except ValueError: + return None + return str(target) if target.is_file() else None + try: + audio = loosefolder_mod.find_audio(resolved) + except Exception: + audio = None + return str(audio) if audio and Path(str(audio)).is_file() else None + + def _mb_lookup_recording(mbid: str) -> dict | None: """Direct lookup for a manifest-carried recording MBID (tier 0).""" body = _mb_http_get( @@ -7493,6 +7672,93 @@ def api_enrichment_search(artist: str = "", title: str = "", limit: int = 8, return {"candidates": mb_match.rank_candidates(ref, cands)} +@app.post("/api/enrichment/identify") +async def api_enrichment_identify(request: Request): + """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. 412 `needs_setup` when the user hasn't + opted in / has no key (the UI nudges them to Settings); 503 when it's set up + but the fpcalc Chromaprint binary is missing or the network is off. Async so + the multipart is size-capped BEFORE spooling; the blocking fpcalc subprocess + + AcoustID HTTP run in the threadpool via run_in_executor.""" + gate = _acoustid_gate() + if gate is not None: + return gate + # Pre-parse Content-Length guard — reject an oversized body before Starlette + # spools the multipart to temp disk (mirrors the song-upload endpoint). The + # per-part cap below is the authoritative limit; this is the fast up-front no. + cl = request.headers.get("content-length") + if cl is not None: + try: + cl_int = int(cl) + except ValueError: + return JSONResponse({"error": "Invalid Content-Length header"}, status_code=400) + if cl_int > _ACOUSTID_MAX_UPLOAD_BYTES + _MULTIPART_OVERHEAD_SLACK: + return JSONResponse({"error": "audio upload too large (256 MB max)"}, status_code=413) + try: + form = await request.form(max_part_size=_ACOUSTID_MAX_UPLOAD_BYTES) + except Exception: + return JSONResponse({"error": "audio upload too large (256 MB max)"}, status_code=413) + file = form.get("file") + if not isinstance(file, UploadFile): + raise HTTPException(status_code=400, detail="missing file 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: + total = 0 + with open(tmp, "wb") as fh: + while True: + chunk = await file.read(1024 * 1024) + if not chunk: + break + total += len(chunk) + if total > _ACOUSTID_MAX_UPLOAD_BYTES: + return JSONResponse( + {"error": "audio upload too large (256 MB max)"}, status_code=413) + fh.write(chunk) + if total == 0: + raise HTTPException(status_code=400, detail="empty upload") + # fpcalc subprocess + AcoustID HTTP are blocking — off the event loop. + cands = await asyncio.get_event_loop().run_in_executor( + None, _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.post("/api/enrichment/identify/{filename:path}") +def api_enrichment_identify_song(filename: str): + """Identify an EXISTING library song by AUDIO FINGERPRINT — the library-side + counterpart to /api/enrichment/identify (which takes an upload). Fingerprints + the song's own master audio on disk (the manual "Identify by audio" action in + the Fix-metadata / match-review flow). Same candidate shape as /search, so the + review UI renders fingerprint hits like text hits. Same 412/503 gating; 404 + when the song has no full-mix audio to fingerprint.""" + gate = _acoustid_gate() + if gate is not None: + return gate + audio = _song_audio_file(filename) + if not audio: + return JSONResponse( + {"error": "no audio", + "detail": "couldn't find this song's master audio to fingerprint " + "(a stems-only pack has no full mix to identify)."}, + status_code=404) + try: + cands = _identify_by_fingerprint(audio) + except EnrichTransportError as e: + return JSONResponse({"error": "acoustid unavailable", "detail": str(e)}, + status_code=503) + return {"candidates": cands} + + @app.get("/api/startup-status") def startup_status(): return _get_startup_status() @@ -9839,6 +10105,17 @@ def _default_settings(): # the external browser, never media delivered in-app. "artist_pages_enabled": True, "artist_external_links": False, + # Audio fingerprinting (AcoustID + Chromaprint). OPT-IN, default OFF. + # Text matching (MusicBrainz) can't reliably pick the exact recording + # for a song with many comp/live/reissue takes (especially a + # non-title-track — the title can't find the album); fingerprinting + # reads the audio itself and resolves the EXACT recording. Needs the + # user's own free AcoustID application key + # (https://acoustid.org/new-application) plus the `fpcalc` binary. The + # key lives here (settings) — not only an env var — so a user can set it + # themselves in the UI; $ACOUSTID_API_KEY stays a server-wide fallback. + "acoustid_enabled": False, + "acoustid_api_key": "", } @@ -10013,13 +10290,24 @@ def save_settings(data: dict): "enrich_apply_names", "enrich_apply_year", "enrich_apply_genres", "enrich_apply_art", # Artist pages (PR-B): page on/off + external-links opt-in. - "artist_pages_enabled", "artist_external_links"): + "artist_pages_enabled", "artist_external_links", + # AcoustID audio-fingerprinting opt-in (default off). + "acoustid_enabled"): if _bool_key in data: raw = data[_bool_key] if raw is not None: if not isinstance(raw, bool): return {"error": f"{_bool_key} must be a boolean"} updates[_bool_key] = raw + if "acoustid_api_key" in data: + # Free AcoustID application key (opaque token). null is a no-op, empty + # string clears; length-capped so a bad POST can't bloat config.json. + # Never logged. The matcher trims + validates presence at read time. + raw = data["acoustid_api_key"] + if raw is not None: + if not isinstance(raw, str) or len(raw) > 128: + return {"error": "acoustid_api_key must be a string (at most 128 chars)"} + updates["acoustid_api_key"] = raw.strip() if "enrich_review_order" in data: raw = data["enrich_review_order"] if raw is not None: diff --git a/static/v3/index.html b/static/v3/index.html index 2196535..32aee83 100644 --- a/static/v3/index.html +++ b/static/v3/index.html @@ -788,6 +788,13 @@
What a confident match may fill in on its own — matches you confirm in the review queue always apply in full.
+ +
+
Audio fingerprint (AcoustID)
+ + +
Opt-in. Get a free key at acoustid.org/new-application; the fpcalc (Chromaprint) binary must be on the server's PATH.
+