#!/usr/bin/env python3
"""
import-dj-collection.py — bulk import the DJ collection from $MUSIC_DATA_DIR/djstuff/
into the beets library + Navidrome dj-* playlists, deduping incoming files against
the library via Chromaprint fingerprints.

Two phases:
  --dry-run (default): fingerprint each DJ file, compare against library, classify
                       as DUPLICATE / NEW / AMBIGUOUS. Write decisions to a state
                       SQLite DB. Emit a per-folder summary. Make no changes.
  --apply            : for each pending state-DB entry, take action:
                         DUPLICATE  → add existing library track to dj-<playlist>
                                      in Navidrome via Subsonic API
                         NEW        → stage in /sldl-dropbox/djstuff/<slug>/,
                                      tag with GROUPING=djstuff + a unique COMMENT
                                      token, run beets import, find the resulting
                                      library path, add to dj-<playlist>
                         AMBIGUOUS  → skip (listed in CSV for manual handling)

The LDSxSupreme/ folder is hard-skipped per plan.

Usage:
    import-dj-collection.py [--dry-run | --apply] [--folder NAME]
                            [--workers N] [--limit N] [--reset]

Examples:
    import-dj-collection.py --dry-run --folder Bass
    import-dj-collection.py --apply --folder Bass
    import-dj-collection.py --dry-run            # all folders, dry
    import-dj-collection.py --apply              # all folders, apply
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import multiprocessing
import os
import re
import shutil
import sqlite3
import subprocess
import sys
import time
import urllib.parse
import urllib.request
import uuid
import xml.etree.ElementTree as ET
from dataclasses import dataclass
from typing import Iterable

import numpy as np

# ============================================================================
# Paths and constants
# ============================================================================

_ALEMBIC_CONFIG_DIR = os.environ.get("ALEMBIC_CONFIG_DIR", "/config")
_MUSIC_DATA_DIR = os.environ.get("MUSIC_DATA_DIR", "/data/music")

DJSTUFF_ROOT = f"{_MUSIC_DATA_DIR}/djstuff"
SKIP_FOLDERS = {"LDSxSupreme"}  # hard-skip per plan

# beets' directory: config may still be the transitional "/music" mount
# (Stages 0-3) or the post-migration MUSIC_DATA_DIR/Library path (Stage 4+);
# these two translators bridge whichever one beets/Navidrome are using to
# this container's own filesystem view. Once beets' directory: matches
# LIBRARY_HOST_ROOT (post Stage-4), container_to_host() becomes a no-op.
LIBRARY_HOST_ROOT = f"{_MUSIC_DATA_DIR}/Library/"
LIBRARY_CONTAINER_ROOT = "/music/"

DROPBOX_HOST_ROOT = f"{_MUSIC_DATA_DIR}/sldl-dropbox"
DROPBOX_DJSTUFF_HOST = f"{DROPBOX_HOST_ROOT}/djstuff"

FINGERPRINT_DB = f"{_ALEMBIC_CONFIG_DIR}/beets/fingerprints.db"
STATE_DB = f"{_ALEMBIC_CONFIG_DIR}/beets/dj-import-state.db"
BEETS_DB = f"{_ALEMBIC_CONFIG_DIR}/beets/data/beets_library.db"
NAVIDROME_DB = f"{_ALEMBIC_CONFIG_DIR}/navidrome/navidrome.db"

LOG_DIR = f"{_ALEMBIC_CONFIG_DIR}/logs"

# Load Navidrome admin creds the same way the shell scripts do.
def _load_navidrome_env():
    path = f"{_ALEMBIC_CONFIG_DIR}/pipeline/navidrome/admin.env"
    env = {}
    if os.path.exists(path):
        for line in open(path):
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, v = line.split("=", 1)
            env[k] = v.strip().strip("'").strip('"')
    return env

_nd_env = _load_navidrome_env()

# Subsonic
ND_BASE = _nd_env.get("ND_BASE", "http://navidrome:4533")
ND_USER = _nd_env.get("ND_USER", "andrew")
ND_PASS = _nd_env.get("ND_PASS", "")
ND_API_VERSION = "1.16.0"
ND_CLIENT = "dj-import"

# Audio extensions we'll process (everything else is ignored)
AUDIO_EXTS = {".flac", ".mp3", ".wav", ".m4a", ".aiff", ".aif", ".ogg"}

# Classification thresholds (acoustid.compare_fingerprints returns 0.0..1.0)
DUPLICATE_THRESHOLD = 0.92
AMBIGUOUS_THRESHOLD = 0.80

DEFAULT_WORKERS = 8


# ============================================================================
# Helpers
# ============================================================================

def container_to_host(p: str) -> str:
    if p.startswith(LIBRARY_CONTAINER_ROOT):
        return LIBRARY_HOST_ROOT + p[len(LIBRARY_CONTAINER_ROOT):]
    return p


def host_to_navidrome_path(host_path: str) -> str:
    """Navidrome stores `media_file.path` relative to its music root."""
    if host_path.startswith(LIBRARY_HOST_ROOT):
        return host_path[len(LIBRARY_HOST_ROOT):]
    if host_path.startswith(LIBRARY_CONTAINER_ROOT):
        return host_path[len(LIBRARY_CONTAINER_ROOT):]
    return host_path


def slugify_folder(name: str) -> str:
    """Folder name → 'dj-<slug>'. Lowercase, &→and, non-alnum→-, collapse repeats."""
    s = name.replace("&", "and").lower()
    s = re.sub(r"[^a-z0-9]+", "-", s)
    s = s.strip("-")
    return f"dj-{s}"


def short_hash(path: str) -> str:
    """Hash of the first 1 MB of a file — used to key state entries stably."""
    h = hashlib.sha256()
    try:
        with open(path, "rb") as f:
            h.update(f.read(1024 * 1024))
    except OSError:
        return ""
    return h.hexdigest()[:16]


def log(msg: str) -> None:
    print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)


# ============================================================================
# Subsonic client (token + salt auth)
# ============================================================================

class Subsonic:
    def __init__(self, base: str, user: str, password: str):
        self.base = base.rstrip("/")
        self.user = user
        self.password = password

    def _auth_params(self) -> dict[str, str]:
        # Use plain-password auth — same convention as existing scripts.
        return {
            "u": self.user,
            "p": self.password,
            "v": ND_API_VERSION,
            "c": ND_CLIENT,
            "f": "xml",
        }

    def _call(self, endpoint: str, params: dict[str, str | list[str]]) -> ET.Element:
        qs_parts: list[tuple[str, str]] = []
        for k, v in {**self._auth_params(), **params}.items():
            if isinstance(v, list):
                for vv in v:
                    qs_parts.append((k, vv))
            else:
                qs_parts.append((k, v))
        url = f"{self.base}/rest/{endpoint}.view?{urllib.parse.urlencode(qs_parts)}"
        with urllib.request.urlopen(url, timeout=60) as r:
            body = r.read()
        root = ET.fromstring(body)
        # Subsonic status check
        if root.attrib.get("status") != "ok":
            err = root.find("{http://subsonic.org/restapi}error")
            msg = err.attrib.get("message", "unknown") if err is not None else "no error element"
            raise RuntimeError(f"Subsonic {endpoint} failed: {msg}")
        return root

    def get_playlists(self) -> list[tuple[str, str]]:
        root = self._call("getPlaylists", {})
        ns = "{http://subsonic.org/restapi}"
        return [
            (p.attrib["id"], p.attrib["name"])
            for p in root.iter(f"{ns}playlist")
        ]

    def create_playlist(self, name: str) -> str:
        root = self._call("createPlaylist", {"name": name})
        ns = "{http://subsonic.org/restapi}"
        pl = root.find(f"{ns}playlist")
        if pl is None or "id" not in pl.attrib:
            raise RuntimeError(f"createPlaylist returned no id for {name!r}")
        return pl.attrib["id"]

    def add_songs(self, playlist_id: str, song_ids: list[str]) -> None:
        if not song_ids:
            return
        # Subsonic supports multiple songIdToAdd params; batch in chunks for URL length safety.
        for i in range(0, len(song_ids), 50):
            chunk = song_ids[i:i + 50]
            self._call("updatePlaylist", {"playlistId": playlist_id, "songIdToAdd": chunk})

    def start_scan(self) -> None:
        self._call("startScan", {})

    def get_scan_status(self) -> tuple[bool, int]:
        root = self._call("getScanStatus", {})
        ns = "{http://subsonic.org/restapi}"
        s = root.find(f"{ns}scanStatus")
        if s is None:
            return False, 0
        scanning = s.attrib.get("scanning", "false") == "true"
        count = int(s.attrib.get("count", "0"))
        return scanning, count


# ============================================================================
# Navidrome track-id lookup
# ============================================================================

def navidrome_path_to_id() -> dict[str, str]:
    """Snapshot Navidrome media_file table: relative_path → id."""
    uri = f"file:{NAVIDROME_DB}?mode=ro"
    conn = sqlite3.connect(uri, uri=True, timeout=30)
    out: dict[str, str] = {}
    for row in conn.execute("SELECT id, path FROM media_file WHERE missing = 0 OR missing IS NULL"):
        out[row[1]] = row[0]
    conn.close()
    return out


# ============================================================================
# Fingerprint matching
# ============================================================================

@dataclass
class LibraryEntry:
    beets_id: int
    host_path: str
    duration: int
    fingerprint: np.ndarray  # uint32


def load_library_fingerprints() -> list[LibraryEntry]:
    if not os.path.exists(FINGERPRINT_DB):
        raise SystemExit(f"fingerprint index missing: {FINGERPRINT_DB}. "
                         f"Run build-fingerprint-index.py first.")
    conn = sqlite3.connect(f"file:{FINGERPRINT_DB}?mode=ro", uri=True)
    rows = list(conn.execute(
        "SELECT beets_id, path, duration, fingerprint FROM fingerprints"
    ))
    conn.close()
    entries: list[LibraryEntry] = []
    for beets_id, path, duration, fp in rows:
        fp_arr = np.fromstring(fp, dtype=np.uint32, sep=",")
        if fp_arr.size == 0:
            continue
        entries.append(LibraryEntry(beets_id, path, duration, fp_arr))
    return entries


# Chromaprint comparison — bitwise popcount over aligned overlap, scanning
# offsets. Each fingerprint chunk is a 32-bit hash of an audio segment;
# matching bit fraction (1 - hamming/32) at the best alignment is the score.
def _compare_fingerprints(fp1: np.ndarray, fp2: np.ndarray, max_offset: int = 80) -> float:
    if fp1.size < 50 or fp2.size < 50:
        return 0.0
    best = 0.0
    for offset in range(-max_offset, max_offset + 1):
        if offset >= 0:
            x = fp1[:max(0, fp1.size - offset)]
            y = fp2[offset:offset + x.size]
        else:
            x = fp1[-offset:]
            y = fp2[:x.size]
        n = int(min(x.size, y.size))
        if n < 100:
            continue
        xor = np.bitwise_xor(x[:n], y[:n])
        v = xor - ((xor >> 1) & np.uint32(0x55555555))
        v = (v & np.uint32(0x33333333)) + ((v >> 2) & np.uint32(0x33333333))
        v = (v + (v >> 4)) & np.uint32(0x0F0F0F0F)
        popcount = ((v * np.uint32(0x01010101)) >> 24)
        diff_bits = int(popcount.sum())
        sim = 1.0 - diff_bits / (n * 32.0)
        if sim > best:
            best = sim
    return best


def fpcalc_raw(path: str) -> tuple[int, np.ndarray] | None:
    try:
        proc = subprocess.run(
            ["fpcalc", "-raw", path],
            capture_output=True, text=True, timeout=120,
        )
    except (subprocess.TimeoutExpired, FileNotFoundError):
        return None
    if proc.returncode != 0:
        return None
    duration = 0
    fp_arr = np.empty(0, dtype=np.uint32)
    for line in proc.stdout.splitlines():
        if line.startswith("DURATION="):
            try:
                duration = int(line.split("=", 1)[1])
            except ValueError:
                pass
        elif line.startswith("FINGERPRINT="):
            fp_arr = np.fromstring(line.split("=", 1)[1], dtype=np.uint32, sep=",")
    return (duration, fp_arr) if fp_arr.size > 0 else None


def best_match(
    incoming: tuple[int, np.ndarray],
    library: list[LibraryEntry],
) -> tuple[float, LibraryEntry | None]:
    """Return (best_score, best_entry). Filters candidates by duration proximity
    (±15s) before scoring to skip obviously-unrelated tracks."""
    incoming_dur, incoming_fp = incoming
    best_score = 0.0
    best_entry: LibraryEntry | None = None
    for entry in library:
        if abs(entry.duration - incoming_dur) > 15:
            continue
        try:
            score = _compare_fingerprints(incoming_fp, entry.fingerprint)
        except Exception:
            continue
        if score > best_score:
            best_score = score
            best_entry = entry
            if score >= 0.99:
                break
    return best_score, best_entry


# ============================================================================
# State DB
# ============================================================================

def init_state_db(conn: sqlite3.Connection) -> None:
    conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS files (
            incoming_path TEXT NOT NULL,
            content_hash TEXT NOT NULL,
            folder TEXT NOT NULL,
            target_playlist TEXT NOT NULL,
            classification TEXT NOT NULL,
            match_score REAL,
            match_beets_id INTEGER,
            match_path TEXT,
            applied INTEGER NOT NULL DEFAULT 0,
            apply_error TEXT,
            applied_at TEXT,
            PRIMARY KEY (incoming_path, content_hash)
        );
        CREATE INDEX IF NOT EXISTS idx_files_classification ON files(classification);
        CREATE INDEX IF NOT EXISTS idx_files_applied ON files(applied);
        """
    )
    conn.commit()


