beets-setup/scripts/audit/audit-library-paths

#!/usr/bin/env python3
"""
Audit a beets-managed music library for path layout violations.

Allowed layouts:

  Normal music:
    <root>/<AlbumArtist>/{Live,Remixes,Albums,EPs,Singles,Anthology}/...

  OSTs:
    <root>/OSTs/<Album>/...

Examples:

  PASS:
    /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:
    /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

import argparse
import csv
import os
import sqlite3
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Optional


VALID_BUCKETS = {
    "Live",
    "Remixes",
    "Albums",
    "EPs",
    "Singles",
    "Anthology",
}

OST_TOP_LEVEL = "OSTs"

DEFAULT_AUDIO_EXTS = {
    ".flac",
    ".mp3",
    ".m4a",
    ".aac",
    ".ogg",
    ".opus",
    ".wav",
    ".aiff",
    ".aif",
    ".wma",
    ".alac",
    ".ape",
    ".wv",
    ".mpc",
    ".dsf",
    ".dff",
}


@dataclass
class BeetsItem:
    item_id: int
    path: Path
    albumartist: str
    artist: str
    album: str
    title: str


def decode_beets_path(value) -> str:
    """
    Beets commonly stores paths as bytes/BLOB values in sqlite.
    Decode them safely into a filesystem string.
    """
    if isinstance(value, bytes):
        return os.fsdecode(value)
    return str(value)


def normalize_path(path: Path) -> str:
    """
    Normalize for comparison without requiring the file to exist.
    """
    return str(path.expanduser().resolve(strict=False))


def validate_path(path: Path, root: Path) -> tuple[Optional[str], str]:
    """
    Return (reason, relative_path).

    reason is None when the path is valid.
    """

    path = path.expanduser().resolve(strict=False)
    root = root.expanduser().resolve(strict=False)

    try:
        rel = path.relative_to(root)
    except ValueError:
        return "OUTSIDE_ROOT", str(path)

    parts = rel.parts

    if not parts:
        return "EMPTY_RELATIVE_PATH", str(rel)

    # Special OST rule:
    #
    #   <root>/OSTs/<Album>/...
    #
    # This intentionally does NOT follow:
    #
    #   <root>/<AlbumArtist>/<Bucket>/...
    #
    if parts[0] == OST_TOP_LEVEL:
        if len(parts) < 3:
            return "OST_TOO_SHALLOW", str(rel)
        return None, str(rel)

    # Normal music rule:
    #
    #   <root>/<AlbumArtist>/<Bucket>/...
    #
    # At minimum, an actual track path needs:
    #
    #   AlbumArtist/Bucket/File
    #
    # or:
    #
    #   AlbumArtist/Bucket/Album/File
    #
    # The validator only enforces the structural prefix.
    if len(parts) < 3:
        return "TOO_SHALLOW", str(rel)

    albumartist = parts[0]
    bucket = parts[1]

    if not albumartist:
        return "MISSING_ALBUMARTIST", str(rel)

    if bucket not in VALID_BUCKETS:
        if bucket == "Remix":
            return "BAD_BUCKET_RENAME_TO_REMIXES", str(rel)
        return "BAD_BUCKET", str(rel)

    return None, str(rel)


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}")

    conn = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row

    try:
        rows = conn.execute(
            """
            SELECT
                id,
                path,
                albumartist,
                artist,
                album,
                title
            FROM items
            ORDER BY path
            """
        ).fetchall()
    finally:
        conn.close()

    items: dict[str, BeetsItem] = {}

    for row in rows:
        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(
            item_id=int(row["id"]),
            path=path,
            albumartist=row["albumartist"] or "",
            artist=row["artist"] or "",
            album=row["album"] or "",
            title=row["title"] or "",
        )

    return items


def scan_disk_audio_files(root: Path, audio_exts: set[str]) -> set[str]:
    root = root.expanduser().resolve(strict=False)

    found: set[str] = set()

    skip_dirs = {
        ".zfs",
        ".snapshot",
        "@eaDir",
    }

    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [
            d for d in dirnames
            if d not in skip_dirs
        ]

        for filename in filenames:
            path = Path(dirpath) / filename

            if path.suffix.lower() not in audio_exts:
                continue

            found.add(normalize_path(path))

    return found


def make_finding(
    *,
    path_key: str,
    root: Path,
    beets_items: dict[str, BeetsItem],
    disk_paths: set[str],
) -> Optional[dict[str, str]]:
    path = Path(path_key)

    reason, relative_path = validate_path(path, root)

    if reason is None:
        return None

    beets_item = beets_items.get(path_key)
    in_beets = beets_item is not None
    on_disk = path_key in disk_paths or path.exists()

    if in_beets and path_key in disk_paths:
        source = "beets+disk"
    elif in_beets:
        source = "beets"
    else:
        source = "disk"

    return {
        "source": source,
        "reason": reason,
        "relative_path": relative_path,
        "path": str(path),
        "in_beets": "yes" if in_beets else "no",
        "exists_on_disk": "yes" if on_disk else "no",
        "beets_id": str(beets_item.item_id) if beets_item else "",
        "albumartist": beets_item.albumartist if beets_item else "",
        "artist": beets_item.artist if beets_item else "",
        "album": beets_item.album if beets_item else "",
        "title": beets_item.title if beets_item else "",
    }


def parse_audio_exts(raw_exts: Optional[str]) -> set[str]:
    if not raw_exts:
        return set(DEFAULT_AUDIO_EXTS)

    exts: set[str] = set()

    for ext in raw_exts.split(","):
        ext = ext.strip().lower()

        if not ext:
            continue

        if not ext.startswith("."):
            ext = "." + ext

        exts.add(ext)

    return exts


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Audit music library paths against the required layout."
    )

    parser.add_argument(
        "--root",
        default="/data/Music/Library/originals",
        help="Music library root. Default: /data/Music/Library/originals",
    )

    parser.add_argument(
        "--db",
        default="/data/Music/Library/musiclibrary.db",
        help="Beets sqlite database path. Default: /data/Music/Library/musiclibrary.db",
    )

    parser.add_argument(
        "--disk",
        action="store_true",
        help="Also scan audio files physically present under --root.",
    )

    parser.add_argument(
        "--csv",
        help="Write findings to this CSV file instead of stdout.",
    )

    parser.add_argument(
        "--audio-exts",
        help=(
            "Comma-separated audio extensions for --disk scan. "
            "Default includes common formats like flac, mp3, m4a, opus, wav."
        ),
    )

    args = parser.parse_args()

    root = Path(args.root)
    db_path = Path(args.db)
    audio_exts = parse_audio_exts(args.audio_exts)

    beets_items = load_beets_items(db_path, root)

    disk_paths: set[str] = set()
    if args.disk:
        disk_paths = scan_disk_audio_files(root, audio_exts)

    paths_to_check = set(beets_items.keys())

    if args.disk:
        paths_to_check |= disk_paths

    findings: list[dict[str, str]] = []

    for path_key in sorted(paths_to_check):
        finding = make_finding(
            path_key=path_key,
            root=root,
            beets_items=beets_items,
            disk_paths=disk_paths,
        )

        if finding is not None:
            findings.append(finding)

    fields = [
        "source",
        "reason",
        "relative_path",
        "path",
        "in_beets",
        "exists_on_disk",
        "beets_id",
        "albumartist",
        "artist",
        "album",
        "title",
    ]

    if args.csv:
        out_path = Path(args.csv)
        out_path.parent.mkdir(parents=True, exist_ok=True)

        with out_path.open("w", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=fields)
            writer.writeheader()
            writer.writerows(findings)

        print(f"Wrote {len(findings)} finding(s) to {out_path}")

    else:
        if findings:
            writer = csv.DictWriter(sys.stdout, fieldnames=fields, delimiter="\t")
            writer.writeheader()
            writer.writerows(findings)
        else:
            print("No invalid music paths found.")

    return 1 if findings else 0


if __name__ == "__main__":
    raise SystemExit(main())