video side: library scanner (server = source of truth) + scan API

Reads the active media server and mirrors it into video.db, adapting the music
scan pattern (ask the server, upsert, prune what's gone) — isolated from music.
- core/video/scanner.py: server-agnostic VideoLibraryScanner. Consumes a media
  source (duck-typed) yielding normalized dicts; upserts movies + show trees,
  prunes removed items, reports progress/state. Skips pruning when a scan
  returns nothing (transient-failure safety). Background thread + scan_sync.
- core/video/sources.py: Plex + Jellyfin adapters that REUSE the shared
  connected clients (MediaServerEngine) but own all video-section logic; produce
  normalized dicts. (Validated against a live server by design; scanner itself
  is fully unit-tested with a fake source.)
- api/video/scan.py: POST /api/video/scan/request, GET /api/video/scan/status.
- .gitignore: video_library.db + sidecars (mirrors music); tests inject a
  tmp DB so none is ever created in the repo.
Tests: scan populate/prune/empty-safety/no-source-error, isolation guard
(core/video imports nothing from music), scan routes registered. 101 green.

.gitignore

@@ -16,6 +16,10 @@ database/music_library.db
 database/music_library.db-shm
 database/music_library.db-wal
 database/music_library.db.backup_*
+database/video_library.db
+database/video_library.db-shm
+database/video_library.db-wal
+database/video_library.db.backup_*
 database/api_call_history.json
 storage/image_cache/
 logs/*.log

api/video/__init__.py

@@ -38,6 +38,8 @@ def create_video_blueprint() -> Blueprint:
     bp = Blueprint("video_api", __name__)
 
     from .dashboard import register_routes as reg_dashboard
+    from .scan import register_routes as reg_scan
     reg_dashboard(bp)
+    reg_scan(bp)
 
     return bp

api/video/scan.py

@@ -0,0 +1,32 @@
+"""Video library scan endpoints.
+
+POST /api/video/scan/request -> start a background scan of the active server
+GET  /api/video/scan/status  -> current scan progress/state
+
+The scan READS the media server (source of truth) into video.db. Triggering the
+server's own rescan (post-download) is wired separately into the download flow.
+"""
+
+from __future__ import annotations
+
+from flask import jsonify
+
+from utils.logging_config import get_logger
+
+logger = get_logger("video_api.scan")
+
+
+def register_routes(bp):
+    @bp.route("/scan/request", methods=["POST"])
+    def video_scan_request():
+        from . import get_video_db
+        from core.video.scanner import get_video_scanner
+        from core.video.sources import get_active_video_source
+        scanner = get_video_scanner(get_video_db())
+        return jsonify(scanner.request_scan(get_active_video_source))
+
+    @bp.route("/scan/status", methods=["GET"])
+    def video_scan_status():
+        from . import get_video_db
+        from core.video.scanner import get_video_scanner
+        return jsonify(get_video_scanner(get_video_db()).get_status())

core/video/__init__.py

@@ -0,0 +1,5 @@
+"""SoulSync — isolated VIDEO core package.
+
+Holds the video library scanner and media-server adapters. Music never imports
+this package; this package never imports the music database layer.
+"""

core/video/scanner.py

@@ -0,0 +1,125 @@
+"""SoulSync — video library scanner.
+
+The media SERVER (Plex/Jellyfin) is the source of truth, exactly like the music
+side: we ask the server what it has and mirror it into video.db. This module is
+server-agnostic — it consumes a "video media source" (duck-typed) that yields
+normalized dicts, so it never touches a media-server SDK directly. The Plex /
+Jellyfin adapters live in core/video/sources.py.
+
+A source must provide:
+    source.server_name -> 'plex' | 'jellyfin'
+    source.iter_movies() -> iterable of normalized movie dicts
+    source.iter_shows()  -> iterable of normalized show dicts (with seasons/episodes)
+
+ISOLATION: imports only video.db + shared infra; music never imports this.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+
+from utils.logging_config import get_logger
+
+logger = get_logger("video_scanner")
+
+
+class VideoLibraryScanner:
+    """Reads the active media server and upserts movies/shows into video.db."""
+
+    def __init__(self, db):
+        self.db = db
+        self._lock = threading.Lock()
+        self._status = {"state": "idle"}
+        self._thread = None
+
+    def get_status(self) -> dict:
+        with self._lock:
+            return dict(self._status)
+
+    def _set(self, **kw) -> None:
+        with self._lock:
+            self._status.update(kw)
+
+    def request_scan(self, source_factory) -> dict:
+        """Kick off a background scan. ``source_factory()`` returns a media
+        source (or None if no video-capable server is connected)."""
+        with self._lock:
+            if self._status.get("state") == "scanning":
+                return {"status": "in_progress"}
+            self._status = {"state": "scanning", "phase": "starting",
+                            "started_at": time.time(),
+                            "movies": 0, "shows": 0, "episodes": 0}
+        self._thread = threading.Thread(
+            target=self._run, args=(source_factory,), daemon=True)
+        self._thread.start()
+        return {"status": "started"}
+
+    def scan_sync(self, source_factory) -> dict:
+        """Run a scan inline (used by tests / callers that want to block)."""
+        with self._lock:
+            self._status = {"state": "scanning", "phase": "starting",
+                            "started_at": time.time(),
+                            "movies": 0, "shows": 0, "episodes": 0}
+        self._run(source_factory)
+        return self.get_status()
+
+    def _run(self, source_factory) -> None:
+        try:
+            source = source_factory()
+            if source is None:
+                self._set(state="error", phase="no video server",
+                          error="No connected Plex/Jellyfin video server")
+                return
+            server = source.server_name
+
+            # ── Movies ──
+            self._set(phase="scanning movies")
+            seen_movies: set[str] = set()
+            movies = 0
+            for item in source.iter_movies():
+                self.db.upsert_movie(server, item)
+                seen_movies.add(str(item["server_id"]))
+                movies += 1
+                if movies % 25 == 0:
+                    self._set(movies=movies)
+            self._set(movies=movies)
+            # Prune only when we actually saw items — avoids wiping the library
+            # if the server returned nothing due to a transient failure.
+            removed_m = self.db.prune_missing("movies", server, seen_movies) if seen_movies else 0
+
+            # ── Shows ──
+            self._set(phase="scanning shows")
+            seen_shows: set[str] = set()
+            shows = 0
+            episodes = 0
+            for show in source.iter_shows():
+                self.db.upsert_show_tree(server, show)
+                seen_shows.add(str(show["server_id"]))
+                shows += 1
+                episodes += sum(len(s.get("episodes", [])) for s in show.get("seasons", []))
+                self._set(shows=shows, episodes=episodes)
+            removed_s = self.db.prune_missing("shows", server, seen_shows) if seen_shows else 0
+
+            self._set(state="done", phase="complete", finished_at=time.time(),
+                      movies=movies, shows=shows, episodes=episodes,
+                      removed=removed_m + removed_s)
+            logger.info("Video scan complete: %d movies, %d shows, %d episodes (%d pruned)",
+                        movies, shows, episodes, removed_m + removed_s)
+        except Exception as e:  # noqa: BLE001 - report any failure to the UI
+            logger.exception("Video library scan failed")
+            self._set(state="error", phase="failed", error=str(e))
+
+
+# Module-level singleton, bound to the (single) video DB.
+_scanner = None
+_scanner_lock = threading.Lock()
+
+
+def get_video_scanner(db) -> VideoLibraryScanner:
+    global _scanner
+    if _scanner is None:
+        with _scanner_lock:
+            if _scanner is None:
+                _scanner = VideoLibraryScanner(db)
+    return _scanner

core/video/sources.py

@@ -0,0 +1,256 @@
+"""SoulSync — video media-server adapters (Plex / Jellyfin).
+
+Turn the live, already-connected media clients (owned by the shared
+MediaServerEngine) into normalized dicts the scanner understands. We REUSE the
+shared connection/auth (don't reinvent it) but keep all video-section logic here
+so music code is untouched.
+
+NOTE: these talk to real Plex/Jellyfin servers and can only be fully validated
+against a live server. The scanner itself is server-agnostic and unit-tested
+with a fake source; bugs found here against a real library are localized to
+these adapters.
+"""
+
+from __future__ import annotations
+
+from utils.logging_config import get_logger
+
+logger = get_logger("video_sources")
+
+
+def get_active_video_source():
+    """Return a media source for the active server, or None when the active
+    server has no video support (Navidrome/standalone) or isn't connected."""
+    try:
+        from config.settings import config_manager
+        from core.media_server.engine import get_media_server_engine
+    except Exception:
+        logger.exception("video sources: shared infra unavailable")
+        return None
+
+    server = config_manager.get_active_media_server()
+    engine = get_media_server_engine()
+    if not engine:
+        return None
+    client = engine.client(server) if server in ("plex", "jellyfin") else None
+    if not client:
+        return None
+    try:
+        if not client.ensure_connection():
+            return None
+    except Exception:
+        logger.exception("video sources: %s ensure_connection failed", server)
+        return None
+
+    if server == "plex" and getattr(client, "server", None) is not None:
+        return PlexVideoSource(client)
+    if server == "jellyfin" and getattr(client, "user_id", None):
+        return JellyfinVideoSource(client)
+    return None
+
+
+# ── Plex ──────────────────────────────────────────────────────────────────────
+class PlexVideoSource:
+    server_name = "plex"
+
+    def __init__(self, client):
+        self._server = client.server
+
+    def _sections(self, kind: str):
+        return [s for s in self._server.library.sections() if s.type == kind]
+
+    def iter_movies(self):
+        for section in self._sections("movie"):
+            for m in section.all():
+                yield self._movie(m)
+
+    def iter_shows(self):
+        for section in self._sections("show"):
+            for sh in section.all():
+                yield self._show(sh)
+
+    @staticmethod
+    def _part_file(obj):
+        try:
+            media = obj.media[0]
+            part = media.parts[0]
+            return {
+                "relative_path": part.file,
+                "size_bytes": getattr(part, "size", None),
+                "resolution": getattr(media, "videoResolution", None),
+                "video_codec": getattr(media, "videoCodec", None),
+                "audio_codec": getattr(media, "audioCodec", None),
+                "runtime_seconds": int(obj.duration / 1000) if getattr(obj, "duration", None) else None,
+            }
+        except Exception:
+            return None
+
+    def _movie(self, m) -> dict:
+        dur = getattr(m, "duration", None)
+        return {
+            "server_id": str(m.ratingKey),
+            "title": m.title,
+            "year": getattr(m, "year", None),
+            "overview": getattr(m, "summary", None),
+            "poster_url": getattr(m, "thumb", None),
+            "content_rating": getattr(m, "contentRating", None),
+            "studio": getattr(m, "studio", None),
+            "runtime_minutes": int(dur / 60000) if dur else None,
+            "file": self._part_file(m),
+        }
+
+    def _show(self, sh) -> dict:
+        seasons = []
+        try:
+            for se in sh.seasons():
+                episodes = []
+                for ep in se.episodes():
+                    dur = getattr(ep, "duration", None)
+                    aired = getattr(ep, "originallyAvailableAt", None)
+                    episodes.append({
+                        "server_id": str(ep.ratingKey),
+                        "season_number": ep.parentIndex if getattr(ep, "parentIndex", None) is not None
+                        else getattr(se, "seasonNumber", 0),
+                        "episode_number": getattr(ep, "index", 0),
+                        "title": ep.title,
+                        "overview": getattr(ep, "summary", None),
+                        "air_date": aired.date().isoformat() if aired else None,
+                        "runtime_minutes": int(dur / 60000) if dur else None,
+                        "file": self._part_file(ep),
+                    })
+                seasons.append({
+                    "server_id": str(se.ratingKey),
+                    "season_number": getattr(se, "seasonNumber", 0),
+                    "title": se.title,
+                    "overview": getattr(se, "summary", None),
+                    "poster_url": getattr(se, "thumb", None),
+                    "episodes": episodes,
+                })
+        except Exception:
+            logger.exception("Plex: failed reading seasons/episodes for %s",
+                             getattr(sh, "title", "?"))
+        return {
+            "server_id": str(sh.ratingKey),
+            "title": sh.title,
+            "year": getattr(sh, "year", None),
+            "overview": getattr(sh, "summary", None),
+            "poster_url": getattr(sh, "thumb", None),
+            "status": None,
+            "network": getattr(sh, "network", None),
+            "content_rating": getattr(sh, "contentRating", None),
+            "seasons": seasons,
+        }
+
+
+# ── Jellyfin ────────────────────────────────────────────────────────────────
+_JF_MOVIE_FIELDS = "Overview,Path,MediaSources,ProductionYear,OfficialRating,RunTimeTicks,Studios"
+_JF_EP_FIELDS = "Overview,Path,MediaSources,PremiereDate,RunTimeTicks,IndexNumber,ParentIndexNumber"
+
+
+class JellyfinVideoSource:
+    server_name = "jellyfin"
+
+    def __init__(self, client):
+        self._c = client
+        self.uid = client.user_id
+
+    def _req(self, path, params=None):
+        return self._c._make_request(path, params=params)
+
+    def _views(self, collection_type: str):
+        resp = self._req(f"/Users/{self.uid}/Views") or {}
+        return [v for v in resp.get("Items", [])
+                if (v.get("CollectionType") or "").lower() == collection_type]
+
+    @staticmethod
+    def _ticks_to_seconds(ticks):
+        return int(ticks / 10_000_000) if ticks else None
+
+    @staticmethod
+    def _file(item):
+        sources = item.get("MediaSources") or []
+        if not sources:
+            path = item.get("Path")
+            return {"relative_path": path} if path else None
+        src = sources[0]
+        streams = src.get("MediaStreams") or []
+        vid = next((s for s in streams if s.get("Type") == "Video"), {})
+        aud = next((s for s in streams if s.get("Type") == "Audio"), {})
+        return {
+            "relative_path": src.get("Path") or item.get("Path") or "",
+            "size_bytes": src.get("Size"),
+            "resolution": (str(vid.get("Height")) + "p") if vid.get("Height") else None,
+            "video_codec": vid.get("Codec"),
+            "audio_codec": aud.get("Codec"),
+            "runtime_seconds": JellyfinVideoSource._ticks_to_seconds(item.get("RunTimeTicks")),
+        }
+
+    def iter_movies(self):
+        for view in self._views("movies"):
+            resp = self._req(f"/Users/{self.uid}/Items", {
+                "ParentId": view["Id"], "IncludeItemTypes": "Movie",
+                "Recursive": "true", "Fields": _JF_MOVIE_FIELDS}) or {}
+            for it in resp.get("Items", []):
+                yield self._movie(it)
+
+    def _movie(self, it) -> dict:
+        studios = it.get("Studios") or []
+        ticks = it.get("RunTimeTicks")
+        return {
+            "server_id": str(it["Id"]),
+            "title": it.get("Name"),
+            "year": it.get("ProductionYear"),
+            "overview": it.get("Overview"),
+            "poster_url": (it.get("ImageTags") or {}).get("Primary"),
+            "content_rating": it.get("OfficialRating"),
+            "studio": studios[0].get("Name") if studios else None,
+            "runtime_minutes": int(ticks / 600_000_000) if ticks else None,
+            "file": self._file(it),
+        }
+
+    def iter_shows(self):
+        for view in self._views("tvshows"):
+            resp = self._req(f"/Users/{self.uid}/Items", {
+                "ParentId": view["Id"], "IncludeItemTypes": "Series",
+                "Recursive": "true", "Fields": "Overview,ProductionYear,OfficialRating"}) or {}
+            for it in resp.get("Items", []):
+                yield self._show(it)
+
+    def _show(self, it) -> dict:
+        series_id = str(it["Id"])
+        seasons = []
+        try:
+            eps_resp = self._req(f"/Shows/{series_id}/Episodes", {
+                "UserId": self.uid, "Fields": _JF_EP_FIELDS}) or {}
+            by_season: dict[int, list] = {}
+            for ep in eps_resp.get("Items", []):
+                snum = ep.get("ParentIndexNumber") or 0
+                aired = ep.get("PremiereDate")
+                ticks = ep.get("RunTimeTicks")
+                by_season.setdefault(snum, []).append({
+                    "server_id": str(ep["Id"]),
+                    "season_number": snum,
+                    "episode_number": ep.get("IndexNumber") or 0,
+                    "title": ep.get("Name"),
+                    "overview": ep.get("Overview"),
+                    "air_date": aired[:10] if aired else None,
+                    "runtime_minutes": int(ticks / 600_000_000) if ticks else None,
+                    "file": self._file(ep),
+                })
+            for snum, eps in sorted(by_season.items()):
+                seasons.append({"server_id": None, "season_number": snum,
+                                "title": None, "overview": None,
+                                "poster_url": None, "episodes": eps})
+        except Exception:
+            logger.exception("Jellyfin: failed reading episodes for %s", it.get("Name", "?"))
+        return {
+            "server_id": series_id,
+            "title": it.get("Name"),
+            "year": it.get("ProductionYear"),
+            "overview": it.get("Overview"),
+            "poster_url": (it.get("ImageTags") or {}).get("Primary"),
+            "status": None,
+            "network": None,
+            "content_rating": it.get("OfficialRating"),
+            "seasons": seasons,
+        }

tests/test_video_api.py

@@ -12,10 +12,12 @@ from pathlib import Path
 from flask import Flask
 
 
-def _make_client(tmp_path, monkeypatch):
-    monkeypatch.setenv("VIDEO_DATABASE_PATH", str(tmp_path / "video_library.db"))
+def _make_client(tmp_path):
+    # Inject a tmp-backed DB directly so the endpoint never falls back to the
+    # real default path (no stray database/video_library.db in the repo).
     import api.video as videoapi
-    videoapi._video_db = None  # drop any cached handle so the env path is used
+    from database.video_database import VideoDatabase
+    videoapi._video_db = VideoDatabase(database_path=str(tmp_path / "video_library.db"))
     app = Flask(__name__)
     app.register_blueprint(videoapi.create_video_blueprint(), url_prefix="/api/video")
     return app.test_client(), videoapi
@@ -27,10 +29,12 @@ def test_blueprint_exposes_dashboard_route():
     app.register_blueprint(create_video_blueprint(), url_prefix="/api/video")
     rules = {r.rule for r in app.url_map.iter_rules()}
     assert "/api/video/dashboard" in rules
+    assert "/api/video/scan/request" in rules
+    assert "/api/video/scan/status" in rules
 
 
-def test_dashboard_endpoint_returns_zeroed_json(tmp_path, monkeypatch):
-    client, videoapi = _make_client(tmp_path, monkeypatch)
+def test_dashboard_endpoint_returns_zeroed_json(tmp_path):
+    client, videoapi = _make_client(tmp_path)
     try:
         resp = client.get("/api/video/dashboard")
         assert resp.status_code == 200

tests/test_video_scanner.py

@@ -0,0 +1,77 @@
+"""Seam tests for the video library scanner (experimental branch).
+
+The scanner is server-agnostic: it consumes a media source that yields
+normalized dicts. We drive it with a fake source so the scan/prune/error logic
+is fully tested without a live Plex/Jellyfin. Also guards that core/video/
+imports nothing from the music database.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from database.video_database import VideoDatabase
+from core.video.scanner import VideoLibraryScanner
+
+
+class FakeSource:
+    server_name = "plex"
+
+    def __init__(self, movies, shows):
+        self._movies, self._shows = movies, shows
+
+    def iter_movies(self):
+        return iter(self._movies)
+
+    def iter_shows(self):
+        return iter(self._shows)
+
+
+@pytest.fixture()
+def db(tmp_path):
+    return VideoDatabase(database_path=str(tmp_path / "video_library.db"))
+
+
+def test_scan_sync_populates_library(db):
+    movies = [{"server_id": "m1", "title": "A", "file": {"relative_path": "a.mkv", "size_bytes": 5}}]
+    shows = [{"server_id": "s1", "title": "Show", "seasons": [
+        {"season_number": 1, "episodes": [
+            {"episode_number": 1, "title": "E1", "file": {"relative_path": "e1.mkv"}}]}]}]
+    st = VideoLibraryScanner(db).scan_sync(lambda: FakeSource(movies, shows))
+    assert st["state"] == "done"
+    assert (st["movies"], st["shows"], st["episodes"]) == (1, 1, 1)
+    lib = db.dashboard_stats()["library"]
+    assert (lib["movies"], lib["shows"], lib["episodes"]) == (1, 1, 1)
+
+
+def test_scan_sync_prunes_removed_items(db):
+    scanner = VideoLibraryScanner(db)
+    scanner.scan_sync(lambda: FakeSource(
+        [{"server_id": "m1", "title": "A"}, {"server_id": "m2", "title": "B"}], []))
+    assert db.dashboard_stats()["library"]["movies"] == 2
+    scanner.scan_sync(lambda: FakeSource([{"server_id": "m1", "title": "A"}], []))
+    assert db.dashboard_stats()["library"]["movies"] == 1
+
+
+def test_empty_scan_does_not_wipe_library(db):
+    # Safety: a scan that returns nothing (transient failure) must NOT prune.
+    scanner = VideoLibraryScanner(db)
+    scanner.scan_sync(lambda: FakeSource([{"server_id": "m1", "title": "A"}], []))
+    scanner.scan_sync(lambda: FakeSource([], []))
+    assert db.dashboard_stats()["library"]["movies"] == 1
+
+
+def test_scan_sync_no_source_reports_error(db):
+    st = VideoLibraryScanner(db).scan_sync(lambda: None)
+    assert st["state"] == "error" and "error" in st
+
+
+def test_core_video_imports_nothing_from_music():
+    base = Path(__file__).resolve().parent.parent / "core" / "video"
+    for py in base.glob("*.py"):
+        for line in py.read_text(encoding="utf-8").splitlines():
+            s = line.strip()
+            if s.startswith("import ") or s.startswith("from "):
+                assert "music" not in s.lower(), f"{py.name}: music import leaked: {s!r}"