Improve/pyview

buzz/core/events.py

@@ -3,6 +3,7 @@
 import json
 import threading
 from collections import deque
+from collections.abc import Callable
 from typing import Any
 
 from .utils import utc_now_iso
@@ -20,6 +21,7 @@ def __init__(
         self.events = deque(maxlen=maxlen)
         self.lock = threading.Lock()
         self.default_source = default_source
+        self.listeners: list[Callable[[dict[str, Any]], None]] = []
 
     def record(
         self,
@@ -39,6 +41,13 @@ def record(
             del event["source"]
         with self.lock:
             self.events.append(event)
+            listeners = list(self.listeners)
+
+        for listener in listeners:
+            try:
+                listener(event)
+            except Exception:
+                pass
 
         # Also print to stdout for legacy logging and visibility
         prefix = f"[{level.upper()}]" if level != "info" else ""
@@ -57,6 +66,11 @@ def reconfigure(self, maxlen: int) -> None:
         with self.lock:
             self.events = deque(self.events, maxlen=maxlen)
 
+    def add_listener(self, listener: Callable[[dict[str, Any]], None]) -> None:
+        """Register a callback invoked after each event is recorded."""
+        with self.lock:
+            self.listeners.append(listener)
+
 
 # Global registry for the process
 registry = EventRegistry()

buzz/core/state.py

@@ -297,7 +297,12 @@ def _etag(self, *parts: Any) -> str:
 class BuzzState:
     """Thread-safe cache of torrent state, snapshot, and Real-Debrid sync."""
 
-    def __init__(self, config: DavConfig, client: Any) -> None:
+    def __init__(
+        self,
+        config: DavConfig,
+        client: Any,
+        on_ui_change: Any | None = None,
+    ) -> None:
         """Initialize state storage and load persisted data from disk."""
         self.config = config
         self.client = client
@@ -329,6 +334,7 @@ def __init__(self, config: DavConfig, client: Any) -> None:
         self.hook_lock = threading.Lock()
         self.hook_task_active = False
         self._closed = False
+        self.on_ui_change = on_ui_change
 
     def _snapshot_exists_in_db(self) -> bool:
         row = self.conn.execute(
@@ -486,6 +492,7 @@ def _classified_changed_roots(
     def sync(self, *, trigger_hook: bool = True) -> SyncReport:
         """Sync torrent state with Real-Debrid and rebuild the snapshot."""
         hook_paths: list[str] = []
+        should_notify = False
         with self.lock:
             self.sync_in_progress = True
         try:
@@ -560,6 +567,7 @@ def sync(self, *, trigger_hook: bool = True) -> SyncReport:
                     self.snapshot_digest = digest
                     self._save_snapshot(self.snapshot, self.snapshot_digest)
                     self.snapshot_loaded = True
+                    should_notify = True
                     if trigger_hook and (
                         self.config.hook_command or self.config.curator_url
                     ):
@@ -570,6 +578,8 @@ def sync(self, *, trigger_hook: bool = True) -> SyncReport:
                 self.last_error = None
             if hook_paths:
                 self._enqueue_hook(hook_paths)
+            if should_notify:
+                self._notify_ui_change("sync")
             return report
         except Exception as exc:
             with self.lock:
@@ -789,6 +799,7 @@ def mark_startup_sync_complete(self) -> None:
         """Flag that the initial startup sync has finished."""
         with self.lock:
             self.startup_sync_complete = True
+        self._notify_ui_change("sync")
 
     def is_ready(self) -> bool:
         """Return True when the library is ready to serve DAV requests."""
@@ -918,6 +929,7 @@ def delete_torrent(self, torrent_id: str) -> OperationResult:
             if torrent_id in self.cache:
                 del self.cache[torrent_id]
                 self._delete_cache_entry(torrent_id)
+        self._notify_ui_change("archive")
         return {"status": "success"}
 
     def _add_to_archive(self, info: TorrentInfo, magnet: str | None = None) -> None:
@@ -983,6 +995,7 @@ def restore_trash(self, thash: str) -> OperationResult:
             if thash in self.trashcan:
                 del self.trashcan[thash]
                 self._delete_archive_entry(thash)
+        self._notify_ui_change("archive")
 
         return {"status": "success", "id": torrent_id}
 
@@ -992,8 +1005,17 @@ def delete_trash_permanently(self, thash: str) -> OperationResult:
             if thash in self.trashcan:
                 del self.trashcan[thash]
                 self._delete_archive_entry(thash)
+        self._notify_ui_change("archive")
         return {"status": "success"}
 
+    def _notify_ui_change(self, topic: str) -> None:
+        if self.on_ui_change is None:
+            return
+        try:
+            self.on_ui_change(topic)
+        except Exception:
+            pass
+
     def select_files(
         self, torrent_id: str, file_ids: list[str]
     ) -> OperationResult:

buzz/curator_app.py

@@ -1,9 +1,11 @@
 """FastAPI application for the curator service."""
 
+import json
 import traceback
 from contextlib import asynccontextmanager
+from urllib import request
 
-from fastapi import FastAPI
+from fastapi import BackgroundTasks, FastAPI
 from fastapi.responses import JSONResponse
 
 from .core.curator import Curator, RebuildError, build_library
@@ -29,6 +31,7 @@ def __init__(self, config: CuratorConfig) -> None:
 
         self.config = config
         self.curator = Curator(config)
+        registry.add_listener(self._notify_dav_ui)
 
         @asynccontextmanager
         async def lifespan(app: FastAPI):
@@ -72,38 +75,13 @@ def get_logs_count():
                 return {"count": len(registry.events)}
 
         @self.app.post("/rebuild")
-        async def rebuild(payload: dict | None = None):
+        async def rebuild(
+            background_tasks: BackgroundTasks,
+            payload: dict | None = None,
+        ):
             changed_roots = (payload or {}).get("changed_roots", [])
-            try:
-                report = self.curator.handle_rebuild(changed_roots)
-                return report
-            except Exception as exc:
-                payload = {"error": str(exc)}
-                if isinstance(exc, RebuildError):
-                    payload.update(exc.payload)
-
-                from urllib.error import HTTPError
-
-                cause = exc.__cause__
-                if isinstance(cause, HTTPError) and cause.code in (401, 403):
-                    record_event(
-                        "curator rebuild failed: "
-                        "Jellyfin API Token is invalid or unauthorized",
-                        level="error",
-                    )
-                    payload["error"] = (
-                        "Jellyfin API Token is invalid or unauthorized"
-                    )
-                    return JSONResponse(
-                        status_code=403, content=payload
-                    )
-
-                record_event(
-                    f"curator rebuild failed: {exc}\n"
-                    f"{traceback.format_exc()}",
-                    level="error",
-                )
-                return JSONResponse(status_code=500, content=payload)
+            background_tasks.add_task(self._run_rebuild, changed_roots)
+            return {"status": "rebuilding"}
 
         @self.app.get("/api/subtitles/status")
         def get_subtitles_status():
@@ -134,6 +112,61 @@ def trigger_subtitles_fetch(payload: dict | None = None):
             )
             return {"status": "triggered"}
 
+    def _run_rebuild(self, changed_roots: list[str]) -> None:
+        try:
+            self.curator.handle_rebuild(changed_roots)
+        except Exception as exc:
+            if isinstance(exc, RebuildError):
+                cause = exc.__cause__
+                from urllib.error import HTTPError
+
+                if isinstance(cause, HTTPError) and cause.code in (401, 403):
+                    record_event(
+                        "curator rebuild failed: "
+                        "Jellyfin API Token is invalid or unauthorized",
+                        level="error",
+                    )
+                    return
+
+            record_event(
+                f"curator rebuild failed: {exc}\n"
+                f"{traceback.format_exc()}",
+                level="error",
+            )
+
+    def _notify_dav_ui(self, event: dict) -> None:
+        if not self.config.dav_ui_notify_url:
+            return
+
+        payload = {
+            "topics": ["logs", "status"],
+            "message": {
+                "source": "curator",
+                "event": event.get("event"),
+                "level": event.get("level"),
+                "message": event.get("message", ""),
+            },
+        }
+        data = json.dumps(payload).encode("utf-8")
+        req = request.Request(
+            self.config.dav_ui_notify_url,
+            data=data,
+            method="POST",
+            headers={"Content-Type": "application/json"},
+        )
+        try:
+            with request.urlopen(req, timeout=2) as response:
+                if response.status not in (200, 204):
+                    print(
+                        f"[WARN] DAV UI notify returned HTTP {response.status}",
+                        flush=True,
+                    )
+        except Exception as exc:
+            print(
+                f"[WARN] DAV UI notify failed: {exc}",
+                flush=True,
+            )
+
 
 def run_curator_server(config: CuratorConfig) -> None:
     """Start the curator HTTP server."""