From a0673ab79ecec81bf43fb6097c1cbe9d68b48f40 Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Fri, 18 Sep 2026 06:53:57 +0800 Subject: [PATCH] feat: add NAS management UI and operation history --- README.md | 20 ++++- am_deduper/cli.py | 9 +++ am_deduper/db.py | 19 +++++ am_deduper/exact.py | 1 + am_deduper/l2.py | 1 + am_deduper/scanner.py | 7 +- am_deduper/web.py | 170 ++++++++++++++++++++++++++++++++++++++++++ docker-compose.yml | 27 ++++++- 8 files changed, 247 insertions(+), 7 deletions(-) create mode 100644 am_deduper/web.py diff --git a/README.md b/README.md index ea2d093..309ef53 100644 --- a/README.md +++ b/README.md @@ -92,7 +92,7 @@ docker pull gitea.king.nyc.mn/openclaw/am-deduper:latest mkdir -p /AppData/am-deduper ``` -然后使用 Compose 启动容器；首次执行建议先运行 scan 和 dry-run： +然后使用 Compose 启动容器；当前 Compose 已配置 NAS 实际目录，并提供独立 Web 管理界面。首次执行建议先运行 scan 和 dry-run： ```bash docker compose run --rm am-deduper scan /music/AllMusic --db /data/dedupe.db @@ -105,12 +105,28 @@ docker compose run --rm am-deduper dedupe /music/AllMusic --db /data/dedupe.db - docker compose run --rm am-deduper dedupe /music/AllMusic --db /data/dedupe.db ``` -然后启动长期监听： +然后启动长期监听和 Web 管理界面： ```bash docker compose up -d ``` +Web UI 默认端口为 `18080`： + +```text +http://:18080/ +``` + +管理界面包含： +- Dashboard：文件数、Hard Link、审核数、索引大小、操作记录。 +- 重复组：查看 SHA-256 完全重复组及估算节省空间。 +- 待审核：查看相似度、候选文件，可人工确认或保留。 +- 任务：查看 scan / dry-run / dedupe 状态。 +- 历史：查看 Hard Link 操作，并对仍保持同 inode 的操作执行安全回滚。 +- 扫描会把数据库中已经不存在的文件标记为 `missing`，避免脏索引继续计入统计。 + +> Web UI 当前面向 NAS 内网使用，未启用账号密码；不要直接暴露到公网。 + 更新版本： ```bash diff --git a/am_deduper/cli.py b/am_deduper/cli.py index 33e4658..ad2ebfb 100644 --- a/am_deduper/cli.py +++ b/am_deduper/cli.py @@ -7,6 +7,7 @@ from .exact import exact_dedupe from .l2 import acoustic_dedupe from .scanner import scan from .watcher import watch +from .web import serve def _db_path(root: Path, explicit: Path | None) -> Path: @@ -28,12 +29,20 @@ def main() -> None: if name != "scan": sp.add_argument("--dry-run", action="store_true") + srv = sub.add_parser("server") + srv.add_argument("--host", default="0.0.0.0") + srv.add_argument("--port", type=int, default=8080) + 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() + if args.cmd == "server": + serve(args.host, args.port) + return + db = _db_path(args.root, args.db) if args.cmd == "watch": diff --git a/am_deduper/db.py b/am_deduper/db.py index fadd2a8..a159613 100644 --- a/am_deduper/db.py +++ b/am_deduper/db.py @@ -33,6 +33,25 @@ 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); +CREATE TABLE IF NOT EXISTS operations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + action TEXT NOT NULL, + file_path TEXT NOT NULL, + winner_path TEXT, + details TEXT, + created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP +); +CREATE INDEX IF NOT EXISTS idx_operations_created ON operations(created_at); +CREATE TABLE IF NOT EXISTS jobs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + kind TEXT NOT NULL, + status TEXT NOT NULL, + started_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + finished_at TEXT, + result TEXT, + error TEXT +); +CREATE INDEX IF NOT EXISTS idx_jobs_started ON jobs(started_at); """ diff --git a/am_deduper/exact.py b/am_deduper/exact.py index ab775ee..1b119da 100644 --- a/am_deduper/exact.py +++ b/am_deduper/exact.py @@ -99,6 +99,7 @@ def exact_dedupe(db_path: Path, root: Path, dry_run: bool = False) -> list[tuple 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"],), diff --git a/am_deduper/l2.py b/am_deduper/l2.py index ad1679e..4c21397 100644 --- a/am_deduper/l2.py +++ b/am_deduper/l2.py @@ -53,6 +53,7 @@ def acoustic_dedupe(db_path: Path, root: Path, dry_run: bool = False) -> dict: 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(?,?,?,?)", ('acoustic-hardlink',str(lp),str(wp),f'chromaprint similarity={sim:.3f}')) conn.execute("UPDATE audio_tracks SET status='hardlinked',review_status=?,updated_at=CURRENT_TIMESTAMP WHERE id=?", (f"acoustic={sim:.3f};winner={wp}", loser["id"])) stats["equivalent"]+=1 return stats diff --git a/am_deduper/scanner.py b/am_deduper/scanner.py index 4c276f6..58de541 100644 --- a/am_deduper/scanner.py +++ b/am_deduper/scanner.py @@ -34,7 +34,12 @@ def index_file(conn, path: Path) -> None: def scan(root: Path, db_path: Path) -> int: count = 0 + seen = set() with connect(db_path) as conn: for path in iter_music(root): - index_file(conn, path); count += 1 + index_file(conn, path); seen.add(str(path)); count += 1 + rows = conn.execute("SELECT id,file_path,status FROM audio_tracks WHERE status!='missing'").fetchall() + for row in rows: + if row["file_path"] not in seen and not Path(row["file_path"]).exists(): + conn.execute("UPDATE audio_tracks SET status='missing',updated_at=CURRENT_TIMESTAMP WHERE id=?", (row["id"],)) return count diff --git a/am_deduper/web.py b/am_deduper/web.py new file mode 100644 index 0000000..054f6cb --- /dev/null +++ b/am_deduper/web.py @@ -0,0 +1,170 @@ +from __future__ import annotations +import json, os, shutil, threading, 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 +from .l2 import acoustic_dedupe + +ROOT=Path("/music/AllMusic"); DB=Path("/data/dedupe.db") +LOCK=threading.Lock() + +HTML=r''' +AM Deduper + +
AM Deduper · NAS Music Library
+ +
+

