video side: library mapping backend (pick Movies/TV library)

The scan no longer blindly grabs every movie/show section — it reads the
libraries you map, like music's 'pick your Music library'.
- GET /api/video/libraries: discover the active server's Movies/TV libraries
  (Plex sections by type / Jellyfin views by CollectionType) + current
  selection. POST: save {movies, tv} per server into video_settings.
- sources.py: _build_source(movies_lib, tv_lib) filters to the mapped library;
  get_active_video_source() (used by the scanner) loads the saved selection;
  list_video_libraries() lists them unfiltered for the UI. Falls back to all
  libraries when nothing is mapped yet.
- VideoDatabase.get/set_library_selection (per-server). 6 tests added; 33 green.

api/video/__init__.py

@@ -40,8 +40,10 @@ def create_video_blueprint() -> Blueprint:
     from .dashboard import register_routes as reg_dashboard
     from .scan import register_routes as reg_scan
     from .library import register_routes as reg_library
+    from .libraries import register_routes as reg_libraries
     reg_dashboard(bp)
     reg_scan(bp)
     reg_library(bp)
+    reg_libraries(bp)
 
     return bp

api/video/libraries.py

@@ -0,0 +1,45 @@
+"""Video library mapping endpoints.
+
+GET  /api/video/libraries -> discover the active server's Movies/TV libraries
+                             + the user's current selection.
+POST /api/video/libraries -> save {movies, tv} (library titles) for the active
+                             server. The scanner then reads only those.
+"""
+
+from __future__ import annotations
+
+from flask import jsonify, request
+
+from utils.logging_config import get_logger
+
+logger = get_logger("video_api.libraries")
+
+
+def register_routes(bp):
+    @bp.route("/libraries", methods=["GET"])
+    def video_libraries():
+        from . import get_video_db
+        try:
+            from core.video.sources import list_video_libraries
+            from config.settings import config_manager
+            libs = list_video_libraries() or {"server": None, "movies": [], "tv": []}
+            server = libs.get("server") or config_manager.get_active_media_server()
+            libs["selected"] = (get_video_db().get_library_selection(server)
+                                if server else {"movies": None, "tv": None})
+            return jsonify(libs)
+        except Exception:
+            logger.exception("Failed to list video libraries")
+            return jsonify({"error": "Failed to list video libraries"}), 500
+
+    @bp.route("/libraries", methods=["POST"])
+    def save_video_libraries():
+        from . import get_video_db
+        try:
+            from config.settings import config_manager
+            body = request.get_json(silent=True) or {}
+            server = config_manager.get_active_media_server()
+            get_video_db().set_library_selection(server, body.get("movies"), body.get("tv"))
+            return jsonify({"status": "saved", "server": server})
+        except Exception:
+            logger.exception("Failed to save video library selection")
+            return jsonify({"error": "Failed to save selection"}), 500

core/video/sources.py

@@ -22,13 +22,10 @@ logger = get_logger("video_sources")
 PLEX_SCAN_TIMEOUT = 120
 
 
-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 configured.
-
-    Reuses the SHARED connection config (config_manager) — we don't reinvent
-    auth — but Plex gets a dedicated, long-timeout connection for the bulk scan
-    so it never inherits the shared client's short interactive timeout."""
+def _build_source(movies_lib=None, tv_lib=None):
+    """Build a media source for the active server, restricted to the named
+    Movies/TV libraries when given. Reuses the SHARED connection config — but
+    Plex gets a dedicated long-timeout connection for the bulk scan."""
     try:
         from config.settings import config_manager
     except Exception:
@@ -45,7 +42,7 @@ def get_active_video_source():
         try:
             from plexapi.server import PlexServer
             srv = PlexServer(base_url, token, timeout=PLEX_SCAN_TIMEOUT)
-            return PlexVideoSource(srv)
+            return PlexVideoSource(srv, movies_lib=movies_lib, tv_lib=tv_lib)
         except Exception:
             logger.exception("video sources: Plex connect failed")
             return None
