video downloads: monitor thread + grab/active/clear endpoints

Phase 2 — the engine:
- core/video/download_monitor.py: a daemon thread polls slskd for active video
  downloads, updates progress, and on completion MOVES the file from the shared
  download folder into the per-type library folder + marks it completed. The
  per-download decision (process_download) is pure (fs + slskd injected) — 6 tests.
- POST /downloads/grab: validates (Soulseek-only v1), resolves the target library by
  kind, starts the slskd download, records the row, lazily starts the monitor.
- GET /downloads/active (list + ensure monitor running) · POST /downloads/clear
  (drop finished).
14 tests, isolation guard + ruff clean. Grab button + Downloads page next.

api/video/downloads.py

@@ -177,6 +177,60 @@ def register_routes(bp):
             r.pop("_avail", None)
         return jsonify({"scope": scope, "results": results[:40], "live": live})
 
+    @bp.route("/downloads/grab", methods=["POST"])
+    def video_downloads_grab():
+        """Start a real download of a chosen release and track it. v1: Soulseek only.
+        Body: {kind, title, release_title, source, username, filename, size_bytes,
+        quality_label}."""
+        from . import get_video_db
+        from config.settings import config_manager
+        from core.video.download_monitor import ensure_started
+        from core.video.download_pipeline import target_dir_for
+        from core.video.slskd_download import start_download
+
+        body = request.get_json(silent=True) or {}
+        source = str(body.get("source") or "soulseek").lower()
+        if source != "soulseek":
+            return jsonify({"ok": False, "error": "Only Soulseek grabs are wired up so far."}), 400
+        username, filename = body.get("username"), body.get("filename")
+        if not username or not filename:
+            return jsonify({"ok": False, "error": "Missing the release's source info."}), 400
+
+        db = get_video_db()
+        paths = {k: db.get_setting(k) or "" for k in ("movies_path", "tv_path", "youtube_path")}
+        if not paths["movies_path"]:
+            paths["movies_path"] = db.get_setting("transfer_path") or ""
+        target = target_dir_for(body.get("kind"), paths)
+        if not target:
+            return jsonify({"ok": False, "error": "Set the library folder for this type on Settings → Downloads."}), 400
+
+        started = start_download(username, filename, body.get("size_bytes") or 0)
+        if not started.get("ok"):
+            return jsonify({"ok": False, "error": started.get("error") or "slskd refused the download."}), 502
+
+        dl_id = db.add_video_download({
+            "kind": str(body.get("kind") or "movie"), "title": body.get("title"),
+            "release_title": body.get("release_title") or body.get("filename"),
+            "source": "soulseek", "username": username, "filename": filename,
+            "size_bytes": int(body.get("size_bytes") or 0), "quality_label": body.get("quality_label"),
+            "target_dir": target, "status": "downloading",
+        })
+        ensure_started(get_video_db)
+        return jsonify({"ok": True, "id": dl_id})
+
+    @bp.route("/downloads/active", methods=["GET"])
+    def video_downloads_active():
+        from . import get_video_db
+        from core.video.download_monitor import ensure_started
+        db = get_video_db()
+        ensure_started(get_video_db)   # also (re)start the monitor when the page is open
+        return jsonify({"downloads": db.list_video_downloads()})
+
+    @bp.route("/downloads/clear", methods=["POST"])
+    def video_downloads_clear():
+        from . import get_video_db
+        return jsonify({"cleared": get_video_db().clear_finished_video_downloads()})
+
     @bp.route("/downloads/youtube-quality", methods=["GET"])
     def video_youtube_quality():
         # Separate, smaller profile — YouTube is yt-dlp, not scene/p2p releases.

core/video/download_monitor.py