重复组

+

待人工审核

+

任务

+

最近操作

+
+ +''' + +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}]} + +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: out.append({"type":"SHA256 完全重复","savings":round(sum(x["file_size"] for x in items[1:])/1024/1024,2),"items":items}) + 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=[] + for r in rows: + 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) + 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() + 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() + return {"items":[dict(r) for r in rows]} + +def run_job(kind): + with connect(DB) as c: + cur=c.execute("INSERT INTO jobs(kind,status) VALUES(?,?)",(kind,"running")); jid=cur.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)} + 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)) + 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)) + 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() + 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) + 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}")) + +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() + 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) + 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 do_GET(self): + p=urlparse(self.path).path + try: + if p=="/": return self.send(200,HTML,"text/html") + 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()) + 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 + 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/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"}) + 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() + diff --git a/docker-compose.yml b/docker-compose.yml index 54f5000..e5f4922 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,7 +1,6 @@ services: am-deduper: - build: . - image: am-music-deduper:local + image: gitea.king.nyc.mn/openclaw/am-deduper:latest container_name: am-deduper restart: unless-stopped command: @@ -13,6 +12,26 @@ services: TZ: Asia/Shanghai PYTHONUNBUFFERED: "1" volumes: - - ${ALLMUSIC_PATH:?Set ALLMUSIC_PATH to the NAS AllMusic directory}:/music/AllMusic - - ${DEDUPE_DATA_PATH:?Set DEDUPE_DATA_PATH to the NAS app-data directory}:/data + - /vol2/1000/Docker/chfs/data/Backup/AllMusic:/music/AllMusic + - /vol2/1000/Docker/am-deduper/data:/data stop_grace_period: 30s + + am-deduper-web: + image: gitea.king.nyc.mn/openclaw/am-deduper:latest + container_name: am-deduper-web + restart: unless-stopped + command: + - server + - --host + - 0.0.0.0 + - --port + - "8080" + environment: + TZ: Asia/Shanghai + PYTHONUNBUFFERED: "1" + volumes: + - /vol2/1000/Docker/chfs/data/Backup/AllMusic:/music/AllMusic + - /vol2/1000/Docker/am-deduper/data:/data + ports: + - "18080:8080" + stop_grace_period: 10s