From f10e92a5cdb494f22c3be91ee81fa1a3538f171e Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Fri, 18 Sep 2026 07:30:38 +0800 Subject: [PATCH] feat: automate exact dedupe and improve web feedback --- README.md | 6 +- am_deduper/exact.py | 81 ++++++++------- am_deduper/locking.py | 18 ++++ am_deduper/scanner.py | 15 ++- am_deduper/watcher.py | 79 ++++++++++++--- am_deduper/web.py | 224 ++++++++++++++++++++++-------------------- pyproject.toml | 2 +- tests/test_scanner.py | 21 ++++ 8 files changed, 283 insertions(+), 163 deletions(-) create mode 100644 am_deduper/locking.py create mode 100644 tests/test_scanner.py diff --git a/README.md b/README.md index 936c0e4..28d1123 100644 --- a/README.md +++ b/README.md @@ -43,7 +43,7 @@ am-deduper dedupe /music/AllMusic --db /data/dedupe.db --dry-run # 确认后执行 am-deduper dedupe /music/AllMusic --db /data/dedupe.db -# 长驻监听：默认只自动执行安全的 Exact hard-link +# 长驻监听：启动时先做一次全量索引 + Exact 去重，之后自动执行安全的 Exact hard-link am-deduper watch /music/AllMusic --db /data/dedupe.db # 开启 L2 acoustic 匹配；建议先 dry-run 验证 @@ -123,7 +123,11 @@ http://:18080/ - 待审核：查看相似度、候选文件，可人工确认或保留。 - 任务：查看 scan / dry-run / dedupe 状态。 - 历史：查看 Hard Link 操作，并对仍保持同 inode 的操作执行安全回滚。 +- 监听服务启动时会先做一次全量扫描与 Exact 去重；新文件稳定后也会自动索引并执行 Exact 去重。 +- 跨 Web 与 watcher 进程使用数据库锁，避免两个容器同时执行 Hard Link。 +- 文件被替换时会自动清空旧 SHA-256 / Chromaprint，防止旧内容哈希污染新文件。 - 扫描会把数据库中已经不存在的文件标记为 `missing`，避免脏索引继续计入统计。 +- Web UI 的每个操作都有明确的启动、运行、完成/失败反馈，并保留任务结果。 > Web UI 支持可选管理 Token。生产环境建议在 NAS 的 Compose 同目录 `.env` 中设置 `AM_DEDUPER_TOKEN=<随机长字符串>`；不要把 Token 提交到 Git。设置后 Web UI 会在首次访问 API 时要求输入 Token。若未设置 Token，则仅适合可信内网环境，不要直接暴露到公网。 diff --git a/am_deduper/exact.py b/am_deduper/exact.py index 1b119da..ebedb32 100644 --- a/am_deduper/exact.py +++ b/am_deduper/exact.py @@ -5,6 +5,7 @@ from pathlib import Path from .db import connect from .hashing import sha256 +from .locking import exclusive def confirm_sha256(conn, row) -> str | None: @@ -59,50 +60,48 @@ def choose_winner(rows): def exact_dedupe(db_path: Path, root: Path, dry_run: bool = False) -> list[tuple[str, str]]: actions: list[tuple[str, str]] = [] root = root.resolve() - with connect(db_path) as conn: - rows = conn.execute( - "SELECT * FROM audio_tracks WHERE status IN ('active','hardlinked')" - ).fetchall() - groups: dict[tuple[int, str], list] = {} - for row in rows: - groups.setdefault((row["file_size"], row["fast_hash"]), []).append(row) - - for candidates in groups.values(): - if len(candidates) < 2: - continue - for row in candidates: - if not row["sha256"]: - confirm_sha256(conn, row) - ids = [row["id"] for row in candidates] - placeholders = ",".join("?" * len(ids)) - refreshed = conn.execute( - f"SELECT * FROM audio_tracks WHERE id IN ({placeholders})", ids + with exclusive(db_path): + with connect(db_path) as conn: + rows = conn.execute( + "SELECT * FROM audio_tracks WHERE status IN ('active','hardlinked')" ).fetchall() - by_sha: dict[str, list] = {} - for row in refreshed: - if row["sha256"]: - by_sha.setdefault(row["sha256"], []).append(row) + groups: dict[tuple[int, str], list] = {} + for row in rows: + groups.setdefault((row["file_size"], row["fast_hash"]), []).append(row) - for same in by_sha.values(): - if len(same) < 2: + for candidates in groups.values(): + if len(candidates) < 2: continue - winner = choose_winner(same) - wp = Path(winner["file_path"]) - if not wp.exists() or not wp.is_relative_to(root): - continue - for row in same: - if row["id"] == winner["id"]: + for row in candidates: + if not row["sha256"]: + confirm_sha256(conn, row) + ids = [row["id"] for row in candidates] + placeholders = ",".join("?" * len(ids)) + refreshed = conn.execute( + f"SELECT * FROM audio_tracks WHERE id IN ({placeholders})", ids + ).fetchall() + by_sha: dict[str, list] = {} + for row in refreshed: + if row["sha256"]: + by_sha.setdefault(row["sha256"], []).append(row) + + for same in by_sha.values(): + if len(same) < 2: continue - lp = Path(row["file_path"]) - if not lp.exists() or not lp.is_relative_to(root): + winner = choose_winner(same) + wp = Path(winner["file_path"]) + if not wp.exists() or not wp.is_relative_to(root): continue - if lp.stat().st_ino != wp.stat().st_ino: - if not dry_run: - hardlink_duplicate(lp, wp) - conn.execute("INSERT INTO operations(action,file_path,winner_path,details) VALUES(?,?,?,?)", ('exact-hardlink',str(lp),str(wp),'sha256 exact duplicate')) - conn.execute( - "UPDATE audio_tracks SET status='hardlinked',updated_at=CURRENT_TIMESTAMP WHERE id=?", - (row["id"],), - ) - actions.append((str(lp), str(wp))) + for row in same: + if row["id"] == winner["id"]: + continue + lp = Path(row["file_path"]) + if not lp.exists() or not lp.is_relative_to(root): + continue + if lp.stat().st_ino != wp.stat().st_ino: + if not dry_run: + hardlink_duplicate(lp, wp) + conn.execute("INSERT INTO operations(action,file_path,winner_path,details) VALUES(?,?,?,?)", ('exact-hardlink',str(lp),str(wp),'sha256 exact duplicate')) + conn.execute("UPDATE audio_tracks SET status='hardlinked',updated_at=CURRENT_TIMESTAMP WHERE id=?", (row["id"],)) + actions.append((str(lp), str(wp))) return actions diff --git a/am_deduper/locking.py b/am_deduper/locking.py new file mode 100644 index 0000000..2177d29 --- /dev/null +++ b/am_deduper/locking.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from contextlib import contextmanager +from pathlib import Path +import fcntl + + +@contextmanager +def exclusive(db_path: Path): + """Cross-process lock for destructive dedupe operations.""" + lock_path = Path(str(db_path) + ".lock") + lock_path.parent.mkdir(parents=True, exist_ok=True) + with lock_path.open("w") as handle: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) diff --git a/am_deduper/scanner.py b/am_deduper/scanner.py index 58de541..c058687 100644 --- a/am_deduper/scanner.py +++ b/am_deduper/scanner.py @@ -22,14 +22,27 @@ def index_file(conn, path: Path) -> None: fh = fast_hash(path) meta = read_metadata(path) flags = ",".join(meta.get("version_flags", [])) + old = conn.execute("SELECT file_size,fast_hash,inode,device FROM audio_tracks WHERE file_path=?", (str(path),)).fetchone() + identity_changed = old is not None and (old["file_size"], old["fast_hash"], old["inode"], old["device"]) != (st.st_size, fh, st.st_ino, st.st_dev) 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' + inode=excluded.inode,device=excluded.device,updated_at=CURRENT_TIMESTAMP,status='active', + sha256=CASE WHEN excluded.file_size!=audio_tracks.file_size OR excluded.fast_hash!=audio_tracks.fast_hash + OR excluded.inode!=audio_tracks.inode OR excluded.device!=audio_tracks.device + THEN NULL ELSE audio_tracks.sha256 END, + chromaprint=CASE WHEN excluded.file_size!=audio_tracks.file_size OR excluded.fast_hash!=audio_tracks.fast_hash + OR excluded.inode!=audio_tracks.inode OR excluded.device!=audio_tracks.device + THEN NULL ELSE audio_tracks.chromaprint END, + review_status=CASE WHEN excluded.file_size!=audio_tracks.file_size OR excluded.fast_hash!=audio_tracks.fast_hash + OR excluded.inode!=audio_tracks.inode OR excluded.device!=audio_tracks.device + THEN NULL ELSE audio_tracks.review_status END """, (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)) + if identity_changed: + conn.execute("UPDATE audio_tracks SET updated_at=CURRENT_TIMESTAMP WHERE file_path=?", (str(path),)) def scan(root: Path, db_path: Path) -> int: diff --git a/am_deduper/watcher.py b/am_deduper/watcher.py index 6c1f76b..fab44c8 100644 --- a/am_deduper/watcher.py +++ b/am_deduper/watcher.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json import threading import time from pathlib import Path @@ -11,7 +12,7 @@ from .constants import MUSIC_EXTENSIONS, STABILITY_CHECKS, STABILITY_INTERVAL_SE from .db import connect from .exact import exact_dedupe from .l2 import acoustic_dedupe -from .scanner import index_file +from .scanner import index_file, scan class Handler(FileSystemEventHandler): @@ -20,6 +21,17 @@ class Handler(FileSystemEventHandler): self.db_path = db_path self.l2 = l2 self._lock = threading.Lock() + self._timers: dict[str, threading.Timer] = {} + self._timers_lock = threading.Lock() + + def _record_job(self, kind: str, status: str, job_id: int, result=None, error=None): + with connect(self.db_path) as conn: + conn.execute("UPDATE jobs SET status=?,finished_at=CURRENT_TIMESTAMP,result=?,error=? WHERE id=?", + (status, json.dumps(result, ensure_ascii=False) if result is not None else None, error, job_id)) + + def _new_job(self, kind: str) -> int: + with connect(self.db_path) as conn: + return conn.execute("INSERT INTO jobs(kind,status) VALUES(?,?)", (kind, "running")).lastrowid def _process(self, filename: str) -> None: path = Path(filename) @@ -32,40 +44,75 @@ class Handler(FileSystemEventHandler): if ".dedupe" in path.parts or path.name.startswith("."): return - # watchdog may emit several events while a downloader is still writing. - # Serialize the worker so a file is never indexed while another event - # is changing the same database or replacing a duplicate. with self._lock: + job_id = self._new_job("auto-exact") try: previous = path.stat() stable = 0 while stable < STABILITY_CHECKS - 1: time.sleep(STABILITY_INTERVAL_SECONDS) current = path.stat() - if (current.st_size, current.st_mtime_ns) == ( - previous.st_size, - previous.st_mtime_ns, - ): + if (current.st_size, current.st_mtime_ns) == (previous.st_size, previous.st_mtime_ns): stable += 1 else: previous = current stable = 0 with connect(self.db_path) as conn: index_file(conn, path) - # Exact byte-for-byte dedupe is always safe. L2 remains opt-in. - exact_dedupe(self.db_path, self.root) + actions = exact_dedupe(self.db_path, self.root) + result = {"indexed": 1, "exact_hardlinked": len(actions), "path": str(path)} if self.l2: - acoustic_dedupe(self.db_path, self.root) - except (FileNotFoundError, PermissionError, OSError): - return + result["acoustic"] = acoustic_dedupe(self.db_path, self.root) + self._record_job("auto-exact", "success", job_id, result=result) + except (FileNotFoundError, PermissionError, OSError) as exc: + self._record_job("auto-exact", "skipped", job_id, error=str(exc)) + except Exception as exc: + self._record_job("auto-exact", "failed", job_id, error=str(exc)) + + def _schedule(self, filename: str) -> None: + with self._timers_lock: + old = self._timers.pop(filename, None) + if old: + old.cancel() + timer = threading.Timer(1.0, self._run_scheduled, args=(filename,)) + timer.daemon = True + self._timers[filename] = timer + timer.start() + + def _run_scheduled(self, filename: str) -> None: + with self._timers_lock: + self._timers.pop(filename, None) + self._process(filename) def on_created(self, event): if not event.is_directory: - self._process(event.src_path) + self._schedule(event.src_path) + + def on_modified(self, event): + if not event.is_directory: + self._schedule(event.src_path) def on_moved(self, event): if not event.is_directory: - self._process(event.dest_path) + self._schedule(event.dest_path) + + +def _initial_reconcile(root: Path, db_path: Path, l2: bool) -> None: + with connect(db_path) as conn: + job_id = conn.execute("INSERT INTO jobs(kind,status) VALUES(?,?)", ("startup-reconcile", "running")).lastrowid + try: + indexed = scan(root, db_path) + actions = exact_dedupe(db_path, root) + result = {"indexed": indexed, "exact_hardlinked": len(actions)} + if l2: + result["acoustic"] = acoustic_dedupe(db_path, root) + with connect(db_path) as conn: + conn.execute("UPDATE jobs SET status='success',finished_at=CURRENT_TIMESTAMP,result=? WHERE id=?", + (json.dumps(result, ensure_ascii=False), job_id)) + except Exception as exc: + with connect(db_path) as conn: + conn.execute("UPDATE jobs SET status='failed',finished_at=CURRENT_TIMESTAMP,error=? WHERE id=?", (str(exc), job_id)) + raise def watch(root: Path, db_path: Path, l2: bool = False) -> None: @@ -73,9 +120,11 @@ def watch(root: Path, db_path: Path, l2: bool = False) -> None: if not root.is_dir(): raise ValueError(f"music root does not exist or is not a directory: {root}") db_path.parent.mkdir(parents=True, exist_ok=True) + _initial_reconcile(root, db_path, l2) observer = Observer() observer.schedule(Handler(root, db_path, l2=l2), str(root), recursive=True) observer.start() + print(f"AM Deduper watcher active: {root}", flush=True) try: while True: time.sleep(1) diff --git a/am_deduper/web.py b/am_deduper/web.py index 73508ec..ef5b95f 100644 --- a/am_deduper/web.py +++ b/am_deduper/web.py @@ -1,191 +1,207 @@ from __future__ import annotations -import json, os, shutil, threading, uuid + +import json +import os +import shutil +import threading +import uuid from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlparse + from .db import connect from .scanner import scan from .exact import exact_dedupe, hardlink_duplicate, choose_winner from .l2 import acoustic_dedupe -ROOT=Path(os.environ.get("AM_DEDUPER_ROOT","/music/AllMusic")).resolve() -DB=Path(os.environ.get("AM_DEDUPER_DB","/data/dedupe.db")).resolve() -TOKEN=os.environ.get("AM_DEDUPER_TOKEN","").strip() -LOCK=threading.Lock() +ROOT = Path(os.environ.get("AM_DEDUPER_ROOT", "/music/AllMusic")).resolve() +DB = Path(os.environ.get("AM_DEDUPER_DB", "/data/dedupe.db")).resolve() +TOKEN = os.environ.get("AM_DEDUPER_TOKEN", "").strip() +LOCK = threading.Lock() -HTML=r''' +HTML = r''' AM Deduper -
AM Deduper · NAS Music Library
- +
AM Deduper · NAS Music Library
自动监听 · SHA-256 精确去重 · Hard Link · 人工审核
+
+
+
正在加载…
读取任务和索引状态
自动刷新 5 秒
-

