feat: make NAS self-build deployment canonical

am_deduper/exact.py

@@ -2,68 +2,106 @@ 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
+    try:
+        st = path.stat()
+    except (FileNotFoundError, PermissionError):
+        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"]))
+    except (FileNotFoundError, PermissionError):
+        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:
+    """Atomically replace loser with a hard link to winner."""
+    if loser.stat().st_dev != winner.stat().st_dev:
+        raise OSError("hard link requires both files on the same filesystem")
     tmp = loser.with_name(loser.name + ".dedupe-linking")
     try:
-        if tmp.exists(): tmp.unlink()
+        if tmp.exists():
+            tmp.unlink()
         os.link(winner, tmp)
         os.replace(tmp, loser)
     finally:
         if tmp.exists():
-            try: tmp.unlink()
-            except OSError: pass
+            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")
+        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)
+
+    return min(rows, key=key)
 
 
-def exact_dedupe(db_path: Path, root: Path, dry_run: bool = False) -> list[tuple[str,str]]:
-    actions=[]
+def exact_dedupe(db_path: Path, root: Path, dry_run: bool = False) -> list[tuple[str, str]]:
+    actions: list[tuple[str, str]] = []
+    root = root.resolve()
     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)
+        rows = conn.execute(
+            "SELECT * FROM audio_tracks WHERE status IN ('active','hardlinked')"
+        ).fetchall()
+        groups: dict[tuple[int, str], list] = {}
+        for row in rows:
+            groups.setdefault((row["file_size"], row["fast_hash"]), []).append(row)
+
         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)))
+            if len(candidates) < 2:
+                continue
+            for row in candidates:
+                if not row["sha256"]:
+                    confirm_sha256(conn, row)
+            ids = [row["id"] for row in candidates]
+            placeholders = ",".join("?" * len(ids))
+            refreshed = conn.execute(
+                f"SELECT * FROM audio_tracks WHERE id IN ({placeholders})", ids
+            ).fetchall()
+            by_sha: dict[str, list] = {}
+            for row in refreshed:
+                if row["sha256"]:
+                    by_sha.setdefault(row["sha256"], []).append(row)
+
+            for same in by_sha.values():
+                if len(same) < 2:
+                    continue
+                winner = choose_winner(same)
+                wp = Path(winner["file_path"])
+                if not wp.exists() or not wp.is_relative_to(root):
+                    continue
+                for row in same:
+                    if row["id"] == winner["id"]:
+                        continue
+                    lp = Path(row["file_path"])
+                    if not lp.exists() or not lp.is_relative_to(root):
+                        continue
+                    if lp.stat().st_ino != wp.stat().st_ino:
+                        if not dry_run:
+                            hardlink_duplicate(lp, wp)
+                            conn.execute(
+                                "UPDATE audio_tracks SET status='hardlinked',updated_at=CURRENT_TIMESTAMP WHERE id=?",
+                                (row["id"],),
+                            )
+                        actions.append((str(lp), str(wp)))
     return actions