# ============================================================================
# Worker for parallel fingerprint+match
# ============================================================================

_library_cache: list[LibraryEntry] | None = None


def _worker_init(_library: list[LibraryEntry]) -> None:
    global _library_cache
    _library_cache = _library


def _worker_classify(args: tuple[str, str]) -> dict:
    """Run in worker pool. Returns dict with classification details for one file."""
    incoming_path, folder = args
    result = {
        "incoming_path": incoming_path,
        "folder": folder,
        "content_hash": short_hash(incoming_path),
        "classification": "ERROR",
        "match_score": 0.0,
        "match_beets_id": None,
        "match_path": None,
    }
    fp = fpcalc_raw(incoming_path)
    if fp is None:
        result["classification"] = "ERROR"
        return result
    assert _library_cache is not None
    score, entry = best_match(fp, _library_cache)
    result["match_score"] = score
    if entry is not None:
        result["match_beets_id"] = entry.beets_id
        result["match_path"] = entry.host_path
    if score >= DUPLICATE_THRESHOLD:
        result["classification"] = "DUPLICATE"
    elif score >= AMBIGUOUS_THRESHOLD:
        result["classification"] = "AMBIGUOUS"
    else:
        result["classification"] = "NEW"
    return result


# ============================================================================
# Folder enumeration
# ============================================================================