@@ -0,0 +1,109 @@
+"""Background monitor that drives video downloads to completion.
+
+A daemon thread polls slskd for the active video downloads, updates their progress,
+and when one finishes MOVES the file from the shared download folder into the right
+per-type library folder (Movies / TV / YouTube) and marks it completed. Simple v1:
+slskd source only, flat move by basename.
+
+The per-download decision (``process_download``) is pure — filesystem + slskd are
+injected — so it's unit-tested; the thread loop is thin glue.
+
+Isolated: stdlib + the sibling video modules + shared config_manager; no music imports.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import threading
+import time
+
+from utils.logging_config import get_logger
+
+from core.video.download_pipeline import dest_path_for, find_completed_file
+from core.video.slskd_download import classify_state, find_transfer, list_downloads, progress_pct
+
+logger = get_logger("video.download_monitor")
+
+_INTERVAL = 3            # seconds between polls
+_started = False
+_lock = threading.Lock()
+
+
+def process_download(dl: dict, transfers: list, download_dir: str, *, lister, mover) -> dict | None:
+    """Decide the next state for one active download given the current slskd transfers.
+    Returns a patch dict for the DB row (or None to leave it untouched this tick)."""
+    t = find_transfer(transfers, dl.get("username"), dl.get("filename"))
+    if not t:
+        return None                                  # not registered yet (or cleared) — wait
+    state = classify_state(t.get("state"))
+    if state == "active":
+        return {"status": "downloading", "progress": progress_pct(t)}
+    if state == "failed":
+        return {"status": "failed", "error": "Soulseek transfer " + str(t.get("state") or "failed")}
+    # completed → locate the file on disk and move it into the library folder
+    src = find_completed_file(download_dir, dl.get("filename"), lister)
+    if not src:
+        return {"progress": 100.0}                   # slskd done; file still settling — retry
+    dest = dest_path_for(dl.get("target_dir"), src)
+    try:
+        mover(src, dest)
+    except Exception as e:   # noqa: BLE001 - any move failure marks the download failed
+        return {"status": "failed", "error": "Move failed: " + str(e)}
+    return {"status": "completed", "progress": 100.0, "dest_path": dest}
+
+
+def _move(src: str, dest: str) -> None:
+    os.makedirs(os.path.dirname(dest) or ".", exist_ok=True)
+    shutil.move(src, dest)
+
+
+def _walk(root: str):
+    for dirpath, _dirs, files in os.walk(str(root or ".")):
+        for f in files:
+            yield os.path.join(dirpath, f)
+
+
+def _tick(db) -> None:
+    active = db.get_active_video_downloads()
+    if not active:
+        return
+    from config.settings import config_manager
+    download_dir = str(config_manager.get("soulseek.download_path", "") or "")
+    transfers = list_downloads()
+    for dl in active:
+        upd = process_download(dl, transfers, download_dir, lister=_walk, mover=_move)
+        if not upd:
+            continue
+        if upd.get("status") in ("completed", "failed"):
+            upd.setdefault("completed_at", time.strftime("%Y-%m-%d %H:%M:%S"))
+        try:
+            db.update_video_download(dl["id"], **upd)
+        except Exception:
+            logger.exception("video download %s: failed to persist update", dl.get("id"))
+
+
+def _run(db_provider) -> None:
+    logger.info("video download monitor started")
+    while True:
+        try:
+            db = db_provider()
+            if db is not None:
+                _tick(db)
+        except Exception:
+            logger.exception("video download monitor tick failed")
+        time.sleep(_INTERVAL)
+
+
+def ensure_started(db_provider) -> None:
+    """Start the monitor thread once (idempotent). Called when the first grab happens."""
+    global _started
+    with _lock:
+        if _started:
+            return
+        _started = True
+        threading.Thread(target=_run, args=(db_provider,), daemon=True,
+                         name="video-download-monitor").start()
+
+
+__all__ = ["process_download", "ensure_started"]

tests/test_video_api.py

@@ -503,6 +503,44 @@ def test_downloads_search_endpoint_ranks_and_filters(tmp_path):
         videoapi._video_db = None
 
 
+def test_downloads_grab_and_active(tmp_path, monkeypatch):
+    import api.video as videoapi
+    import core.video.download_monitor as mon
+    import core.video.slskd_download as slskd
+    from database.video_database import VideoDatabase
+
+    monkeypatch.setattr(slskd, "start_download", lambda *a, **k: {"ok": True})
+    monkeypatch.setattr(mon, "ensure_started", lambda *a, **k: None)   # don't spawn the thread
+
+    db = VideoDatabase(database_path=str(tmp_path / "video_library.db"))
+    db.set_setting("movies_path", "/media/movies")
+    videoapi._video_db = db
+    app = Flask(__name__)
+    app.register_blueprint(videoapi.create_video_blueprint(), url_prefix="/api/video")
+    client = app.test_client()
+    try:
+        r = client.post("/api/video/downloads/grab", json={
+            "kind": "movie", "title": "The Matrix", "source": "soulseek",
+            "username": "neo", "filename": r"@@a\x\m.mkv", "size_bytes": 8, "quality_label": "1080p"})
+        out = r.get_json()
+        assert out["ok"] is True and out["id"] > 0
+        # tracked + routed to the Movies library.
+        act = client.get("/api/video/downloads/active").get_json()["downloads"]
+        assert len(act) == 1 and act[0]["status"] == "downloading"
+        assert act[0]["target_dir"] == "/media/movies" and act[0]["kind"] == "movie"
+        # missing source info → 400.
+        assert client.post("/api/video/downloads/grab",
+                           json={"kind": "movie", "source": "soulseek"}).status_code == 400
+        # non-soulseek not wired yet → 400.
+        assert client.post("/api/video/downloads/grab",
+                           json={"kind": "movie", "source": "torrent", "username": "u",
+                                 "filename": "f"}).status_code == 400
+        # clear removes only finished (none yet).
+        assert client.post("/api/video/downloads/clear").get_json()["cleared"] == 0
+    finally:
+        videoapi._video_db = None
+
+
 def test_slskd_config_shared_via_config_manager(tmp_path, monkeypatch):
     import api.video as videoapi
     import config.settings as cfg

tests/test_video_download_pipeline.py

@@ -83,3 +83,64 @@ def test_video_downloads_crud(tmp_path):
     assert listed[0]["status"] == "completed" and listed[0]["dest_path"] == "/media/movies/m.mkv"
     assert db.clear_finished_video_downloads() == 1
     assert db.list_video_downloads() == []
+
+
+# ── monitor decision (pure, injected fs) ──────────────────────────────────────
+def _dl(**kw):
+    base = {"id": 1, "username": "neo", "filename": r"@@a\Folder\movie.mkv", "target_dir": "/media/movies"}
+    base.update(kw)
+    return base
+
+
+def _xfer(state, **kw):
+    base = {"username": "neo", "filename": r"@@a\Folder\movie.mkv", "state": state, "size": 100, "transferred": 40}
+    base.update(kw)
+    return base
+
+
+def test_process_download_active_reports_progress():
+    from core.video.download_monitor import process_download
+    upd = process_download(_dl(), [_xfer("InProgress")], "/dl",
+                           lister=lambda d: [], mover=lambda s, d: None)
+    assert upd == {"status": "downloading", "progress": 40.0}
+
+
+def test_process_download_failed():
+    from core.video.download_monitor import process_download
+    upd = process_download(_dl(), [_xfer("Completed, Errored")], "/dl",
+                           lister=lambda d: [], mover=lambda s, d: None)
+    assert upd["status"] == "failed"
+
+
+def test_process_download_no_transfer_yet():
+    from core.video.download_monitor import process_download
+    assert process_download(_dl(), [], "/dl", lister=lambda d: [], mover=lambda s, d: None) is None
+
+
+def test_process_download_completed_moves_file():
+    from core.video.download_monitor import process_download
+    moved = {}
+    upd = process_download(
+        _dl(), [_xfer("Completed, Succeeded")], "/dl",
+        lister=lambda d: ["/dl/Folder/movie.mkv"],
+        mover=lambda s, d: moved.update(src=s, dest=d))
+    assert upd == {"status": "completed", "progress": 100.0, "dest_path": "/media/movies/movie.mkv"}
+    assert moved == {"src": "/dl/Folder/movie.mkv", "dest": "/media/movies/movie.mkv"}
+
+
+def test_process_download_completed_but_file_not_settled():
+    from core.video.download_monitor import process_download
+    upd = process_download(_dl(), [_xfer("Completed, Succeeded")], "/dl",
+                           lister=lambda d: [], mover=lambda s, d: None)
+    assert upd == {"progress": 100.0}   # no status change — retries next tick
+
+
+def test_process_download_move_failure_marks_failed():
+    from core.video.download_monitor import process_download
+
+    def boom(s, d):
+        raise OSError("disk full")
+
+    upd = process_download(_dl(), [_xfer("Completed, Succeeded")], "/dl",
+                           lister=lambda d: ["/dl/Folder/movie.mkv"], mover=boom)
+    assert upd["status"] == "failed" and "disk full" in upd["error"]