344 lines
9.7 KiB
Python
344 lines
9.7 KiB
Python
#!/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())
|