def find_audio_in_folder(folder_path: str) -> list[str]:
    """Return audio file paths in folder_path. No recursion (kept folders are flat)."""
    out: list[str] = []
    for entry in os.scandir(folder_path):
        if not entry.is_file():
            continue
        ext = os.path.splitext(entry.name)[1].lower()
        if ext in AUDIO_EXTS:
            out.append(entry.path)
    return sorted(out)


def select_folders(filter_name: str | None) -> list[str]:
    """Return [top-level folder paths] under djstuff to process, skipping SKIP_FOLDERS."""
    out: list[str] = []
    for entry in sorted(os.scandir(DJSTUFF_ROOT), key=lambda e: e.name):
        if not entry.is_dir():
            continue
        if entry.name in SKIP_FOLDERS:
            continue
        if filter_name is not None and entry.name != filter_name:
            continue
        out.append(entry.path)
    return out


# ============================================================================
# Dry-run pass
# ============================================================================

def cmd_dry_run(args: argparse.Namespace) -> int:
    log(f"loading library fingerprint index from {FINGERPRINT_DB}")
    library = load_library_fingerprints()
    log(f"library has {len(library)} fingerprints")
    if len(library) < 100:
        log("WARNING: library fingerprint index is unexpectedly small — "
            "run build-fingerprint-index.py first.")

    folders = select_folders(args.folder)
    if not folders:
        log("no folders matched")
        return 1
    log(f"processing folders: {[os.path.basename(f) for f in folders]}")

    conn = sqlite3.connect(STATE_DB)
    init_state_db(conn)

    if args.reset:
        conn.execute("DELETE FROM files WHERE applied = 0")
        conn.commit()
        log("reset: deleted unapplied state rows")

    jobs: list[tuple[str, str]] = []
    skipped_already_classified = 0
    for folder_path in folders:
        folder_name = os.path.basename(folder_path)
        for f in find_audio_in_folder(folder_path):
            content_hash = short_hash(f)
            if not content_hash:
                continue
            row = conn.execute(
                "SELECT classification FROM files WHERE incoming_path = ? AND content_hash = ?",
                (f, content_hash),
            ).fetchone()
            if row is not None and not args.reclassify:
                skipped_already_classified += 1
                continue
            jobs.append((f, folder_name))
    log(f"{len(jobs)} files to fingerprint+classify "
        f"({skipped_already_classified} already in state DB; --reclassify to redo)")
    if args.limit > 0:
        jobs = jobs[:args.limit]
        log(f"--limit {args.limit} applied")
    if not jobs:
        conn.close()
        return 0

    start = time.time()
    done = 0
    with multiprocessing.Pool(
        args.workers, initializer=_worker_init, initargs=(library,)
    ) as pool:
        for r in pool.imap_unordered(_worker_classify, jobs, chunksize=1):
            done += 1
            folder = r["folder"]
            target = slugify_folder(folder)
            conn.execute(
                """
                INSERT INTO files (incoming_path, content_hash, folder, target_playlist,
                                   classification, match_score, match_beets_id, match_path,
                                   applied)
                VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0)
                ON CONFLICT(incoming_path, content_hash) DO UPDATE SET
                  folder=excluded.folder,
                  target_playlist=excluded.target_playlist,
                  classification=excluded.classification,
                  match_score=excluded.match_score,
                  match_beets_id=excluded.match_beets_id,
                  match_path=excluded.match_path
                """,
                (r["incoming_path"], r["content_hash"], folder, target,
                 r["classification"], r["match_score"],
                 r["match_beets_id"], r["match_path"]),
            )
            if done % 50 == 0:
                conn.commit()
                rate = done / max(time.time() - start, 0.01)
                log(f"  {done}/{len(jobs)} ({rate:.1f}/s)")
    conn.commit()
    log(f"classified {done} files in {time.time() - start:.0f}s")

    write_dry_run_summary(conn, args.folder)
    conn.close()
    return 0


