diff --git a/.gitignore b/.gitignore index 1a9740f..985108b 100644 --- a/.gitignore +++ b/.gitignore @@ -35,8 +35,8 @@ __pycache__/ *.pyd # Beets/audit generated output -audit/ -sample/ +./audit +./sample # Tool build products and large tool data tools/**/build/ diff --git a/scripts/unsorted/find-bad-music-folders.py b/scripts/audit/audit-library-paths similarity index 86% rename from scripts/unsorted/find-bad-music-folders.py rename to scripts/audit/audit-library-paths index 06a7d1f..2e51c84 100755 --- a/scripts/unsorted/find-bad-music-folders.py +++ b/scripts/audit/audit-library-paths @@ -13,15 +13,15 @@ Allowed layouts: Examples: PASS: - /music/library/originals/Metallica/Albums/Master of Puppets/01 Battery.flac - /music/library/originals/Daft Punk/Remixes/TRON Legacy Reconfigured/01 Derezzed.flac - /music/library/originals/OSTs/Final Fantasy VII/01 Prelude.flac + /data/Music/Library/originals/Metallica/Albums/Master of Puppets/01 Battery.flac + /data/Music/Library/originals/Daft Punk/Remixes/TRON Legacy Reconfigured/01 Derezzed.flac + /data/Music/Library/originals/OSTs/Final Fantasy VII/01 Prelude.flac FAIL: - /music/library/originals/Metallica/Remix/Some Album/01 Track.flac - /music/library/originals/Metallica/Bootlegs/Some Album/01 Track.flac - /music/library/originals/OSTs/01 Track.flac - /music/library/originals/Some Artist/OSTs/Some Album/01 Track.flac + /data/Music/Library/originals/Metallica/Remix/Some Album/01 Track.flac + /data/Music/Library/originals/Metallica/Bootlegs/Some Album/01 Track.flac + /data/Music/Library/originals/OSTs/01 Track.flac + /data/Music/Library/originals/Some Artist/OSTs/Some Album/01 Track.flac """ from __future__ import annotations @@ -157,8 +157,9 @@ def validate_path(path: Path, root: Path) -> tuple[Optional[str], str]: return None, str(rel) -def load_beets_items(db_path: Path) -> dict[str, BeetsItem]: +def load_beets_items(db_path: Path, root: Path) -> dict[str, BeetsItem]: db_path = db_path.expanduser().resolve(strict=False) + root = root.expanduser().resolve(strict=False) if not db_path.exists(): raise FileNotFoundError(f"Beets database does not exist: {db_path}") @@ -186,7 +187,16 @@ def load_beets_items(db_path: Path) -> dict[str, BeetsItem]: items: dict[str, BeetsItem] = {} for row in rows: - path = Path(decode_beets_path(row["path"])) + raw_path = Path(decode_beets_path(row["path"])) + + # Critical fix: + # Beets may store item paths relative to the configured music directory. + # Those must be resolved relative to --root, not relative to cwd. + if raw_path.is_absolute(): + path = raw_path + else: + path = root / raw_path + key = normalize_path(path) items[key] = BeetsItem( @@ -296,14 +306,14 @@ def main() -> int: parser.add_argument( "--root", - default="/music/library/originals", - help="Music library root. Default: /music/library/originals", + default="/data/Music/Library/originals", + help="Music library root. Default: /data/Music/Library/originals", ) parser.add_argument( "--db", - default="/music/beets/musiclibrary.db", - help="Beets sqlite database path. Default: /music/beets/musiclibrary.db", + default="/data/Music/Library/musiclibrary.db", + help="Beets sqlite database path. Default: /data/Music/Library/musiclibrary.db", ) parser.add_argument( @@ -331,7 +341,7 @@ def main() -> int: db_path = Path(args.db) audio_exts = parse_audio_exts(args.audio_exts) - beets_items = load_beets_items(db_path) + beets_items = load_beets_items(db_path, root) disk_paths: set[str] = set() if args.disk: diff --git a/scripts/audit/check-audio-decode b/scripts/audit/check-audio-decode new file mode 100755 index 0000000..cf817e3 --- /dev/null +++ b/scripts/audit/check-audio-decode @@ -0,0 +1,13 @@ +#!/usr/bin/env bash +set -u + +file="${1:?missing file path}" + +ffmpeg \ + -hide_banner \ + -nostdin \ + -v error \ + -i "$file" \ + -map 0:a:0 \ + -f null - \ + >/dev/null diff --git a/scripts/unsorted/audit/album_coverage.py b/scripts/unsorted/audit/album_coverage.py new file mode 100644 index 0000000..b436a85 --- /dev/null +++ b/scripts/unsorted/audit/album_coverage.py @@ -0,0 +1,1168 @@ +#!/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()) diff --git a/scripts/unsorted/audit/album_coverage.sh b/scripts/unsorted/audit/album_coverage.sh new file mode 100755 index 0000000..90f5816 --- /dev/null +++ b/scripts/unsorted/audit/album_coverage.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" + +exec python3 "$SCRIPT_DIR/album_coverage.py" "$@" diff --git a/scripts/unsorted/audit/audit-lyrics.py b/scripts/unsorted/audit/audit-lyrics.py new file mode 100644 index 0000000..dc1438f --- /dev/null +++ b/scripts/unsorted/audit/audit-lyrics.py @@ -0,0 +1,346 @@ +#!/usr/bin/env python3 + +import argparse +import csv +import re +import sys +from pathlib import Path + +from beets.library import Library +from mutagen import File as MutagenFile + + +URL_RE = re.compile(r"https?://\S+|www\.\S+", re.IGNORECASE) + +HTML_OR_WEB_ERROR_RE = re.compile( + r"""(?ix) + | + | + cloudflare | + just\ a\ moment | + enable\ javascript | + javascript\ required | + access\ denied | + 403\ forbidden | + 404\ not\ found | + page\ not\ found | + captcha | + request\ unsuccessful | + service\ unavailable + """ +) + +API_GARBAGE_RE = re.compile( + r"""(?ix) + ^\s*(null|none|undefined|nan|\[object\s+object\])\s*$ | + "error"\s*: | + "status"\s*:\s*(403|404|429|500|503) + """ +) + +PLACEHOLDER_RE = re.compile( + r"""(?ix) + ^\s* + \[? + ( + instrumental | + no\ lyrics | + lyrics\ not\ found | + lyrics\ unavailable | + not\ available | + unknown | + intro | + outro | + interlude + ) + \]? + \s*$ + """ +) + +SCOREISH_RE = re.compile( + r"""(?ix) + soundtrack | + original\ score | + motion\ picture\ score | + score | + ost | + bgm | + instrumental | + karaoke | + game\ soundtrack | + video\ game + """ +) + + +def clean_text(value): + if value is None: + return "" + if isinstance(value, bytes): + value = value.decode("utf-8", "replace") + return str(value).replace("\r\n", "\n").replace("\r", "\n").strip() + + +def snippet(text, width=220): + one_line = re.sub(r"\s+", " ", text).strip() + return one_line[:width] + ("…" if len(one_line) > width else "") + + +def is_scoreish(item): + haystack = " ".join( + clean_text(getattr(item, field, "")) + for field in ("title", "album", "albumartist", "artist", "genre", "albumtype") + ) + return bool(SCOREISH_RE.search(haystack)) + + +def classify(item, lyrics): + text = clean_text(lyrics) + if not text: + return None + + lower = text.lower() + urls = URL_RE.findall(text) + leftover_without_urls = URL_RE.sub("", text).strip() + line_count = len([line for line in text.splitlines() if line.strip()]) + char_count = len(text) + + if "/pmedia" in lower: + return ("AUTO_STRIP", "pmedia_url_garbage") + + if urls and char_count <= 500 and len(leftover_without_urls) <= 40: + return ("AUTO_STRIP", "url_only_or_url_stub") + + if HTML_OR_WEB_ERROR_RE.search(text): + return ("AUTO_STRIP", "html_or_web_error_page") + + if API_GARBAGE_RE.search(text): + return ("AUTO_STRIP", "api_or_placeholder_garbage") + + if PLACEHOLDER_RE.search(text): + if is_scoreish(item): + return ("REVIEW", "placeholder_but_score_or_soundtrack_like") + return ("REVIEW", "placeholder_text") + + if char_count <= 40 and line_count <= 2: + if is_scoreish(item): + return ("REVIEW", "very_short_but_score_or_soundtrack_like") + return ("REVIEW", "very_short_lyrics") + + return None + + +def strip_file_lyrics(path): + audio = MutagenFile(path) + if audio is None or audio.tags is None: + return "no_tags" + + tags = audio.tags + removed = [] + + # Vorbis-style tags: FLAC, Ogg Vorbis, Opus, etc. + for key in list(tags.keys()): + if key.lower() in { + "lyrics", + "unsyncedlyrics", + "syncedlyrics", + "lyric", + "lyricist", + }: + del tags[key] + removed.append(key) + + # ID3: MP3, AIFF, sometimes WAV. + for key in list(tags.keys()): + if key.startswith("USLT") or key.startswith("SYLT"): + del tags[key] + removed.append(key) + + # Some tools store lyrics in TXXX frames. + for key in list(tags.keys()): + lowered = key.lower() + if lowered.startswith("txxx:") and any( + marker in lowered for marker in ("lyrics", "unsyncedlyrics", "syncedlyrics") + ): + del tags[key] + removed.append(key) + + # MP4/M4A. + for key in list(tags.keys()): + lowered = key.lower() + if lowered in {"©lyr", "\xa9lyr"} or "lyrics" in lowered: + del tags[key] + removed.append(key) + + if removed: + audio.save() + return ",".join(removed) + + return "no_lyrics_tag_found" + + +def clear_beets_lyrics(item, clear_meta=False): + item["lyrics"] = "" + + if clear_meta: + for field in ( + "lyrics_backend", + "lyrics_url", + "lyrics_language", + "lyrics_translation_language", + ): + if field in item: + item[field] = "" + + item.store() + + +def main(): + parser = argparse.ArgumentParser( + description="Find and optionally strip obviously broken lyrics from a beets library." + ) + parser.add_argument( + "--library", + default="/music/beets/musiclibrary.db", + help="Path to beets library DB. Default: /music/beets/musiclibrary.db", + ) + parser.add_argument( + "--report", + default="bad_lyrics_report.tsv", + help="TSV report path.", + ) + parser.add_argument( + "--backup", + default="bad_lyrics_backup.tsv", + help="TSV backup of original lyrics for matched candidates.", + ) + parser.add_argument( + "--apply", + action="store_true", + help="Actually strip AUTO_STRIP matches from beets DB and file tags.", + ) + parser.add_argument( + "--apply-review", + action="store_true", + help="Also strip REVIEW matches. Not recommended for first pass.", + ) + parser.add_argument( + "--clear-lyrics-meta", + action="store_true", + help="Also clear lyrics_backend, lyrics_url, and lyrics language fields in beets.", + ) + args = parser.parse_args() + + lib_path = Path(args.library) + if not lib_path.exists(): + print(f"ERROR: beets library DB not found: {lib_path}", file=sys.stderr) + sys.exit(2) + + lib = Library(str(lib_path)) + + candidates = [] + stripped = 0 + failed = 0 + + with open(args.report, "w", newline="", encoding="utf-8") as report_f, \ + open(args.backup, "w", newline="", encoding="utf-8") as backup_f: + + report = csv.writer(report_f, delimiter="\t") + backup = csv.writer(backup_f, delimiter="\t") + + report.writerow([ + "action", + "reason", + "id", + "artist", + "album", + "title", + "path", + "chars", + "lines", + "snippet", + "strip_result", + ]) + + backup.writerow([ + "id", + "artist", + "album", + "title", + "path", + "original_lyrics", + ]) + + for item in lib.items(): + lyrics = clean_text(getattr(item, "lyrics", "")) + result = classify(item, lyrics) + if result is None: + continue + + action, reason = result + path = bytes(item.path).decode("utf-8", "replace") if isinstance(item.path, bytes) else str(item.path) + line_count = len([line for line in lyrics.splitlines() if line.strip()]) + + strip_result = "" + + should_strip = args.apply and ( + action == "AUTO_STRIP" or (action == "REVIEW" and args.apply_review) + ) + + if should_strip: + try: + backup.writerow([ + item.id, + item.artist, + item.album, + item.title, + path, + lyrics, + ]) + + clear_beets_lyrics(item, clear_meta=args.clear_lyrics_meta) + strip_result = strip_file_lyrics(path) + stripped += 1 + except Exception as exc: + failed += 1 + strip_result = f"ERROR: {type(exc).__name__}: {exc}" + + report.writerow([ + action, + reason, + item.id, + item.artist, + item.album, + item.title, + path, + len(lyrics), + line_count, + snippet(lyrics), + strip_result, + ]) + + candidates.append((action, reason, item.id, path)) + + auto_count = sum(1 for action, _, _, _ in candidates if action == "AUTO_STRIP") + review_count = sum(1 for action, _, _, _ in candidates if action == "REVIEW") + + print(f"Report written: {args.report}") + print(f"Backup written: {args.backup}") + print(f"AUTO_STRIP candidates: {auto_count}") + print(f"REVIEW candidates: {review_count}") + + if args.apply: + print(f"Stripped: {stripped}") + print(f"Failed: {failed}") + else: + print("Dry run only. Re-run with --apply to strip AUTO_STRIP matches.") + + +if __name__ == "__main__": + main() diff --git a/scripts/unsorted/audit/audit_duplicates.py b/scripts/unsorted/audit/audit_duplicates.py new file mode 100644 index 0000000..7d95ada --- /dev/null +++ b/scripts/unsorted/audit/audit_duplicates.py @@ -0,0 +1,273 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import csv +import re +import subprocess +import unicodedata +from collections import defaultdict +from pathlib import Path +from typing import Dict, Iterable, List + + +SCRIPT_DIR = Path(__file__).resolve().parent +BEETS_DIR = SCRIPT_DIR.parent +MUSIC_ROOT = BEETS_DIR.parent +LIBRARY_DIR = MUSIC_ROOT / "library" +OUT_DIR = SCRIPT_DIR + +# Internal delimiter only. You never have to type or think about this. +SEP = "\x1f" + +ALBUM_FIELDS = [ + "id", + "albumartist", + "album", + "year", + "albumtype", + "albumtotal", + "mb_albumid", + "mb_releasegroupid", + "albumdisambig", + "label", + "catalognum", + "country", + "media", + "path", +] + + +def normalize_text(value: str) -> str: + value = unicodedata.normalize("NFKC", value) + + replacements = { + "’": "'", + "‘": "'", + "`": "'", + "´": "'", + "‐": "-", + "-": "-", + "‒": "-", + "–": "-", + "—": "-", + "“": '"', + "”": '"', + } + + for old, new in replacements.items(): + value = value.replace(old, new) + + value = re.sub(r"\s+", " ", value).strip() + return value.casefold() + + +def relative_path(path: str) -> str: + if not path: + return "" + + p = Path(path) + + try: + return str(p.relative_to(MUSIC_ROOT)) + except ValueError: + return path + + +def run_beet_album_export() -> List[Dict[str, str]]: + fmt = SEP.join(f"${field}" for field in ALBUM_FIELDS) + + result = subprocess.run( + ["beet", "ls", "-a", "-f", fmt], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=True, + ) + + if result.stderr.strip(): + (OUT_DIR / "audit.duplicates.beet.stderr.txt").write_text(result.stderr) + + bad_lines: List[str] = [] + rows: List[Dict[str, str]] = [] + + for line_number, line in enumerate(result.stdout.splitlines(), start=1): + parts = line.split(SEP) + + if len(parts) != len(ALBUM_FIELDS): + bad_lines.append(f"{line_number}: {line}") + continue + + row = dict(zip(ALBUM_FIELDS, parts)) + row["relpath"] = relative_path(row["path"]) + rows.append(row) + + if bad_lines: + (OUT_DIR / "audit.duplicates.bad-lines.txt").write_text("\n".join(bad_lines) + "\n") + + return rows + + +def write_tsv(path: Path, rows: Iterable[Dict[str, str]], fields: List[str]) -> None: + with path.open("w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=fields, delimiter="\t", extrasaction="ignore") + writer.writeheader() + for row in rows: + writer.writerow(row) + + +def format_album(row: Dict[str, str]) -> str: + return "\n".join( + [ + f"id={row['id']}", + f" albumartist: {row['albumartist']}", + f" album: {row['album']}", + f" year/type: {row['year']} / {row['albumtype']}", + f" tracks: {row['albumtotal']}", + f" mb_albumid: {row['mb_albumid']}", + f" rgid: {row['mb_releasegroupid']}", + f" disambig: {row['albumdisambig']}", + f" label/cat: {row['label']} / {row['catalognum']}", + f" country: {row['country']}", + f" media: {row['media']}", + f" path: {row['relpath']}", + ] + ) + + +def write_normalized_duplicate_report(rows: List[Dict[str, str]]) -> int: + groups: Dict[tuple[str, str], List[Dict[str, str]]] = defaultdict(list) + + for row in rows: + key = (normalize_text(row["albumartist"]), normalize_text(row["album"])) + groups[key].append(row) + + duplicate_groups = {key: value for key, value in groups.items() if len(value) > 1} + + lines: List[str] = [] + + for key, group_rows in sorted(duplicate_groups.items()): + lines.append("=" * 100) + lines.append(f"NORMALIZED DUPLICATE: {key[0]} / {key[1]}") + + for row in sorted(group_rows, key=lambda r: int(r["id"])): + lines.append(format_album(row)) + + output = "\n".join(lines) + if output: + output += "\n" + + (OUT_DIR / "audit.normalized-album-dupes.txt").write_text(output, encoding="utf-8") + + flat_rows: List[Dict[str, str]] = [] + for key, group_rows in sorted(duplicate_groups.items()): + for row in group_rows: + row = dict(row) + row["normalized_albumartist"] = key[0] + row["normalized_album"] = key[1] + flat_rows.append(row) + + write_tsv( + OUT_DIR / "audit.normalized-album-dupes.tsv", + flat_rows, + ["normalized_albumartist", "normalized_album", *ALBUM_FIELDS, "relpath"], + ) + + return len(duplicate_groups) + + +def classify_release_group(group_rows: List[Dict[str, str]]) -> str: + albums = {normalize_text(row["album"]) for row in group_rows} + albumartists = {normalize_text(row["albumartist"]) for row in group_rows} + types = {normalize_text(row["albumtype"]) for row in group_rows} + titles = " ".join(row["album"] for row in group_rows).casefold() + + if len(albums) == 1 and len(albumartists) == 1: + return "likely duplicate: same normalized artist and album title" + + if "remix" in titles or "remixes" in titles: + return "probably okay: original/remix-family releases" + + if "deluxe" in titles or "anniversary" in titles or re.search(r"\bxxv\b", titles): + return "curation choice: standard vs deluxe/anniversary edition" + + if types == {"single"}: + return "curation choice: related single variants" + + if "compilation" in types or "album" in types: + return "inspect: same release group with different album titles/editions" + + return "inspect" + + +def write_release_group_duplicate_report(rows: List[Dict[str, str]]) -> int: + groups: Dict[str, List[Dict[str, str]]] = defaultdict(list) + + for row in rows: + rgid = row["mb_releasegroupid"] + if rgid: + groups[rgid].append(row) + + duplicate_groups = {key: value for key, value in groups.items() if len(value) > 1} + + lines: List[str] = [] + + for rgid, group_rows in sorted(duplicate_groups.items()): + lines.append("=" * 100) + lines.append(f"RELEASE GROUP DUPLICATE: {rgid}") + lines.append(f"HINT: {classify_release_group(group_rows)}") + + for row in sorted(group_rows, key=lambda r: (r["albumartist"], r["album"], int(r["id"]))): + lines.append(format_album(row)) + + output = "\n".join(lines) + if output: + output += "\n" + + (OUT_DIR / "audit.releasegroup-dupes.txt").write_text(output, encoding="utf-8") + + flat_rows: List[Dict[str, str]] = [] + for rgid, group_rows in sorted(duplicate_groups.items()): + hint = classify_release_group(group_rows) + for row in group_rows: + row = dict(row) + row["releasegroup_duplicate_hint"] = hint + flat_rows.append(row) + + write_tsv( + OUT_DIR / "audit.releasegroup-dupes.tsv", + flat_rows, + ["releasegroup_duplicate_hint", *ALBUM_FIELDS, "relpath"], + ) + + return len(duplicate_groups) + + +def main() -> int: + OUT_DIR.mkdir(parents=True, exist_ok=True) + + rows = run_beet_album_export() + + write_tsv( + OUT_DIR / "audit.albums.snapshot.tsv", + rows, + [*ALBUM_FIELDS, "relpath"], + ) + + normalized_count = write_normalized_duplicate_report(rows) + releasegroup_count = write_release_group_duplicate_report(rows) + + print(f"Albums scanned: {len(rows)}") + print(f"Normalized duplicate groups: {normalized_count}") + print(f"Release-group duplicate groups: {releasegroup_count}") + print() + print(f"Wrote: {OUT_DIR / 'audit.normalized-album-dupes.txt'}") + print(f"Wrote: {OUT_DIR / 'audit.normalized-album-dupes.tsv'}") + print(f"Wrote: {OUT_DIR / 'audit.releasegroup-dupes.txt'}") + print(f"Wrote: {OUT_DIR / 'audit.releasegroup-dupes.tsv'}") + print(f"Wrote: {OUT_DIR / 'audit.albums.snapshot.tsv'}") + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/unsorted/audit/audit_duplicates.sh b/scripts/unsorted/audit/audit_duplicates.sh new file mode 100755 index 0000000..2121e0b --- /dev/null +++ b/scripts/unsorted/audit/audit_duplicates.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +python3 "$SCRIPT_DIR/audit_duplicates.py" diff --git a/scripts/unsorted/audit/compare.py b/scripts/unsorted/audit/compare.py new file mode 100755 index 0000000..7cd3e2b --- /dev/null +++ b/scripts/unsorted/audit/compare.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +import csv +import json +import os +from pathlib import Path +from collections import defaultdict + +BEETS_JSON = "beets.json" +NAVIDROME_CSV = "navidrome_tracks.csv" +LIBRARY_ROOT = Path("/music/library").resolve() + +def norm_rel(p: str) -> str: + p = p.strip() + if not p: + return "" + path = Path(p) + + # Navidrome often stores relative paths; beets often stores absolute paths. + if path.is_absolute(): + try: + path = path.resolve().relative_to(LIBRARY_ROOT) + except Exception: + path = Path(os.path.relpath(path.resolve(), LIBRARY_ROOT)) + return str(path).replace("\\", "/") + +def load_beets(): + data = json.load(open(BEETS_JSON, "r", encoding="utf-8")) + rows = [] + for row in data: + rel = norm_rel(row.get("path", "")) + rows.append({ + "relpath": rel, + "title": row.get("title", ""), + "artist": row.get("artist", ""), + "album": row.get("album", ""), + "albumartist": row.get("albumartist", ""), + "format": row.get("format", ""), + "exists_on_disk": (LIBRARY_ROOT / rel).exists() if rel else False, + }) + return rows + +def load_navidrome(): + rows = [] + with open(NAVIDROME_CSV, "r", encoding="utf-8", newline="") as f: + for row in csv.DictReader(f): + rel = norm_rel(row.get("path", "")) + rows.append({ + "relpath": rel, + "title": row.get("title", ""), + "artist": row.get("artist", ""), + "album": row.get("album", ""), + "exists_on_disk": (LIBRARY_ROOT / rel).exists() if rel else False, + }) + return rows + +beets = load_beets() +nd = load_navidrome() + +beets_by_path = defaultdict(list) +nd_by_path = defaultdict(list) + +for r in beets: + beets_by_path[r["relpath"]].append(r) +for r in nd: + nd_by_path[r["relpath"]].append(r) + +beets_paths = set(beets_by_path) +nd_paths = set(nd_by_path) + +only_beets = sorted(beets_paths - nd_paths) +only_nd = sorted(nd_paths - beets_paths) +dup_nd = sorted([p for p, rows in nd_by_path.items() if len(rows) > 1]) +beets_missing_on_disk = sorted([r["relpath"] for r in beets if r["relpath"] and not r["exists_on_disk"]]) +nd_missing_on_disk = sorted([r["relpath"] for r in nd if r["relpath"] and not r["exists_on_disk"]]) + +def dump(name, items): + print(f"\n=== {name}: {len(items)} ===") + for x in items[:200]: + print(x) + if len(items) > 200: + print(f"... and {len(items)-200} more") + +dump("In beets but not Navidrome", only_beets) +dump("In Navidrome but not beets", only_nd) +dump("Navidrome duplicate live paths", dup_nd) +dump("beets rows whose files are missing on disk", beets_missing_on_disk) +dump("Navidrome live rows whose files are missing on disk", nd_missing_on_disk) diff --git a/scripts/unsorted/audit/compare.sh b/scripts/unsorted/audit/compare.sh new file mode 100755 index 0000000..0832381 --- /dev/null +++ b/scripts/unsorted/audit/compare.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash + +ROOT="/music" +BEETS="$ROOT/beets" +LIBRARY="$ROOT/library" +NAVIDROME="$ROOT/navidrome" +NAVIDROME_DATA="$NAVIDROME/data" +NAVIDROME_DB="$NAVIDROME_DATA/navidrome.db" + +echo "Exporting Navidrome Tracks to navidrome_tracks.csv" +sqlite3 "$NAVIDROME_DB" <<'SQL' +.headers on +.mode csv +.output navidrome_tracks.csv +SELECT path, title, album, artist, missing +FROM media_file +WHERE missing = 0 +ORDER BY path; +.output stdout +SQL + +echo "Exporting Beets Tracks to beets.json" +beet export --library \ + --include-keys path,title,artist,album,albumartist,track,disc,format,year \ + --output beets.json + +echo "Performing Comparison..." +python3 ./compare.py | tee -a ./comparison.txt + diff --git a/scripts/unsorted/audit/compare_album_tracks.py b/scripts/unsorted/audit/compare_album_tracks.py new file mode 100644 index 0000000..1a05dad --- /dev/null +++ b/scripts/unsorted/audit/compare_album_tracks.py @@ -0,0 +1,207 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import re +import subprocess +import sys +import unicodedata +from collections import defaultdict +from pathlib import Path +from typing import Dict, List + + +SCRIPT_DIR = Path(__file__).resolve().parent +BEETS_DIR = SCRIPT_DIR.parent +MUSIC_ROOT = BEETS_DIR.parent +OUT_DIR = SCRIPT_DIR + +SEP = "\x1f" + +FIELDS = [ + "id", + "album_id", + "albumartist", + "album", + "disc", + "track", + "title", + "artist", + "format", + "bitdepth", + "samplerate", + "bitrate", + "path", +] + + +def relpath(path: str) -> str: + try: + return str(Path(path).relative_to(MUSIC_ROOT)) + except Exception: + return path + + +def normalize(value: str) -> str: + value = unicodedata.normalize("NFKC", value) + value = value.replace("’", "'").replace("‘", "'").replace("`", "'").replace("´", "'") + value = value.replace("‐", "-").replace("–", "-").replace("—", "-") + value = re.sub(r"\s+", " ", value).strip() + return value.casefold() + + +def normalize_loose_title(value: str) -> str: + value = normalize(value) + + # Remove common version/remix/detail parentheticals for "same underlying song" checks. + value = re.sub(r"\s*[\(\[].*?[\)\]]\s*", " ", value) + + # Normalize censorship differences. + value = value.replace("f**k", "fuck") + value = value.replace("f--k", "fuck") + + value = re.sub(r"\s+", " ", value).strip() + return value + + +def run_beet_for_album(album_id: str) -> List[Dict[str, str]]: + fmt = SEP.join(f"${field}" for field in FIELDS) + + result = subprocess.run( + ["beet", "ls", f"album_id:{album_id}", "-f", fmt], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=True, + ) + + rows: List[Dict[str, str]] = [] + + for line in result.stdout.splitlines(): + parts = line.split(SEP) + if len(parts) != len(FIELDS): + continue + + row = dict(zip(FIELDS, parts)) + row["relpath"] = relpath(row["path"]) + row["title_norm"] = normalize(row["title"]) + row["title_loose"] = normalize_loose_title(row["title"]) + rows.append(row) + + rows.sort(key=lambda r: (int_or_zero(r["disc"]), int_or_zero(r["track"]), int_or_zero(r["id"]))) + return rows + + +def int_or_zero(value: str) -> int: + try: + return int(float(value)) + except Exception: + return 0 + + +def album_label(rows: List[Dict[str, str]]) -> str: + if not rows: + return "" + + first = rows[0] + return f"{first['album_id']} | {first['albumartist']} - {first['album']}" + + +def render_track(row: Dict[str, str]) -> str: + return ( + f"{row['album_id']}:{row['disc']}-{row['track']} | " + f"{row['title']} | " + f"{row['format']} | " + f"{row['bitdepth']}bit | " + f"{row['samplerate']} | " + f"{row['bitrate']}" + ) + + +def render_overlap(title: str, rows: List[Dict[str, str]]) -> List[str]: + lines = [f"- {title}"] + for row in rows: + lines.append(f" {render_track(row)}") + return lines + + +def main(argv: List[str]) -> int: + album_ids = argv[1:] + + if len(album_ids) < 2: + print("Usage: compare_album_tracks.py [ ...]", file=sys.stderr) + return 2 + + all_rows: List[Dict[str, str]] = [] + + for album_id in album_ids: + all_rows.extend(run_beet_for_album(album_id)) + + by_album: Dict[str, List[Dict[str, str]]] = defaultdict(list) + by_exact: Dict[str, List[Dict[str, str]]] = defaultdict(list) + by_loose: Dict[str, List[Dict[str, str]]] = defaultdict(list) + + for row in all_rows: + by_album[row["album_id"]].append(row) + by_exact[row["title_norm"]].append(row) + by_loose[row["title_loose"]].append(row) + + exact_overlaps = { + title: rows + for title, rows in by_exact.items() + if len({row["album_id"] for row in rows}) > 1 + } + + loose_overlaps = { + title: rows + for title, rows in by_loose.items() + if len({row["album_id"] for row in rows}) > 1 + } + + lines: List[str] = [] + lines.append(f"Album track comparison: {' '.join(album_ids)}") + lines.append("") + + lines.append("ALBUMS") + for album_id in album_ids: + rows = by_album.get(album_id, []) + lines.append(f" {album_label(rows)}") + lines.append("") + + lines.append(f"EXACT TITLE OVERLAPS: {len(exact_overlaps)}") + for title, rows in sorted(exact_overlaps.items()): + lines.extend(render_overlap(title, rows)) + lines.append("") + + lines.append(f"LOOSE TITLE OVERLAPS: {len(loose_overlaps)}") + lines.append("Loose overlap ignores parenthetical/version text, so use it as a clue, not proof.") + for title, rows in sorted(loose_overlaps.items()): + lines.extend(render_overlap(title, rows)) + lines.append("") + + lines.append("UNIQUE EXACT TITLES BY ALBUM") + for album_id in album_ids: + rows = by_album.get(album_id, []) + unique_rows = [ + row + for row in rows + if len({r["album_id"] for r in by_exact[row["title_norm"]]}) == 1 + ] + + lines.append(f"album_id={album_id} unique_exact_count={len(unique_rows)}") + for row in unique_rows: + lines.append(f" {render_track(row)}") + lines.append("") + + output = "\n".join(lines) + safe_ids = "-".join(album_ids) + output_path = OUT_DIR / f"audit.track-overlap-{safe_ids}.txt" + output_path.write_text(output, encoding="utf-8") + + print(output) + print(f"Wrote: {output_path}") + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv)) diff --git a/scripts/unsorted/audit/compare_album_tracks.sh b/scripts/unsorted/audit/compare_album_tracks.sh new file mode 100755 index 0000000..e9f24a2 --- /dev/null +++ b/scripts/unsorted/audit/compare_album_tracks.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +python3 "$SCRIPT_DIR/compare_album_tracks.py" "$@" diff --git a/scripts/unsorted/audit/compare_albums.py b/scripts/unsorted/audit/compare_albums.py new file mode 100644 index 0000000..9d271e7 --- /dev/null +++ b/scripts/unsorted/audit/compare_albums.py @@ -0,0 +1,294 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import statistics +import subprocess +import sys +from pathlib import Path +from typing import Dict, List + + +SCRIPT_DIR = Path(__file__).resolve().parent +BEETS_DIR = SCRIPT_DIR.parent +MUSIC_ROOT = BEETS_DIR.parent +OUT_DIR = SCRIPT_DIR + +SEP = "\x1f" + +ALBUM_FIELDS = [ + "id", + "albumartist", + "album", + "year", + "albumtype", + "albumtotal", + "mb_albumid", + "mb_releasegroupid", + "albumdisambig", + "label", + "catalognum", + "country", + "media", + "path", +] + +TRACK_FIELDS = [ + "id", + "album_id", + "disc", + "track", + "tracktotal", + "title", + "artist", + "format", + "bitdepth", + "samplerate", + "bitrate", + "channels", + "length", + "mb_trackid", + "path", +] + +LOSSLESS_FORMATS = { + "flac", + "alac", + "wav", + "aiff", + "aif", + "ape", + "wavpack", + "wv", + "dsf", + "dff", +} + + +def relative_path(path: str) -> str: + if not path: + return "" + + p = Path(path) + + try: + return str(p.relative_to(MUSIC_ROOT)) + except ValueError: + return path + + +def run_beet(fields: List[str], args: List[str]) -> List[Dict[str, str]]: + fmt = SEP.join(f"${field}" for field in fields) + + result = subprocess.run( + ["beet", *args, "-f", fmt], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + check=True, + ) + + rows: List[Dict[str, str]] = [] + + for line in result.stdout.splitlines(): + parts = line.split(SEP) + + if len(parts) != len(fields): + continue + + row = dict(zip(fields, parts)) + row["relpath"] = relative_path(row.get("path", "")) + rows.append(row) + + return rows + + +def get_album(album_id: str) -> Dict[str, str]: + rows = run_beet(ALBUM_FIELDS, ["ls", "-a", f"id:{album_id}"]) + + if not rows: + raise SystemExit(f"No album found with album id {album_id}") + + if len(rows) > 1: + raise SystemExit(f"More than one album matched id {album_id}; this should not happen") + + return rows[0] + + +def get_tracks(album_id: str) -> List[Dict[str, str]]: + rows = run_beet(TRACK_FIELDS, ["ls", f"album_id:{album_id}"]) + + def sort_key(row: Dict[str, str]) -> tuple[int, int, int]: + return ( + int_or_zero(row.get("disc", "")), + int_or_zero(row.get("track", "")), + int_or_zero(row.get("id", "")), + ) + + return sorted(rows, key=sort_key) + + +def int_or_zero(value: str) -> int: + try: + return int(float(value)) + except Exception: + return 0 + + +def float_or_zero(value: str) -> float: + try: + return float(value) + except Exception: + return 0.0 + + +def summarize_quality(tracks: List[Dict[str, str]]) -> Dict[str, object]: + formats = sorted({row["format"] for row in tracks if row.get("format")}) + bitdepths = [int_or_zero(row.get("bitdepth", "")) for row in tracks if int_or_zero(row.get("bitdepth", ""))] + samplerates = [int_or_zero(row.get("samplerate", "")) for row in tracks if int_or_zero(row.get("samplerate", ""))] + bitrates = [int_or_zero(row.get("bitrate", "")) for row in tracks if int_or_zero(row.get("bitrate", ""))] + lengths = [float_or_zero(row.get("length", "")) for row in tracks if float_or_zero(row.get("length", ""))] + + normalized_formats = {fmt.casefold() for fmt in formats} + all_lossless = bool(normalized_formats) and normalized_formats.issubset(LOSSLESS_FORMATS) + any_lossless = bool(normalized_formats.intersection(LOSSLESS_FORMATS)) + + return { + "track_count": len(tracks), + "formats": ", ".join(formats) if formats else "", + "all_lossless": all_lossless, + "any_lossless": any_lossless, + "max_bitdepth": max(bitdepths) if bitdepths else "", + "max_samplerate": max(samplerates) if samplerates else "", + "avg_bitrate": round(statistics.mean(bitrates)) if bitrates else "", + "total_length": round(sum(lengths), 2) if lengths else "", + } + + +def album_quality_score(summary: Dict[str, object]) -> tuple: + return ( + 1 if summary["all_lossless"] else 0, + 1 if summary["any_lossless"] else 0, + int(summary["max_bitdepth"] or 0), + int(summary["max_samplerate"] or 0), + int(summary["avg_bitrate"] or 0), + int(summary["track_count"] or 0), + ) + + +def format_seconds(seconds: object) -> str: + try: + total = int(float(seconds)) + except Exception: + return "" + + minutes, sec = divmod(total, 60) + hours, minutes = divmod(minutes, 60) + + if hours: + return f"{hours}:{minutes:02d}:{sec:02d}" + + return f"{minutes}:{sec:02d}" + + +def render_album(album: Dict[str, str], tracks: List[Dict[str, str]]) -> str: + summary = summarize_quality(tracks) + + lines: List[str] = [] + + lines.append("=" * 100) + lines.append(f"id={album['id']}") + lines.append(f"albumartist: {album['albumartist']}") + lines.append(f"album: {album['album']}") + lines.append(f"year/type: {album['year']} / {album['albumtype']}") + lines.append(f"tracks: album field={album['albumtotal']} actual={summary['track_count']}") + lines.append(f"mb_albumid: {album['mb_albumid']}") + lines.append(f"rgid: {album['mb_releasegroupid']}") + lines.append(f"disambig: {album['albumdisambig']}") + lines.append(f"label/cat: {album['label']} / {album['catalognum']}") + lines.append(f"country: {album['country']}") + lines.append(f"media: {album['media']}") + lines.append(f"path: {album['relpath']}") + lines.append("") + lines.append("QUALITY") + lines.append(f" formats: {summary['formats']}") + lines.append(f" all lossless: {summary['all_lossless']}") + lines.append(f" any lossless: {summary['any_lossless']}") + lines.append(f" max bitdepth: {summary['max_bitdepth']}") + lines.append(f" max samplerate: {summary['max_samplerate']}") + lines.append(f" avg bitrate: {summary['avg_bitrate']}") + lines.append(f" total length: {format_seconds(summary['total_length'])}") + lines.append("") + lines.append("TRACKS") + + for row in tracks: + lines.append( + f" {row['disc']}-{row['track']}/{row['tracktotal']} | " + f"{row['title']} | " + f"{row['format']} | " + f"{row['bitdepth']}bit | " + f"{row['samplerate']} | " + f"{row['bitrate']} | " + f"{format_seconds(row['length'])}" + ) + + return "\n".join(lines) + + +def main(argv: List[str]) -> int: + album_ids = argv[1:] + + if not album_ids: + print("Usage: compare_albums.py [ ...]", file=sys.stderr) + return 2 + + albums = [] + track_groups = [] + + for album_id in album_ids: + album = get_album(album_id) + tracks = get_tracks(album_id) + albums.append(album) + track_groups.append(tracks) + + lines: List[str] = [] + lines.append(f"Album comparison: {' '.join(album_ids)}") + lines.append("") + + scored = [] + + for album, tracks in zip(albums, track_groups): + summary = summarize_quality(tracks) + scored.append((album_quality_score(summary), album["id"], album, summary)) + + lines.append("QUALITY RANKING") + for score, album_id, album, summary in sorted(scored, reverse=True): + lines.append( + f" id={album_id} | " + f"score={score} | " + f"{album['albumartist']} - {album['album']} | " + f"{summary['formats']} | " + f"{summary['max_bitdepth']}bit | " + f"{summary['max_samplerate']}Hz | " + f"{summary['track_count']} tracks" + ) + + lines.append("") + + for album, tracks in zip(albums, track_groups): + lines.append(render_album(album, tracks)) + lines.append("") + + output = "\n".join(lines) + + safe_ids = "-".join(album_ids) + output_path = OUT_DIR / f"audit.compare-albums-{safe_ids}.txt" + output_path.write_text(output, encoding="utf-8") + + print(output) + print(f"Wrote: {output_path}") + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv)) diff --git a/scripts/unsorted/audit/compare_albums.sh b/scripts/unsorted/audit/compare_albums.sh new file mode 100755 index 0000000..f83992b --- /dev/null +++ b/scripts/unsorted/audit/compare_albums.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +python3 "$SCRIPT_DIR/compare_albums.py" "$@" diff --git a/scripts/unsorted/audit/field_audit.py b/scripts/unsorted/audit/field_audit.py new file mode 100644 index 0000000..e86f47e --- /dev/null +++ b/scripts/unsorted/audit/field_audit.py @@ -0,0 +1,319 @@ +#!/usr/bin/env python3 +import argparse +import csv +import datetime as dt +import os +import sqlite3 +from pathlib import Path + +COMMON_PURPOSE = { + "id": "Internal beets item/album row id.", + "path": "Canonical file path tracked by beets.", + "album_id": "Internal link from item to album.", + "title": "Track title.", + "artist": "Track artist.", + "artist_sort": "Sort name for track artist.", + "artist_credit": "Displayed track artist credit.", + "album": "Album/release title.", + "albumartist": "Album-level artist.", + "albumartist_sort": "Sort name for album artist.", + "albumartist_credit": "Displayed album artist credit.", + "genre": "Genre field, often populated by lastgenre or source tags.", + "composer": "Composer credit.", + "grouping": "Grouping/work/category field; often inconsistent unless intentionally used.", + "year": "Release year for the specific release.", + "month": "Release month for the specific release.", + "day": "Release day for the specific release.", + "original_year": "Original release year.", + "original_month": "Original release month.", + "original_day": "Original release day.", + "track": "Track number.", + "tracktotal": "Total track count.", + "disc": "Disc number.", + "disctotal": "Total disc count.", + "lyrics": "Embedded/synced lyrics text.", + "comments": "Freeform comments; frequently contains ripper/source cruft.", + "bpm": "Tempo in beats per minute.", + "comp": "Compilation flag.", + "albumtype": "MusicBrainz release-group/release type, e.g. album/single/ep.", + "label": "Release label.", + "asin": "Amazon ASIN; usually low-value unless you care about legacy retail IDs.", + "catalognum": "Label catalog number.", + "script": "Language script code.", + "language": "Release/track language code.", + "country": "Release country.", + "albumstatus": "Release status, e.g. official/bootleg/promotion.", + "media": "Release medium, e.g. CD, Digital Media, Vinyl.", + "albumdisambig": "MusicBrainz album disambiguation comment.", + "releasegroupdisambig": "MusicBrainz release group disambiguation comment.", + "trackdisambig": "MusicBrainz recording/track disambiguation comment.", + "disctitle": "Disc subtitle.", + "encoder": "Encoder tag; often file-origin cruft.", + "length": "Audio duration.", + "bitrate": "Audio bitrate.", + "bitrate_mode": "MP3 bitrate mode.", + "encoder_info": "Encoder information detected from file.", + "encoder_settings": "Encoder settings detected from file.", + "format": "Audio container/codec format.", + "channels": "Audio channel count.", + "bitdepth": "Audio bit depth where available.", + "samplerate": "Audio sample rate.", + "mb_trackid": "MusicBrainz recording id.", + "mb_releasetrackid": "MusicBrainz release-track id.", + "mb_albumid": "MusicBrainz release id.", + "mb_artistid": "MusicBrainz track artist id.", + "mb_albumartistid": "MusicBrainz album artist id.", + "mb_releasegroupid": "MusicBrainz release group id.", + "acoustid_fingerprint": "AcoustID fingerprint; large and rarely useful to keep embedded.", + "acoustid_id": "AcoustID id.", + "mtime": "File modification time tracked by beets.", + "added": "Date/time item was added to beets.", +} + +KEEP_FIELDS = { + "id", "path", "album_id", + "title", "artist", "artist_sort", "artist_credit", + "album", "albumartist", "albumartist_sort", "albumartist_credit", + "year", "month", "day", "original_year", "original_month", "original_day", + "track", "tracktotal", "disc", "disctotal", + "comp", "albumtype", "label", "catalognum", "country", "albumstatus", + "media", "albumdisambig", "releasegroupdisambig", "trackdisambig", + "disctitle", + "length", "bitrate", "bitrate_mode", "format", "channels", "bitdepth", + "samplerate", + "mb_trackid", "mb_releasetrackid", "mb_albumid", "mb_artistid", + "mb_albumartistid", "mb_releasegroupid", + "mtime", "added", +} + +REVIEW_FIELDS = { + "genre", "comments", "grouping", "composer", "lyrics", "bpm", + "encoder", "encoder_info", "encoder_settings", + "asin", "script", "language", + "acoustid_fingerprint", "acoustid_id", +} + +def qident(name: str) -> str: + return '"' + name.replace('"', '""') + '"' + +def as_text(value, limit=120) -> str: + if value is None: + return "" + if isinstance(value, bytes): + value = value.decode("utf-8", errors="replace") + value = str(value).replace("\n", "\\n").replace("\r", "\\r") + if len(value) > limit: + value = value[:limit] + "…" + return value + +def get_tables(conn): + return [ + r[0] for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name" + ) + ] + +def get_columns(conn, table): + return [r[1] for r in conn.execute(f"PRAGMA table_info({qident(table)})")] + +def nonempty_expr(col): + qc = qident(col) + return f"{qc} IS NOT NULL AND TRIM(CAST({qc} AS TEXT)) <> ''" + +def count_fixed_field(conn, table, field, total): + qt = qident(table) + qf = qident(field) + where = nonempty_expr(field) + + populated = conn.execute( + f"SELECT COUNT(*) FROM {qt} WHERE {where}" + ).fetchone()[0] + + distinct = conn.execute( + f"SELECT COUNT(DISTINCT {qf}) FROM {qt} WHERE {where}" + ).fetchone()[0] + + samples = conn.execute( + f""" + SELECT {qf}, COUNT(*) AS n + FROM {qt} + WHERE {where} + GROUP BY {qf} + ORDER BY n DESC + LIMIT 8 + """ + ).fetchall() + + return populated, distinct, "; ".join( + f"{as_text(v)} ({n})" for v, n in samples + ) + +def detect_flex_tables(conn): + flex = [] + for table in get_tables(conn): + cols = set(get_columns(conn, table)) + if {"entity_id", "key", "value"}.issubset(cols): + if "item" in table: + scope = "item" + entity_table = "items" + elif "album" in table: + scope = "album" + entity_table = "albums" + else: + scope = table + entity_table = None + flex.append((table, scope, entity_table)) + return flex + +def suggest_action(field, storage, populated): + if populated == 0: + return "empty / ignore" + if storage.endswith(".flex"): + return "review flexible attr" + if field in KEEP_FIELDS: + return "keep" + if field in REVIEW_FIELDS: + return "review" + return "review unknown" + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--db", required=True, help="Path to beets musiclibrary.db") + ap.add_argument("--outdir", default=".", help="Output directory") + args = ap.parse_args() + + db_path = Path(args.db) + outdir = Path(args.outdir) + outdir.mkdir(parents=True, exist_ok=True) + + conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) + conn.row_factory = sqlite3.Row + + rows = [] + totals = {} + for table, scope in [("items", "item"), ("albums", "album")]: + if table not in get_tables(conn): + continue + total = conn.execute(f"SELECT COUNT(*) FROM {qident(table)}").fetchone()[0] + totals[table] = total + for field in get_columns(conn, table): + populated, distinct, samples = count_fixed_field(conn, table, field, total) + rows.append({ + "scope": scope, + "field": field, + "storage": f"{table}.fixed", + "total_entities": total, + "populated_count": populated, + "populated_pct": round((populated / total * 100), 2) if total else 0, + "distinct_count": distinct, + "sample_values": samples, + "purpose": COMMON_PURPOSE.get(field, "No built-in note. Review manually."), + "suggested_action": suggest_action(field, f"{table}.fixed", populated), + }) + + for flex_table, scope, entity_table in detect_flex_tables(conn): + total = totals.get(entity_table) if entity_table else None + qft = qident(flex_table) + keys = [ + r[0] for r in conn.execute( + f""" + SELECT key + FROM {qft} + WHERE value IS NOT NULL AND TRIM(CAST(value AS TEXT)) <> '' + GROUP BY key + ORDER BY key + """ + ) + ] + for key in keys: + populated = conn.execute( + f""" + SELECT COUNT(DISTINCT entity_id) + FROM {qft} + WHERE key = ? + AND value IS NOT NULL + AND TRIM(CAST(value AS TEXT)) <> '' + """, + (key,), + ).fetchone()[0] + + distinct = conn.execute( + f""" + SELECT COUNT(DISTINCT value) + FROM {qft} + WHERE key = ? + AND value IS NOT NULL + AND TRIM(CAST(value AS TEXT)) <> '' + """, + (key,), + ).fetchone()[0] + + samples = conn.execute( + f""" + SELECT value, COUNT(*) AS n + FROM {qft} + WHERE key = ? + AND value IS NOT NULL + AND TRIM(CAST(value AS TEXT)) <> '' + GROUP BY value + ORDER BY n DESC + LIMIT 8 + """, + (key,), + ).fetchall() + + rows.append({ + "scope": scope, + "field": key, + "storage": f"{flex_table}.flex", + "total_entities": total or "", + "populated_count": populated, + "populated_pct": round((populated / total * 100), 2) if total else "", + "distinct_count": distinct, + "sample_values": "; ".join(f"{as_text(v)} ({n})" for v, n in samples), + "purpose": COMMON_PURPOSE.get(key, "Flexible/custom field. Review whether you intentionally use it."), + "suggested_action": suggest_action(key, f"{flex_table}.flex", populated), + }) + + rows.sort(key=lambda r: (r["scope"], r["suggested_action"], r["field"])) + + csv_path = outdir / "beets_db_field_counts.csv" + with csv_path.open("w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=[ + "scope", "field", "storage", "total_entities", + "populated_count", "populated_pct", "distinct_count", + "suggested_action", "purpose", "sample_values", + ]) + writer.writeheader() + writer.writerows(rows) + + md_path = outdir / "beets_tag_review.md" + with md_path.open("w", encoding="utf-8") as f: + f.write("# Beets Tag Review\n\n") + f.write(f"Generated: {dt.datetime.now().isoformat(timespec='seconds')}\n\n") + f.write("This document audits fields stored in the beets SQLite database. It does not prove which raw tag frames are embedded in the audio files.\n\n") + f.write("## Summary\n\n") + for scope in sorted(set(r["scope"] for r in rows)): + sr = [r for r in rows if r["scope"] == scope] + f.write(f"- {scope}: {len(sr)} fields/flexible attributes\n") + f.write("\n## Fields\n\n") + + for r in rows: + f.write(f"> [!info]- `{r['field']}` — {r['scope']} — {r['suggested_action']}\n") + f.write(f"> Storage: `{r['storage']}` \n") + f.write(f"> Populated: `{r['populated_count']}` / `{r['total_entities']}`") + if r["populated_pct"] != "": + f.write(f" (`{r['populated_pct']}%`)") + f.write(" \n") + f.write(f"> Distinct values: `{r['distinct_count']}` \n") + f.write(f"> Purpose: {r['purpose']} \n") + if r["sample_values"]: + f.write(f"> Samples: {r['sample_values']} \n") + f.write(">\n") + f.write("> Decision: keep / strip-from-files / remove-flexattr / ignore\n\n") + + print(f"Wrote {csv_path}") + print(f"Wrote {md_path}") + +if __name__ == "__main__": + main() diff --git a/scripts/unsorted/audit/find_duplicates.py b/scripts/unsorted/audit/find_duplicates.py new file mode 100644 index 0000000..180515f --- /dev/null +++ b/scripts/unsorted/audit/find_duplicates.py @@ -0,0 +1,47 @@ +#1/usr/bin/env python3 + +import sys +import re +import unicodedata +from collections import defaultdict + +def norm(s: str) -> str: + s = unicodedata.normalize("NFKC", s) + s = s.replace("’", "'").replace("‘", "'").replace("`", "'").replace("´", "'") + s = s.replace("‐", "-").replace("-", "-").replace("‒", "-").replace("–", "-").replace("—", "-") + s = s.replace("“", '"').replace("”", '"') + s = re.sub(r"\s+", " ", s).strip() + return s.casefold() + +groups = defaultdict(list) + +for line in sys.stdin: + line = line.rstrip("\n") + if not line: + continue + + parts = line.split("\t") + if len(parts) != 9: + print(f"BAD LINE: {line!r}", file=sys.stderr) + continue + + album_id, albumartist, album, year, albumtype, albumtotal, mb_albumid, mb_releasegroupid, path = parts + key = (norm(albumartist), norm(album)) + groups[key].append(parts) + +for key, rows in sorted(groups.items()): + if len(rows) <= 1: + continue + + print("=" * 100) + print(f"NORMALIZED DUPLICATE: {key[0]} / {key[1]}") + for row in rows: + album_id, albumartist, album, year, albumtype, albumtotal, mb_albumid, mb_releasegroupid, path = row + print(f"id={album_id}") + print(f" albumartist: {albumartist}") + print(f" album: {album}") + print(f" year/type: {year} / {albumtype}") + print(f" tracks: {albumtotal}") + print(f" mb_albumid: {mb_albumid}") + print(f" rgid: {mb_releasegroupid}") + print(f" path: {path}") diff --git a/scripts/unsorted/audit/find_duplicates.sh b/scripts/unsorted/audit/find_duplicates.sh new file mode 100755 index 0000000..665bf4f --- /dev/null +++ b/scripts/unsorted/audit/find_duplicates.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash + +beet ls -a -f '$id $albumartist $album $year $albumtype $albumtotal $mb_albumid $mb_releasegroupid $path' \ + | python3 /music/beets/audit/find_duplicates.py \ + | tee /music/beets/audit/audit.normalized-album-dupes.txt diff --git a/scripts/unsorted/audit/quality_audit.py b/scripts/unsorted/audit/quality_audit.py new file mode 100644 index 0000000..6ed6676 --- /dev/null +++ b/scripts/unsorted/audit/quality_audit.py @@ -0,0 +1,343 @@ +#!/usr/bin/env python3 + +import argparse +import csv +import json +import subprocess +import sys +from pathlib import Path +from typing import Any, Optional + + +AUDIO_EXTS = { + ".flac", ".mp3", ".m4a", ".mp4", ".aac", ".wav", ".aiff", ".aif", + ".ogg", ".opus", ".wma", ".ape", ".wv", ".dsf", ".dff", +} + +LOSSLESS_CODECS = { + "flac", + "alac", + "wavpack", + "ape", + "tta", + "pcm_s16le", "pcm_s24le", "pcm_s32le", + "pcm_s16be", "pcm_s24be", "pcm_s32be", + "pcm_f32le", "pcm_f64le", +} + +LOSSY_CODECS = { + "mp3", + "aac", + "opus", + "vorbis", + "wma", +} + + +def parse_int(value: Any) -> Optional[int]: + if value in (None, "", "N/A"): + return None + try: + return int(float(value)) + except (TypeError, ValueError): + return None + + +def parse_float(value: Any) -> Optional[float]: + if value in (None, "", "N/A"): + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def ffprobe(path: Path) -> tuple[Optional[dict[str, Any]], Optional[str]]: + cmd = [ + "ffprobe", + "-hide_banner", + "-v", "error", + "-show_error", + "-show_format", + "-show_streams", + "-of", "json", + str(path), + ] + + proc = subprocess.run( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + if proc.returncode != 0: + return None, proc.stderr.strip() or f"ffprobe exited {proc.returncode}" + + try: + return json.loads(proc.stdout), None + except json.JSONDecodeError as exc: + return None, f"ffprobe JSON parse error: {exc}" + + +def first_audio_stream(data: dict[str, Any]) -> Optional[dict[str, Any]]: + for stream in data.get("streams", []): + if stream.get("codec_type") == "audio": + return stream + return None + + +def get_bit_depth(stream: dict[str, Any]) -> Optional[int]: + for key in ("bits_per_raw_sample", "bits_per_sample"): + value = parse_int(stream.get(key)) + if value and value > 0: + return value + + # Useful fallback for PCM. Avoid pretending lossy AAC/MP3 sample_fmt is source bit depth. + codec = stream.get("codec_name") + sample_fmt = stream.get("sample_fmt") or "" + if codec and codec.startswith("pcm_"): + if "s16" in sample_fmt or "16" in codec: + return 16 + if "s24" in sample_fmt or "24" in codec: + return 24 + if "s32" in sample_fmt or "32" in codec: + return 32 + + return None + + +def issue_join(issues: list[str]) -> str: + return ";".join(issues) + + +def audit_file(path: Path) -> dict[str, Any]: + stat_size = path.stat().st_size + data, probe_error = ffprobe(path) + + row: dict[str, Any] = { + "path": str(path), + "extension": path.suffix.lower(), + "probe_error": probe_error or "", + "codec": "", + "container": "", + "duration_sec": "", + "size_bytes": stat_size, + "actual_kbps": "", + "stream_kbps": "", + "sample_rate": "", + "channels": "", + "bit_depth": "", + "lossless": "", + "pcm_kbps": "", + "compression_ratio": "", + "issues": "", + } + + issues: list[str] = [] + + if probe_error: + issues.append("probe_error") + row["issues"] = issue_join(issues) + return row + + assert data is not None + fmt = data.get("format", {}) + stream = first_audio_stream(data) + + row["container"] = fmt.get("format_name", "") + + if stream is None: + issues.append("no_audio_stream") + row["issues"] = issue_join(issues) + return row + + codec = stream.get("codec_name") or "" + row["codec"] = codec + + duration = parse_float(stream.get("duration")) or parse_float(fmt.get("duration")) + sample_rate = parse_int(stream.get("sample_rate")) + channels = parse_int(stream.get("channels")) + bit_depth = get_bit_depth(stream) + stream_kbps = None + + stream_br = parse_int(stream.get("bit_rate")) + if stream_br: + stream_kbps = stream_br / 1000.0 + + actual_kbps = None + if duration and duration > 0: + actual_kbps = stat_size * 8 / duration / 1000.0 + + is_lossless = codec in LOSSLESS_CODECS or codec.startswith("pcm_") + is_lossy = codec in LOSSY_CODECS + + pcm_kbps = None + compression_ratio = None + if is_lossless and sample_rate and channels and bit_depth: + pcm_kbps = sample_rate * channels * bit_depth / 1000.0 + if actual_kbps and pcm_kbps > 0: + compression_ratio = actual_kbps / pcm_kbps + + row["duration_sec"] = f"{duration:.3f}" if duration else "" + row["actual_kbps"] = f"{actual_kbps:.1f}" if actual_kbps else "" + row["stream_kbps"] = f"{stream_kbps:.1f}" if stream_kbps else "" + row["sample_rate"] = sample_rate or "" + row["channels"] = channels or "" + row["bit_depth"] = bit_depth or "" + row["lossless"] = "yes" if is_lossless else "no" + row["pcm_kbps"] = f"{pcm_kbps:.1f}" if pcm_kbps else "" + row["compression_ratio"] = f"{compression_ratio:.3f}" if compression_ratio else "" + + ext = path.suffix.lower() + + # Container/extension mismatch-ish checks. + if ext == ".flac" and codec != "flac": + issues.append("extension_flac_but_codec_not_flac") + + if ext in {".m4a", ".mp4"} and codec not in {"aac", "alac"}: + issues.append("m4a_mp4_unexpected_audio_codec") + + if ext == ".wav" and not codec.startswith("pcm_"): + issues.append("wav_not_pcm_codec") + + # Basic structural oddities. + if not duration or duration <= 0: + issues.append("bad_or_missing_duration") + + if duration and duration < 3: + issues.append("very_short_track") + + if channels and channels > 2: + issues.append("multichannel_audio") + + if sample_rate and sample_rate not in {22050, 32000, 44100, 48000, 88200, 96000, 176400, 192000}: + issues.append("unusual_sample_rate") + + if is_lossless and not bit_depth: + issues.append("lossless_missing_bit_depth") + + if bit_depth and bit_depth not in {16, 24, 32}: + issues.append("unusual_bit_depth") + + # Suspicious low-bitrate lossless files. + if ( + is_lossless + and actual_kbps + and channels + and sample_rate + and channels >= 2 + and sample_rate >= 44100 + ): + if actual_kbps < 512: + issues.append("suspicious_low_lossless_bitrate_under_512kbps") + + if compression_ratio and compression_ratio < 0.35: + issues.append("suspicious_low_lossless_compression_ratio_under_0.35") + + # Hi-res that compresses suspiciously tiny. + if ( + is_lossless + and actual_kbps + and sample_rate + and bit_depth + and sample_rate >= 88200 + and bit_depth >= 24 + and actual_kbps < 900 + ): + issues.append("suspicious_low_hires_lossless_bitrate") + + # Near-PCM size can indicate noise, bad compression, or just unusual content. + if is_lossless and compression_ratio and compression_ratio > 0.95: + issues.append("lossless_near_uncompressed_size") + + # Lossy oddities. + if is_lossy and actual_kbps and actual_kbps < 96 and duration and duration > 30: + issues.append("very_low_lossy_bitrate_under_96kbps") + + if codec == "mp3" and actual_kbps and actual_kbps > 360: + issues.append("mp3_actual_bitrate_over_360kbps_possible_artwork_or_bad_metadata") + + if codec == "aac" and actual_kbps and actual_kbps > 500: + issues.append("aac_actual_bitrate_over_500kbps_possible_artwork_or_bad_metadata") + + # Stream/container overhead. This catches files with huge embedded art or weird metadata. + if stream_kbps and actual_kbps and actual_kbps > stream_kbps * 1.35 and stat_size > 10_000_000: + issues.append("large_container_overhead_possible_embedded_artwork") + + row["issues"] = issue_join(issues) + return row + + +def iter_audio_files(root: Path): + for path in root.rglob("*"): + if path.is_file() and path.suffix.lower() in AUDIO_EXTS: + yield path + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("root", type=Path, help="music library root, e.g. /music/library") + parser.add_argument("--out", type=Path, default=Path("quality-audit.csv")) + parser.add_argument("--issues-out", type=Path, default=Path("quality-issues.csv")) + args = parser.parse_args() + + files = sorted(iter_audio_files(args.root)) + if not files: + print(f"No audio files found under {args.root}", file=sys.stderr) + return 1 + + fieldnames = [ + "path", + "extension", + "probe_error", + "codec", + "container", + "duration_sec", + "size_bytes", + "actual_kbps", + "stream_kbps", + "sample_rate", + "channels", + "bit_depth", + "lossless", + "pcm_kbps", + "compression_ratio", + "issues", + ] + + all_rows = [] + issue_rows = [] + + total = len(files) + for i, path in enumerate(files, 1): + print(f"[{i}/{total}] {path}", file=sys.stderr) + row = audit_file(path) + all_rows.append(row) + if row["issues"]: + issue_rows.append(row) + + args.out.parent.mkdir(parents=True, exist_ok=True) + args.issues_out.parent.mkdir(parents=True, exist_ok=True) + + with args.out.open("w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(all_rows) + + with args.issues_out.open("w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(issue_rows) + + print() + print(f"Wrote full audit: {args.out}") + print(f"Wrote issue audit: {args.issues_out}") + print(f"Files scanned: {len(all_rows)}") + print(f"Files with issues: {len(issue_rows)}") + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/unsorted/audit/quality_audit.sh b/scripts/unsorted/audit/quality_audit.sh new file mode 100755 index 0000000..37b6a55 --- /dev/null +++ b/scripts/unsorted/audit/quality_audit.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash +python3 /music/beets/audit/quality_audit.py \ + /music/library \ + --out /music/beets/audit/quality-audit.csv \ + --issues-out /music/beets/audit/quality-issues.csv diff --git a/scripts/unsorted/audit/quality_check.py b/scripts/unsorted/audit/quality_check.py new file mode 100755 index 0000000..f267aee --- /dev/null +++ b/scripts/unsorted/audit/quality_check.py @@ -0,0 +1,22 @@ +#!/usr/bin/env python3 +import csv + +path = "/music/beets/audit/quality-issues.csv" + +with open(path, newline="", encoding="utf-8") as f: + rows = list(csv.DictReader(f)) + +wanted = [ + r for r in rows + if "suspicious_low_lossless" in r["issues"] + or "suspicious_low_hires" in r["issues"] + or "extension_flac_but_codec_not_flac" in r["issues"] +] + +for r in wanted[:100]: + print() + print(r["issues"]) + print(f"{r['actual_kbps']} kbps | ratio={r['compression_ratio']} | {r['sample_rate']} Hz | {r['bit_depth']}-bit | {r['channels']} ch | {r['codec']}") + print(r["path"]) + +print(f"\nshown: {min(len(wanted), 100)} / {len(wanted)}") diff --git a/scripts/unsorted/audit/raw_tag_audit.py b/scripts/unsorted/audit/raw_tag_audit.py new file mode 100644 index 0000000..6820370 --- /dev/null +++ b/scripts/unsorted/audit/raw_tag_audit.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python3 +import argparse +import csv +import sqlite3 +from collections import defaultdict +from pathlib import Path + +try: + from mutagen import File as MutagenFile +except ImportError: + raise SystemExit("Missing mutagen. Install it in your beets environment: pip install mutagen") + +def qident(name: str) -> str: + return '"' + name.replace('"', '""') + '"' + +def decode_path(value): + if isinstance(value, bytes): + return value.decode("utf-8", errors="surrogateescape") + return str(value) + +def short(value, limit=160): + if value is None: + return "" + if isinstance(value, (list, tuple)): + value = "; ".join(map(str, value[:3])) + value = str(value).replace("\n", "\\n").replace("\r", "\\r") + return value[:limit] + ("…" if len(value) > limit else "") + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--db", required=True) + ap.add_argument("--outdir", default=".") + args = ap.parse_args() + + outdir = Path(args.outdir) + outdir.mkdir(parents=True, exist_ok=True) + + conn = sqlite3.connect(f"file:{Path(args.db)}?mode=ro", uri=True) + conn.row_factory = sqlite3.Row + + rows = conn.execute("SELECT id, path, format FROM items ORDER BY id").fetchall() + + counts = defaultdict(int) + samples = {} + errors = [] + + for row in rows: + item_id = row["id"] + fmt = row["format"] or "" + path = decode_path(row["path"]) + + try: + audio = MutagenFile(path, easy=False) + if audio is None: + errors.append((item_id, path, "mutagen could not identify file")) + continue + + if getattr(audio, "tags", None): + for key, value in audio.tags.items(): + k = (fmt, str(key)) + counts[k] += 1 + samples.setdefault(k, short(value)) + + if hasattr(audio, "pictures") and audio.pictures: + k = (fmt, "__embedded_picture__") + counts[k] += len(audio.pictures) + samples.setdefault(k, f"{len(audio.pictures)} picture(s) in sample file") + + except Exception as e: + errors.append((item_id, path, repr(e))) + + counts_path = outdir / "raw_file_tag_counts.csv" + with counts_path.open("w", newline="", encoding="utf-8") as f: + writer = csv.writer(f) + writer.writerow(["format", "raw_tag_key", "files_or_occurrences", "sample_value"]) + for (fmt, key), n in sorted(counts.items(), key=lambda x: (-x[1], x[0][0], x[0][1])): + writer.writerow([fmt, key, n, samples.get((fmt, key), "")]) + + errors_path = outdir / "raw_file_tag_errors.csv" + with errors_path.open("w", newline="", encoding="utf-8") as f: + writer = csv.writer(f) + writer.writerow(["item_id", "path", "error"]) + writer.writerows(errors) + + print(f"Wrote {counts_path}") + print(f"Wrote {errors_path}") + +if __name__ == "__main__": + main() diff --git a/scripts/unsorted/audit/release_dupes.py b/scripts/unsorted/audit/release_dupes.py new file mode 100644 index 0000000..6bc279e --- /dev/null +++ b/scripts/unsorted/audit/release_dupes.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python3 + +import sys +from collections import defaultdict + +groups = defaultdict(list) + +for line in sys.stdin: + line = line.rstrip("\n") + if not line: + continue + + parts = line.split("\t") + if len(parts) != 9: + print(f"BAD LINE: {line!r}", file=sys.stderr) + continue + + album_id, albumartist, album, year, albumtype, albumtotal, mb_albumid, mb_releasegroupid, path = parts + + if not mb_releasegroupid: + continue + + groups[mb_releasegroupid].append(parts) + +for rgid, rows in sorted(groups.items()): + if len(rows) <= 1: + continue + + print("=" * 100) + print(f"RELEASE GROUP DUPLICATE: {rgid}") + for row in rows: + album_id, albumartist, album, year, albumtype, albumtotal, mb_albumid, mb_releasegroupid, path = row + print(f"id={album_id}") + print(f" albumartist: {albumartist}") + print(f" album: {album}") + print(f" year/type: {year} / {albumtype}") + print(f" tracks: {albumtotal}") + print(f" mb_albumid: {mb_albumid}") + print(f" path: {path}") diff --git a/scripts/unsorted/audit/release_dupes.sh b/scripts/unsorted/audit/release_dupes.sh new file mode 100755 index 0000000..851c80d --- /dev/null +++ b/scripts/unsorted/audit/release_dupes.sh @@ -0,0 +1,5 @@ +#!/usr/bin/env bash + +beet ls -a -f '$id $albumartist $album $year $albumtype $albumtotal $mb_albumid $mb_releasegroupid $path' \ + | python3 /music/beets/audit/release_dupes.py \ + | tee /music/beets/audit/audit.releasegroup-dupes.txt diff --git a/scripts/unsorted/audit/remove-dup.sh b/scripts/unsorted/audit/remove-dup.sh new file mode 100755 index 0000000..3063b6a --- /dev/null +++ b/scripts/unsorted/audit/remove-dup.sh @@ -0,0 +1,18 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="/music/incoming" +TRASH="/music/duplicates" + +find "$ROOT" -type f -iname '*.flac' -print0 | +while IFS= read -r -d '' flac; do + stem="${flac%.*}" + for other in "$stem".*; do + [ "$other" = "$flac" ] && continue + [ -f "$other" ] || continue + + rel="${other#$ROOT/}" + mkdir -p -- "$TRASH/$(dirname "$rel")" + mv -v -- "$other" "$TRASH/$rel" + done +done