重复组

-

待人工审核

-

任务

-

最近操作

-

设置

加载中...
-
- +

重复组

+

待人工审核

+

任务执行记录

包含自动去重任务
+

最近操作

可回滚仍保持同 inode 的 Hard Link
+

运行设置

加载中...
+
''' + def stats(): with connect(DB) as c: - total=c.execute("SELECT COUNT(*) n FROM audio_tracks WHERE status IN ('active','hardlinked','review')").fetchone()["n"] - hard=c.execute("SELECT COUNT(*) n FROM audio_tracks WHERE status='hardlinked'").fetchone()["n"] - review=c.execute("SELECT COUNT(*) n FROM audio_tracks WHERE status='review'").fetchone()["n"] - size=c.execute("SELECT COALESCE(SUM(file_size),0) n FROM audio_tracks WHERE status IN ('active','hardlinked','review')").fetchone()["n"] - ops=c.execute("SELECT COUNT(*) n FROM operations").fetchone()["n"] - return {"cards":[{"name":"音乐文件","value":total},{"name":"Hard Link","value":hard},{"name":"待人工审核","value":review},{"name":"索引大小","value":f"{size/1024/1024/1024:.2f} GB"},{"name":"操作记录","value":ops}]} + total = c.execute("SELECT COUNT(*) n FROM audio_tracks WHERE status IN ('active','hardlinked','review')").fetchone()["n"] + hard = c.execute("SELECT COUNT(*) n FROM audio_tracks WHERE status='hardlinked'").fetchone()["n"] + review = c.execute("SELECT COUNT(*) n FROM audio_tracks WHERE status='review'").fetchone()["n"] + size = c.execute("SELECT COALESCE(SUM(file_size),0) n FROM audio_tracks WHERE status IN ('active','hardlinked','review')").fetchone()["n"] + ops = c.execute("SELECT COUNT(*) n FROM operations").fetchone()["n"] + return {"cards":[ + {"name":"逻辑音乐文件","value":total,"hint":"每个用户路径各算 1 个"}, + {"name":"Hard Link 文件","value":hard,"hint":"物理数据已共享"}, + {"name":"待人工审核","value":review,"hint":"L2 相似或版本冲突"}, + {"name":"索引逻辑大小","value":f"{size/1024/1024/1024:.2f} GB","hint":"按路径统计，非实际占用"}, + {"name":"操作记录","value":ops,"hint":"可追溯/可回滚"}]} + def duplicates(): with connect(DB) as c: - rows=c.execute("SELECT * FROM audio_tracks WHERE status IN ('active','hardlinked') AND sha256 IS NOT NULL ORDER BY sha256,id").fetchall() - groups={} - for r in rows: groups.setdefault(r["sha256"],[]).append(dict(r)) - out=[] - for _,items in groups.items(): - if len(items)>1: - winner=choose_winner(items) - ordered=[winner]+[x for x in items if x["id"]!=winner["id"]] - savings=round(sum(x["file_size"] for x in ordered[1:])/1024/1024,2) - out.append({"type":"SHA256 完全重复","savings":savings,"winner_id":winner["id"],"items":ordered}) + rows = c.execute("SELECT * FROM audio_tracks WHERE status IN ('active','hardlinked') AND sha256 IS NOT NULL ORDER BY sha256,id").fetchall() + groups = {} + for r in rows: groups.setdefault(r["sha256"], []).append(dict(r)) + out = [] + for _, items in groups.items(): + if len(items) > 1: + winner = choose_winner(items) + ordered = [winner] + [x for x in items if x["id"] != winner["id"]] + savings = round(sum(x["file_size"] for x in ordered[1:]) / 1024 / 1024, 2) + out.append({"type":"SHA256 完全重复", "savings":savings, "winner_id":winner["id"], "items":ordered}) return {"groups":out[:100]} + def review_items(): - with connect(DB) as c: rows=c.execute("SELECT * FROM audio_tracks WHERE status='review' ORDER BY updated_at DESC").fetchall() - out=[] + with connect(DB) as c: rows = c.execute("SELECT * FROM audio_tracks WHERE status='review' ORDER BY updated_at DESC").fetchall() + out = [] for r in rows: - d=dict(r); d["similarity"]=(r["review_status"] or "").replace("similarity=","") + d = dict(r); d["similarity"] = (r["review_status"] or "").replace("similarity=", "") with connect(DB) as c2: - cand=c2.execute("SELECT id,file_path FROM audio_tracks WHERE id!=? AND status IN ('active','hardlinked') AND duration BETWEEN ? AND ? ORDER BY ABS(duration-?) LIMIT 1",(r["id"],(r["duration"] or 0)-3,(r["duration"] or 0)+3,r["duration"] or 0)).fetchone() - d["candidate_id"]=cand["id"] if cand else None; d["candidate_path"]=cand["file_path"] if cand else None; out.append(d) + cand = c2.execute("SELECT id,file_path FROM audio_tracks WHERE id!=? AND status IN ('active','hardlinked') AND duration BETWEEN ? AND ? ORDER BY ABS(duration-?) LIMIT 1", (r["id"], (r["duration"] or 0)-3, (r["duration"] or 0)+3, r["duration"] or 0)).fetchone() + d["candidate_id"] = cand["id"] if cand else None; d["candidate_path"] = cand["file_path"] if cand else None; out.append(d) return {"items":out} + def history(): - with connect(DB) as c: rows=c.execute("SELECT id,created_at,action,file_path,winner_path,details FROM operations ORDER BY id DESC LIMIT 100").fetchall() + with connect(DB) as c: rows = c.execute("SELECT id,created_at,action,file_path,winner_path,details FROM operations ORDER BY id DESC LIMIT 100").fetchall() return {"items":[dict(r, reversible=r["action"] in ("exact-hardlink","acoustic-hardlink","manual-hardlink")) for r in rows]} + def jobs(): - with connect(DB) as c: rows=c.execute("SELECT * FROM jobs ORDER BY id DESC LIMIT 30").fetchall() + with connect(DB) as c: rows = c.execute("SELECT * FROM jobs ORDER BY id DESC LIMIT 30").fetchall() return {"items":[dict(r) for r in rows]} + def settings(): from .l2 import MIN_SIMILARITY, REVIEW_SIMILARITY, DURATION_DELTA - return {"root":str(ROOT),"db":str(DB),"auth":"Token 已启用" if TOKEN else "未启用（仅建议内网）", - "l2_min":MIN_SIMILARITY,"l2_review":REVIEW_SIMILARITY,"duration_delta":DURATION_DELTA} + return {"root":str(ROOT),"db":str(DB),"auth":"Token 已启用" if TOKEN else "未启用（仅建议内网）","l2_min":MIN_SIMILARITY,"l2_review":REVIEW_SIMILARITY,"duration_delta":DURATION_DELTA} + def run_job(kind): with connect(DB) as c: - cur=c.execute("INSERT INTO jobs(kind,status) VALUES(?,?)",(kind,"running")); jid=cur.lastrowid + jid = c.execute("INSERT INTO jobs(kind,status) VALUES(?,?)", (kind,"running")).lastrowid def work(): try: with LOCK: - if kind=="scan": result={"count":scan(ROOT,DB)} - elif kind=="dry": - scan(ROOT,DB); result={"exact":len(exact_dedupe(DB,ROOT,True)),"acoustic":acoustic_dedupe(DB,ROOT,True)} + indexed = scan(ROOT, DB) + if kind == "scan": result = {"indexed": indexed} + elif kind == "dry": + result = {"indexed": indexed, "exact": len(exact_dedupe(DB, ROOT, True)), "acoustic": acoustic_dedupe(DB, ROOT, True)} else: - scan(ROOT,DB); result={"exact":len(exact_dedupe(DB,ROOT)),"acoustic":acoustic_dedupe(DB,ROOT)} - with connect(DB) as c: c.execute("UPDATE jobs SET status='success',finished_at=CURRENT_TIMESTAMP,result=? WHERE id=?",(json.dumps(result,ensure_ascii=False),jid)) + result = {"indexed": indexed, "exact": len(exact_dedupe(DB, ROOT)), "acoustic": acoustic_dedupe(DB, ROOT)} + with connect(DB) as c: c.execute("UPDATE jobs SET status='success',finished_at=CURRENT_TIMESTAMP,result=? WHERE id=?", (json.dumps(result,ensure_ascii=False),jid)) except Exception as e: - with connect(DB) as c: c.execute("UPDATE jobs SET status='failed',finished_at=CURRENT_TIMESTAMP,error=? WHERE id=?",(str(e),jid)) + with connect(DB) as c: c.execute("UPDATE jobs SET status='failed',finished_at=CURRENT_TIMESTAMP,error=? WHERE id=?", (str(e),jid)) threading.Thread(target=work,daemon=True).start() return jid + def rollback(opid): - with connect(DB) as c: row=c.execute("SELECT * FROM operations WHERE id=?",(opid,)).fetchone() + with connect(DB) as c: row = c.execute("SELECT * FROM operations WHERE id=?", (opid,)).fetchone() if not row or row["action"] not in ("exact-hardlink","acoustic-hardlink","manual-hardlink"): raise ValueError("operation not reversible") - loser,winner=Path(row["file_path"]),Path(row["winner_path"]) - if not loser.exists() or not winner.exists() or loser.stat().st_ino!=winner.stat().st_ino: raise ValueError("current file state changed; refusing rollback") - tmp=loser.with_name(loser.name+".dedupe-rollback-"+uuid.uuid4().hex) + loser,winner = Path(row["file_path"]),Path(row["winner_path"]) + if not loser.exists() or not winner.exists() or loser.stat().st_ino != winner.stat().st_ino: raise ValueError("current file state changed; refusing rollback") + tmp = loser.with_name(loser.name+".dedupe-rollback-"+uuid.uuid4().hex) shutil.copy2(winner,tmp); os.replace(tmp,loser) with connect(DB) as c: - c.execute("UPDATE audio_tracks SET status='active',review_status='rolled-back',updated_at=CURRENT_TIMESTAMP WHERE file_path=?",(str(loser),)) - c.execute("INSERT INTO operations(action,file_path,winner_path,details) VALUES(?,?,?,?)",("rollback",str(loser),str(winner),f"rollback of operation {opid}")) + c.execute("UPDATE audio_tracks SET status='active',review_status='rolled-back',updated_at=CURRENT_TIMESTAMP WHERE file_path=?", (str(loser),)) + c.execute("INSERT INTO operations(action,file_path,winner_path,details) VALUES(?,?,?,?)", ("rollback",str(loser),str(winner),f"rollback of operation {opid}")) + def approve(rid,cid): with connect(DB) as c: - r=c.execute("SELECT * FROM audio_tracks WHERE id=? AND status='review'",(rid,)).fetchone() - w=c.execute("SELECT * FROM audio_tracks WHERE id=? AND status IN ('active','hardlinked')",(cid,)).fetchone() + r=c.execute("SELECT * FROM audio_tracks WHERE id=? AND status='review'",(rid,)).fetchone(); w=c.execute("SELECT * FROM audio_tracks WHERE id=? AND status IN ('active','hardlinked')",(cid,)).fetchone() if not r or not w: raise ValueError("review item or candidate not found") lp,wp=Path(r["file_path"]),Path(w["file_path"]) if not lp.exists() or not wp.exists() or not lp.is_relative_to(ROOT) or not wp.is_relative_to(ROOT): raise ValueError("invalid library path") - if lp.stat().st_dev!=wp.stat().st_dev: raise ValueError("different filesystem") - if lp.stat().st_ino!=wp.stat().st_ino: - hardlink_duplicate(lp,wp) + if lp.stat().st_dev != wp.stat().st_dev: raise ValueError("different filesystem") + if lp.stat().st_ino != wp.stat().st_ino: hardlink_duplicate(lp,wp) with connect(DB) as c: c.execute("UPDATE audio_tracks SET status='hardlinked',review_status='manual-approved',updated_at=CURRENT_TIMESTAMP WHERE id=?",(rid,)) c.execute("INSERT INTO operations(action,file_path,winner_path,details) VALUES(?,?,?,?)",("manual-hardlink",str(lp),str(wp),"manual review approval")) + class Handler(BaseHTTPRequestHandler): def send(self,status,data,ctype="application/json"): - raw=data.encode() if isinstance(data,str) else json.dumps(data,ensure_ascii=False).encode() - self.send_response(status); self.send_header("Content-Type",ctype+"; charset=utf-8"); self.send_header("Content-Length",str(len(raw))); self.end_headers(); self.wfile.write(raw) - def authorized(self): - if not TOKEN: return True - return self.headers.get("Authorization","") == "Bearer "+TOKEN - + raw=data.encode() if isinstance(data,str) else json.dumps(data,ensure_ascii=False).encode(); self.send_response(status); self.send_header("Content-Type",ctype+"; charset=utf-8"); self.send_header("Content-Length",str(len(raw))); self.end_headers(); self.wfile.write(raw) + def authorized(self): return not TOKEN or self.headers.get("Authorization","") == "Bearer "+TOKEN def do_GET(self): p=urlparse(self.path).path try: if p=="/": return self.send(200,HTML,"text/html") if not self.authorized(): return self.send(401,{"error":"unauthorized"}) - if p=="/api/stats": return self.send(200,stats()) - if p=="/api/duplicates": return self.send(200,duplicates()) - if p=="/api/review": return self.send(200,review_items()) - if p=="/api/history": return self.send(200,history()) - if p=="/api/jobs": return self.send(200,jobs()) - if p=="/api/settings": return self.send(200,settings()) + routes={"/api/stats":stats,"/api/duplicates":duplicates,"/api/review":review_items,"/api/history":history,"/api/jobs":jobs,"/api/settings":settings} + if p in routes: return self.send(200,routes[p]()) return self.send(404,{"error":"not found"}) except Exception as e: return self.send(500,{"error":str(e)}) - def do_POST(self): p=urlparse(self.path).path if not self.authorized(): return self.send(401,{"error":"unauthorized"}) try: body=json.loads(self.rfile.read(int(self.headers.get("content-length","0"))) or b"{}") except Exception: return self.send(400,{"error":"invalid json"}) try: - if p=="/api/jobs": - if body.get("kind") not in {"scan","dry","dedupe"}: return self.send(400,{"error":"invalid job"}) - return self.send(202,{"ok":True,"job_id":run_job(body["kind"])}) + if p=="/api/jobs" and body.get("kind") in {"scan","dry","dedupe"}: return self.send(202,{"ok":True,"job_id":run_job(body["kind"])}) if p=="/api/review": if body.get("action")=="reject": with connect(DB) as c: c.execute("UPDATE audio_tracks SET status='active',review_status='rejected',updated_at=CURRENT_TIMESTAMP WHERE id=? AND status='review'",(int(body["id"]),)) return self.send(200,{"ok":True}) if body.get("action")=="approve": approve(int(body["id"]),int(body["candidate_id"])); return self.send(200,{"ok":True}) - return self.send(400,{"error":"invalid action"}) if p=="/api/rollback": rollback(int(body["id"])); return self.send(200,{"ok":True}) - return self.send(404,{"error":"not found"}) + return self.send(400,{"error":"invalid request"}) except Exception as e: return self.send(409,{"error":str(e)}) def log_message(self,*args): pass -def serve(host="0.0.0.0",port=8080): - print(f"AM Deduper UI listening on {host}:{port}",flush=True) - ThreadingHTTPServer((host,port),Handler).serve_forever() +def serve(host="0.0.0.0",port=8080): + print(f"AM Deduper UI listening on {host}:{port}",flush=True); ThreadingHTTPServer((host,port),Handler).serve_forever() diff --git a/pyproject.toml b/pyproject.toml index 8d44f98..1474a86 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "am-music-deduper" -version = "0.1.0" +version = "0.2.0" description = "Lightweight NAS music deduplication service for AllMusic" requires-python = ">=3.10" dependencies = [ diff --git a/tests/test_scanner.py b/tests/test_scanner.py new file mode 100644 index 0000000..e3905ae --- /dev/null +++ b/tests/test_scanner.py @@ -0,0 +1,21 @@ +from pathlib import Path + +from am_deduper.db import connect +from am_deduper.scanner import scan + + +def test_replacing_file_invalidates_content_hashes(tmp_path: Path): + root = tmp_path / "AllMusic" + (root / "admin").mkdir(parents=True) + path = root / "admin" / "song.flac" + db = root / ".dedupe" / "dedupe.db" + path.write_bytes(b"first" * 1000) + scan(root, db) + with connect(db) as c: + c.execute("UPDATE audio_tracks SET sha256='old-sha', chromaprint='old-fp' WHERE file_path=?", (str(path),)) + path.write_bytes(b"second" * 1000) + scan(root, db) + with connect(db) as c: + row = c.execute("SELECT sha256,chromaprint FROM audio_tracks WHERE file_path=?", (str(path),)).fetchone() + assert row["sha256"] is None + assert row["chromaprint"] is None