def write_dry_run_summary(conn: sqlite3.Connection, folder_filter: str | None) -> None:
    """Write a per-folder summary + ambiguous CSV to $ALEMBIC_CONFIG_DIR/logs/."""
    os.makedirs(LOG_DIR, exist_ok=True)
    stamp = time.strftime("%Y%m%d-%H%M%S")
    summary_path = f"{LOG_DIR}/dj-import-dry-{stamp}.log"
    ambig_csv_path = f"{LOG_DIR}/dj-import-ambiguous-{stamp}.csv"

    where = ""
    params: tuple = ()
    if folder_filter is not None:
        where = "WHERE folder = ?"
        params = (folder_filter,)

    rows = list(conn.execute(
        f"SELECT folder, classification, COUNT(*) FROM files {where} "
        f"GROUP BY folder, classification ORDER BY folder, classification",
        params,
    ))

    with open(summary_path, "w") as f:
        f.write(f"# DJ import dry-run summary — {stamp}\n")
        if folder_filter:
            f.write(f"# Folder filter: {folder_filter}\n")
        f.write("\n")
        totals = {"DUPLICATE": 0, "NEW": 0, "AMBIGUOUS": 0, "ERROR": 0}
        cur_folder = None
        for folder, classification, n in rows:
            if folder != cur_folder:
                f.write(f"\n## {folder} → {slugify_folder(folder)}\n")
                cur_folder = folder
            f.write(f"  {classification:10s}  {n:5d}\n")
            totals[classification] = totals.get(classification, 0) + n
        f.write("\n## Totals\n")
        for k in ("DUPLICATE", "NEW", "AMBIGUOUS", "ERROR"):
            f.write(f"  {k:10s}  {totals[k]:5d}\n")

    ambig_rows = list(conn.execute(
        f"SELECT incoming_path, folder, target_playlist, match_score, match_path "
        f"FROM files {where + ' AND ' if where else 'WHERE '}classification = 'AMBIGUOUS' "
        f"ORDER BY match_score DESC",
        params,
    ))
    with open(ambig_csv_path, "w", newline="") as f:
        w = csv.writer(f)
        w.writerow(["incoming_path", "folder", "target_playlist", "match_score", "match_path"])
        for r in ambig_rows:
            w.writerow(r)

    log(f"summary written to {summary_path}")
    log(f"ambiguous list ({len(ambig_rows)} rows) written to {ambig_csv_path}")


