feat(verification): shared evaluate() PASS/SKIP/FAIL decision core

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 14:46:32 +02:00 · 2026-06-10 14:46:32 +02:00 · d989f25220
commit d989f25220
parent cf0a17c14a
2 changed files with 242 additions and 1 deletions
--- a/core/matching/audio_verification.py
+++ b/core/matching/audio_verification.py
@ -11,12 +11,32 @@ gate and duration guard are defined exactly once.
 """

 import re
+from dataclasses import dataclass
 from difflib import SequenceMatcher
+from enum import Enum
+from typing import Any, List, Optional

 # Thresholds — the single definition both paths share.
 MIN_ACOUSTID_SCORE = 0.80       # Minimum fingerprint score to trust a match.
 TITLE_MATCH_THRESHOLD = 0.70    # Title similarity to consider a match.
 ARTIST_MATCH_THRESHOLD = 0.60   # Artist similarity to consider a match.
+CLEAR_MISMATCH_THRESHOLD = 0.30  # Below this artist sim = clear wrong song.
+
+
+class Decision(Enum):
+    PASS = "pass"
+    SKIP = "skip"
+    FAIL = "fail"
+
+
+@dataclass
+class Outcome:
+    decision: Decision
+    title_sim: float = 0.0
+    artist_sim: float = 0.0
+    matched_title: str = ""
+    matched_artist: str = ""
+    reason: str = ""


 def normalize(text: str) -> str:
@ -57,3 +77,172 @@ def similarity(a: str, b: str) -> float:
    if na == nb:
        return 1.0
    return SequenceMatcher(None, na, nb).ratio()
+
+
+_match_engine = None
+
+
+def _detect_title_version(title: str) -> str:
+    """Version label ('original'/'instrumental'/'live'/'remix'/...) for a title."""
+    global _match_engine
+    if not title:
+        return 'original'
+    if _match_engine is None:
+        from core.matching_engine import MusicMatchingEngine
+        _match_engine = MusicMatchingEngine()
+    version_type, _ = _match_engine.detect_version_type(title)
+    return version_type
+
+
+def _alias_aware_artist_sim(expected_artist: str, actual_artist: str,
+                            aliases: Optional[Any] = None) -> float:
+    """Best artist similarity across (expected, *aliases) vs actual.
+
+    Bridges cross-script artist comparisons (kanji↔romaji etc) when MusicBrainz
+    aliases are available. ``aliases`` is an iterable of alias strings, or a
+    callable resolving them lazily (only invoked when direct similarity falls
+    below threshold — keeps the happy path lookup-free).
+    """
+    from core.matching.artist_aliases import artist_names_match
+
+    direct = similarity(expected_artist, actual_artist)
+    if aliases is None:
+        return direct
+    if direct >= ARTIST_MATCH_THRESHOLD:
+        return direct
+    resolved = aliases() if callable(aliases) else aliases
+    if not resolved:
+        return direct
+    _matched, score = artist_names_match(
+        expected_artist, actual_artist, aliases=resolved,
+        threshold=ARTIST_MATCH_THRESHOLD, similarity=similarity,
+    )
+    return score
+
+
+def _find_best_title_artist_match(recordings, expected_title, expected_artist,
+                                  aliases=None):
+    """Return (best_recording, title_sim, artist_sim) — title weighted higher."""
+    best_rec = None
+    best_title_sim = 0.0
+    best_artist_sim = 0.0
+    best_combined = 0.0
+    for rec in recordings:
+        title = rec.get('title') or ''
+        artist = rec.get('artist') or ''
+        title_sim = similarity(expected_title, title)
+        artist_sim = _alias_aware_artist_sim(expected_artist, artist, aliases)
+        combined = (title_sim * 0.6) + (artist_sim * 0.4)
+        if combined > best_combined:
+            best_combined = combined
+            best_rec = rec
+            best_title_sim = title_sim
+            best_artist_sim = artist_sim
+    return best_rec, best_title_sim, best_artist_sim
+
+
+def evaluate(expected_title: str, expected_artist: str,
+             recordings: List[dict], *, fingerprint_score: float,
+             file_duration_s: Optional[float] = None,
+             aliases_provider: Optional[Any] = None) -> Outcome:
+    """Decide PASS / SKIP / FAIL for a fingerprinted file against expected
+    title/artist. Pure: no I/O. Shared by import verification and library scan.
+
+    ``aliases_provider``: iterable or callable of expected-artist aliases
+    (kanji/cyrillic/etc) used to bridge cross-script comparisons.
+    ``file_duration_s``: when provided, a strong duration mismatch downgrades a
+    would-be FAIL to SKIP (fingerprint hash collision guard, used by the scan).
+    """
+    from core.matching.script_compat import is_cross_script_mismatch
+    from core.matching.acoustid_candidates import (
+        duration_mismatches_strongly, find_matching_recording,
+    )
+    from core.matching.version_mismatch import is_acceptable_version_mismatch
+
+    best_rec, title_sim, artist_sim = _find_best_title_artist_match(
+        recordings, expected_title, expected_artist, aliases_provider,
+    )
+    if not best_rec:
+        return Outcome(Decision.SKIP, reason="No recordings with title/artist info")
+
+    matched_title = best_rec.get('title', '?') or '?'
+    matched_artist = best_rec.get('artist', '?') or '?'
+
+    def out(dec, reason):
+        return Outcome(dec, title_sim, artist_sim, matched_title, matched_artist, reason)
+
+    # Version gate: original vs instrumental/live/remix is a real difference.
+    expected_version = _detect_title_version(expected_title)
+    matched_version = _detect_title_version(matched_title)
+    if expected_version != matched_version:
+        if not is_acceptable_version_mismatch(
+            expected_version, matched_version,
+            fingerprint_score=fingerprint_score,
+            title_similarity=title_sim, artist_similarity=artist_sim,
+        ):
+            return out(Decision.FAIL,
+                       f"Version mismatch: expected ({expected_version}) "
+                       f"but file is ({matched_version})")
+
+    # Clean match.
+    if title_sim >= TITLE_MATCH_THRESHOLD and artist_sim >= ARTIST_MATCH_THRESHOLD:
+        return out(Decision.PASS, "Audio verified")
+
+    # Title matches, artist doesn't — cover/collab vs genuinely wrong.
+    if title_sim >= TITLE_MATCH_THRESHOLD and artist_sim < ARTIST_MATCH_THRESHOLD:
+        for rec in recordings:
+            if _alias_aware_artist_sim(
+                expected_artist, rec.get('artist', ''), aliases_provider,
+            ) >= ARTIST_MATCH_THRESHOLD:
+                return out(Decision.PASS, "Expected artist found in AcoustID results")
+        if artist_sim < CLEAR_MISMATCH_THRESHOLD:
+            if file_duration_s and duration_mismatches_strongly(
+                file_duration_s, best_rec.get('duration') or best_rec.get('length')):
+                return out(Decision.SKIP, "Duration mismatch (fingerprint collision)")
+            return out(Decision.FAIL,
+                       f"Audio mismatch: '{matched_title}' by '{matched_artist}' "
+                       f"— expected artist not found")
+        return out(Decision.SKIP, "Title matches but artist ambiguous (cover/collab?)")
+
+    # Title doesn't match — scan all recordings for a version-matched hit.
+    def _title_sim(a, b):
+        return similarity(a, b)
+
+    def _artist_sim(ea, aa):
+        return _alias_aware_artist_sim(ea, aa, aliases_provider)
+
+    candidate = None
+    for rec in recordings:
+        if _detect_title_version(rec.get('title') or '') != expected_version:
+            continue
+        if (similarity(expected_title, rec.get('title') or '') >= TITLE_MATCH_THRESHOLD
+                and _alias_aware_artist_sim(
+                    expected_artist, rec.get('artist', ''), aliases_provider,
+                ) >= ARTIST_MATCH_THRESHOLD):
+            candidate = rec
+            break
+    if candidate is not None:
+        return out(Decision.PASS, "Scan match found in AcoustID results")
+
+    # High-confidence / cross-script skips (don't quarantine a correct file).
+    has_non_ascii = (any(ord(c) > 127 for c in (expected_title or ''))
+                     or any(ord(c) > 127 for c in matched_title))
+    language_script_skip = (fingerprint_score >= 0.95 and has_non_ascii
+                            and artist_sim >= ARTIST_MATCH_THRESHOLD)
+    high_confidence_strong_match_skip = (fingerprint_score >= 0.95
+                                         and title_sim >= 0.80
+                                         and artist_sim >= ARTIST_MATCH_THRESHOLD)
+    cross_script_artist_skip = (fingerprint_score >= MIN_ACOUSTID_SCORE
+                                and artist_sim >= ARTIST_MATCH_THRESHOLD
+                                and is_cross_script_mismatch(expected_artist, matched_artist))
+    if (language_script_skip or high_confidence_strong_match_skip
+            or cross_script_artist_skip):
+        return out(Decision.SKIP, "Likely same song in different language/script")
+
+    if file_duration_s and duration_mismatches_strongly(
+            file_duration_s, best_rec.get('duration') or best_rec.get('length')):
+        return out(Decision.SKIP, "Duration mismatch (fingerprint collision)")
+
+    return out(Decision.FAIL,
+               f"Audio mismatch: file identified as '{matched_title}' by "
+               f"'{matched_artist}', expected '{expected_title}' by '{expected_artist}'")
--- a/tests/matching/test_audio_verification_core.py
+++ b/tests/matching/test_audio_verification_core.py
@ -4,7 +4,59 @@ One place for normalization + the PASS/SKIP/FAIL decision used by BOTH import-ti
 verification and the library AcoustID scan, so the two paths can't drift apart.
 """