@@ -56,7 +53,7 @@ def get_active_video_source():
             engine = get_media_server_engine()
             client = engine.client("jellyfin") if engine else None
             if client and client.ensure_connection() and getattr(client, "user_id", None):
-                return JellyfinVideoSource(client)
+                return JellyfinVideoSource(client, movies_lib=movies_lib, tv_lib=tv_lib)
         except Exception:
             logger.exception("video sources: Jellyfin connect failed")
         return None
@@ -64,18 +61,59 @@ def get_active_video_source():
     return None
 
 
+def _load_selection():
+    """The user's Movies/TV library choice for the active server (or {})."""
+    try:
+        from config.settings import config_manager
+        from database.video_database import VideoDatabase
+        server = config_manager.get_active_media_server()
+        return VideoDatabase().get_library_selection(server)
+    except Exception:
+        logger.exception("video sources: could not load library selection")
+        return {}
+
+
+def get_active_video_source():
+    """Source for SCANNING — restricted to the user-mapped Movies/TV libraries.
+    Falls back to all libraries when nothing is mapped yet."""
+    sel = _load_selection() or {}
+    return _build_source(sel.get("movies") or None, sel.get("tv") or None)
+
+
+def list_video_libraries():
+    """Discover the active server's video libraries for the mapping UI:
+    {'server', 'movies': [{'title'}], 'tv': [{'title'}]} or None."""
+    src = _build_source()
+    if src is None:
+        return None
+    out = src.available_libraries()
+    out["server"] = src.server_name
+    return out
+
+
 # ── Plex ──────────────────────────────────────────────────────────────────────
 class PlexVideoSource:
     server_name = "plex"
 
-    def __init__(self, server):
+    def __init__(self, server, movies_lib=None, tv_lib=None):
         self._server = server
+        self._movies_lib = movies_lib
+        self._tv_lib = tv_lib
 
-    def _sections(self, kind: str):
-        return [s for s in self._server.library.sections() if s.type == kind]
+    def _sections(self, kind: str, name=None):
+        secs = [s for s in self._server.library.sections() if s.type == kind]
+        if name:
+            secs = [s for s in secs if s.title == name]
+        return secs
+
+    def available_libraries(self) -> dict:
+        return {
+            "movies": [{"title": s.title} for s in self._sections("movie")],
+            "tv": [{"title": s.title} for s in self._sections("show")],
+        }
 
     def iter_movies(self, incremental=False):
-        for section in self._sections("movie"):
+        for section in self._sections("movie", self._movies_lib):
             items = section.search(sort="addedAt:desc", maxresults=100) if incremental else section.all()
             for m in items:
                 try:
@@ -84,7 +122,7 @@ class PlexVideoSource:
                     logger.exception("Plex: skipping movie %s", getattr(m, "title", "?"))
 
     def iter_shows(self, incremental=False):
-        for section in self._sections("show"):
+        for section in self._sections("show", self._tv_lib):
             items = section.search(sort="addedAt:desc", maxresults=50) if incremental else section.all()
             for sh in items:
                 try:
@@ -174,17 +212,28 @@ _JF_EP_FIELDS = "Overview,Path,MediaSources,PremiereDate,RunTimeTicks,IndexNumbe
 class JellyfinVideoSource:
     server_name = "jellyfin"
 
-    def __init__(self, client):
+    def __init__(self, client, movies_lib=None, tv_lib=None):
         self._c = client
         self.uid = client.user_id
+        self._movies_lib = movies_lib
+        self._tv_lib = tv_lib
 
     def _req(self, path, params=None):
         return self._c._make_request(path, params=params)
 
-    def _views(self, collection_type: str):
+    def _views(self, collection_type: str, name=None):
         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]
+        views = [v for v in resp.get("Items", [])
+                 if (v.get("CollectionType") or "").lower() == collection_type]
+        if name:
+            views = [v for v in views if v.get("Name") == name]
+        return views
+
+    def available_libraries(self) -> dict:
+        return {
+            "movies": [{"title": v.get("Name")} for v in self._views("movies")],
+            "tv": [{"title": v.get("Name")} for v in self._views("tvshows")],
+        }
 
     @staticmethod
     def _ticks_to_seconds(ticks):