# ============================================================================
# Apply pass
# ============================================================================

def cmd_apply(args: argparse.Namespace) -> int:
    if not os.path.exists(STATE_DB):
        log(f"state DB {STATE_DB} missing — run --dry-run first")
        return 1

    log("snapshotting Navidrome + beets DBs for rollback")
    for src in [NAVIDROME_DB, BEETS_DB]:
        dst = f"{src}.pre-dj-import.bak"
        if not os.path.exists(dst):
            shutil.copy2(src, dst)
            log(f"  {src} → {dst}")
        else:
            log(f"  {dst} already exists; leaving as-is")

    sub = Subsonic(ND_BASE, ND_USER, ND_PASS)
    log("fetching existing playlists from Navidrome")
    name_to_id = {name: pid for pid, name in sub.get_playlists()}

    conn = sqlite3.connect(STATE_DB)
    init_state_db(conn)

    where = "applied = 0 AND classification IN ('DUPLICATE','NEW')"
    params: tuple = ()
    if args.folder is not None:
        where += " AND folder = ?"
        params = (args.folder,)

    pending = list(conn.execute(
        f"SELECT incoming_path, content_hash, folder, target_playlist, "
        f"classification, match_path FROM files WHERE {where} ORDER BY folder, incoming_path",
        params,
    ))
    if args.limit > 0:
        pending = pending[:args.limit]
    log(f"{len(pending)} pending entries to apply")
    if not pending:
        conn.close()
        return 0

    # Group by classification for batched flow
    dupes = [p for p in pending if p[4] == "DUPLICATE"]
    news = [p for p in pending if p[4] == "NEW"]
    log(f"  {len(dupes)} duplicates → playlist add only")
    log(f"  {len(news)} new tracks → stage + beets import + playlist add")

    # --- Pass 1: duplicates (no library write needed) -----------------------
    apply_duplicates(sub, name_to_id, conn, dupes)

    # --- Pass 2: new tracks --------------------------------------------------
    if news:
        apply_new_tracks(sub, name_to_id, conn, news)

    # --- Post-pass housekeeping ---------------------------------------------
    log("triggering Navidrome scan + waiting for completion")
    sub.start_scan()
    wait_for_scan(sub)

    conn.close()
    log("apply pass complete")
    return 0


