From d989f252204126fc2ee29388586b78173e6f03f9 Mon Sep 17 00:00:00 2001 From: dev Date: Wed, 10 Jun 2026 14:46:32 +0200 Subject: [PATCH] feat(verification): shared evaluate() PASS/SKIP/FAIL decision core Co-Authored-By: Claude Opus 4.8 --- core/matching/audio_verification.py | 189 ++++++++++++++++++ .../matching/test_audio_verification_core.py | 54 ++++- 2 files changed, 242 insertions(+), 1 deletion(-) diff --git a/core/matching/audio_verification.py b/core/matching/audio_verification.py index 45ff48e6..72da6e8b 100644 --- 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}'") diff --git a/tests/matching/test_audio_verification_core.py b/tests/matching/test_audio_verification_core.py index a9dd2e85..efc8162c 100644 --- 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 / 澤野弘之 — <> 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", "澤野弘之 ")], + 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():