diff --git a/.dockerignore b/.dockerignore index b742bd3..f257c43 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,5 +1,5 @@ .git -.dedupe __pycache__ -.pytest_cache -.venv +*.pyc +tests +.dedupe diff --git a/Dockerfile b/Dockerfile index d33724e..2617b76 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,12 +1,7 @@ FROM python:3.12-slim - -RUN apt-get update \ - && apt-get install -y --no-install-recommends chromaprint-tools \ - && rm -rf /var/lib/apt/lists/* - +RUN apt-get update && apt-get install -y --no-install-recommends chromaprint-tools && rm -rf /var/lib/apt/lists/* WORKDIR /app COPY pyproject.toml README.md ./ COPY am_deduper ./am_deduper RUN pip install --no-cache-dir . - ENTRYPOINT ["am-deduper"] diff --git a/am_deduper/acoustic.py b/am_deduper/acoustic.py new file mode 100644 index 0000000..18003d6 --- /dev/null +++ b/am_deduper/acoustic.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +MAX_SECONDS = 120 + + +def chromaprint(path: Path) -> str | None: + try: + p = subprocess.run( + ["fpcalc", "-length", str(MAX_SECONDS), "-json", str(path)], + capture_output=True, text=True, timeout=180, check=True, + ) + return json.loads(p.stdout).get("fingerprint") + except (OSError, subprocess.SubprocessError, ValueError, json.JSONDecodeError): + return None + + +def similarity(a: str, b: str) -> float: + """Conservative normalized positional similarity for equal-length prefixes.""" + if not a or not b: + return 0.0 + n = min(len(a), len(b)) + if not n: + return 0.0 + # Chromaprint JSON fingerprints are comma-separated integer features. + try: + aa, bb = [int(x) for x in a.split(",")], [int(x) for x in b.split(",")] + except ValueError: + return 0.0 + n = min(len(aa), len(bb)) + if not n: + return 0.0 + # Feature-wise equality is intentionally conservative. A future backend + # can replace this with a library-specific fingerprint distance. + return sum(x == y for x, y in zip(aa[:n], bb[:n])) / n diff --git a/am_deduper/arbitration.py b/am_deduper/arbitration.py new file mode 100644 index 0000000..b63c005 --- /dev/null +++ b/am_deduper/arbitration.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from pathlib import Path + +FORMAT_SCORE = {"flac": 100, "alac": 100, "wav": 100, "m4a": 60, "aac": 60, "ogg": 50, "opus": 50, "mp3": 40} + +def score(row) -> int: + value = FORMAT_SCORE.get((row["format"] or "").lower(), 0) + value += int((row["bitrate"] or 0) * 0.1) + if row["sample_rate"]: + value += min(int(row["sample_rate"] / 1000), 20) + if row["bit_depth"]: + value += min(int(row["bit_depth"]), 32) + return value + +def is_version_conflict(a, b) -> bool: + af = set((a["version_flags"] or "").split(",")) + bf = set((b["version_flags"] or "").split(",")) + return bool(af.symmetric_difference(bf)) + +def choose_winner(rows): + def key(r): + path = str(r["file_path"]) + admin = "/admin/" in path or path.endswith("/admin") + return (-score(r), 0 if admin else 1, r["created_at"], path) + return min(rows, key=key) diff --git a/am_deduper/cli.py b/am_deduper/cli.py index 93b7bf9..a6163e0 100644 --- a/am_deduper/cli.py +++ b/am_deduper/cli.py @@ -1,28 +1,18 @@ from __future__ import annotations - import argparse from pathlib import Path - from .exact import exact_dedupe from .scanner import scan +from .l2 import acoustic_dedupe from .watcher import watch - -def main() -> None: - p = argparse.ArgumentParser(prog="am-deduper") - sub = p.add_subparsers(dest="cmd", required=True) - for name in ("scan", "dedupe", "reconcile", "watch"): - sp = sub.add_parser(name) - sp.add_argument("root", type=Path) - sp.add_argument("--db", type=Path, default=None) - args = p.parse_args() - db = args.db or (args.root / ".dedupe" / "dedupe.db") - if args.cmd == "watch": - watch(args.root, db) - return - if args.cmd in {"scan", "reconcile"}: - print(f"indexed={scan(args.root, db)} db={db}") - if args.cmd == "dedupe": - scan(args.root, db) - actions = exact_dedupe(db, args.root) - print(f"hardlinked={len(actions)}") +def main(): + p=argparse.ArgumentParser(prog='am-deduper'); sub=p.add_subparsers(dest='cmd',required=True) + for name in ('scan','dedupe','reconcile','watch'): + sp=sub.add_parser(name); sp.add_argument('root',type=Path); sp.add_argument('--db',type=Path,default=None) + a=p.parse_args(); db=a.db or a.root/'.dedupe'/'dedupe.db' + if a.cmd=='watch': watch(a.root,db); return + print(f'indexed={scan(a.root,db)} db={db}') + if a.cmd in ('dedupe','reconcile'): + ex=exact_dedupe(db,a.root); print(f'exact_hardlinked={len(ex)}') + l2=acoustic_dedupe(db,a.root); print(f'l2={l2}') diff --git a/am_deduper/constants.py b/am_deduper/constants.py index f84171a..cbb9fc4 100644 --- a/am_deduper/constants.py +++ b/am_deduper/constants.py @@ -9,3 +9,7 @@ STABILITY_INTERVAL_SECONDS = 2 TRASH_DIRNAME = ".dedupe" TRASH_SUBDIR = "trash" QUARANTINE_SUBDIR = "quarantine" +VERSION_MARKERS = frozenset({ + "live", "remix", "acoustic", "karaoke", "instrumental", "demo", + "radio edit", "edit", "version", "mono", "stereo", "concert", +}) diff --git a/am_deduper/db.py b/am_deduper/db.py index 4d95887..d705a50 100644 --- a/am_deduper/db.py +++ b/am_deduper/db.py @@ -32,6 +32,7 @@ CREATE INDEX IF NOT EXISTS idx_fast_hash ON audio_tracks(fast_hash); CREATE INDEX IF NOT EXISTS idx_sha256 ON audio_tracks(sha256); CREATE INDEX IF NOT EXISTS idx_duration ON audio_tracks(duration); CREATE INDEX IF NOT EXISTS idx_status ON audio_tracks(status); +CREATE INDEX IF NOT EXISTS idx_duration_size ON audio_tracks(duration, file_size); """ @@ -41,5 +42,6 @@ def connect(path: Path) -> sqlite3.Connection: conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=ON") + conn.execute("PRAGMA busy_timeout=5000") conn.executescript(SCHEMA) return conn diff --git a/am_deduper/exact.py b/am_deduper/exact.py index 16d6644..549bc3d 100644 --- a/am_deduper/exact.py +++ b/am_deduper/exact.py @@ -2,100 +2,67 @@ from __future__ import annotations import os from pathlib import Path - from .db import connect from .hashing import sha256 def confirm_sha256(conn, row) -> str | None: path = Path(row["file_path"]) - try: - st = path.stat() - except FileNotFoundError: - return None - if st.st_size != row["file_size"]: - return None - # If the file changed after indexing, refresh the fast hash first and refuse - # this pass. The next incremental/reconcile pass will safely retry it. + try: st = path.stat() + except FileNotFoundError: return None + if st.st_size != row["file_size"]: return None digest = sha256(path) - conn.execute( - "UPDATE audio_tracks SET sha256=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", - (digest, row["id"]), - ) + try: + st2 = path.stat() + except FileNotFoundError: return None + if st2.st_size != st.st_size or st2.st_mtime_ns != st.st_mtime_ns: return None + conn.execute("UPDATE audio_tracks SET sha256=?,updated_at=CURRENT_TIMESTAMP WHERE id=?", (digest,row["id"])) return digest -def hardlink_duplicate(conn, loser: Path, winner: Path) -> None: - """Make loser a hard link to winner, atomically replacing loser.""" +def hardlink_duplicate(loser: Path, winner: Path) -> None: tmp = loser.with_name(loser.name + ".dedupe-linking") - if tmp.exists(): - tmp.unlink() - os.link(winner, tmp) - os.replace(tmp, loser) + try: + if tmp.exists(): tmp.unlink() + os.link(winner, tmp) + os.replace(tmp, loser) + finally: + if tmp.exists(): + try: tmp.unlink() + except OSError: pass def choose_winner(rows): - """Prefer public admin paths, then older records, then lexical path.""" def key(r): - path = r["file_path"] - is_admin = "/admin/" in path or path.endswith("/admin") - return (0 if is_admin else 1, r["created_at"], path) - return min(rows, key=key) + p=str(r["file_path"]); admin="/admin/" in p or p.endswith("/admin") + return (0 if admin else 1, r["created_at"], p) + return min(rows,key=key) -def exact_dedupe(db_path: Path, root: Path) -> list[tuple[str, str]]: - """Hard-link byte-identical files while preserving every logical user path.""" - actions = [] +def exact_dedupe(db_path: Path, root: Path) -> list[tuple[str,str]]: + actions=[] with connect(db_path) as conn: - candidates = conn.execute( - """ - SELECT * FROM audio_tracks - WHERE status='active' - ORDER BY id - """ - ).fetchall() - # Only full-hash files that share the cheap L1 key need to be compared. - groups = {} - for row in candidates: - groups.setdefault((row["file_size"], row["fast_hash"]), []).append(row) - for rows in groups.values(): - if len(rows) < 2: - # L1 has no collision candidate. Do not read the whole file; - # the full SHA-256 will be computed only when a future file - # lands in the same L1 bucket. - continue - for row in rows: - if row["sha256"] is None: - confirm_sha256(conn, row) - refreshed = conn.execute( - "SELECT * FROM audio_tracks WHERE id IN (%s)" % ",".join("?" * len(rows)), - [r["id"] for r in rows], - ).fetchall() - by_hash = {} - for row in refreshed: - if row["sha256"]: - by_hash.setdefault(row["sha256"], []).append(row) - for same in by_hash.values(): - if len(same) < 2: - continue - winner = choose_winner(same) - winpath = Path(winner["file_path"]) - if not winpath.exists(): - continue - for row in same: - if row["id"] == winner["id"]: - continue - loser = Path(row["file_path"]) - if not loser.exists(): - continue - # Preserve the user's logical path. Exact duplicates are NOT - # moved to trash: all users should continue seeing/downloading - # their own path, while the inode/data is shared. - if loser.stat().st_ino != winpath.stat().st_ino: - hardlink_duplicate(conn, loser, winpath) - conn.execute( - "UPDATE audio_tracks SET status='hardlinked', updated_at=CURRENT_TIMESTAMP WHERE id=?", - (row["id"],), - ) - actions.append((str(loser), str(winpath))) + rows=conn.execute("SELECT * FROM audio_tracks WHERE status IN ('active','hardlinked')").fetchall() + groups={} + for r in rows: groups.setdefault((r["file_size"],r["fast_hash"]),[]).append(r) + for candidates in groups.values(): + if len(candidates)<2: continue + for r in candidates: + if not r["sha256"]: confirm_sha256(conn,r) + ids=[r["id"] for r in candidates] + refreshed=conn.execute("SELECT * FROM audio_tracks WHERE id IN (%s)"%','.join('?'*len(ids)),ids).fetchall() + by={} + for r in refreshed: + if r["sha256"]: by.setdefault(r["sha256"],[]).append(r) + for same in by.values(): + if len(same)<2: continue + winner=choose_winner(same); wp=Path(winner["file_path"]) + if not wp.exists(): continue + for r in same: + if r["id"]==winner["id"]: continue + lp=Path(r["file_path"]) + if not lp.exists(): continue + if lp.stat().st_ino != wp.stat().st_ino: hardlink_duplicate(lp,wp) + conn.execute("UPDATE audio_tracks SET status='hardlinked',updated_at=CURRENT_TIMESTAMP WHERE id=?",(r["id"],)) + actions.append((str(lp),str(wp))) return actions diff --git a/am_deduper/l2.py b/am_deduper/l2.py new file mode 100644 index 0000000..7e6ca62 --- /dev/null +++ b/am_deduper/l2.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +from pathlib import Path +from .acoustic import chromaprint, similarity +from .arbitration import choose_winner, is_version_conflict, score +from .db import connect +from .constants import QUARANTINE_SUBDIR, TRASH_SUBDIR + +MIN_SIMILARITY = 0.90 +HIGH_SIMILARITY = 0.95 +REVIEW_SIMILARITY = 0.80 +DURATION_DELTA = 3.0 + + +def _set_meta(conn, row, fp): + conn.execute("UPDATE audio_tracks SET chromaprint=?,score=?,updated_at=CURRENT_TIMESTAMP WHERE id=?", (fp,score(row),row["id"])) + + +def find_candidates(conn,row): + d=row["duration"] + if d is None: return [] + return conn.execute("SELECT * FROM audio_tracks WHERE id!=? AND status IN ('active','hardlinked') AND duration IS NOT NULL AND duration BETWEEN ? AND ?", (row["id"],d-DURATION_DELTA,d+DURATION_DELTA)).fetchall() + + +def acoustic_dedupe(db_path: Path, root: Path) -> dict: + stats={"checked":0,"equivalent":0,"review":0,"unique":0,"errors":0} + with connect(db_path) as conn: + rows=conn.execute("SELECT * FROM audio_tracks WHERE status='active' ORDER BY id").fetchall() + for row in rows: + if row["chromaprint"]: continue + fp=chromaprint(Path(row["file_path"])); stats["checked"]+=1 + if not fp: stats["errors"]+=1; continue + _set_meta(conn,row,fp) + candidates=find_candidates(conn,row); best=None + for c in candidates: + if not c["chromaprint"]: + cfp=chromaprint(Path(c["file_path"])) + if cfp: _set_meta(conn,c,cfp); c=dict(c); c["chromaprint"]=cfp + sim=similarity(fp,c["chromaprint"] or "") + if sim>=REVIEW_SIMILARITY and (best is None or sim>best[0]): best=(sim,c) + if not best or best[0] dict: + data = {"format": path.suffix.lower().lstrip(".")} if TinyTag is None: - return {} - tag = TinyTag.get(str(path), image=False) - return { + return data + try: + tag = TinyTag.get(str(path), image=False) + except Exception: + return data + data.update({ "duration": tag.duration, "artist": tag.artist, "title": tag.title, "album": tag.album, - "format": path.suffix.lower().lstrip("."), "bitrate": int(tag.bitrate) if tag.bitrate else None, "sample_rate": int(tag.samplerate) if tag.samplerate else None, "bit_depth": int(tag.bitdepth) if tag.bitdepth else None, - } + }) + text = " ".join(str(data.get(k) or "") for k in ("title", "album", "file_name")) + data["version_flags"] = [m for m in VERSION_MARKERS if re.search(rf"(? Iterator[Path]: for path in root.rglob("*"): - if not path.is_file() or path.is_symlink(): - continue - if path.name.startswith(".") or ".dedupe" in path.parts: + if not path.is_file() or path.is_symlink() or path.name.startswith(".") or ".dedupe" in path.parts: continue if path.suffix.lower() in MUSIC_EXTENSIONS: yield path @@ -22,26 +20,21 @@ def iter_music(root: Path) -> Iterator[Path]: def index_file(conn, path: Path) -> None: st = path.stat() fh = fast_hash(path) - conn.execute( - """ - INSERT INTO audio_tracks(file_path, file_size, fast_hash, inode, device, status) - VALUES (?, ?, ?, ?, ?, 'active') - ON CONFLICT(file_path) DO UPDATE SET - file_size=excluded.file_size, - fast_hash=excluded.fast_hash, - inode=excluded.inode, - device=excluded.device, - updated_at=CURRENT_TIMESTAMP, - status='active' - """, - (str(path), st.st_size, fh, st.st_ino, st.st_dev), - ) + meta = read_metadata(path) + flags = ",".join(meta.get("version_flags", [])) + conn.execute(""" + INSERT INTO audio_tracks(file_path,file_size,fast_hash,duration,artist,title,album,format,bitrate,sample_rate,bit_depth,version_flags,inode,device,status) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,'active') + ON CONFLICT(file_path) DO UPDATE SET file_size=excluded.file_size, fast_hash=excluded.fast_hash, + duration=excluded.duration,artist=excluded.artist,title=excluded.title,album=excluded.album,format=excluded.format, + bitrate=excluded.bitrate,sample_rate=excluded.sample_rate,bit_depth=excluded.bit_depth,version_flags=excluded.version_flags, + inode=excluded.inode,device=excluded.device,updated_at=CURRENT_TIMESTAMP,status='active' + """, (str(path),st.st_size,fh,meta.get("duration"),meta.get("artist"),meta.get("title"),meta.get("album"),meta.get("format"),meta.get("bitrate"),meta.get("sample_rate"),meta.get("bit_depth"),flags,st.st_ino,st.st_dev)) def scan(root: Path, db_path: Path) -> int: count = 0 with connect(db_path) as conn: for path in iter_music(root): - index_file(conn, path) - count += 1 + index_file(conn, path); count += 1 return count diff --git a/docker-compose.yml b/docker-compose.yml index 25dd9fb..86936e2 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -6,6 +6,7 @@ services: command: ["watch", "/music/AllMusic"] environment: TZ: Asia/Shanghai + PYTHONUNBUFFERED: "1" volumes: - /your/nas/path/AllMusic:/music/AllMusic - - ./data:/music/AllMusic/.dedupe + stop_grace_period: 30s