feat: automate exact dedupe and improve web feedback

am_deduper/watcher.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import threading
 import time
 from pathlib import Path
@@ -11,7 +12,7 @@ from .constants import MUSIC_EXTENSIONS, STABILITY_CHECKS, STABILITY_INTERVAL_SE
 from .db import connect
 from .exact import exact_dedupe
 from .l2 import acoustic_dedupe
-from .scanner import index_file
+from .scanner import index_file, scan
 
 
 class Handler(FileSystemEventHandler):
@@ -20,6 +21,17 @@ class Handler(FileSystemEventHandler):
         self.db_path = db_path
         self.l2 = l2
         self._lock = threading.Lock()
+        self._timers: dict[str, threading.Timer] = {}
+        self._timers_lock = threading.Lock()
+
+    def _record_job(self, kind: str, status: str, job_id: int, result=None, error=None):
+        with connect(self.db_path) as conn:
+            conn.execute("UPDATE jobs SET status=?,finished_at=CURRENT_TIMESTAMP,result=?,error=? WHERE id=?",
+                         (status, json.dumps(result, ensure_ascii=False) if result is not None else None, error, job_id))
+
+    def _new_job(self, kind: str) -> int:
+        with connect(self.db_path) as conn:
+            return conn.execute("INSERT INTO jobs(kind,status) VALUES(?,?)", (kind, "running")).lastrowid
 
     def _process(self, filename: str) -> None:
         path = Path(filename)
@@ -32,40 +44,75 @@ class Handler(FileSystemEventHandler):
         if ".dedupe" in path.parts or path.name.startswith("."):
             return
 
-        # watchdog may emit several events while a downloader is still writing.
-        # Serialize the worker so a file is never indexed while another event
-        # is changing the same database or replacing a duplicate.
         with self._lock:
+            job_id = self._new_job("auto-exact")
             try:
                 previous = path.stat()
                 stable = 0
                 while stable < STABILITY_CHECKS - 1:
                     time.sleep(STABILITY_INTERVAL_SECONDS)
                     current = path.stat()
-                    if (current.st_size, current.st_mtime_ns) == (
-                        previous.st_size,
-                        previous.st_mtime_ns,
-                    ):
+                    if (current.st_size, current.st_mtime_ns) == (previous.st_size, previous.st_mtime_ns):
                         stable += 1
                     else:
                         previous = current
                         stable = 0
                 with connect(self.db_path) as conn:
                     index_file(conn, path)
-                # Exact byte-for-byte dedupe is always safe. L2 remains opt-in.
-                exact_dedupe(self.db_path, self.root)
+                actions = exact_dedupe(self.db_path, self.root)
+                result = {"indexed": 1, "exact_hardlinked": len(actions), "path": str(path)}
                 if self.l2:
-                    acoustic_dedupe(self.db_path, self.root)
-            except (FileNotFoundError, PermissionError, OSError):
-                return
+                    result["acoustic"] = acoustic_dedupe(self.db_path, self.root)
+                self._record_job("auto-exact", "success", job_id, result=result)
+            except (FileNotFoundError, PermissionError, OSError) as exc:
+                self._record_job("auto-exact", "skipped", job_id, error=str(exc))
+            except Exception as exc:
+                self._record_job("auto-exact", "failed", job_id, error=str(exc))
+
+    def _schedule(self, filename: str) -> None:
+        with self._timers_lock:
+            old = self._timers.pop(filename, None)
+            if old:
+                old.cancel()
+            timer = threading.Timer(1.0, self._run_scheduled, args=(filename,))
+            timer.daemon = True
+            self._timers[filename] = timer
+            timer.start()
+
+    def _run_scheduled(self, filename: str) -> None:
+        with self._timers_lock:
+            self._timers.pop(filename, None)
+        self._process(filename)
 
     def on_created(self, event):
         if not event.is_directory:
-            self._process(event.src_path)
+            self._schedule(event.src_path)
+
+    def on_modified(self, event):
+        if not event.is_directory:
+            self._schedule(event.src_path)
 
     def on_moved(self, event):
         if not event.is_directory:
-            self._process(event.dest_path)
+            self._schedule(event.dest_path)
+
+
+def _initial_reconcile(root: Path, db_path: Path, l2: bool) -> None:
+    with connect(db_path) as conn:
+        job_id = conn.execute("INSERT INTO jobs(kind,status) VALUES(?,?)", ("startup-reconcile", "running")).lastrowid
+    try:
+        indexed = scan(root, db_path)
+        actions = exact_dedupe(db_path, root)
+        result = {"indexed": indexed, "exact_hardlinked": len(actions)}
+        if l2:
+            result["acoustic"] = acoustic_dedupe(db_path, root)
+        with connect(db_path) as conn:
+            conn.execute("UPDATE jobs SET status='success',finished_at=CURRENT_TIMESTAMP,result=? WHERE id=?",
+                         (json.dumps(result, ensure_ascii=False), job_id))
+    except Exception as exc:
+        with connect(db_path) as conn:
+            conn.execute("UPDATE jobs SET status='failed',finished_at=CURRENT_TIMESTAMP,error=? WHERE id=?", (str(exc), job_id))
+        raise
 
 
 def watch(root: Path, db_path: Path, l2: bool = False) -> None:
@@ -73,9 +120,11 @@ def watch(root: Path, db_path: Path, l2: bool = False) -> None:
     if not root.is_dir():
         raise ValueError(f"music root does not exist or is not a directory: {root}")
     db_path.parent.mkdir(parents=True, exist_ok=True)
+    _initial_reconcile(root, db_path, l2)
     observer = Observer()
     observer.schedule(Handler(root, db_path, l2=l2), str(root), recursive=True)
     observer.start()
+    print(f"AM Deduper watcher active: {root}", flush=True)
     try:
         while True:
             time.sleep(1)