59 lines
2.8 KiB
Python
59 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from .acoustic import chromaprint, similarity
|
|
from .arbitration import choose_winner, is_version_conflict, score
|
|
from .db import connect
|
|
from .constants import QUARANTINE_SUBDIR, TRASH_SUBDIR
|
|
|
|
MIN_SIMILARITY = 0.90
|
|
HIGH_SIMILARITY = 0.95
|
|
REVIEW_SIMILARITY = 0.80
|
|
DURATION_DELTA = 3.0
|
|
|
|
|
|
def _set_meta(conn, row, fp):
|
|
conn.execute("UPDATE audio_tracks SET chromaprint=?,score=?,updated_at=CURRENT_TIMESTAMP WHERE id=?", (fp,score(row),row["id"]))
|
|
|
|
|
|
def find_candidates(conn,row):
|
|
d=row["duration"]
|
|
if d is None: return []
|
|
return conn.execute("SELECT * FROM audio_tracks WHERE id!=? AND status IN ('active','hardlinked') AND duration IS NOT NULL AND duration BETWEEN ? AND ?", (row["id"],d-DURATION_DELTA,d+DURATION_DELTA)).fetchall()
|
|
|
|
|
|
def acoustic_dedupe(db_path: Path, root: Path) -> dict:
|
|
stats={"checked":0,"equivalent":0,"review":0,"unique":0,"errors":0}
|
|
with connect(db_path) as conn:
|
|
rows=conn.execute("SELECT * FROM audio_tracks WHERE status='active' ORDER BY id").fetchall()
|
|
for row in rows:
|
|
if row["chromaprint"]: continue
|
|
fp=chromaprint(Path(row["file_path"])); stats["checked"]+=1
|
|
if not fp: stats["errors"]+=1; continue
|
|
_set_meta(conn,row,fp)
|
|
candidates=find_candidates(conn,row); best=None
|
|
for c in candidates:
|
|
if not c["chromaprint"]:
|
|
cfp=chromaprint(Path(c["file_path"]))
|
|
if cfp: _set_meta(conn,c,cfp); c=dict(c); c["chromaprint"]=cfp
|
|
sim=similarity(fp,c["chromaprint"] or "")
|
|
if sim>=REVIEW_SIMILARITY and (best is None or sim>best[0]): best=(sim,c)
|
|
if not best or best[0]<REVIEW_SIMILARITY:
|
|
stats["unique"]+=1; continue
|
|
sim,c=best
|
|
if sim<MIN_SIMILARITY or is_version_conflict(row,c):
|
|
conn.execute("UPDATE audio_tracks SET status='review',review_status=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(f"similarity={sim:.3f}",row["id"]))
|
|
stats["review"]+=1; continue
|
|
winner=choose_winner([row,c]); loser=c if winner["id"]==row["id"] else row
|
|
lp=Path(loser["file_path"]); wp=Path(winner["file_path"])
|
|
if lp.exists() and wp.exists():
|
|
trash=root/TRASH_SUBDIR
|
|
trash.mkdir(parents=True,exist_ok=True)
|
|
target=trash/lp.name
|
|
n=1
|
|
while target.exists(): target=trash/f"{lp.stem}.{n}{lp.suffix}"; n+=1
|
|
lp.replace(target)
|
|
conn.execute("UPDATE audio_tracks SET status='trashed',file_path=?,trashed_at=CURRENT_TIMESTAMP,review_status=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(str(target),f"equivalent={sim:.3f};winner={wp}",loser["id"]))
|
|
stats["equivalent"]+=1
|
|
return stats
|