from __future__ import annotations 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 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() 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 # 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: 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) # Exact byte-for-byte dedupe is always safe. L2 remains opt-in. exact_dedupe(self.db_path, self.root) if self.l2: acoustic_dedupe(self.db_path, self.root) except (FileNotFoundError, PermissionError, OSError): return def on_created(self, event): if not event.is_directory: self._process(event.src_path) def on_moved(self, event): if not event.is_directory: self._process(event.dest_path) 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) observer = Observer() observer.schedule(Handler(root, db_path, l2=l2), str(root), recursive=True) observer.start() try: while True: time.sleep(1) finally: observer.stop() observer.join()