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, 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 DeduperAM 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:
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:
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 authorized(self):
if not TOKEN: return True
return 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())
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/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()