#!/usr/bin/env python3 """ Audit beets MusicBrainz matches using stored Chromaprint fingerprints. Key behavior: - Read-only against the beets library. - Uses stored acoustid_fingerprint from beets. - Queries AcoustID only for tracks that need work. - Resumes interrupted runs. - Skips previously audited tracks unless: --force --rerun-problems --rerun-status STATUS[,STATUS] --rerun-score-below FLOAT - Writes a clean CSV snapshot incrementally during the run. - Opens suspicious rows in fzf after processing. Result validity: A previous result is reused only if these still match: item id fingerprint hash rounded duration beets mb_trackid beets mb_albumid min-score policy audit script version """ from __future__ import annotations import argparse import csv import datetime as _dt import hashlib import json import os import shlex import shutil import sqlite3 import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass from pathlib import Path from typing import Any AUDIT_VERSION = 2 ACOUSTID_LOOKUP_URL = "https://api.acoustid.org/v2/lookup" ITEM_FIELDS = [ "id", "album_id", "path", "albumartist", "artist", "album", "title", "disc", "track", "tracktotal", "disctotal", "length", "mb_trackid", "mb_albumid", "acoustid_id", "acoustid_fingerprint", ] CSV_FIELDS = [ "item_id", "album_id", "albumartist", "artist", "album", "title", "disc", "track", "tracktotal", "disctotal", "length", "duration_rounded", "beets_mb_trackid", "beets_mb_albumid", "beets_acoustid_id", "top_score", "top_acoustid_id", "top_recording_id", "top_recording_title", "top_artist_credit", "matched_recording_ids", "matched_release_ids", "candidate_summary", "status", "severity", "reason", "album_checked_tracks", "album_wrong_recording_tracks", "album_problem_tracks", "album_suspicion_pct", "album_verdict", "audit_version", "audited_at", "reused_from_state", "path", ] REVIEW_STATUSES = { "WRONG_RECORDING", "LOW_CONFIDENCE", "NO_RESULTS", "NO_MB_TRACKID", "RECORDING_OK_RELEASE_UNSEEN", "API_ERROR", "NO_FINGERPRINT", "NO_DURATION", } @dataclass class Item: id: int album_id: int | None path: str albumartist: str artist: str album: str title: str disc: int track: int tracktotal: int disctotal: int length: float mb_trackid: str mb_albumid: str acoustid_id: str acoustid_fingerprint: str def eprint(*args: object) -> None: print(*args, file=sys.stderr, flush=True) def now_iso() -> str: return _dt.datetime.now(_dt.UTC).replace(microsecond=0).isoformat() def decode_db_value(value: Any) -> Any: if value is None: return "" if isinstance(value, bytes): return os.fsdecode(value) return value def clean_text(value: Any) -> str: value = decode_db_value(value) if value is None: return "" return str(value) def clean_int(value: Any) -> int: value = decode_db_value(value) if value in ("", None): return 0 try: return int(value) except (TypeError, ValueError): return 0 def clean_float(value: Any) -> float: value = decode_db_value(value) if value in ("", None): return 0.0 try: return float(value) except (TypeError, ValueError): return 0.0 def parse_float_or_none(value: Any) -> float | None: value = clean_text(value).strip() if not value: return None try: return float(value) except ValueError: return None def sanitize_tsv(value: Any) -> str: text = clean_text(value) return text.replace("\t", " ").replace("\r", " ").replace("\n", " ").strip() def require_readable_file(path: str, label: str) -> str: path = os.path.abspath(os.path.expanduser(path)) if not os.path.isfile(path): raise SystemExit(f"{label} does not exist or is not a file: {path}") if not os.access(path, os.R_OK): raise SystemExit(f"{label} is not readable: {path}") return path def get_beets_library_path(args: argparse.Namespace) -> str: if args.library: return require_readable_file(args.library, "beets library") if os.environ.get("BEET_LIBRARY"): return require_readable_file(os.environ["BEET_LIBRARY"], "BEET_LIBRARY") try: from beets import config # type: ignore if args.config: config.set_file(args.config) library = os.fspath(config["library"].as_filename()) return require_readable_file(library, "beets configured library") except Exception as exc: raise SystemExit( "Could not determine beets library path automatically.\n" "Pass it explicitly with:\n" " --library /music/beets/musiclibrary.db\n\n" f"Auto-detection error: {exc}" ) def get_acoustid_client_key(args: argparse.Namespace) -> str: if args.client: return args.client.strip() for env_name in ("ACOUSTID_API_KEY", "ACOUSTID_CLIENT"): if os.environ.get(env_name): return os.environ[env_name].strip() try: from beets import config # type: ignore if args.config: config.set_file(args.config) key = config["acoustid"]["apikey"].as_str() if key: return key.strip() except Exception: pass raise SystemExit( "No AcoustID client key found.\n\n" "Provide one with one of these:\n" " --client YOUR_ACOUSTID_CLIENT_KEY\n" " ACOUSTID_API_KEY=YOUR_KEY\n" " acoustid.apikey in your beets config" ) def table_columns(conn: sqlite3.Connection, table: str) -> set[str]: rows = conn.execute(f"PRAGMA table_info({table})").fetchall() return {str(row[1]) for row in rows} def sql_select_expr(col: str, columns: set[str]) -> str: if col in columns: return f'"{col}"' return f"NULL AS \"{col}\"" def get_item_ids_from_beets_query(query_args: list[str]) -> set[int]: if not query_args: return set() beet = shutil.which("beet") if not beet: raise SystemExit("--beets-query requires the `beet` command to be available on PATH.") cmd = [beet, "ls", "-f", "$id", *query_args] eprint("Resolving beets query:", " ".join(shlex.quote(x) for x in cmd)) proc = subprocess.run( cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) if proc.returncode != 0: raise SystemExit( "beet query failed.\n\n" f"Command: {' '.join(shlex.quote(x) for x in cmd)}\n" f"stderr:\n{proc.stderr}" ) ids: set[int] = set() for line in proc.stdout.splitlines(): line = line.strip() if not line: continue try: ids.add(int(line)) except ValueError: eprint(f"WARNING: ignoring non-integer beets id from query output: {line!r}") return ids def load_items(db_path: str, wanted_ids: set[int], limit: int | None) -> list[Item]: conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row columns = table_columns(conn, "items") missing_core = {"id", "path"} - columns if missing_core: raise SystemExit(f"Not a usable beets items table. Missing columns: {sorted(missing_core)}") select_list = ", ".join(sql_select_expr(c, columns) for c in ITEM_FIELDS) where_parts: list[str] = [] params: list[Any] = [] if wanted_ids: placeholders = ",".join("?" for _ in wanted_ids) where_parts.append(f"id IN ({placeholders})") params.extend(sorted(wanted_ids)) where_sql = "" if where_parts: where_sql = "WHERE " + " AND ".join(where_parts) limit_sql = "" if limit is not None: limit_sql = "LIMIT ?" params.append(limit) sql = f""" SELECT {select_list} FROM items {where_sql} ORDER BY albumartist, album, disc, track, title, id {limit_sql} """ rows = conn.execute(sql, params).fetchall() conn.close() items: list[Item] = [] for row in rows: items.append( Item( id=clean_int(row["id"]), album_id=clean_int(row["album_id"]) or None, path=clean_text(row["path"]), albumartist=clean_text(row["albumartist"]), artist=clean_text(row["artist"]), album=clean_text(row["album"]), title=clean_text(row["title"]), disc=clean_int(row["disc"]), track=clean_int(row["track"]), tracktotal=clean_int(row["tracktotal"]), disctotal=clean_int(row["disctotal"]), length=clean_float(row["length"]), mb_trackid=clean_text(row["mb_trackid"]).strip(), mb_albumid=clean_text(row["mb_albumid"]).strip(), acoustid_id=clean_text(row["acoustid_id"]).strip(), acoustid_fingerprint=clean_text(row["acoustid_fingerprint"]).strip(), ) ) return items def sha1_text(text: str) -> str: return hashlib.sha1(text.encode("utf-8", "replace")).hexdigest() def sha256_json(obj: dict[str, Any]) -> str: payload = json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":")) return hashlib.sha256(payload.encode("utf-8", "replace")).hexdigest() def duration_rounded(item: Item) -> int: return int(round(item.length)) if item.length > 0 else 0 def item_signature(item: Item, min_score: float) -> str: return sha256_json( { "audit_version": AUDIT_VERSION, "item_id": item.id, "fingerprint_sha1": sha1_text(item.acoustid_fingerprint) if item.acoustid_fingerprint else "", "duration_rounded": duration_rounded(item), "mb_trackid": item.mb_trackid, "mb_albumid": item.mb_albumid, "min_score": f"{min_score:.6f}", } ) def lookup_cache_key(fingerprint: str, duration: int) -> str: return sha256_json( { "fingerprint_sha1": sha1_text(fingerprint), "duration": duration, } ) def init_state_db(state_path: str) -> sqlite3.Connection: conn = sqlite3.connect(state_path) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode = WAL") conn.execute("PRAGMA synchronous = NORMAL") conn.execute( """ CREATE TABLE IF NOT EXISTS lookup_cache ( cache_key TEXT PRIMARY KEY, fingerprint_sha1 TEXT NOT NULL, duration INTEGER NOT NULL, status TEXT NOT NULL, response_json TEXT, error TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS audit_results ( item_id INTEGER PRIMARY KEY, item_signature TEXT NOT NULL, audit_version INTEGER NOT NULL, min_score REAL NOT NULL, status TEXT NOT NULL, top_score REAL, result_json TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """ ) conn.execute( """ CREATE INDEX IF NOT EXISTS idx_audit_results_status ON audit_results(status) """ ) conn.execute( """ CREATE INDEX IF NOT EXISTS idx_audit_results_top_score ON audit_results(top_score) """ ) conn.commit() return conn def lookup_cache_get(conn: sqlite3.Connection, key: str) -> dict[str, Any] | None: row = conn.execute( """ SELECT status, response_json, error FROM lookup_cache WHERE cache_key = ? """, (key,), ).fetchone() if not row: return None if row["status"] == "ok" and row["response_json"]: try: return {"status": "ok", "data": json.loads(row["response_json"]), "error": ""} except json.JSONDecodeError: return None return {"status": row["status"], "data": None, "error": row["error"] or ""} def lookup_cache_put( conn: sqlite3.Connection, key: str, fingerprint: str, duration: int, status: str, data: dict[str, Any] | None, error: str, ) -> None: timestamp = now_iso() response_json = json.dumps(data, ensure_ascii=False, sort_keys=True) if data is not None else None conn.execute( """ INSERT INTO lookup_cache ( cache_key, fingerprint_sha1, duration, status, response_json, error, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(cache_key) DO UPDATE SET status = excluded.status, response_json = excluded.response_json, error = excluded.error, updated_at = excluded.updated_at """, ( key, sha1_text(fingerprint), duration, status, response_json, error, timestamp, timestamp, ), ) conn.commit() def audit_result_get(conn: sqlite3.Connection, item_id: int) -> dict[str, Any] | None: row = conn.execute( """ SELECT item_signature, status, top_score, result_json FROM audit_results WHERE item_id = ? """, (item_id,), ).fetchone() if not row: return None try: result = json.loads(row["result_json"]) except json.JSONDecodeError: return None return { "item_signature": row["item_signature"], "status": row["status"], "top_score": row["top_score"], "result": result, } def audit_result_put( conn: sqlite3.Connection, item: Item, sig: str, min_score: float, row: dict[str, Any], ) -> None: timestamp = now_iso() result_json = json.dumps(row, ensure_ascii=False, sort_keys=True) top_score = parse_float_or_none(row.get("top_score")) conn.execute( """ INSERT INTO audit_results ( item_id, item_signature, audit_version, min_score, status, top_score, result_json, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(item_id) DO UPDATE SET item_signature = excluded.item_signature, audit_version = excluded.audit_version, min_score = excluded.min_score, status = excluded.status, top_score = excluded.top_score, result_json = excluded.result_json, updated_at = excluded.updated_at """, ( item.id, sig, AUDIT_VERSION, min_score, row["status"], top_score, result_json, timestamp, timestamp, ), ) conn.commit() def acoustid_lookup( client_key: str, fingerprint: str, duration: int, timeout: float, retries: int, ) -> tuple[str, dict[str, Any] | None, str]: payload = { "client": client_key, "meta": "recordings releases releasegroups", "duration": str(duration), "fingerprint": fingerprint, } encoded = urllib.parse.urlencode(payload).encode("utf-8") for attempt in range(retries + 1): req = urllib.request.Request( ACOUSTID_LOOKUP_URL, data=encoded, method="POST", headers={ "Content-Type": "application/x-www-form-urlencoded", "User-Agent": f"beet-fingerprint-audit/{AUDIT_VERSION}", }, ) try: with urllib.request.urlopen(req, timeout=timeout) as response: raw = response.read().decode("utf-8", "replace") data = json.loads(raw) if data.get("status") == "ok": return "ok", data, "" return "api_error", data, f"AcoustID status={data.get('status')!r}; error={data.get('error')!r}" except urllib.error.HTTPError as exc: body = "" try: body = exc.read().decode("utf-8", "replace") except Exception: pass if exc.code in (429, 500, 502, 503, 504) and attempt < retries: time.sleep(2.0 * (attempt + 1)) continue return "http_error", None, f"HTTP {exc.code}: {body[:500]}" except urllib.error.URLError as exc: if attempt < retries: time.sleep(2.0 * (attempt + 1)) continue return "network_error", None, str(exc) except TimeoutError as exc: if attempt < retries: time.sleep(2.0 * (attempt + 1)) continue return "timeout", None, str(exc) except json.JSONDecodeError as exc: return "json_error", None, str(exc) return "unknown_error", None, "unexpected retry exhaustion" def base_row_for_item(item: Item) -> dict[str, Any]: return { "item_id": item.id, "album_id": item.album_id or "", "albumartist": item.albumartist, "artist": item.artist, "album": item.album, "title": item.title, "disc": item.disc, "track": item.track, "tracktotal": item.tracktotal, "disctotal": item.disctotal, "length": item.length, "duration_rounded": duration_rounded(item) or "", "beets_mb_trackid": item.mb_trackid, "beets_mb_albumid": item.mb_albumid, "beets_acoustid_id": item.acoustid_id, "top_score": "", "top_acoustid_id": "", "top_recording_id": "", "top_recording_title": "", "top_artist_credit": "", "matched_recording_ids": "", "matched_release_ids": "", "candidate_summary": "", "status": "", "severity": "", "reason": "", "album_checked_tracks": "", "album_wrong_recording_tracks": "", "album_problem_tracks": "", "album_suspicion_pct": "", "album_verdict": "", "audit_version": AUDIT_VERSION, "audited_at": now_iso(), "reused_from_state": "no", "path": item.path, } def refresh_current_metadata(row: dict[str, Any], item: Item) -> dict[str, Any]: """ Preserve old audit result fields, but refresh display metadata from current beets DB. This lets skipped/reused rows reflect harmless title/path/albumartist edits. """ updated = dict(row) updated.update( { "item_id": item.id, "album_id": item.album_id or "", "albumartist": item.albumartist, "artist": item.artist, "album": item.album, "title": item.title, "disc": item.disc, "track": item.track, "tracktotal": item.tracktotal, "disctotal": item.disctotal, "length": item.length, "duration_rounded": duration_rounded(item) or "", "beets_mb_trackid": item.mb_trackid, "beets_mb_albumid": item.mb_albumid, "beets_acoustid_id": item.acoustid_id, "audit_version": AUDIT_VERSION, "reused_from_state": "yes", "path": item.path, } ) return updated def extract_candidates(data: dict[str, Any], min_score: float) -> dict[str, Any]: results = data.get("results") or [] if not isinstance(results, list): results = [] all_recording_ids: set[str] = set() high_recording_ids: set[str] = set() all_release_ids: set[str] = set() high_release_ids: set[str] = set() candidate_summaries: list[str] = [] top_score = 0.0 top_acoustid_id = "" top_recording_id = "" top_recording_title = "" top_artist_credit = "" sorted_results = sorted( results, key=lambda r: float(r.get("score") or 0.0), reverse=True, ) for result in sorted_results: score = float(result.get("score") or 0.0) acoustid_id = clean_text(result.get("id")).strip() recordings = result.get("recordings") or [] if score > top_score: top_score = score top_acoustid_id = acoustid_id if not isinstance(recordings, list): continue for rec in recordings: if not isinstance(rec, dict): continue rec_id = clean_text(rec.get("id")).strip() rec_title = clean_text(rec.get("title")).strip() artists = rec.get("artists") or [] artist_credit = [] if isinstance(artists, list): for artist in artists: if isinstance(artist, dict): name = clean_text(artist.get("name")).strip() if name: artist_credit.append(name) artist_credit_s = ", ".join(artist_credit) if rec_id: all_recording_ids.add(rec_id) if score >= min_score: high_recording_ids.add(rec_id) releases = rec.get("releases") or [] if isinstance(releases, list): for rel in releases: if isinstance(rel, dict): rel_id = clean_text(rel.get("id")).strip() if rel_id: all_release_ids.add(rel_id) if score >= min_score: high_release_ids.add(rel_id) if not top_recording_id and rec_id: top_recording_id = rec_id top_recording_title = rec_title top_artist_credit = artist_credit_s if rec_id: summary = f"{score:.3f}:{rec_id}:{artist_credit_s} - {rec_title}" candidate_summaries.append(summary) return { "result_count": len(sorted_results), "top_score": top_score, "top_acoustid_id": top_acoustid_id, "top_recording_id": top_recording_id, "top_recording_title": top_recording_title, "top_artist_credit": top_artist_credit, "all_recording_ids": sorted(all_recording_ids), "high_recording_ids": sorted(high_recording_ids), "all_release_ids": sorted(all_release_ids), "high_release_ids": sorted(high_release_ids), "candidate_summary": " | ".join(candidate_summaries[:12]), } def classify_item( item: Item, lookup: dict[str, Any] | None, error_status: str, error: str, min_score: float, ) -> dict[str, Any]: row = base_row_for_item(item) if not item.acoustid_fingerprint: row.update( status="NO_FINGERPRINT", severity="skip", reason="No acoustid_fingerprint in beets DB.", ) return row if item.length <= 0: row.update( status="NO_DURATION", severity="skip", reason="No usable track length in beets DB.", ) return row if error_status != "ok": row.update( status="API_ERROR", severity="error", reason=f"{error_status}: {error}", ) return row if not lookup: row.update( status="API_ERROR", severity="error", reason="Lookup returned no data.", ) return row candidates = extract_candidates(lookup, min_score=min_score) row.update( top_score=f"{candidates['top_score']:.3f}" if candidates["top_score"] else "", top_acoustid_id=candidates["top_acoustid_id"], top_recording_id=candidates["top_recording_id"], top_recording_title=candidates["top_recording_title"], top_artist_credit=candidates["top_artist_credit"], matched_recording_ids=";".join(candidates["high_recording_ids"]), matched_release_ids=";".join(candidates["high_release_ids"]), candidate_summary=candidates["candidate_summary"], ) if candidates["result_count"] == 0: row.update( status="NO_RESULTS", severity="error", reason="AcoustID returned no candidates.", ) return row if candidates["top_score"] < min_score: row.update( status="LOW_CONFIDENCE", severity="warning", reason=f"Best AcoustID score {candidates['top_score']:.3f} is below threshold {min_score:.3f}.", ) return row if not item.mb_trackid: row.update( status="NO_MB_TRACKID", severity="warning", reason="Track has no beets mb_trackid, so fingerprint cannot validate the stored MB recording.", ) return row if item.mb_trackid in candidates["high_recording_ids"]: if item.mb_albumid and item.mb_albumid not in candidates["high_release_ids"]: row.update( status="RECORDING_OK_RELEASE_UNSEEN", severity="notice", reason=( "Fingerprint validates the MB recording, but the beets mb_albumid was not present " "among AcoustID release candidates. Treat as edition-review hint, not proof of a bad match." ), ) return row row.update( status="OK", severity="ok", reason="Fingerprint resolves to the same MB recording stored in beets.", ) return row row.update( status="WRONG_RECORDING", severity="error", reason="Fingerprint did not resolve to the beets mb_trackid above the configured score threshold.", ) return row def parse_status_set(raw_values: list[str]) -> set[str]: statuses: set[str] = set() for raw in raw_values: for part in raw.split(","): part = part.strip().upper() if part: statuses.add(part) return statuses def should_rerun_existing( existing: dict[str, Any], args: argparse.Namespace, rerun_statuses: set[str], ) -> tuple[bool, str]: result = existing["result"] status = clean_text(result.get("status")).upper() top_score = parse_float_or_none(result.get("top_score")) if args.force: return True, "--force" if args.rerun_problems and status != "OK": return True, "--rerun-problems" if rerun_statuses and status in rerun_statuses: return True, f"--rerun-status matched {status}" if args.rerun_score_below is not None: if top_score is None: return True, "--rerun-score-below matched missing score" if top_score < args.rerun_score_below: return True, f"--rerun-score-below matched {top_score:.3f}" return False, "" def album_key(row: dict[str, Any]) -> tuple[str, str, str]: album_id = clean_text(row.get("album_id")).strip() if album_id: return ("id", album_id, "") return ( "text", clean_text(row.get("albumartist")).casefold(), clean_text(row.get("album")).casefold(), ) def add_album_verdicts( rows: list[dict[str, Any]], min_album_wrong: int, min_album_wrong_pct: float, ) -> None: groups: dict[tuple[str, str, str], list[dict[str, Any]]] = {} for row in rows: groups.setdefault(album_key(row), []).append(row) for group_rows in groups.values(): checked = 0 wrong = 0 problem = 0 for row in group_rows: status = row["status"] if status not in {"NO_FINGERPRINT", "NO_DURATION", "API_ERROR"}: checked += 1 if status == "WRONG_RECORDING": wrong += 1 if status != "OK": problem += 1 pct = (wrong / checked) if checked else 0.0 if checked == 0: verdict = "ALBUM_UNKNOWN" elif wrong >= min_album_wrong and pct >= min_album_wrong_pct: verdict = "ALBUM_SUSPICIOUS" elif wrong > 0: verdict = "ALBUM_REVIEW" elif problem > 0: verdict = "ALBUM_MINOR_REVIEW" else: verdict = "ALBUM_OK" for row in group_rows: row["album_checked_tracks"] = checked row["album_wrong_recording_tracks"] = wrong row["album_problem_tracks"] = problem row["album_suspicion_pct"] = f"{pct:.3f}" if checked else "" row["album_verdict"] = verdict def sorted_rows(rows_by_item_id: dict[int, dict[str, Any]]) -> list[dict[str, Any]]: rows = list(rows_by_item_id.values()) rows.sort( key=lambda r: ( clean_text(r.get("albumartist")).casefold(), clean_text(r.get("album")).casefold(), clean_int(r.get("disc")), clean_int(r.get("track")), clean_text(r.get("title")).casefold(), clean_int(r.get("item_id")), ) ) return rows def write_csv_snapshot( rows_by_item_id: dict[int, dict[str, Any]], out_path: str, min_album_wrong: int, min_album_wrong_pct: float, ) -> None: rows = sorted_rows(rows_by_item_id) add_album_verdicts(rows, min_album_wrong, min_album_wrong_pct) out = Path(out_path) out.parent.mkdir(parents=True, exist_ok=True) tmp = out.with_name(out.name + ".tmp") with open(tmp, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=CSV_FIELDS, extrasaction="ignore") writer.writeheader() for row in rows: writer.writerow(row) os.replace(tmp, out) def format_preview(row: dict[str, Any]) -> str: lines = [ f"Status: {row['status']}", f"Severity: {row['severity']}", f"Reason: {row['reason']}", "", f"Album verdict: {row['album_verdict']}", f"Album wrong: {row['album_wrong_recording_tracks']} / {row['album_checked_tracks']}", f"Album pct: {row['album_suspicion_pct']}", "", f"Item ID: {row['item_id']}", f"Album ID: {row['album_id']}", f"Album artist: {row['albumartist']}", f"Artist: {row['artist']}", f"Album: {row['album']}", f"Track: disc {row['disc']} track {row['track']}", f"Title: {row['title']}", f"Length: {row['length']}", "", "Beets MB recording:", f" {row['beets_mb_trackid']}", "", "Beets MB release:", f" {row['beets_mb_albumid']}", "", "Beets AcoustID:", f" {row['beets_acoustid_id']}", "", "Top AcoustID score:", f" {row['top_score']}", "", "Top candidate recording:", f" {row['top_recording_id']}", f" {row['top_artist_credit']} - {row['top_recording_title']}", "", "Matched recording IDs above threshold:", ] matched_recordings = clean_text(row.get("matched_recording_ids")) if matched_recordings: for rec_id in matched_recordings.split(";"): lines.append(f" {rec_id}") else: lines.append(" ") lines += ["", "Candidate summary:"] summary = clean_text(row.get("candidate_summary")) if summary: for part in summary.split(" | "): lines.append(f" {part}") else: lines.append(" ") lines += [ "", f"Audited at: {row.get('audited_at', '')}", f"Reused: {row.get('reused_from_state', '')}", "", "Path:", f" {row['path']}", ] return "\n".join(lines) + "\n" def run_fzf(rows: list[dict[str, Any]], out_csv: str, include_ok: bool) -> None: if not shutil.which("fzf"): eprint("fzf not found on PATH; skipping interactive review.") return if include_ok: review_rows = rows else: review_rows = [r for r in rows if r["status"] != "OK"] if not review_rows: eprint("No non-OK rows to review in fzf.") return selected_path = str(Path(out_csv).with_suffix(".selected.tsv")) with tempfile.TemporaryDirectory(prefix="beet-acoustid-audit-") as td: td_path = Path(td) lines: list[str] = [] for i, row in enumerate(review_rows): preview_path = td_path / f"{i:08d}.txt" preview_path.write_text(format_preview(row), encoding="utf-8") track_s = f"{clean_int(row['disc'])}.{clean_int(row['track']):02d}" score_s = sanitize_tsv(row["top_score"]) or "-" candidate_s = sanitize_tsv( f"{row['top_artist_credit']} - {row['top_recording_title']}" ).strip(" -") or "-" cols = [ str(preview_path), sanitize_tsv(row["status"]), sanitize_tsv(row["severity"]), sanitize_tsv(row["album_verdict"]), sanitize_tsv(row["albumartist"]), sanitize_tsv(row["album"]), track_s, sanitize_tsv(row["title"]), score_s, candidate_s, sanitize_tsv(row["path"]), ] lines.append("\t".join(cols)) input_text = "\n".join(lines) + "\n" cmd = [ "fzf", "--multi", "--delimiter=\t", "--with-nth=2..", "--preview=cat {1}", "--preview-window=down,60%,wrap", "--header=TAB selects multiple. ENTER writes selected rows to " + selected_path + ". Columns: status severity album_verdict albumartist album track title score top_candidate path", ] proc = subprocess.run( cmd, input=input_text, text=True, stdout=subprocess.PIPE, stderr=None, check=False, ) if proc.returncode == 130: eprint("fzf cancelled.") return if proc.returncode != 0: eprint(f"fzf exited with status {proc.returncode}.") return selected = [line for line in proc.stdout.splitlines() if line.strip()] if not selected: eprint("No rows selected in fzf.") return with open(selected_path, "w", encoding="utf-8", newline="") as f: for line in selected: parts = line.split("\t") f.write("\t".join(parts[1:]) + "\n") eprint(f"Selected rows written to: {selected_path}") def summarize(rows: list[dict[str, Any]], pending_count: int, reused_count: int, processed_count: int) -> None: status_counts: dict[str, int] = {} album_counts: dict[str, int] = {} seen_album_keys: set[tuple[str, str, str]] = set() for row in rows: status_counts[row["status"]] = status_counts.get(row["status"], 0) + 1 key = album_key(row) if key not in seen_album_keys: seen_album_keys.add(key) verdict = row["album_verdict"] album_counts[verdict] = album_counts.get(verdict, 0) + 1 eprint("") eprint("Run summary:") eprint(f" reused from previous audit: {reused_count}") eprint(f" processed this run: {processed_count}") eprint(f" still pending/not written: {pending_count}") eprint("") eprint("Track statuses:") for status, count in sorted(status_counts.items(), key=lambda kv: (-kv[1], kv[0])): eprint(f" {status}: {count}") eprint("") eprint("Album verdicts:") for verdict, count in sorted(album_counts.items(), key=lambda kv: (-kv[1], kv[0])): eprint(f" {verdict}: {count}") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Audit beets MB recording matches using stored Chromaprint fingerprints." ) parser.add_argument( "--library", help="Path to beets SQLite library. Example: /music/beets/musiclibrary.db", ) parser.add_argument( "--config", help="Optional beets config path, used only for auto-detecting library/API key.", ) parser.add_argument( "--client", help="AcoustID client key. Can also use ACOUSTID_API_KEY or acoustid.apikey in beets config.", ) parser.add_argument( "--out", default="beet-fingerprint-audit.csv", help="Output CSV path. Default: beet-fingerprint-audit.csv", ) parser.add_argument( "--state", default=".beet-fingerprint-audit-state.sqlite", help="State/cache SQLite DB. Default: .beet-fingerprint-audit-state.sqlite", ) parser.add_argument( "--force", action="store_true", help="Recompute audit results for all selected tracks. Still uses lookup cache unless --refresh-lookups is also used.", ) parser.add_argument( "--refresh-lookups", action="store_true", help="Ignore cached AcoustID API responses and query AcoustID again.", ) parser.add_argument( "--rerun-problems", action="store_true", help="Rerun previously audited tracks whose status is not OK.", ) parser.add_argument( "--rerun-status", action="append", default=[], help=( "Rerun previous rows with matching status. Can be repeated or comma-separated. " "Example: --rerun-status WRONG_RECORDING,LOW_CONFIDENCE" ), ) parser.add_argument( "--rerun-score-below", type=float, help=( "Rerun previous rows whose stored top_score is missing or below this value. " "Example: --rerun-score-below 0.90" ), ) parser.add_argument( "--min-score", type=float, default=0.80, help="Minimum AcoustID score to treat a candidate as usable. Default: 0.80", ) parser.add_argument( "--sleep", type=float, default=0.50, help="Delay between uncached AcoustID requests. Default: 0.50", ) parser.add_argument( "--timeout", type=float, default=30.0, help="HTTP timeout in seconds. Default: 30", ) parser.add_argument( "--retries", type=int, default=2, help="Retries for transient HTTP/network errors. Default: 2", ) parser.add_argument( "--limit", type=int, help="Limit number of beets items processed. Useful for testing.", ) parser.add_argument( "--beets-query", nargs=argparse.REMAINDER, help=( "Optional beets query to restrict items. Must be last argument. " "Example: --beets-query albumartist:Disturbed" ), ) parser.add_argument( "--album-min-wrong", type=int, default=2, help="Minimum wrong-recording tracks before album can be ALBUM_SUSPICIOUS. Default: 2", ) parser.add_argument( "--album-min-wrong-pct", type=float, default=0.30, help="Minimum wrong-recording fraction before album can be ALBUM_SUSPICIOUS. Default: 0.30", ) parser.add_argument( "--csv-flush-every", type=int, default=25, help=( "Refresh the clean CSV snapshot after this many newly processed tracks. " "Use 1 for maximum mid-run visibility. Default: 25" ), ) parser.add_argument( "--no-fzf", action="store_true", help="Do not launch fzf after writing CSV.", ) parser.add_argument( "--fzf-all", action="store_true", help="Show all rows in fzf, not just non-OK rows.", ) return parser.parse_args() def main() -> int: args = parse_args() db_path = get_beets_library_path(args) client_key = get_acoustid_client_key(args) out_path = os.path.abspath(os.path.expanduser(args.out)) state_path = os.path.abspath(os.path.expanduser(args.state)) wanted_ids = get_item_ids_from_beets_query(args.beets_query or []) items = load_items(db_path, wanted_ids=wanted_ids, limit=args.limit) if not items: eprint("No beets items found for the selected scope.") return 1 rerun_statuses = parse_status_set(args.rerun_status) state_conn = init_state_db(state_path) eprint(f"Beets library: {db_path}") eprint(f"Items loaded: {len(items)}") eprint(f"CSV output: {out_path}") eprint(f"State DB: {state_path}") eprint(f"Min score: {args.min_score:.3f}") eprint(f"Sleep: {args.sleep:.2f}s between uncached API requests") rows_by_item_id: dict[int, dict[str, Any]] = {} pending: list[tuple[Item, str, str]] = [] reused_count = 0 for item in items: sig = item_signature(item, args.min_score) existing = audit_result_get(state_conn, item.id) if existing and existing["item_signature"] == sig: rerun, reason = should_rerun_existing(existing, args, rerun_statuses) if not rerun: row = refresh_current_metadata(existing["result"], item) rows_by_item_id[item.id] = row reused_count += 1 continue pending.append((item, sig, reason)) else: reason = "new item or changed item signature" pending.append((item, sig, reason)) eprint("") eprint(f"Reusable previous results: {reused_count}") eprint(f"Pending audit work: {len(pending)}") if rows_by_item_id: write_csv_snapshot( rows_by_item_id, out_path, min_album_wrong=args.album_min_wrong, min_album_wrong_pct=args.album_min_wrong_pct, ) eprint(f"Initial CSV snapshot written with reused rows: {out_path}") processed_count = 0 uncached_requests = 0 interrupted = False try: for index, (item, sig, rerun_reason) in enumerate(pending, start=1): eprint( f"Processing {index}/{len(pending)} " f"id={item.id} " f"{item.albumartist} - {item.album} - {item.title} " f"({rerun_reason})" ) if not item.acoustid_fingerprint or item.length <= 0: row = classify_item( item, lookup=None, error_status="skipped", error="", min_score=args.min_score, ) audit_result_put(state_conn, item, sig, args.min_score, row) rows_by_item_id[item.id] = row processed_count += 1 else: duration = duration_rounded(item) lk_key = lookup_cache_key(item.acoustid_fingerprint, duration) cached = None if args.refresh_lookups else lookup_cache_get(state_conn, lk_key) if cached is not None: status = cached["status"] data = cached["data"] error = cached["error"] else: if uncached_requests > 0 and args.sleep > 0: time.sleep(args.sleep) status, data, error = acoustid_lookup( client_key=client_key, fingerprint=item.acoustid_fingerprint, duration=duration, timeout=args.timeout, retries=args.retries, ) lookup_cache_put( state_conn, key=lk_key, fingerprint=item.acoustid_fingerprint, duration=duration, status=status, data=data, error=error, ) uncached_requests += 1 row = classify_item( item, lookup=data, error_status=status, error=error, min_score=args.min_score, ) audit_result_put(state_conn, item, sig, args.min_score, row) rows_by_item_id[item.id] = row processed_count += 1 if args.csv_flush_every > 0 and processed_count % args.csv_flush_every == 0: write_csv_snapshot( rows_by_item_id, out_path, min_album_wrong=args.album_min_wrong, min_album_wrong_pct=args.album_min_wrong_pct, ) eprint(f"CSV snapshot refreshed: {out_path}") except KeyboardInterrupt: interrupted = True eprint("") eprint("Interrupted. Saving final CSV snapshot from completed rows...") finally: write_csv_snapshot( rows_by_item_id, out_path, min_album_wrong=args.album_min_wrong, min_album_wrong_pct=args.album_min_wrong_pct, ) state_conn.close() rows = sorted_rows(rows_by_item_id) add_album_verdicts(rows, args.album_min_wrong, args.album_min_wrong_pct) pending_count = len(items) - len(rows_by_item_id) summarize( rows, pending_count=pending_count, reused_count=reused_count, processed_count=processed_count, ) eprint("") eprint(f"Wrote CSV snapshot: {out_path}") if interrupted: eprint("Run was interrupted. Re-run the same command to resume.") return 130 if not args.no_fzf: run_fzf(rows, out_csv=out_path, include_ok=args.fzf_all) return 0 if __name__ == "__main__": raise SystemExit(main())