def apply_duplicates(sub: Subsonic, name_to_id: dict[str, str],
                     conn: sqlite3.Connection,
                     dupes: list[tuple]) -> None:
    if not dupes:
        return
    log(f"=== duplicates pass: {len(dupes)} entries ===")
    # Group by target playlist
    by_pl: dict[str, list[tuple]] = {}
    for row in dupes:
        by_pl.setdefault(row[3], []).append(row)

    path_to_nd = navidrome_path_to_id()
    log(f"loaded {len(path_to_nd)} Navidrome track paths")

    for target_pl, rows in by_pl.items():
        pid = name_to_id.get(target_pl)
        if pid is None:
            log(f"  creating playlist {target_pl}")
            pid = sub.create_playlist(target_pl)
            name_to_id[target_pl] = pid

        # Collect Navidrome IDs to add
        song_ids: list[str] = []
        missing: list[str] = []
        for (incoming_path, content_hash, folder, _tp, _cl, match_path) in rows:
            nd_path = host_to_navidrome_path(match_path)
            sid = path_to_nd.get(nd_path)
            if sid is None:
                missing.append(nd_path)
                conn.execute(
                    "UPDATE files SET apply_error = ? WHERE incoming_path = ? AND content_hash = ?",
                    (f"navidrome id not found for {nd_path}", incoming_path, content_hash),
                )
                continue
            song_ids.append(sid)
        if missing:
            log(f"  {target_pl}: {len(missing)} library tracks not yet in Navidrome "
                f"(scan needed; will retry on next apply run)")

        if song_ids:
            sub.add_songs(pid, song_ids)
            log(f"  {target_pl}: added {len(song_ids)} duplicate matches")

        # Mark applied for the rows we successfully added
        applied_at = time.strftime("%Y-%m-%dT%H:%M:%S")
        applied_paths: set[tuple[str, str]] = set()
        for (incoming_path, content_hash, _f, _tp, _cl, match_path) in rows:
            nd_path = host_to_navidrome_path(match_path)
            if path_to_nd.get(nd_path) is not None:
                applied_paths.add((incoming_path, content_hash))
        for ip, ch in applied_paths:
            conn.execute(
                "UPDATE files SET applied = 1, applied_at = ?, apply_error = NULL "
                "WHERE incoming_path = ? AND content_hash = ?",
                (applied_at, ip, ch),
            )
        conn.commit()


