initial commit

backend/app/services/watcher.py

@@ -0,0 +1,140 @@
+import asyncio
+import logging
+from pathlib import Path
+from datetime import datetime
+from watchdog.observers import Observer
+from watchdog.events import FileSystemEventHandler, FileMovedEvent, FileCreatedEvent, FileDeletedEvent, DirMovedEvent
+from sqlalchemy import select
+
+from app.database import SessionLocal
+from app.models import Library, MediaItem
+from app.services.scanner import classify, hash_file
+
+log = logging.getLogger(__name__)
+
+_observers: dict[int, Observer] = {}
+
+
+class LibraryEventHandler(FileSystemEventHandler):
+    def __init__(self, library_id: int, library_path: str):
+        self.library_id = library_id
+        self.root = Path(library_path)
+
+    def _rel(self, abs_path: str) -> str:
+        return str(Path(abs_path).relative_to(self.root))
+
+    def on_moved(self, event):
+        asyncio.run(self._handle_move(event))
+
+    def on_created(self, event):
+        if not event.is_directory:
+            asyncio.run(self._handle_create(event.src_path))
+
+    def on_deleted(self, event):
+        if not event.is_directory:
+            asyncio.run(self._handle_delete(event.src_path))
+
+    async def _handle_move(self, event):
+        async with SessionLocal() as db:
+            if isinstance(event, DirMovedEvent):
+                old_prefix = self._rel(event.src_path)
+                new_prefix = self._rel(event.dest_path)
+                result = await db.execute(
+                    select(MediaItem).where(
+                        MediaItem.library_id == self.library_id,
+                        MediaItem.rel_path.startswith(old_prefix + "/"),
+                    )
+                )
+                for item in result.scalars().all():
+                    item.rel_path = new_prefix + item.rel_path[len(old_prefix):]
+                    item.updated_at = datetime.utcnow()
+            else:
+                src_rel = self._rel(event.src_path)
+                dest_rel = self._rel(event.dest_path)
+                result = await db.execute(
+                    select(MediaItem).where(
+                        MediaItem.library_id == self.library_id,
+                        MediaItem.rel_path == src_rel,
+                    )
+                )
+                item = result.scalars().first()
+                if item:
+                    item.rel_path = dest_rel
+                    item.filename = Path(event.dest_path).name
+                    item.missing = False
+                    item.updated_at = datetime.utcnow()
+            await db.commit()
+
+    async def _handle_create(self, abs_path: str):
+        path = Path(abs_path)
+        media_type = classify(path)
+        if not media_type:
+            return
+        rel = self._rel(abs_path)
+        async with SessionLocal() as db:
+            existing = await db.execute(
+                select(MediaItem).where(
+                    MediaItem.library_id == self.library_id,
+                    MediaItem.rel_path == rel,
+                )
+            )
+            if existing.scalars().first():
+                return
+            file_hash = hash_file(path)
+            item = MediaItem(
+                library_id=self.library_id,
+                rel_path=rel,
+                filename=path.name,
+                file_hash=file_hash,
+                media_type=media_type,
+                size_bytes=path.stat().st_size,
+            )
+            db.add(item)
+            await db.commit()
+
+    async def _handle_delete(self, abs_path: str):
+        rel = self._rel(abs_path)
+        async with SessionLocal() as db:
+            result = await db.execute(
+                select(MediaItem).where(
+                    MediaItem.library_id == self.library_id,
+                    MediaItem.rel_path == rel,
+                )
+            )
+            item = result.scalars().first()
+            if item:
+                item.missing = True
+                item.updated_at = datetime.utcnow()
+                await db.commit()
+
+
+def start_watcher(library_id: int, library_path: str):
+    if library_id in _observers:
+        return
+    handler = LibraryEventHandler(library_id, library_path)
+    observer = Observer()
+    observer.schedule(handler, library_path, recursive=True)
+    observer.start()
+    _observers[library_id] = observer
+    log.info("Started watcher for library %d at %s", library_id, library_path)
+
+
+def stop_watcher(library_id: int):
+    observer = _observers.pop(library_id, None)
+    if observer:
+        observer.stop()
+        observer.join()
+        log.info("Stopped watcher for library %d", library_id)
+
+
+async def start_all():
+    async with SessionLocal() as db:
+        result = await db.execute(select(Library))
+        libraries = result.scalars().all()
+    for lib in libraries:
+        start_watcher(lib.id, lib.path)
+
+
+async def stop_all():
+    for library_id in list(_observers.keys()):
+        stop_watcher(library_id)