from __future__ import annotations
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()
HTML = r'''
AM Deduper
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,"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})
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 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}
def run_job(kind):
with connect(DB) as c:
jid = c.execute("INSERT INTO jobs(kind,status) VALUES(?,?)", (kind,"running")).lastrowid
def work():
try:
with LOCK:
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:
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))
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 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"})
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" 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})
if p=="/api/rollback": rollback(int(body["id"])); return self.send(200,{"ok":True})
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()