def apply_new_tracks(sub: Subsonic, name_to_id: dict[str, str],
                     conn: sqlite3.Connection,
                     news: list[tuple]) -> None:
    """Stage + beets-import + playlist-add for files classified NEW."""
    log(f"=== new tracks pass: {len(news)} entries ===")
    os.makedirs(DROPBOX_DJSTUFF_HOST, exist_ok=True)

    # Group by target playlist so beets imports a folder at a time.
    by_pl: dict[str, list[tuple]] = {}
    for row in news:
        by_pl.setdefault(row[3], []).append(row)

    staged: list[tuple[str, str, str, str, str]] = []
    # (incoming_path, content_hash, target_playlist, token, staged_host_path)

    for target_pl, rows in by_pl.items():
        slug = target_pl  # already 'dj-xxx'
        stage_dir = f"{DROPBOX_DJSTUFF_HOST}/{slug}"
        os.makedirs(stage_dir, exist_ok=True)
        for (incoming_path, content_hash, folder, _tp, _cl, _mp) in rows:
            token = f"djstuff-{uuid.uuid4().hex[:12]}"
            staged_path = stage_track(incoming_path, stage_dir, token)
            if staged_path is None:
                conn.execute(
                    "UPDATE files SET apply_error = ? WHERE incoming_path = ? AND content_hash = ?",
                    ("staging failed", incoming_path, content_hash),
                )
                conn.commit()
                continue
            staged.append((incoming_path, content_hash, target_pl, token, staged_path))
    log(f"staged {len(staged)} files in {DROPBOX_DJSTUFF_HOST}/")

    if not staged:
        return

    # beets runs in-process in this same container now, so it sees
    # DROPBOX_DJSTUFF_HOST directly — no container path needed.
    log(f"running beets import on {DROPBOX_DJSTUFF_HOST} (no autotag, no dup-check)")
    proc = subprocess.run(
        ["beet", "import", "-q", "-s", "-A", DROPBOX_DJSTUFF_HOST],
        capture_output=True, text=True,
    )
    log(f"  beet import exit={proc.returncode}")
    if proc.returncode != 0:
        log(f"  stderr: {proc.stderr[-500:]}")

    # For each staged file, locate the resulting library track by token.
    log("resolving imported tracks by COMMENT token")
    resolved: dict[str, tuple[int, str]] = {}  # token → (beets_id, container_path)
    for (incoming_path, content_hash, target_pl, token, staged_path) in staged:
        beets_id, container_path = lookup_by_comment_token(token)
        if beets_id is None:
            conn.execute(
                "UPDATE files SET apply_error = ? WHERE incoming_path = ? AND content_hash = ?",
                (f"beets did not import (token {token} missing)", incoming_path, content_hash),
            )
            continue
        resolved[token] = (beets_id, container_path)
    conn.commit()
    log(f"resolved {len(resolved)}/{len(staged)} imported tracks")

    # Trigger a Navidrome scan + wait so new tracks become addressable.
    log("triggering Navidrome scan so new tracks get IDs")
    sub.start_scan()
    wait_for_scan(sub)

    path_to_nd = navidrome_path_to_id()

    # Add each new track to its target dj-* playlist.
    log("adding new tracks to dj-* playlists in Navidrome")
    by_pl_resolved: dict[str, list[tuple[str, str, str, str, str, int, str]]] = {}
    for (incoming_path, content_hash, target_pl, token, staged_path) in staged:
        r = resolved.get(token)
        if r is None:
            continue
        beets_id, container_path = r
        by_pl_resolved.setdefault(target_pl, []).append(
            (incoming_path, content_hash, target_pl, token, staged_path, beets_id, container_path)
        )

    applied_at = time.strftime("%Y-%m-%dT%H:%M:%S")
    for target_pl, rows in by_pl_resolved.items():
        pid = name_to_id.get(target_pl)
        if pid is None:
            pid = sub.create_playlist(target_pl)
            name_to_id[target_pl] = pid
        song_ids: list[str] = []
        for (_ip, _ch, _tp, _tok, _sp, _bid, container_path) in rows:
            nd_path = host_to_navidrome_path(container_path)
            sid = path_to_nd.get(nd_path)
            if sid is not None:
                song_ids.append(sid)
        if song_ids:
            sub.add_songs(pid, song_ids)
            log(f"  {target_pl}: added {len(song_ids)} new tracks")
        # Mark applied
        for (ip, ch, _tp, _tok, _sp, _bid, container_path) in rows:
            nd_path = host_to_navidrome_path(container_path)
            if path_to_nd.get(nd_path) is not None:
                conn.execute(
                    "UPDATE files SET applied = 1, applied_at = ?, apply_error = NULL "
                    "WHERE incoming_path = ? AND content_hash = ?",
                    (applied_at, ip, ch),
                )
            else:
                conn.execute(
                    "UPDATE files SET apply_error = ? "
                    "WHERE incoming_path = ? AND content_hash = ?",
                    (f"navidrome id not found post-scan for {nd_path}", ip, ch),
                )
        conn.commit()


def stage_track(src: str, stage_dir: str, token: str) -> str | None:
    """Copy src into stage_dir, convert WAV→FLAC, tag with GROUPING=djstuff and a
    unique COMMENT token. Return the staged path on success, None on failure."""
    base = os.path.basename(src)
    dst = os.path.join(stage_dir, base)
    try:
        shutil.copy2(src, dst)
    except OSError as e:
        log(f"  stage: copy failed for {src}: {e}")
        return None
    ext = os.path.splitext(dst)[1].lower()
    # WAV → FLAC
    if ext == ".wav":
        flac_dst = dst[:-4] + ".flac"
        proc = subprocess.run(
            ["flac", "--silent", "--best", "--delete-input-file", dst, "-o", flac_dst],
            capture_output=True, text=True,
        )
        if proc.returncode != 0 or not os.path.exists(flac_dst):
            log(f"  stage: WAV→FLAC failed for {dst}: {proc.stderr[-200:]}")
            return None
        # Try to parse "Artist - Title" from filename for basic tags
        stem = os.path.basename(flac_dst)[:-5]
        if " - " in stem:
            artist, title = stem.split(" - ", 1)
            subprocess.run(
                ["metaflac",
                 f"--set-tag=ARTIST={artist}",
                 f"--set-tag=TITLE={title}",
                 f"--set-tag=ALBUM={title}",
                 flac_dst],
                capture_output=True, text=True,
            )
        dst = flac_dst
        ext = ".flac"
    # Tag GROUPING=djstuff + COMMENT token
    return tag_dj_track(dst, ext, token)


