Add background Jellyfin indexing job manager

scene_search/index_jobs.py

@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+import threading
+from copy import deepcopy
+from datetime import datetime, timezone
+from typing import Any
+
+from .database import connect, init_db
+from .jellyfin_index import (
+    build_jellyfin_client,
+    index_libraries,
+    resolve_library_ids,
+)
+from .search_service import SearchService
+from .sync import rebuild_meili
+
+
+class IndexJobManager:
+    """Runs one plugin-triggered Jellyfin indexing job at a time."""
+
+    def __init__(self, search_service: SearchService) -> None:
+        self.search_service = search_service
+        self._state_lock = threading.Lock()
+        self._thread: threading.Thread | None = None
+        self._state: dict[str, Any] = {
+            "running": False,
+            "stage": "idle",
+            "started_at": None,
+            "finished_at": None,
+            "error": None,
+        }
+
+    @property
+    def running(self) -> bool:
+        with self._state_lock:
+            return bool(self._state.get("running"))
+
+    def status(self) -> dict[str, Any]:
+        with self._state_lock:
+            return deepcopy(self._state)
+
+    def libraries(self) -> list[dict[str, Any]]:
+        client = build_jellyfin_client()
+        libraries = client.media_folders()
+        init_db(self.search_service.db_path)
+
+        counts: dict[str, tuple[int, int]] = {}
+        with connect(self.search_service.db_path) as con:
+            rows = con.execute(
+                """
+                SELECT m.jellyfin_library_id,
+                       COUNT(DISTINCT m.id) AS indexed_items,
+                       COUNT(f.id) AS indexed_frames
+                FROM media m
+                LEFT JOIN frames f ON f.media_id = m.id
+                WHERE m.source_kind = 'jellyfin'
+                  AND m.jellyfin_library_id IS NOT NULL
+                GROUP BY m.jellyfin_library_id
+                """
+            ).fetchall()
+            for row in rows:
+                counts[str(row["jellyfin_library_id"])] = (
+                    int(row["indexed_items"]),
+                    int(row["indexed_frames"]),
+                )
+
+        result = []
+        for library in sorted(
+            libraries,
+            key=lambda item: str(item.get("Name") or "").casefold(),
+        ):
+            library_id = str(library.get("Id") or "")
+            indexed_items, indexed_frames = counts.get(library_id, (0, 0))
+            result.append(
+                {
+                    "id": library_id,
+                    "name": str(library.get("Name") or library_id),
+                    "indexed_items": indexed_items,
+                    "indexed_frames": indexed_frames,
+                }
+            )
+        return result
+
+    def start(self, library_ids: list[str], *, force: bool = False) -> dict[str, Any]:
+        with self._state_lock:
+            if self._state.get("running"):
+                raise RuntimeError("A Jellyfin indexing job is already running.")
+
+        client = build_jellyfin_client()
+        libraries = resolve_library_ids(client, library_ids)
+
+        with self._state_lock:
+            self._state = {
+                "running": True,
+                "stage": "starting",
+                "started_at": datetime.now(timezone.utc).isoformat(),
+                "finished_at": None,
+                "error": None,
+                "library_total": len(libraries),
+                "library_ids": [str(item.get("Id") or "") for item in libraries],
+            }
+
+        self._thread = threading.Thread(
+            target=self._run,
+            args=(libraries, force),
+            name="scene-search-index",
+            daemon=True,
+        )
+        self._thread.start()
+        return self.status()
+
+    def _progress(self, update: dict) -> None:
+        with self._state_lock:
+            self._state.update(update)
+
+    def _run(self, libraries: list[dict], force: bool) -> None:
+        try:
+            # Search and indexing share one model instance and never use it concurrently.
+            with self.search_service.model_lock:
+                embedder = self.search_service.get_embedder()
+                _, summary = index_libraries(
+                    build_jellyfin_client(),
+                    libraries,
+                    embedder=embedder,
+                    force=force,
+                    progress=self._progress,
+                )
+                self._progress({"stage": "syncing", **summary})
+                documents = rebuild_meili(progress=self._progress)
+
+            self._progress(
+                {
+                    "running": False,
+                    "stage": "complete",
+                    "finished_at": datetime.now(timezone.utc).isoformat(),
+                    "documents_processed": documents,
+                    **summary,
+                }
+            )
+        except Exception as exc:
+            self._progress(
+                {
+                    "running": False,
+                    "stage": "failed",
+                    "finished_at": datetime.now(timezone.utc).isoformat(),
+                    "error": str(exc),
+                }
+            )