From 8e6820dbdf6cf6d5368c3372a1aefb9df4756fe0 Mon Sep 17 00:00:00 2001 From: dev Date: Wed, 10 Jun 2026 18:30:49 +0200 Subject: [PATCH] feat(verification): status vocabulary, DB column, SOULSYNC_VERIFICATION tag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Also: evaluate() treats an empty expected artist as title-only comparison (old scanner behaviour — a missing DB artist is no evidence of a wrong file), and the thresholds are now defined once in the core and re-exported. Co-Authored-By: Claude Opus 4.8 --- core/acoustid_verification.py | 9 ++-- core/matching/audio_verification.py | 7 +++ core/matching/verification_status.py | 48 ++++++++++++++++++ core/tag_writer.py | 46 +++++++++++++++++ database/music_database.py | 5 ++ .../matching/test_audio_verification_core.py | 9 ++++ tests/matching/test_verification_status.py | 26 ++++++++++ tests/test_verification_tag_roundtrip.py | 50 +++++++++++++++++++ 8 files changed, 196 insertions(+), 4 deletions(-) create mode 100644 core/matching/verification_status.py create mode 100644 tests/matching/test_verification_status.py create mode 100644 tests/test_verification_tag_roundtrip.py diff --git a/core/acoustid_verification.py b/core/acoustid_verification.py index a399dda6..0d10b613 100644 --- a/core/acoustid_verification.py +++ b/core/acoustid_verification.py @@ -22,10 +22,11 @@ from core.musicbrainz_client import MusicBrainzClient logger = get_logger("acoustid.verification") -# Thresholds -MIN_ACOUSTID_SCORE = 0.80 # Minimum AcoustID fingerprint score to trust -TITLE_MATCH_THRESHOLD = 0.70 # Title similarity needed to consider a match -ARTIST_MATCH_THRESHOLD = 0.60 # Artist similarity needed to consider a match +# Thresholds — single definition lives in the shared core; re-exported here so +# existing importers keep working and the values can't drift between paths. +from core.matching.audio_verification import ( # noqa: E402 + MIN_ACOUSTID_SCORE, TITLE_MATCH_THRESHOLD, ARTIST_MATCH_THRESHOLD, +) # Single matching-engine instance so version detection reuses the same patterns # used by the pre-download Soulseek matcher (remix / live / acoustic / diff --git a/core/matching/audio_verification.py b/core/matching/audio_verification.py index cd1b2c17..06b90949 100644 --- a/core/matching/audio_verification.py +++ b/core/matching/audio_verification.py @@ -173,9 +173,16 @@ def evaluate(expected_title: str, expected_artist: str, from core.matching.script_compat import is_cross_script_mismatch from core.matching.version_mismatch import is_acceptable_version_mismatch + # No expected artist on record (legacy/compilation rows): compare on title + # only — the old scanner treated this as artist-match=1.0 and a missing DB + # value is no evidence the file is wrong. + no_expected_artist = not normalize(expected_artist or '') + best_rec, title_sim, artist_sim = _find_best_title_artist_match( recordings, expected_title, expected_artist, aliases_provider, ) + if no_expected_artist: + artist_sim = 1.0 if not best_rec: return Outcome(Decision.SKIP, reason="No recordings with title/artist info") diff --git a/core/matching/verification_status.py b/core/matching/verification_status.py new file mode 100644 index 00000000..4b7569c9 --- /dev/null +++ b/core/matching/verification_status.py @@ -0,0 +1,48 @@ +"""Verification-status vocabulary for imported tracks. + +Three states, persisted in the DB (``tracks.verification_status``) AND as an +embedded file tag (``SOULSYNC_VERIFICATION``) so the information survives DB +resets and travels with the file: + +- ``verified`` — clean AcoustID PASS at import time. +- ``unverified`` — AcoustID SKIP (cross-script / ambiguous / no match in + the AcoustID DB). Imported, but not hard-confirmed. +- ``force_imported`` — accepted via the version-mismatch fallback after the + retry budget was exhausted (user opted in via + ``post_processing.accept_version_mismatch_fallback``). + A later library scan will still re-check these but + reports them as informational, clearly marked. + +Quarantined files are never imported, so they carry no status. +""" + +VERIFIED = 'verified' +UNVERIFIED = 'unverified' +FORCE_IMPORTED = 'force_imported' + +ALL_STATUSES = (VERIFIED, UNVERIFIED, FORCE_IMPORTED) + +# The file tag name (Vorbis comment key / ID3 TXXX desc / MP4 freeform). +TAG_NAME = 'SOULSYNC_VERIFICATION' + + +def status_from_acoustid_result(result_value): + """Map an AcoustID verification result string ('pass'/'skip'/...) to a + status. 'disabled'/'error'/unknown return None — no claim either way.""" + if result_value == 'pass': + return VERIFIED + if result_value == 'skip': + return UNVERIFIED + return None + + +def status_for_import(context: dict): + """Status for a just-imported file from its pipeline context. + + The version-mismatch fallback flag wins: a force-accepted file is + ``force_imported`` regardless of what the (earlier, failed) verification + said about the candidate. + """ + if context.get('_version_mismatch_fallback'): + return FORCE_IMPORTED + return status_from_acoustid_result(context.get('_acoustid_result')) diff --git a/core/tag_writer.py b/core/tag_writer.py index b0d4c6fa..816e69cd 100644 --- a/core/tag_writer.py +++ b/core/tag_writer.py @@ -44,6 +44,8 @@ def read_file_tags(file_path: str) -> Dict[str, Any]: 'replaygain_track_peak': None, 'replaygain_album_gain': None, 'replaygain_album_peak': None, + # SoulSync verification status ('verified'/'unverified'/'force_imported') + 'verification_status': None, } if not file_path or not os.path.exists(file_path): @@ -80,6 +82,10 @@ def read_file_tags(file_path: str) -> Dict[str, Any]: except (ValueError, TypeError): pass result['has_cover_art'] = bool(audio.tags.getall('APIC')) + for fr in audio.tags.getall('TXXX'): + if getattr(fr, 'desc', '') == 'SOULSYNC_VERIFICATION' and fr.text: + result['verification_status'] = str(fr.text[0]) + break elif isinstance(audio, (FLAC, OggVorbis)) or type(audio).__name__ == 'OggOpus': # FLAC / OGG @@ -102,6 +108,7 @@ def read_file_tags(file_path: str) -> Dict[str, Any]: else: # OGG doesn't have a standard picture field we can easily check result['has_cover_art'] = False + result['verification_status'] = _vorbis_first(audio, 'soulsync_verification') elif isinstance(audio, MP4): # MP4 / M4A @@ -118,6 +125,12 @@ def read_file_tags(file_path: str) -> Dict[str, Any]: if disk: result['disc_number'] = disk[0][0] if isinstance(disk[0], tuple) else None result['has_cover_art'] = bool(audio.tags.get('covr', [])) if audio.tags else False + vs = (audio.tags or {}).get('----:com.soulsync:VERIFICATION') + if vs: + raw = vs[0] + result['verification_status'] = ( + raw.decode('utf-8', 'ignore') if isinstance(raw, bytes) else str(raw) + ) except Exception as e: result['error'] = str(e) @@ -156,6 +169,39 @@ def is_placeholder_meta(value: Any) -> bool: return s == '' or s in _PLACEHOLDER_META_VALUES +def write_verification_status(file_path: str, status: str) -> bool: + """Embed the SoulSync verification status into the file's tags. + + Vorbis comment ``SOULSYNC_VERIFICATION`` (FLAC/OGG/Opus), ID3 + ``TXXX:SOULSYNC_VERIFICATION`` (MP3), MP4 freeform + ``----:com.soulsync:VERIFICATION``. The tag travels with the file so the + status survives DB resets; the AcoustID scan reads it back via + ``read_file_tags`` to refresh the DB column and to mark force-imported + fallbacks in its findings. Never raises; returns success. + """ + if not status or not file_path or not os.path.exists(file_path): + return False + try: + audio = MutagenFile(file_path) + if audio is None: + return False + if getattr(audio, 'tags', None) is None and hasattr(audio, 'add_tags'): + audio.add_tags() + if isinstance(audio.tags, ID3): + audio.tags.delall('TXXX:SOULSYNC_VERIFICATION') + audio.tags.add(TXXX(encoding=3, desc='SOULSYNC_VERIFICATION', text=[status])) + elif isinstance(audio, MP4): + audio.tags['----:com.soulsync:VERIFICATION'] = [status.encode('utf-8')] + else: + # Vorbis-comment family (FLAC / OggVorbis / OggOpus) + audio['SOULSYNC_VERIFICATION'] = [status] + audio.save() + return True + except Exception as e: + logger.debug("write_verification_status failed for %s: %s", file_path, e) + return False + + def guard_placeholder_overwrite(db_val: Any, file_val: Any) -> Any: """#800 guard: never replace a real file value with a placeholder. diff --git a/database/music_database.py b/database/music_database.py index 7b352a34..2b2b3881 100644 --- a/database/music_database.py +++ b/database/music_database.py @@ -2409,6 +2409,11 @@ class MusicDatabase: if 'musicbrainz_match_status' not in tracks_columns: cursor.execute("ALTER TABLE tracks ADD COLUMN musicbrainz_match_status TEXT") added_tracks = True + if 'verification_status' not in tracks_columns: + # 'verified' / 'unverified' / 'force_imported' — set at import, + # refreshed by the AcoustID scan (which reads the file tag). + cursor.execute("ALTER TABLE tracks ADD COLUMN verification_status TEXT") + added_tracks = True if added_tracks: columns_added = True logger.info("Added MusicBrainz columns to tracks table") diff --git a/tests/matching/test_audio_verification_core.py b/tests/matching/test_audio_verification_core.py index efc8162c..09f6a2aa 100644 --- a/tests/matching/test_audio_verification_core.py +++ b/tests/matching/test_audio_verification_core.py @@ -72,3 +72,12 @@ def test_normalize_strips_version_and_featuring(): def test_normalize_keeps_plain_text(): assert normalize("Sawano Hiroyuki") == "sawano hiroyuki" + + +def test_empty_expected_artist_does_not_fail(): + # Old scanner treated a missing expected artist as artist-match=1.0 + # (compare title only). The unified core must not FAIL a track just + # because the DB has no artist value. + out = evaluate("Some Track", "", [_rec("Some Track", "Whoever")], + fingerprint_score=0.95) + assert out.decision == Decision.PASS diff --git a/tests/matching/test_verification_status.py b/tests/matching/test_verification_status.py new file mode 100644 index 00000000..d488710e --- /dev/null +++ b/tests/matching/test_verification_status.py @@ -0,0 +1,26 @@ +"""Verification-status vocabulary + mapping (DB column / file tag / UI badge).""" + +from core.matching.verification_status import ( + VERIFIED, UNVERIFIED, FORCE_IMPORTED, + status_from_acoustid_result, status_for_import, +) + + +def test_acoustid_result_maps_to_status(): + assert status_from_acoustid_result('pass') == VERIFIED + assert status_from_acoustid_result('skip') == UNVERIFIED + # disabled / error / unknown -> no claim either way + assert status_from_acoustid_result('disabled') is None + assert status_from_acoustid_result('error') is None + assert status_from_acoustid_result(None) is None + + +def test_force_import_context_wins_over_acoustid(): + ctx = {'_version_mismatch_fallback': 'instrumental', '_acoustid_result': 'pass'} + assert status_for_import(ctx) == FORCE_IMPORTED + + +def test_status_for_import_falls_back_to_acoustid_result(): + assert status_for_import({'_acoustid_result': 'pass'}) == VERIFIED + assert status_for_import({'_acoustid_result': 'skip'}) == UNVERIFIED + assert status_for_import({}) is None diff --git a/tests/test_verification_tag_roundtrip.py b/tests/test_verification_tag_roundtrip.py new file mode 100644 index 00000000..974dd7b0 --- /dev/null +++ b/tests/test_verification_tag_roundtrip.py @@ -0,0 +1,50 @@ +"""SOULSYNC_VERIFICATION file tag: write + read back (travels with the file, +survives DB resets; the AcoustID scan reads it to refresh the DB column).""" + +import shutil +import subprocess + +import pytest + +from core.tag_writer import read_file_tags, write_verification_status + + +pytestmark = pytest.mark.skipif( + shutil.which('ffmpeg') is None, reason='ffmpeg required to build test audio' +) + + +def _make_flac(path): + subprocess.run( + ['ffmpeg', '-loglevel', 'error', '-y', '-f', 'lavfi', + '-i', 'anullsrc=r=44100:cl=mono', '-t', '0.1', str(path)], + check=True, + ) + + +def test_flac_verification_tag_roundtrip(tmp_path): + f = tmp_path / 'x.flac' + _make_flac(f) + assert write_verification_status(str(f), 'force_imported') is True + tags = read_file_tags(str(f)) + assert tags.get('verification_status') == 'force_imported' + + +def test_overwrite_existing_status(tmp_path): + f = tmp_path / 'y.flac' + _make_flac(f) + write_verification_status(str(f), 'unverified') + write_verification_status(str(f), 'verified') + assert read_file_tags(str(f)).get('verification_status') == 'verified' + + +def test_missing_file_returns_false_not_raises(tmp_path): + assert write_verification_status(str(tmp_path / 'nope.flac'), 'verified') is False + + +def test_db_migration_adds_verification_status_column(tmp_path): + from database.music_database import MusicDatabase + db = MusicDatabase(str(tmp_path / 't.db')) + with db._get_connection() as conn: + cols = [r[1] for r in conn.execute('PRAGMA table_info(tracks)').fetchall()] + assert 'verification_status' in cols