1582 lines
46 KiB
Python
Executable File
1582 lines
46 KiB
Python
Executable File
#!/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(" <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 += [
|
|
"",
|
|
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())
|