feedBack/lib/acoustid_match.py

"""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