@@ -210,7 +259,7 @@ class JellyfinVideoSource:
         }
 
     def iter_movies(self, incremental=False):
-        for view in self._views("movies"):
+        for view in self._views("movies", self._movies_lib):
             params = {"ParentId": view["Id"], "IncludeItemTypes": "Movie",
                       "Recursive": "true", "Fields": _JF_MOVIE_FIELDS}
             if incremental:
@@ -235,7 +284,7 @@ class JellyfinVideoSource:
         }
 
     def iter_shows(self, incremental=False):
-        for view in self._views("tvshows"):
+        for view in self._views("tvshows", self._tv_lib):
             params = {"ParentId": view["Id"], "IncludeItemTypes": "Series",
                       "Recursive": "true", "Fields": "Overview,ProductionYear,OfficialRating"}
             if incremental:

database/video_database.py

@@ -127,6 +127,17 @@ class VideoDatabase:
         finally:
             conn.close()
 
+    # ── library mapping (which server library is Movies / TV) ─────────────────
+    def get_library_selection(self, server: str) -> dict:
+        return {
+            "movies": self.get_setting(server + ".movies_library"),
+            "tv": self.get_setting(server + ".tv_library"),
+        }
+
+    def set_library_selection(self, server: str, movies, tv) -> None:
+        self.set_setting(server + ".movies_library", movies or "")
+        self.set_setting(server + ".tv_library", tv or "")
+
     # ── dashboard ─────────────────────────────────────────────────────────────
     def dashboard_stats(self) -> dict:
         """Live counts for the video dashboard, straight from video.db.

tests/test_video_api.py

@@ -32,6 +32,7 @@ def test_blueprint_exposes_dashboard_route():
     assert "/api/video/scan/request" in rules
     assert "/api/video/scan/status" in rules
     assert "/api/video/library" in rules
+    assert "/api/video/libraries" in rules
 
 
 def test_dashboard_endpoint_returns_zeroed_json(tmp_path):
@@ -60,6 +61,27 @@ def test_library_endpoint_lists_content(tmp_path):
         videoapi._video_db = None
 
 
+def test_libraries_endpoint_lists_and_saves(tmp_path, monkeypatch):
+    client, videoapi = _make_client(tmp_path)
+    try:
+        import core.video.sources as vs
+        monkeypatch.setattr(vs, "list_video_libraries", lambda: {
+            "server": "plex", "movies": [{"title": "Movies"}], "tv": [{"title": "TV"}]})
+        import config.settings as cs
+        monkeypatch.setattr(cs.config_manager, "get_active_media_server", lambda: "plex")
+
+        data = client.get("/api/video/libraries").get_json()
+        assert data["server"] == "plex"
+        assert [m["title"] for m in data["movies"]] == ["Movies"]
+        assert data["selected"]["movies"] is None
+
+        assert client.post("/api/video/libraries", json={"movies": "Movies", "tv": "TV"}).status_code == 200
+        data2 = client.get("/api/video/libraries").get_json()
+        assert data2["selected"] == {"movies": "Movies", "tv": "TV"}
+    finally:
+        videoapi._video_db = None
+
+
 def test_video_api_imports_nothing_from_music():
     base = Path(__file__).resolve().parent.parent / "api" / "video"
     for py in base.glob("*.py"):

tests/test_video_database.py

@@ -162,6 +162,14 @@ def test_dashboard_stats_counts_content_and_downloads(db):
     assert s["watchlist"] == 1 and s["wishlist"] == 1
 
 
+def test_library_selection_roundtrip(db):
+    assert db.get_library_selection("plex") == {"movies": None, "tv": None}
+    db.set_library_selection("plex", "Movies", "TV Shows")
+    assert db.get_library_selection("plex") == {"movies": "Movies", "tv": "TV Shows"}
+    # Per-server keys don't collide.
+    assert db.get_library_selection("jellyfin") == {"movies": None, "tv": None}
+
+
 def test_settings_kv_roundtrip(db):
     assert db.get_setting("download_dir", "unset") == "unset"
     db.set_setting("download_dir", "/data/video")