1169 lines
30 KiB
Python
1169 lines
30 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
album_coverage.py
|
|
|
|
Compact terminal-friendly album deletion/coverage checker for beets.
|
|
|
|
Default output:
|
|
- compact summary
|
|
- album list
|
|
- relationship map showing candidate track -> best match/reason
|
|
- hides exact-covered tracks unless --show-covered is used
|
|
- avoids long paths in terminal output
|
|
|
|
Verbose/detail output:
|
|
--details shows compact candidate/match blocks
|
|
--verbose shows forensic match categories
|
|
--paths includes relative paths in details/verbose output
|
|
|
|
Layout assumption:
|
|
/music/
|
|
beets/
|
|
audit/
|
|
album_coverage.py
|
|
album_coverage.sh
|
|
library/
|
|
|
|
Audit output defaults to /music/beets/audit.
|
|
Displayed paths, when enabled, are relative to /music.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import unicodedata
|
|
from collections import Counter
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
|
|
|
|
US = "\x1f"
|
|
DIVIDER = "-" * 100
|
|
|
|
LOSSLESS_FORMATS = {
|
|
"FLAC",
|
|
"ALAC",
|
|
"WAV",
|
|
"AIFF",
|
|
"AIF",
|
|
"APE",
|
|
"WavPack",
|
|
"WV",
|
|
"DSF",
|
|
"DFF",
|
|
}
|
|
|
|
TRACK_FIELDS = [
|
|
"album_id",
|
|
"id",
|
|
"albumartist",
|
|
"album",
|
|
"artist",
|
|
"title",
|
|
"disc",
|
|
"track",
|
|
"tracktotal",
|
|
"format",
|
|
"bitdepth",
|
|
"samplerate",
|
|
"bitrate",
|
|
"path",
|
|
]
|
|
|
|
ALBUM_FIELDS = [
|
|
"id",
|
|
"albumartist",
|
|
"album",
|
|
"year",
|
|
"albumtype",
|
|
"albumtotal",
|
|
"mb_albumid",
|
|
"mb_releasegroupid",
|
|
"albumdisambig",
|
|
"label",
|
|
"catalognum",
|
|
"country",
|
|
"media",
|
|
"path",
|
|
]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Album:
|
|
id: str
|
|
albumartist: str
|
|
album: str
|
|
year: str
|
|
albumtype: str
|
|
albumtotal: str
|
|
mb_albumid: str
|
|
mb_releasegroupid: str
|
|
albumdisambig: str
|
|
label: str
|
|
catalognum: str
|
|
country: str
|
|
media: str
|
|
path: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Track:
|
|
album_id: str
|
|
id: str
|
|
albumartist: str
|
|
album: str
|
|
artist: str
|
|
title: str
|
|
disc: str
|
|
track: str
|
|
tracktotal: str
|
|
format: str
|
|
bitdepth: str
|
|
samplerate: str
|
|
bitrate: str
|
|
path: str
|
|
|
|
@property
|
|
def exact_title_key(self) -> str:
|
|
return normalize_title_exact(self.title)
|
|
|
|
@property
|
|
def loose_title_key(self) -> str:
|
|
return normalize_title_loose(self.title)
|
|
|
|
@property
|
|
def artist_keys(self) -> set[str]:
|
|
return artist_keys(self.artist, self.albumartist)
|
|
|
|
|
|
@dataclass
|
|
class TrackCoverage:
|
|
candidate: Track
|
|
status: str
|
|
exact_related: list[Track]
|
|
exact_other: list[Track]
|
|
loose_related: list[Track]
|
|
loose_other: list[Track]
|
|
|
|
@property
|
|
def exact_match_count(self) -> int:
|
|
return len(self.exact_related) + len(self.exact_other)
|
|
|
|
@property
|
|
def loose_match_count(self) -> int:
|
|
return len(self.loose_related) + len(self.loose_other)
|
|
|
|
@property
|
|
def is_exactly_covered(self) -> bool:
|
|
return bool(self.exact_related)
|
|
|
|
@property
|
|
def is_attention_needed(self) -> bool:
|
|
return self.status != "EXACT_RELATED"
|
|
|
|
|
|
def default_paths() -> tuple[Path, Path, Path]:
|
|
script_path = Path(__file__).resolve()
|
|
audit_dir = script_path.parent
|
|
beets_dir = audit_dir.parent
|
|
music_root = beets_dir.parent
|
|
library_dir = music_root / "library"
|
|
return audit_dir, music_root, library_dir
|
|
|
|
|
|
def terminal_width(cli_width: int | None) -> int:
|
|
if cli_width:
|
|
return max(72, cli_width)
|
|
|
|
detected = shutil.get_terminal_size(fallback=(100, 24)).columns
|
|
return max(72, min(detected, 120))
|
|
|
|
|
|
def ellipsize(value: str, width: int) -> str:
|
|
value = value or ""
|
|
if width <= 0:
|
|
return ""
|
|
if len(value) <= width:
|
|
return value
|
|
if width == 1:
|
|
return "…"
|
|
return value[: width - 1] + "…"
|
|
|
|
|
|
def relpath(path_value: str, music_root: Path) -> str:
|
|
if not path_value:
|
|
return ""
|
|
|
|
path = Path(path_value)
|
|
|
|
if not path.is_absolute():
|
|
return path_value
|
|
|
|
try:
|
|
return str(path.relative_to(music_root))
|
|
except ValueError:
|
|
return path_value
|
|
|
|
|
|
def run_beet_ls(
|
|
fields: list[str],
|
|
query: list[str] | None = None,
|
|
*,
|
|
album_mode: bool = False,
|
|
) -> list[dict[str, str]]:
|
|
query = query or []
|
|
fmt = US.join(f"${field}" for field in fields)
|
|
|
|
cmd = ["beet", "ls"]
|
|
|
|
if album_mode:
|
|
cmd.append("-a")
|
|
|
|
cmd.extend(["-f", fmt])
|
|
cmd.extend(query)
|
|
|
|
try:
|
|
proc = subprocess.run(
|
|
cmd,
|
|
check=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
except FileNotFoundError:
|
|
print("ERROR: could not find 'beet' on PATH.", file=sys.stderr)
|
|
raise SystemExit(2)
|
|
except subprocess.CalledProcessError as exc:
|
|
print("ERROR: beet command failed:", file=sys.stderr)
|
|
print(" ".join(cmd), file=sys.stderr)
|
|
if exc.stderr:
|
|
print(exc.stderr, file=sys.stderr)
|
|
raise SystemExit(exc.returncode)
|
|
|
|
rows: list[dict[str, str]] = []
|
|
|
|
for line_number, line in enumerate(proc.stdout.splitlines(), start=1):
|
|
if not line.strip():
|
|
continue
|
|
|
|
parts = line.split(US)
|
|
|
|
if len(parts) != len(fields):
|
|
print(
|
|
f"ERROR: could not parse beet output line {line_number}. "
|
|
f"Expected {len(fields)} fields, got {len(parts)}.",
|
|
file=sys.stderr,
|
|
)
|
|
print("Raw line:", repr(line), file=sys.stderr)
|
|
print("Command:", " ".join(cmd), file=sys.stderr)
|
|
raise SystemExit(2)
|
|
|
|
rows.append(dict(zip(fields, parts)))
|
|
|
|
return rows
|
|
|
|
|
|
def load_album(album_id: str) -> Album:
|
|
rows = run_beet_ls(ALBUM_FIELDS, [f"id:{album_id}"], album_mode=True)
|
|
|
|
if not rows:
|
|
raise SystemExit(f"ERROR: no album found for album id {album_id!r}")
|
|
|
|
return Album(**rows[0])
|
|
|
|
|
|
def load_tracks_for_album(album_id: str) -> list[Track]:
|
|
rows = run_beet_ls(TRACK_FIELDS, [f"album_id:{album_id}"], album_mode=False)
|
|
return [Track(**row) for row in rows]
|
|
|
|
|
|
def load_all_tracks() -> list[Track]:
|
|
rows = run_beet_ls(TRACK_FIELDS, [], album_mode=False)
|
|
return [Track(**row) for row in rows]
|
|
|
|
|
|
def normalize_common(value: str) -> str:
|
|
value = unicodedata.normalize("NFKC", value or "")
|
|
|
|
replacements = {
|
|
"\u2018": "'",
|
|
"\u2019": "'",
|
|
"\u201a": "'",
|
|
"\u201b": "'",
|
|
"\u2032": "'",
|
|
"\u2035": "'",
|
|
"\u201c": '"',
|
|
"\u201d": '"',
|
|
"\u201e": '"',
|
|
"\u2033": '"',
|
|
"\u2036": '"',
|
|
"\u2010": "-",
|
|
"\u2011": "-",
|
|
"\u2012": "-",
|
|
"\u2013": "-",
|
|
"\u2014": "-",
|
|
"\u2015": "-",
|
|
"\u2212": "-",
|
|
"\u2026": "...",
|
|
"\u00a0": " ",
|
|
}
|
|
|
|
for old, new in replacements.items():
|
|
value = value.replace(old, new)
|
|
|
|
value = value.casefold()
|
|
value = re.sub(r"\s+", " ", value)
|
|
return value.strip()
|
|
|
|
|
|
def normalize_censored_words(value: str) -> str:
|
|
value = re.sub(r"\bf[\W_]*\*[\W_]*\*[\W_]*k\b", "fuck", value)
|
|
value = re.sub(r"\bf[\W_]*_[\W_]*_[\W_]*k\b", "fuck", value)
|
|
value = re.sub(r"\bf[\W_]*x[\W_]*x[\W_]*k\b", "fuck", value)
|
|
return value
|
|
|
|
|
|
def normalize_title_exact(title: str) -> str:
|
|
title = normalize_common(title)
|
|
title = normalize_censored_words(title)
|
|
return title
|
|
|
|
|
|
def normalize_title_loose(title: str) -> str:
|
|
title = normalize_title_exact(title)
|
|
|
|
previous = None
|
|
while previous != title:
|
|
previous = title
|
|
title = re.sub(r"\s*[\(\[\{（【].*?[\)\]\}）】]\s*", " ", title)
|
|
|
|
title = re.sub(
|
|
r"\s+-\s+(remaster(?:ed)?|radio edit|single version|album version|edit|remix).*$",
|
|
" ",
|
|
title,
|
|
)
|
|
|
|
title = title.replace("_", " ")
|
|
title = re.sub(r"[^\w\s']", " ", title, flags=re.UNICODE)
|
|
title = re.sub(r"\s+", " ", title)
|
|
return title.strip()
|
|
|
|
|
|
def artist_name_common(value: str) -> str:
|
|
value = normalize_common(value)
|
|
value = value.replace("&", " and ")
|
|
value = re.sub(r"\b(featuring|feat\.?|ft\.?)\b", " feat ", value)
|
|
value = re.sub(r"\bwith\b", " with ", value)
|
|
value = re.sub(r"\s+", " ", value)
|
|
return value.strip()
|
|
|
|
|
|
def artist_keys(*names: str) -> set[str]:
|
|
keys: set[str] = set()
|
|
|
|
for raw in names:
|
|
if not raw:
|
|
continue
|
|
|
|
cleaned = artist_name_common(raw)
|
|
|
|
if cleaned:
|
|
keys.add(cleaned)
|
|
|
|
split_ready = cleaned
|
|
split_ready = re.sub(r"\b(feat|with|and|x)\b", ",", split_ready)
|
|
split_ready = re.sub(r"[,/;+|]", ",", split_ready)
|
|
|
|
for part in split_ready.split(","):
|
|
part = part.strip()
|
|
part = re.sub(r"\s+", " ", part)
|
|
|
|
if not part:
|
|
continue
|
|
|
|
if len(part) < 2:
|
|
continue
|
|
|
|
keys.add(part)
|
|
|
|
return keys
|
|
|
|
|
|
def tracks_are_related(candidate: Track, other: Track) -> bool:
|
|
candidate_keys = candidate.artist_keys
|
|
other_keys = other.artist_keys
|
|
|
|
if candidate_keys & other_keys:
|
|
return True
|
|
|
|
for c in candidate_keys:
|
|
for o in other_keys:
|
|
if len(c) >= 5 and len(o) >= 5 and (c in o or o in c):
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def parse_intish(value: str) -> int:
|
|
if not value:
|
|
return 0
|
|
|
|
match = re.search(r"\d+", value)
|
|
|
|
if not match:
|
|
return 0
|
|
|
|
return int(match.group(0))
|
|
|
|
|
|
def parse_samplerate_hz(value: str) -> int:
|
|
value = (value or "").strip()
|
|
|
|
if not value:
|
|
return 0
|
|
|
|
match = re.search(r"(\d+(?:\.\d+)?)\s*k\s*hz", value, flags=re.I)
|
|
|
|
if match:
|
|
return int(float(match.group(1)) * 1000)
|
|
|
|
match = re.search(r"(\d+)", value)
|
|
|
|
if match:
|
|
number = int(match.group(1))
|
|
if number < 1000:
|
|
return number * 1000
|
|
return number
|
|
|
|
return 0
|
|
|
|
|
|
def parse_bitrate_kbps(value: str) -> int:
|
|
value = (value or "").strip()
|
|
|
|
if not value:
|
|
return 0
|
|
|
|
match = re.search(r"(\d+(?:\.\d+)?)\s*kbps", value, flags=re.I)
|
|
|
|
if match:
|
|
return int(float(match.group(1)))
|
|
|
|
match = re.search(r"(\d+)", value)
|
|
|
|
if match:
|
|
return int(match.group(1))
|
|
|
|
return 0
|
|
|
|
|
|
def is_lossless(track: Track) -> bool:
|
|
return track.format in LOSSLESS_FORMATS
|
|
|
|
|
|
def quality_score(track: Track) -> tuple[int, int, int, int]:
|
|
return (
|
|
1 if is_lossless(track) else 0,
|
|
parse_intish(track.bitdepth),
|
|
parse_samplerate_hz(track.samplerate),
|
|
parse_bitrate_kbps(track.bitrate),
|
|
)
|
|
|
|
|
|
def quality_short(track: Track) -> str:
|
|
fmt = track.format or "?"
|
|
bitdepth = parse_intish(track.bitdepth)
|
|
samplerate = parse_samplerate_hz(track.samplerate)
|
|
bitrate = parse_bitrate_kbps(track.bitrate)
|
|
|
|
parts = [fmt]
|
|
|
|
if bitdepth > 0:
|
|
parts.append(f"{bitdepth}b")
|
|
|
|
if samplerate > 0:
|
|
parts.append(f"{round(samplerate / 1000)}kHz")
|
|
|
|
if bitrate > 0:
|
|
parts.append(f"{bitrate}k")
|
|
|
|
return " ".join(parts)
|
|
|
|
|
|
def natural_track_number(value: str) -> int:
|
|
return parse_intish(value)
|
|
|
|
|
|
def track_position(track: Track) -> str:
|
|
disc = parse_intish(track.disc)
|
|
number = parse_intish(track.track)
|
|
|
|
if disc:
|
|
return f"{disc:02d}-{number:02d}" if number else f"{disc:02d}-??"
|
|
|
|
return f"{number:02d}" if number else "??"
|
|
|
|
|
|
def sort_matches(matches: Iterable[Track]) -> list[Track]:
|
|
return sorted(
|
|
matches,
|
|
key=lambda t: (
|
|
t.albumartist.casefold(),
|
|
t.album.casefold(),
|
|
-quality_score(t)[0],
|
|
-quality_score(t)[1],
|
|
-quality_score(t)[2],
|
|
-quality_score(t)[3],
|
|
natural_track_number(t.disc),
|
|
natural_track_number(t.track),
|
|
t.title.casefold(),
|
|
),
|
|
)
|
|
|
|
|
|
def classify_track(candidate: Track, outside_tracks: list[Track]) -> TrackCoverage:
|
|
exact_related: list[Track] = []
|
|
exact_other: list[Track] = []
|
|
loose_related: list[Track] = []
|
|
loose_other: list[Track] = []
|
|
|
|
candidate_exact = candidate.exact_title_key
|
|
candidate_loose = candidate.loose_title_key
|
|
|
|
for other in outside_tracks:
|
|
if not candidate_exact and not candidate_loose:
|
|
continue
|
|
|
|
related = tracks_are_related(candidate, other)
|
|
|
|
exact_match = candidate_exact and candidate_exact == other.exact_title_key
|
|
loose_match = candidate_loose and candidate_loose == other.loose_title_key
|
|
|
|
if exact_match:
|
|
if related:
|
|
exact_related.append(other)
|
|
else:
|
|
exact_other.append(other)
|
|
|
|
if loose_match and not exact_match:
|
|
if related:
|
|
loose_related.append(other)
|
|
else:
|
|
loose_other.append(other)
|
|
|
|
exact_related = sort_matches(exact_related)
|
|
exact_other = sort_matches(exact_other)
|
|
loose_related = sort_matches(loose_related)
|
|
loose_other = sort_matches(loose_other)
|
|
|
|
if exact_related:
|
|
status = "EXACT_RELATED"
|
|
elif loose_related:
|
|
status = "LOOSE_RELATED"
|
|
elif exact_other:
|
|
status = "EXACT_OTHER_ARTIST"
|
|
elif loose_other:
|
|
status = "LOOSE_OTHER_ARTIST"
|
|
else:
|
|
status = "UNCOVERED"
|
|
|
|
return TrackCoverage(
|
|
candidate=candidate,
|
|
status=status,
|
|
exact_related=exact_related,
|
|
exact_other=exact_other,
|
|
loose_related=loose_related,
|
|
loose_other=loose_other,
|
|
)
|
|
|
|
|
|
def status_short(status: str) -> str:
|
|
return {
|
|
"EXACT_RELATED": "EXACT",
|
|
"LOOSE_RELATED": "LOOSE",
|
|
"EXACT_OTHER_ARTIST": "OTHER",
|
|
"LOOSE_OTHER_ARTIST": "OTHER",
|
|
"UNCOVERED": "MISS",
|
|
}.get(status, status[:5])
|
|
|
|
|
|
def status_reason(status: str) -> str:
|
|
return {
|
|
"EXACT_RELATED": "exact title + related artist",
|
|
"LOOSE_RELATED": "base title + related artist",
|
|
"EXACT_OTHER_ARTIST": "same exact title, different artist",
|
|
"LOOSE_OTHER_ARTIST": "same base title, different artist",
|
|
"UNCOVERED": "no title match elsewhere",
|
|
}.get(status, "")
|
|
|
|
|
|
def best_match_for(coverage: TrackCoverage) -> Track | None:
|
|
if coverage.status == "EXACT_RELATED":
|
|
return coverage.exact_related[0] if coverage.exact_related else None
|
|
|
|
if coverage.status == "LOOSE_RELATED":
|
|
return coverage.loose_related[0] if coverage.loose_related else None
|
|
|
|
if coverage.status == "EXACT_OTHER_ARTIST":
|
|
return coverage.exact_other[0] if coverage.exact_other else None
|
|
|
|
if coverage.status == "LOOSE_OTHER_ARTIST":
|
|
return coverage.loose_other[0] if coverage.loose_other else None
|
|
|
|
return None
|
|
|
|
|
|
def candidate_ref(track: Track) -> str:
|
|
return f"A{track.album_id} {track_position(track)} {track.title}"
|
|
|
|
|
|
def match_ref(track: Track) -> str:
|
|
return (
|
|
f"A{track.album_id} {track_position(track)} "
|
|
f"{track.artist} - {track.title} [{track.album}]"
|
|
)
|
|
|
|
|
|
def album_line(album: Album, width: int, music_root: Path, include_paths: bool) -> str:
|
|
parts = []
|
|
|
|
if album.year:
|
|
parts.append(album.year)
|
|
|
|
if album.albumtype:
|
|
parts.append(album.albumtype)
|
|
|
|
if album.albumtotal:
|
|
parts.append(f"{album.albumtotal} tracks")
|
|
|
|
if album.media:
|
|
parts.append(album.media)
|
|
|
|
if album.country:
|
|
parts.append(album.country)
|
|
|
|
if album.albumdisambig:
|
|
parts.append(album.albumdisambig)
|
|
|
|
meta = " | ".join(parts)
|
|
base = f"A{album.id} {album.albumartist} - {album.album}"
|
|
|
|
if meta:
|
|
base += f" ({meta})"
|
|
|
|
if include_paths and album.path:
|
|
base += f" | {relpath(album.path, music_root)}"
|
|
|
|
return ellipsize(base, width)
|
|
|
|
|
|
def build_summary_lines(
|
|
album_ids: list[str],
|
|
albums: list[Album],
|
|
candidate_tracks: list[Track],
|
|
outside_tracks: list[Track],
|
|
coverages: list[TrackCoverage],
|
|
music_root: Path,
|
|
*,
|
|
width: int,
|
|
include_paths: bool,
|
|
) -> list[str]:
|
|
counts = Counter(c.status for c in coverages)
|
|
total = len(candidate_tracks)
|
|
|
|
exact = counts["EXACT_RELATED"]
|
|
loose = counts["LOOSE_RELATED"]
|
|
other = counts["EXACT_OTHER_ARTIST"] + counts["LOOSE_OTHER_ARTIST"]
|
|
miss = counts["UNCOVERED"]
|
|
attention = total - exact
|
|
|
|
safe_delete = "YES" if attention == 0 else "NO"
|
|
|
|
lines: list[str] = []
|
|
|
|
lines.append("SUMMARY")
|
|
lines.append(f" albums: {' '.join(album_ids)}")
|
|
lines.append(f" tracks: {total}")
|
|
lines.append(f" searched elsewhere: {len(outside_tracks)} tracks")
|
|
lines.append(
|
|
f" exact={exact} loose={loose} other-artist={other} missing={miss}"
|
|
)
|
|
lines.append(f" safe-delete-by-exact-coverage: {safe_delete} ({attention} need review)")
|
|
lines.append("")
|
|
lines.append("LEGEND")
|
|
lines.append(" EXACT = same title, related artist; usually safe duplicate coverage")
|
|
lines.append(" LOOSE = same base title, related artist; review version/remix/edit")
|
|
lines.append(" OTHER = same title, unrelated artist; not useful deletion coverage")
|
|
lines.append(" MISS = no title match elsewhere")
|
|
lines.append("")
|
|
lines.append("ALBUMS")
|
|
|
|
for album in albums:
|
|
lines.append(" " + album_line(album, max(10, width - 2), music_root, include_paths))
|
|
|
|
return lines
|
|
|
|
|
|
def build_relationship_map(
|
|
coverages: list[TrackCoverage],
|
|
*,
|
|
width: int,
|
|
show_covered: bool,
|
|
) -> list[str]:
|
|
rows = [
|
|
coverage
|
|
for coverage in coverages
|
|
if show_covered or coverage.status != "EXACT_RELATED"
|
|
]
|
|
|
|
lines: list[str] = []
|
|
lines.append("")
|
|
lines.append("RELATIONSHIP MAP")
|
|
|
|
if not rows:
|
|
lines.append(" all candidate tracks have exact related coverage elsewhere")
|
|
return lines
|
|
|
|
hidden_exact = sum(1 for c in coverages if c.status == "EXACT_RELATED")
|
|
|
|
if hidden_exact and not show_covered:
|
|
lines.append(
|
|
f" {hidden_exact} EXACT rows hidden; use --show-covered to include them"
|
|
)
|
|
|
|
lines.append("")
|
|
|
|
status_w = 5
|
|
gap = 2
|
|
|
|
if width >= 100:
|
|
candidate_w = 40
|
|
elif width >= 86:
|
|
candidate_w = 34
|
|
else:
|
|
candidate_w = 28
|
|
|
|
right_w = width - status_w - gap - candidate_w - gap
|
|
right_w = max(24, right_w)
|
|
|
|
header = (
|
|
f"{'ST':<{status_w}}"
|
|
f"{' ' * gap}"
|
|
f"{'CANDIDATE':<{candidate_w}}"
|
|
f"{' ' * gap}"
|
|
f"BEST MATCH / REASON"
|
|
)
|
|
|
|
lines.append(ellipsize(header, width))
|
|
lines.append(ellipsize("-" * width, width))
|
|
|
|
for coverage in rows:
|
|
candidate = coverage.candidate
|
|
best = best_match_for(coverage)
|
|
|
|
left = candidate_ref(candidate)
|
|
|
|
if best is None:
|
|
right = status_reason(coverage.status)
|
|
else:
|
|
right = f"-> {match_ref(best)}"
|
|
|
|
row = (
|
|
f"{status_short(coverage.status):<{status_w}}"
|
|
f"{' ' * gap}"
|
|
f"{ellipsize(left, candidate_w):<{candidate_w}}"
|
|
f"{' ' * gap}"
|
|
f"{ellipsize(right, right_w)}"
|
|
)
|
|
|
|
lines.append(row)
|
|
|
|
return lines
|
|
|
|
|
|
def compact_detail_block(
|
|
coverage: TrackCoverage,
|
|
music_root: Path,
|
|
*,
|
|
width: int,
|
|
include_paths: bool,
|
|
max_matches: int,
|
|
verbose: bool,
|
|
) -> list[str]:
|
|
lines: list[str] = []
|
|
candidate = coverage.candidate
|
|
|
|
lines.append(ellipsize("-" * width, width))
|
|
lines.append(f"{status_short(coverage.status)} | {status_reason(coverage.status)}")
|
|
lines.append(ellipsize(f"CAND {full_track_ref(candidate)}", width))
|
|
|
|
if include_paths:
|
|
lines.append(ellipsize(f"PATH {relpath(candidate.path, music_root)}", width))
|
|
|
|
lines.append(
|
|
"CNT "
|
|
f"exact_rel={len(coverage.exact_related)} "
|
|
f"exact_other={len(coverage.exact_other)} "
|
|
f"loose_rel={len(coverage.loose_related)} "
|
|
f"loose_other={len(coverage.loose_other)}"
|
|
)
|
|
|
|
categories = [
|
|
("EXACT RELATED", coverage.exact_related, "REL"),
|
|
("EXACT OTHER", coverage.exact_other, "OTH"),
|
|
("LOOSE RELATED", coverage.loose_related, "REL"),
|
|
("LOOSE OTHER", coverage.loose_other, "OTH"),
|
|
]
|
|
|
|
for heading, matches, label in categories:
|
|
if not matches:
|
|
continue
|
|
|
|
lines.append(f"{heading}")
|
|
|
|
shown = matches if verbose else matches[:max_matches]
|
|
|
|
for match in shown:
|
|
lines.append(ellipsize(f" {label} {full_track_ref(match)}", width))
|
|
if include_paths:
|
|
lines.append(ellipsize(f" {relpath(match.path, music_root)}", width))
|
|
|
|
hidden = len(matches) - len(shown)
|
|
if hidden > 0:
|
|
lines.append(f" ... {hidden} more omitted")
|
|
|
|
return lines
|
|
|
|
|
|
def full_track_ref(track: Track) -> str:
|
|
return (
|
|
f"A{track.album_id} {track_position(track)} i{track.id} | "
|
|
f"{track.artist} - {track.title} | "
|
|
f"{track.album} | {quality_short(track)}"
|
|
)
|
|
|
|
|
|
def build_details_output(
|
|
coverages: list[TrackCoverage],
|
|
music_root: Path,
|
|
*,
|
|
width: int,
|
|
show_covered: bool,
|
|
include_paths: bool,
|
|
max_matches: int,
|
|
verbose: bool,
|
|
) -> list[str]:
|
|
rows = [
|
|
coverage
|
|
for coverage in coverages
|
|
if show_covered or verbose or coverage.status != "EXACT_RELATED"
|
|
]
|
|
|
|
lines: list[str] = []
|
|
lines.append("")
|
|
lines.append("DETAILS")
|
|
|
|
if not rows:
|
|
lines.append(" no detail rows; all tracks are exact-covered")
|
|
return lines
|
|
|
|
for coverage in rows:
|
|
lines.extend(
|
|
compact_detail_block(
|
|
coverage,
|
|
music_root,
|
|
width=width,
|
|
include_paths=include_paths,
|
|
max_matches=max_matches,
|
|
verbose=verbose,
|
|
)
|
|
)
|
|
|
|
return lines
|
|
|
|
|
|
def write_tsv(path: Path, coverages: list[TrackCoverage], music_root: Path) -> None:
|
|
columns = [
|
|
"candidate_album_id",
|
|
"candidate_item_id",
|
|
"status",
|
|
"status_reason",
|
|
"exact_related_count",
|
|
"exact_other_count",
|
|
"loose_related_count",
|
|
"loose_other_count",
|
|
"candidate_albumartist",
|
|
"candidate_album",
|
|
"candidate_artist",
|
|
"candidate_title",
|
|
"candidate_disc",
|
|
"candidate_track",
|
|
"candidate_format",
|
|
"candidate_bitdepth",
|
|
"candidate_samplerate",
|
|
"candidate_bitrate",
|
|
"candidate_path",
|
|
"best_match_album_id",
|
|
"best_match_item_id",
|
|
"best_match_albumartist",
|
|
"best_match_album",
|
|
"best_match_artist",
|
|
"best_match_title",
|
|
"best_match_format",
|
|
"best_match_bitdepth",
|
|
"best_match_samplerate",
|
|
"best_match_bitrate",
|
|
"best_match_path",
|
|
]
|
|
|
|
with path.open("w", encoding="utf-8", newline="") as f:
|
|
writer = csv.DictWriter(f, fieldnames=columns, delimiter="\t")
|
|
writer.writeheader()
|
|
|
|
for coverage in coverages:
|
|
candidate = coverage.candidate
|
|
best = best_match_for(coverage)
|
|
|
|
row = {
|
|
"candidate_album_id": candidate.album_id,
|
|
"candidate_item_id": candidate.id,
|
|
"status": coverage.status,
|
|
"status_reason": status_reason(coverage.status),
|
|
"exact_related_count": len(coverage.exact_related),
|
|
"exact_other_count": len(coverage.exact_other),
|
|
"loose_related_count": len(coverage.loose_related),
|
|
"loose_other_count": len(coverage.loose_other),
|
|
"candidate_albumartist": candidate.albumartist,
|
|
"candidate_album": candidate.album,
|
|
"candidate_artist": candidate.artist,
|
|
"candidate_title": candidate.title,
|
|
"candidate_disc": candidate.disc,
|
|
"candidate_track": candidate.track,
|
|
"candidate_format": candidate.format,
|
|
"candidate_bitdepth": candidate.bitdepth,
|
|
"candidate_samplerate": candidate.samplerate,
|
|
"candidate_bitrate": candidate.bitrate,
|
|
"candidate_path": relpath(candidate.path, music_root),
|
|
"best_match_album_id": best.album_id if best else "",
|
|
"best_match_item_id": best.id if best else "",
|
|
"best_match_albumartist": best.albumartist if best else "",
|
|
"best_match_album": best.album if best else "",
|
|
"best_match_artist": best.artist if best else "",
|
|
"best_match_title": best.title if best else "",
|
|
"best_match_format": best.format if best else "",
|
|
"best_match_bitdepth": best.bitdepth if best else "",
|
|
"best_match_samplerate": best.samplerate if best else "",
|
|
"best_match_bitrate": best.bitrate if best else "",
|
|
"best_match_path": relpath(best.path, music_root) if best else "",
|
|
}
|
|
|
|
writer.writerow(row)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
audit_dir, music_root, library_dir = default_paths()
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description="Check whether beets album tracks are covered elsewhere."
|
|
)
|
|
|
|
parser.add_argument(
|
|
"album_ids",
|
|
nargs="+",
|
|
help="Beets album IDs to evaluate, e.g. 454 or 371 372.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--audit-dir",
|
|
default=str(audit_dir),
|
|
help=f"Directory for audit output. Default: {audit_dir}",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--music-root",
|
|
default=str(music_root),
|
|
help=f"Root used for relative paths. Default: {music_root}",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--library-dir",
|
|
default=str(library_dir),
|
|
help=f"Expected library directory. Default: {library_dir}",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--width",
|
|
type=int,
|
|
default=None,
|
|
help="Report width. Default: detected terminal width, capped at 120.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--no-write",
|
|
action="store_true",
|
|
help="Print report but do not write .txt/.tsv files.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--show-covered",
|
|
action="store_true",
|
|
help="Include EXACT rows in the relationship map/details.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--details",
|
|
action="store_true",
|
|
help="Show compact detail blocks for attention rows.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--verbose",
|
|
action="store_true",
|
|
help="Show forensic match blocks. Implies --details.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--paths",
|
|
action="store_true",
|
|
help="Include relative paths in detail/verbose output.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--matches",
|
|
type=int,
|
|
default=3,
|
|
help="Max matches per category in details mode. Default: 3.",
|
|
)
|
|
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
|
|
album_ids = [str(album_id).strip() for album_id in args.album_ids]
|
|
album_ids = [album_id for album_id in album_ids if album_id]
|
|
|
|
invalid = [album_id for album_id in album_ids if not album_id.isdigit()]
|
|
|
|
if invalid:
|
|
print(
|
|
f"ERROR: album IDs should be numeric. Invalid: {', '.join(invalid)}",
|
|
file=sys.stderr,
|
|
)
|
|
return 2
|
|
|
|
audit_dir = Path(args.audit_dir).resolve()
|
|
music_root = Path(args.music_root).resolve()
|
|
library_dir = Path(args.library_dir).resolve()
|
|
width = terminal_width(args.width)
|
|
|
|
if not library_dir.exists():
|
|
print(
|
|
f"WARNING: expected library directory does not exist: {library_dir}",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
albums = [load_album(album_id) for album_id in album_ids]
|
|
|
|
candidate_tracks: list[Track] = []
|
|
|
|
for album_id in album_ids:
|
|
candidate_tracks.extend(load_tracks_for_album(album_id))
|
|
|
|
if not candidate_tracks:
|
|
print(
|
|
f"ERROR: no tracks found for album IDs: {' '.join(album_ids)}",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
candidate_album_id_set = set(album_ids)
|
|
|
|
all_tracks = load_all_tracks()
|
|
|
|
outside_tracks = [
|
|
track for track in all_tracks if track.album_id not in candidate_album_id_set
|
|
]
|
|
|
|
candidate_tracks = sorted(
|
|
candidate_tracks,
|
|
key=lambda t: (
|
|
parse_intish(t.album_id),
|
|
natural_track_number(t.disc),
|
|
natural_track_number(t.track),
|
|
t.title.casefold(),
|
|
),
|
|
)
|
|
|
|
coverages = [
|
|
classify_track(candidate, outside_tracks)
|
|
for candidate in candidate_tracks
|
|
]
|
|
|
|
lines: list[str] = []
|
|
|
|
lines.extend(
|
|
build_summary_lines(
|
|
album_ids,
|
|
albums,
|
|
candidate_tracks,
|
|
outside_tracks,
|
|
coverages,
|
|
music_root,
|
|
width=width,
|
|
include_paths=args.paths,
|
|
)
|
|
)
|
|
|
|
lines.extend(
|
|
build_relationship_map(
|
|
coverages,
|
|
width=width,
|
|
show_covered=args.show_covered,
|
|
)
|
|
)
|
|
|
|
if args.details or args.verbose:
|
|
lines.extend(
|
|
build_details_output(
|
|
coverages,
|
|
music_root,
|
|
width=width,
|
|
show_covered=args.show_covered,
|
|
include_paths=args.paths,
|
|
max_matches=args.matches,
|
|
verbose=args.verbose,
|
|
)
|
|
)
|
|
|
|
suffix = "-".join(album_ids)
|
|
txt_path = audit_dir / f"audit.album-coverage-{suffix}.txt"
|
|
tsv_path = audit_dir / f"audit.album-coverage-{suffix}.tsv"
|
|
|
|
if not args.no_write:
|
|
audit_dir.mkdir(parents=True, exist_ok=True)
|
|
txt_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
write_tsv(tsv_path, coverages, music_root)
|
|
|
|
output_lines = list(lines)
|
|
|
|
if not args.no_write:
|
|
output_lines.append("")
|
|
output_lines.append(f"Wrote: {txt_path}")
|
|
output_lines.append(f"Wrote: {tsv_path}")
|
|
|
|
print("\n".join(output_lines))
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|