feat(verification): status vocabulary, DB column, SOULSYNC_VERIFICATION tag

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 <noreply@anthropic.com>

core/acoustid_verification.py

@@ -22,10 +22,11 @@ from core.musicbrainz_client import MusicBrainzClient
 
 logger = get_logger("acoustid.verification")
 
-# Thresholds
+# Thresholds — single definition lives in the shared core; re-exported here so
-MIN_ACOUSTID_SCORE = 0.80       # Minimum AcoustID fingerprint score to trust
+# existing importers keep working and the values can't drift between paths.
-TITLE_MATCH_THRESHOLD = 0.70    # Title similarity needed to consider a match
+from core.matching.audio_verification import (  # noqa: E402
-ARTIST_MATCH_THRESHOLD = 0.60   # Artist similarity needed to consider a match
+    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 /

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

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'))

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.
 

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

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

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

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