-from core.matching.audio_verification import normalize
+from core.matching.audio_verification import normalize, evaluate, Decision
+
+
+def _rec(title, artist, duration=None):
+    return {"title": title, "artist": artist, "duration": duration}
+
+
+def test_cross_script_vocal_credit_clean_match_passes():
+    # Sawano / 澤野弘之 <Vocal: ...> — <> stripped, alias bridges the artist,
+    # title matches too -> clean PASS (must never FAIL/quarantine).
+    out = evaluate(
+        "Call Your Name", "Sawano Hiroyuki",
+        [_rec("call your name", "澤野弘之 <Vocal: mpi & CASG>")],
+        fingerprint_score=0.95,
+        aliases_provider=lambda: ["澤野弘之"],
+    )
+    assert out.decision == Decision.PASS
+
+
+def test_cross_script_ipa_title_skips_not_fails():
+    # AcoustID returns an IPA-transcribed title that can't string-match, but the
+    # artist bridges cross-script -> SKIP (import anyway), never FAIL.
+    out = evaluate(
+        "Attack on Titan", "Sawano Hiroyuki",
+        [_rec("ətˈæk 0N tάɪtn", "澤野弘之")],
+        fingerprint_score=0.95,
+        aliases_provider=lambda: ["澤野弘之"],
+    )
+    assert out.decision == Decision.SKIP
+
+
+def test_clean_cross_script_match_passes():
+    out = evaluate(
+        "Xl-Tt", "Sawano Hiroyuki",
+        [_rec("xl-tt", "澤野弘之")],
+        fingerprint_score=0.95,
+        aliases_provider=lambda: ["澤野弘之"],
+    )
+    assert out.decision == Decision.PASS
+
+
+def test_genuine_wrong_song_fails():
+    out = evaluate(
+        "Yellow", "Coldplay",
+        [_rec("Rich Interlude", "Kendrick Lamar")],
+        fingerprint_score=0.85,
+    )
+    assert out.decision == Decision.FAIL
+
+
+def test_no_recordings_skips():
+    out = evaluate("Whatever", "Someone", [], fingerprint_score=0.9)
+    assert out.decision == Decision.SKIP


 def test_normalize_strips_paren_bracket_angle_and_keeps_cjk():