def tag_dj_track(path: str, ext: str, token: str) -> str | None:
    """Set GROUPING=djstuff + COMMENT=<token> on the staged file. Return path or None."""
    try:
        if ext == ".flac":
            # beets/mediafile maps Vorbis DESCRIPTION (not COMMENT) to its
            # `comments` field for FLAC. Bandcamp-sourced FLACs ship with
            # DESCRIPTION=https://...bandcamp.com which would mask our token
            # if we only wrote COMMENT. Strip + set both fields.
            subprocess.run(
                ["metaflac",
                 "--remove-tag=GROUPING",
                 "--remove-tag=COMMENT",
                 "--remove-tag=DESCRIPTION",
                 path],
                capture_output=True, text=True,
            )
            r = subprocess.run(
                ["metaflac",
                 f"--set-tag=GROUPING=djstuff",
                 f"--set-tag=COMMENT={token}",
                 f"--set-tag=DESCRIPTION={token}",
                 path],
                capture_output=True, text=True,
            )
            if r.returncode != 0:
                log(f"  tag: metaflac failed on {path}: {r.stderr[-200:]}")
                return None
        elif ext == ".mp3":
            # id3v2 frames: TIT1 = grouping, COMM = comment
            subprocess.run(["id3v2", "--TIT1", "djstuff", path],
                           capture_output=True, text=True)
            subprocess.run(["id3v2", "--COMM", token, path],
                           capture_output=True, text=True)
        else:
            # Use mutagen for other formats
            try:
                import mutagen
                f = mutagen.File(path, easy=False)
                if f is None:
                    log(f"  tag: mutagen could not open {path}")
                    return None
                # Easy-tag for grouping is format-dependent; use raw frames best-effort.
                # For m4a (MP4): ©grp = grouping, ©cmt = comment.
                if hasattr(f, "tags") and f.tags is not None:
                    try:
                        f.tags["©grp"] = ["djstuff"]
                        f.tags["©cmt"] = [token]
                    except Exception:
                        pass
                f.save()
            except Exception as e:
                log(f"  tag: mutagen failed on {path}: {e}")
                return None
        return path
    except Exception as e:
        log(f"  tag: exception on {path}: {e}")
        return None


def lookup_by_comment_token(token: str) -> tuple[int | None, str | None]:
    """Find the beets id + path for a staged file by its COMMENT token."""
    proc = subprocess.run(
        ["beet", "ls", "-f", "$id|||$path", f"comments::{token}"],
        capture_output=True, text=True,
    )
    if proc.returncode != 0:
        return None, None
    for line in proc.stdout.splitlines():
        if "|||" in line:
            id_str, path = line.split("|||", 1)
            try:
                return int(id_str), path.strip()
            except ValueError:
                pass
    return None, None


def wait_for_scan(sub: Subsonic, max_wait: int = 300) -> None:
    start = time.time()
    last_count = -1
    while time.time() - start < max_wait:
        try:
            scanning, count = sub.get_scan_status()
        except Exception as e:
            log(f"  scan status check failed: {e}")
            time.sleep(3)
            continue
        if count != last_count:
            log(f"  scan in progress, count={count}")
            last_count = count
        if not scanning:
            log(f"  scan complete (count={count})")
            return
        time.sleep(3)
    log("  scan wait timed out — proceeding anyway")


# ============================================================================
# CLI
# ============================================================================

def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    grp = ap.add_mutually_exclusive_group()
    grp.add_argument("--dry-run", action="store_true", default=True,
                     help="(default) fingerprint+classify; no changes")
    grp.add_argument("--apply", action="store_true",
                     help="apply classified decisions: playlist add + beets import")
    ap.add_argument("--folder", help="process only this top-level djstuff folder")
    ap.add_argument("--workers", type=int, default=DEFAULT_WORKERS)
    ap.add_argument("--limit", type=int, default=0)
    ap.add_argument("--reset", action="store_true",
                     help="delete unapplied state rows before classifying (keeps applied=1 rows)")
    ap.add_argument("--reclassify", action="store_true",
                     help="re-fingerprint files that already have a classification")
    args = ap.parse_args()

    os.makedirs(LOG_DIR, exist_ok=True)

    if args.apply:
        return cmd_apply(args)
    return cmd_dry_run(args)


if __name__ == "__main__":
    sys.exit(main())