From e08010610e5ebb6944c66f6dc2e03c6e4c04c2d4 Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Fri, 18 Sep 2026 05:13:22 +0800 Subject: [PATCH] feat: make NAS self-build deployment canonical --- README.md | 52 ++++++++++++-------- am_deduper/cli.py | 27 +++++++---- am_deduper/exact.py | 110 ++++++++++++++++++++++++++++-------------- am_deduper/watcher.py | 58 ++++++++++++++-------- docker-compose.yml | 9 +++- tests/test_exact.py | 18 ++++++- 6 files changed, 186 insertions(+), 88 deletions(-) diff --git a/README.md b/README.md index dcd3116..324ee10 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ ```text AllMusic ↓ -文件稳定检测 +文件稳定检测（size + mtime） ↓ L1：size + head/tail 64 KiB ↓ @@ -19,17 +19,17 @@ L1：size + head/tail 64 KiB └─ 低相似 → 保留 ``` -### 关键安全规则 +### 安全规则 - SHA-256 是 Exact Duplicate 的最终依据；L1 只负责缩小候选范围。 - 不根据文件名、歌手、歌名、专辑或来源单独判重。 - 确认相同版本后，优先使用 hard link 保留每个用户原来的逻辑路径。 - 用户删除自己的路径不会影响其他 hard link；最后一个 link 删除后数据才释放。 -- 不跨文件系统创建 hard link。 +- hard link 要求源文件和目标文件位于同一文件系统。 - 正在写入的文件必须先通过稳定性检查。 - L2 对版本标记冲突的文件进入 review，不自动合并。 -- `--dry-run` 只报告/更新索引，不修改音乐文件。 - SQLite 是加速索引，文件系统是最终事实；数据库丢失后可以重新扫描。 +- `.dedupe` 不参与音乐扫描，避免把内部文件再次处理。 ## CLI @@ -46,18 +46,21 @@ am-deduper dedupe /music/AllMusic --db /data/dedupe.db # 长驻监听：默认只自动执行安全的 Exact hard-link am-deduper watch /music/AllMusic --db /data/dedupe.db -# 开启 L2 自动 acoustic 匹配（建议在 dry-run 验证阈值后再启用） +# 开启 L2 acoustic 匹配；建议先 dry-run 验证 am-deduper watch /music/AllMusic --db /data/dedupe.db --l2 ``` -## Docker / Feiniu NAS 部署 +## Feiniu NAS 部署 -wjyhk 负责构建镜像，推送到 Gitea Container Registry；NAS 只负责 pull + run，不在 NAS 上重新 build。 +生产环境采用 **Gitea 源码 → NAS 本地 Docker build → NAS 本地运行**。 +不需要 Docker Registry，也不需要让 wjyhk 构建生产镜像。 -NAS 的 `AllMusic` 与 dedupe 数据建议分开： +建议目录： ```text NAS +├── Gitea 仓库 +│ └── am-deduper/ ├── AllMusic/ │ ├── admin/ │ ├── alice/ @@ -67,29 +70,40 @@ NAS └── dedupe.db ``` -注意：如果以后启用 trash，trash 目录必须位于 `AllMusic` 所在同一文件系统内，不能把需要原子 rename/hard-link 语义的文件直接移动到 `/data`。 - ### NAS `.env` ```dotenv -AM_DEDUPER_IMAGE=//am-music-deduper: ALLMUSIC_PATH=/AllMusic DEDUPE_DATA_PATH=/AppData/am-deduper ``` -### NAS 部署 +### 首次部署 + +```bash +cd /am-deduper +git pull +mkdir -p /AppData/am-deduper +docker compose build +docker compose run --rm am-deduper scan /music/AllMusic --db /data/dedupe.db +docker compose run --rm am-deduper dedupe /music/AllMusic --db /data/dedupe.db --dry-run +``` + +确认 dry-run 输出后，再执行正式 Exact/L2： + +```bash +docker compose run --rm am-deduper dedupe /music/AllMusic --db /data/dedupe.db +``` + +然后启动长期监听： ```bash -mkdir -p "$DEDUPE_DATA_PATH" -docker login -docker compose pull docker compose up -d ``` -查看日志： +更新版本： ```bash -docker logs -f am-deduper +git pull +docker compose build +docker compose up -d ``` - -首次上线建议先暂停 `watch`，执行一次完整 `scan` + `dedupe --dry-run`，确认候选数量与 review 项，再启用正式 dedupe；L2 自动处理单独验证后再打开 `--l2`。 diff --git a/am_deduper/cli.py b/am_deduper/cli.py index 70c545b..33e4658 100644 --- a/am_deduper/cli.py +++ b/am_deduper/cli.py @@ -9,25 +9,32 @@ from .scanner import scan from .watcher import watch +def _db_path(root: Path, explicit: Path | None) -> Path: + if explicit: + return explicit + if str(root).startswith("/music/"): + return Path("/data/dedupe.db") + return root / ".dedupe" / "dedupe.db" + + def main() -> None: parser = argparse.ArgumentParser(prog="am-deduper") sub = parser.add_subparsers(dest="cmd", required=True) - def add_common(sp): + for name in ("scan", "dedupe", "reconcile"): + sp = sub.add_parser(name) sp.add_argument("root", type=Path) sp.add_argument("--db", type=Path, default=None) - sp.add_argument("--dry-run", action="store_true", help="report actions without changing files") + if name != "scan": + sp.add_argument("--dry-run", action="store_true") - add_common(sub.add_parser("scan")) - add_common(sub.add_parser("dedupe")) - add_common(sub.add_parser("reconcile")) - watch_parser = sub.add_parser("watch") - watch_parser.add_argument("root", type=Path) - watch_parser.add_argument("--db", type=Path, default=None) - watch_parser.add_argument("--l2", action="store_true", help="enable acoustic matching in the watcher") + wp = sub.add_parser("watch") + wp.add_argument("root", type=Path) + wp.add_argument("--db", type=Path, default=None) + wp.add_argument("--l2", action="store_true", help="enable acoustic matching") args = parser.parse_args() - db = args.db or (Path("/data/dedupe.db") if args.cmd != "watch" and str(args.root).startswith("/music/") else args.root / ".dedupe" / "dedupe.db") + db = _db_path(args.root, args.db) if args.cmd == "watch": watch(args.root, db, l2=args.l2) diff --git a/am_deduper/exact.py b/am_deduper/exact.py index e86fa60..ab775ee 100644 --- a/am_deduper/exact.py +++ b/am_deduper/exact.py @@ -2,68 +2,106 @@ 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 + try: + st = path.stat() + except (FileNotFoundError, PermissionError): + return None + if st.st_size != row["file_size"]: + return None digest = sha256(path) 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"])) + except (FileNotFoundError, PermissionError): + 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(loser: Path, winner: Path) -> None: + """Atomically replace loser with a hard link to winner.""" + if loser.stat().st_dev != winner.stat().st_dev: + raise OSError("hard link requires both files on the same filesystem") tmp = loser.with_name(loser.name + ".dedupe-linking") try: - if tmp.exists(): tmp.unlink() + if tmp.exists(): + tmp.unlink() os.link(winner, tmp) os.replace(tmp, loser) finally: if tmp.exists(): - try: tmp.unlink() - except OSError: pass + try: + tmp.unlink() + except OSError: + pass def choose_winner(rows): def key(r): - p=str(r["file_path"]); admin="/admin/" in p or p.endswith("/admin") + 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) + + return min(rows, key=key) -def exact_dedupe(db_path: Path, root: Path, dry_run: bool = False) -> list[tuple[str,str]]: - actions=[] +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={} - for r in rows: groups.setdefault((r["file_size"],r["fast_hash"]),[]).append(r) + 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 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 and not dry_run: - 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))) + 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 + ).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 + 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"]: + 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( + "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/watcher.py b/am_deduper/watcher.py index cecffd9..6c1f76b 100644 --- a/am_deduper/watcher.py +++ b/am_deduper/watcher.py @@ -1,5 +1,6 @@ from __future__ import annotations +import threading import time from pathlib import Path @@ -15,33 +16,48 @@ from .scanner import index_file class Handler(FileSystemEventHandler): def __init__(self, root: Path, db_path: Path, l2: bool = False): - self.root = root + self.root = root.resolve() self.db_path = db_path self.l2 = l2 + self._lock = threading.Lock() def _process(self, filename: str) -> None: path = Path(filename) - if path.is_symlink() or path.suffix.lower() not in MUSIC_EXTENSIONS or ".dedupe" in path.parts: + if path.is_symlink() or path.suffix.lower() not in MUSIC_EXTENSIONS: return try: - previous = path.stat().st_size - stable = 0 - while stable < STABILITY_CHECKS - 1: - time.sleep(STABILITY_INTERVAL_SECONDS) - size = path.stat().st_size - if size == previous: - stable += 1 - else: - previous = size - stable = 0 - with connect(self.db_path) as conn: - index_file(conn, path) - # Exact dedupe is byte-for-byte safe. Acoustic dedupe is opt-in. - exact_dedupe(self.db_path, self.root) - if self.l2: - acoustic_dedupe(self.db_path, self.root) - except (FileNotFoundError, PermissionError, OSError): + path.relative_to(self.root) + except ValueError: return + 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: + 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, + ): + 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) + if self.l2: + acoustic_dedupe(self.db_path, self.root) + except (FileNotFoundError, PermissionError, OSError): + return def on_created(self, event): if not event.is_directory: @@ -52,9 +68,11 @@ class Handler(FileSystemEventHandler): self._process(event.dest_path) - def watch(root: Path, db_path: Path, l2: bool = False) -> None: root = root.resolve() + 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) observer = Observer() observer.schedule(Handler(root, db_path, l2=l2), str(root), recursive=True) observer.start() diff --git a/docker-compose.yml b/docker-compose.yml index a5901d3..54f5000 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,9 +1,14 @@ services: am-deduper: - image: ${AM_DEDUPER_IMAGE:?Set AM_DEDUPER_IMAGE to the pushed Gitea registry image} + build: . + image: am-music-deduper:local container_name: am-deduper restart: unless-stopped - command: ["watch", "/music/AllMusic"] + command: + - watch + - /music/AllMusic + - --db + - /data/dedupe.db environment: TZ: Asia/Shanghai PYTHONUNBUFFERED: "1" diff --git a/tests/test_exact.py b/tests/test_exact.py index 22247aa..59e5d0e 100644 --- a/tests/test_exact.py +++ b/tests/test_exact.py @@ -19,4 +19,20 @@ def test_exact_dedupe_creates_hardlink(tmp_path: Path): actions = exact_dedupe(db, root) assert len(actions) == 1 assert a.stat().st_ino == b.stat().st_ino - assert a.read_bytes() == b.read_bytes() + + +def test_deleting_one_hardlink_keeps_data(tmp_path: Path): + root = tmp_path / "AllMusic" + (root / "admin").mkdir(parents=True) + (root / "alice").mkdir(parents=True) + a = root / "admin" / "song.flac" + b = root / "alice" / "song.flac" + a.write_bytes(b"same" * 1000) + b.write_bytes(a.read_bytes()) + db = root / ".dedupe" / "dedupe.db" + scan(root, db) + exact_dedupe(db, root) + inode = a.stat().st_ino + b.unlink() + assert a.exists() and a.stat().st_ino == inode + assert a.read_bytes() == b"same" * 1000