70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
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
|
|
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"]))
|
|
return digest
|
|
|
|
|
|
def hardlink_duplicate(loser: Path, winner: Path) -> None:
|
|
tmp = loser.with_name(loser.name + ".dedupe-linking")
|
|
try:
|
|
if tmp.exists(): tmp.unlink()
|
|
os.link(winner, tmp)
|
|
os.replace(tmp, loser)
|
|
finally:
|
|
if tmp.exists():
|
|
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")
|
|
return (0 if admin else 1, r["created_at"], p)
|
|
return min(rows,key=key)
|
|
|
|
|
|
def exact_dedupe(db_path: Path, root: Path, dry_run: bool = False) -> list[tuple[str,str]]:
|
|
actions=[]
|
|
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)
|
|
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)))
|
|
return actions
|