from __future__ import annotations import json import threading import time from pathlib import Path from watchdog.events import FileSystemEventHandler from watchdog.observers import Observer from .constants import MUSIC_EXTENSIONS, STABILITY_CHECKS, STABILITY_INTERVAL_SECONDS from .db import connect from .exact import exact_dedupe from .l2 import acoustic_dedupe from .scanner import index_file, scan class Handler(FileSystemEventHandler): def __init__(self, root: Path, db_path: Path, l2: bool = False): self.root = root.resolve() 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) if path.is_symlink() or path.suffix.lower() not in MUSIC_EXTENSIONS: return try: path.relative_to(self.root) except ValueError: return if ".dedupe" in path.parts or path.name.startswith("."): return 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): stable += 1 else: previous = current stable = 0 with connect(self.db_path) as conn: index_file(conn, path) actions = exact_dedupe(self.db_path, self.root) result = {"indexed": 1, "exact_hardlinked": len(actions), "path": str(path)} if self.l2: 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._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._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: root = root.resolve() 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) finally: observer.stop() observer.join()