1149 lines
34 KiB
Python
Executable File
1149 lines
34 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Audit beets MusicBrainz matches using existing Chromaprint/AcoustID fingerprints.
|
|
|
|
What it checks:
|
|
- For each beets item with acoustid_fingerprint + mb_trackid:
|
|
ask AcoustID what MusicBrainz recordings the audio fingerprint resolves to.
|
|
- If the beets mb_trackid is not among the returned recording IDs, flag it.
|
|
- Group results by album and compute an album-level suspicion score.
|
|
- Write full CSV.
|
|
- Optionally launch fzf over suspicious rows.
|
|
|
|
This script is read-only against the beets library.
|
|
"""
|
|
|
|
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
|
|
|
|
|
|
ACOUSTID_LOOKUP_URL = "https://api.acoustid.org/v2/lookup"
|
|
|
|
|
|
DEFAULT_FIELDS = [
|
|
"id",
|
|
"album_id",
|
|
"path",
|
|
"albumartist",
|
|
"artist",
|
|
"album",
|
|
"title",
|
|
"disc",
|
|
"track",
|
|
"tracktotal",
|
|
"disctotal",
|
|
"length",
|
|
"mb_trackid",
|
|
"mb_albumid",
|
|
"acoustid_id",
|
|
"acoustid_fingerprint",
|
|
]
|
|
|
|
|
|
@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)
|
|
|
|
|
|
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 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))
|
|
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
except OSError as exc:
|
|
raise SystemExit(f"Failed to run beet: {exc}")
|
|
|
|
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 DEFAULT_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 init_cache(cache_path: str) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(cache_path)
|
|
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.commit()
|
|
return conn
|
|
|
|
|
|
def cache_key(fingerprint: str, duration: int) -> str:
|
|
material = f"{duration}\0{fingerprint}".encode("utf-8", "replace")
|
|
return hashlib.sha256(material).hexdigest()
|
|
|
|
|
|
def fingerprint_sha1(fingerprint: str) -> str:
|
|
return hashlib.sha1(fingerprint.encode("utf-8", "replace")).hexdigest()
|
|
|
|
|
|
def 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
|
|
|
|
status, response_json, error = row
|
|
if status == "ok" and response_json:
|
|
try:
|
|
return {"status": "ok", "data": json.loads(response_json), "error": ""}
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
return {"status": status, "data": None, "error": error or ""}
|
|
|
|
|
|
def cache_put(
|
|
conn: sqlite3.Connection,
|
|
key: str,
|
|
fp: 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,
|
|
fingerprint_sha1(fp),
|
|
duration,
|
|
status,
|
|
response_json,
|
|
error,
|
|
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": "beet-fingerprint-audit/1.0",
|
|
},
|
|
)
|
|
|
|
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 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]:
|
|
base = {
|
|
"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": int(round(item.length)) if item.length > 0 else "",
|
|
"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": "",
|
|
"path": item.path,
|
|
}
|
|
|
|
if not item.acoustid_fingerprint:
|
|
base.update(status="NO_FINGERPRINT", severity="skip", reason="No acoustid_fingerprint in beets DB.")
|
|
return base
|
|
|
|
if item.length <= 0:
|
|
base.update(status="NO_DURATION", severity="skip", reason="No usable track length in beets DB.")
|
|
return base
|
|
|
|
if error_status != "ok":
|
|
base.update(status="API_ERROR", severity="error", reason=f"{error_status}: {error}")
|
|
return base
|
|
|
|
if not lookup:
|
|
base.update(status="API_ERROR", severity="error", reason="Lookup returned no data.")
|
|
return base
|
|
|
|
candidates = extract_candidates(lookup, min_score=min_score)
|
|
|
|
base.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:
|
|
base.update(status="NO_RESULTS", severity="error", reason="AcoustID returned no candidates.")
|
|
return base
|
|
|
|
if candidates["top_score"] < min_score:
|
|
base.update(
|
|
status="LOW_CONFIDENCE",
|
|
severity="warning",
|
|
reason=f"Best AcoustID score {candidates['top_score']:.3f} is below threshold {min_score:.3f}.",
|
|
)
|
|
return base
|
|
|
|
if not item.mb_trackid:
|
|
base.update(
|
|
status="NO_MB_TRACKID",
|
|
severity="warning",
|
|
reason="Track has no beets mb_trackid, so fingerprint cannot validate the stored MB recording.",
|
|
)
|
|
return base
|
|
|
|
if item.mb_trackid in candidates["high_recording_ids"]:
|
|
if item.mb_albumid and item.mb_albumid not in candidates["high_release_ids"]:
|
|
base.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. This is often harmless; use it as an edition-review hint."
|
|
),
|
|
)
|
|
return base
|
|
|
|
base.update(
|
|
status="OK",
|
|
severity="ok",
|
|
reason="Fingerprint resolves to the same MB recording stored in beets.",
|
|
)
|
|
return base
|
|
|
|
base.update(
|
|
status="WRONG_RECORDING",
|
|
severity="error",
|
|
reason="Fingerprint did not resolve to the beets mb_trackid above the configured score threshold.",
|
|
)
|
|
return base
|
|
|
|
|
|
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 not in {"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 write_csv(rows: list[dict[str, Any]], out_path: str) -> None:
|
|
fieldnames = [
|
|
"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",
|
|
"path",
|
|
]
|
|
|
|
with open(out_path, "w", newline="", encoding="utf-8") as f:
|
|
writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore")
|
|
writer.writeheader()
|
|
for row in rows:
|
|
writer.writerow(row)
|
|
|
|
|
|
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']}",
|
|
"",
|
|
f"Beets MB recording:",
|
|
f" {row['beets_mb_trackid']}",
|
|
"",
|
|
f"Beets MB release:",
|
|
f" {row['beets_mb_albumid']}",
|
|
"",
|
|
f"Beets AcoustID:",
|
|
f" {row['beets_acoustid_id']}",
|
|
"",
|
|
f"Top AcoustID score:",
|
|
f" {row['top_score']}",
|
|
"",
|
|
f"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(" <none>")
|
|
|
|
lines += [
|
|
"",
|
|
"Candidate summary:",
|
|
]
|
|
|
|
summary = clean_text(row.get("candidate_summary"))
|
|
if summary:
|
|
for part in summary.split(" | "):
|
|
lines.append(f" {part}")
|
|
else:
|
|
lines.append(" <none>")
|
|
|
|
lines += [
|
|
"",
|
|
"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(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")
|
|
# Drop hidden preview path.
|
|
f.write("\t".join(parts[1:]) + "\n")
|
|
|
|
eprint(f"Selected rows written to: {selected_path}")
|
|
|
|
|
|
def summarize(rows: list[dict[str, Any]]) -> 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("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(
|
|
"--cache",
|
|
default=".beet-fingerprint-audit-cache.sqlite",
|
|
help="Lookup cache DB. Default: .beet-fingerprint-audit-cache.sqlite",
|
|
)
|
|
parser.add_argument(
|
|
"--refresh-cache",
|
|
action="store_true",
|
|
help="Ignore cached AcoustID responses and query again.",
|
|
)
|
|
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.40,
|
|
help="Delay between uncached AcoustID requests. Default: 0.35",
|
|
)
|
|
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(
|
|
"--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))
|
|
cache_path = os.path.abspath(os.path.expanduser(args.cache))
|
|
|
|
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
|
|
|
|
eprint(f"Beets library: {db_path}")
|
|
eprint(f"Items loaded: {len(items)}")
|
|
eprint(f"CSV output: {out_path}")
|
|
eprint(f"Cache DB: {cache_path}")
|
|
|
|
cache_conn = init_cache(cache_path)
|
|
|
|
rows: list[dict[str, Any]] = []
|
|
uncached_requests = 0
|
|
|
|
for index, item in enumerate(items, start=1):
|
|
if index == 1 or index % 25 == 0 or index == len(items):
|
|
eprint(f"Processing {index}/{len(items)}...")
|
|
|
|
if not item.acoustid_fingerprint or item.length <= 0:
|
|
row = classify_item(item, lookup=None, error_status="skipped", error="", min_score=args.min_score)
|
|
rows.append(row)
|
|
continue
|
|
|
|
duration = int(round(item.length))
|
|
key = cache_key(item.acoustid_fingerprint, duration)
|
|
|
|
cached = None if args.refresh_cache else cache_get(cache_conn, 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,
|
|
)
|
|
cache_put(
|
|
cache_conn,
|
|
key=key,
|
|
fp=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,
|
|
)
|
|
rows.append(row)
|
|
|
|
cache_conn.close()
|
|
|
|
add_album_verdicts(
|
|
rows,
|
|
min_album_wrong=args.album_min_wrong,
|
|
min_album_wrong_pct=args.album_min_wrong_pct,
|
|
)
|
|
|
|
write_csv(rows, out_path)
|
|
summarize(rows)
|
|
|
|
eprint("")
|
|
eprint(f"Wrote CSV: {out_path}")
|
|
|
|
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())
|