video downloads pipeline: data + pure logic foundation

Phase 1 of the real grab→transfer pipeline:
- video.db video_downloads table (kind/title/release/source/username/filename/size/
  target_dir/dest_path/status/progress/error/timestamps) + CRUD (add/list/get_active/
  update/clear_finished). Schema v14.
- core/video/slskd_download.py: start_download (POST /transfers/downloads/{user}),
  list_downloads (GET → flattened), + pure classify_state (completed/failed/active),
  progress_pct, find_transfer.
- core/video/download_pipeline.py: pure find_completed_file (locate by basename in the
  download dir), dest_path_for, target_dir_for (kind → movies/tv/youtube library).

All filesystem/HTTP is injected or thin glue; 7 tests, isolation guard + ruff clean.
Monitor + grab endpoint + Downloads page next.

core/video/download_pipeline.py

@@ -0,0 +1,54 @@
+"""Pure helpers for the video download pipeline: locate a finished file in the
+download dir and work out where it should move to.
+
+slskd writes a completed download somewhere under the shared download folder (it
+mirrors the remote folder structure), so we locate the file by basename. Kept pure
+(filesystem access is injected) so it's unit-tested without touching disk.
+
+Isolated: stdlib only; no music imports.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Callable
+
+
+def basename_of(path: Any) -> str:
+    """Final path segment, handling both / and \\ separators (slskd uses Windows-style)."""
+    return str(path or "").replace("\\", "/").rstrip("/").rsplit("/", 1)[-1]
+
+
+def find_completed_file(download_dir: str, filename: str, lister: Callable) -> str | None:
+    """Find the downloaded file on disk by basename. ``lister(dir)`` yields candidate
+    full paths (injected: os.walk-based in the monitor, a list in tests). Returns the
+    largest match (the real video, not a stray same-named bit), or None."""
+    base = basename_of(filename)
+    if not base or not download_dir:
+        return None
+    matches = [p for p in lister(download_dir) if basename_of(p) == base]
+    if not matches:
+        return None
+    return matches[0] if len(matches) == 1 else max(matches, key=len)
+
+
+def dest_path_for(target_dir: str, src_path: str) -> str:
+    """Where a finished file moves to inside its library folder (flat, by basename)."""
+    return os.path.join(str(target_dir or ""), basename_of(src_path))
+
+
+def target_dir_for(kind: str, paths: dict) -> str:
+    """Pick the library folder for a download's kind. ``paths`` = the config dict
+    ({movies_path, tv_path, youtube_path})."""
+    paths = paths or {}
+    k = str(kind or "").lower()
+    if k == "movie":
+        return paths.get("movies_path") or ""
+    if k in ("show", "tv", "episode", "season", "series"):
+        return paths.get("tv_path") or ""
+    if k == "youtube":
+        return paths.get("youtube_path") or ""
+    return ""
+
+
+__all__ = ["basename_of", "find_completed_file", "dest_path_for", "target_dir_for"]

core/video/slskd_download.py

@@ -0,0 +1,104 @@
+"""Start + track Soulseek (slskd) downloads for the video pipeline.
+
+Thin wrapper over slskd's transfer API (the same shared instance the music side uses):
+  - ``start_download(username, filename, size)`` → POST /transfers/downloads/{username}
+  - ``list_downloads()``                         → GET  /transfers/downloads (flattened)
+The flatten + state-classification helpers are pure (unit-tested); the HTTP is glue.
+
+Isolated: stdlib + requests + shared ``config_manager``; no music imports.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from urllib.parse import quote
+
+import requests
+
+
+def _conn():
+    from config.settings import config_manager
+    base = str(config_manager.get("soulseek.slskd_url", "") or "").rstrip("/")
+    key = config_manager.get("soulseek.api_key", "") or ""
+    return base, ({"X-API-Key": key} if key else {})
+
+
+def start_download(username: str, filename: str, size_bytes: int = 0) -> dict:
+    """Ask slskd to download one file from a user. Returns {ok[, error]}."""
+    base, headers = _conn()
+    if not base:
+        return {"ok": False, "error": "slskd isn't configured"}
+    try:
+        r = requests.post(base + "/api/v0/transfers/downloads/" + quote(str(username or "")),
+                          json=[{"filename": filename, "size": int(size_bytes or 0)}],
+                          headers=headers, timeout=15)
+        r.raise_for_status()
+        return {"ok": True}
+    except Exception as e:   # noqa: BLE001 - surface any slskd/network failure to the caller
+        return {"ok": False, "error": str(e)}
+
+
+def list_downloads() -> list:
+    """Current slskd downloads, flattened to one dict per file."""
+    base, headers = _conn()
+    if not base:
+        return []
+    try:
+        r = requests.get(base + "/api/v0/transfers/downloads", headers=headers, timeout=15)
+        if not r.ok:
+            return []
+        data = r.json()
+    except Exception:   # noqa: BLE001, S110 - transient slskd error → no downloads this tick
+        return []
+    return flatten_downloads(data)
+
+
+def flatten_downloads(data: Any) -> list:
+    """slskd's nested users→directories→files → a flat list of file transfers. Pure."""
+    out = []
+    for user in (data if isinstance(data, list) else []):
+        if not isinstance(user, dict):
+            continue
+        un = user.get("username", "")
+        for d in (user.get("directories") or []):
+            for f in (d.get("files") or []):
+                out.append({
+                    "username": un, "filename": f.get("filename", ""), "id": f.get("id", ""),
+                    "state": f.get("state", ""), "size": f.get("size", 0) or 0,
+                    "transferred": f.get("bytesTransferred", 0) or 0,
+                })
+    return out
+
+
+def classify_state(state: Any) -> str:
+    """slskd state string → 'completed' | 'failed' | 'active'. Pure."""
+    s = str(state or "").lower()
+    if "completed" in s and "succeed" in s:
+        return "completed"
+    if any(x in s for x in ("error", "cancel", "timed", "failed", "reject")):
+        return "failed"
+    if "completed" in s:        # completed but not succeeded → treat as failed
+        return "failed"
+    return "active"
+
+
+def progress_pct(transfer: dict) -> float:
+    """0–100 from bytesTransferred/size (100 once completed). Pure."""
+    transfer = transfer or {}
+    if classify_state(transfer.get("state")) == "completed":
+        return 100.0
+    size = transfer.get("size", 0) or 0
+    tr = transfer.get("transferred", 0) or 0
+    return round(min(99.0, (tr / size * 100.0) if size else 0.0), 1)
+
+
+def find_transfer(transfers: list, username: str, filename: str) -> dict:
+    """The transfer matching a started download (by username + filename). Pure."""
+    for t in (transfers or []):
+        if t.get("username") == username and t.get("filename") == filename:
+            return t
+    return {}
+
+
+__all__ = ["start_download", "list_downloads", "flatten_downloads", "classify_state",
+           "progress_pct", "find_transfer"]

database/video_database.py

@@ -31,7 +31,7 @@ logger = get_logger("video_database")
 
 # Bump when video_schema.sql changes in a way worth recording. Stored in
 # PRAGMA user_version as a backstop indicator (nothing gates on it yet).
-SCHEMA_VERSION = 13
+SCHEMA_VERSION = 14
 
 _DEFAULT_DB_PATH = "database/video_library.db"
 _SCHEMA_FILE = Path(__file__).resolve().parent / "video_schema.sql"
@@ -1126,6 +1126,71 @@ class VideoDatabase:
         finally:
             conn.close()
 
+    # ── video downloads (the grab → transfer pipeline) ────────────────────────
+    _DL_FIELDS = ("kind", "title", "release_title", "source", "username", "filename",
+                  "size_bytes", "quality_label", "target_dir", "status")
+
+    def add_video_download(self, rec: dict) -> int:
+        """Insert a download row (status defaults to 'downloading'); returns its id."""
+        rec = rec or {}
+        cols = [f for f in self._DL_FIELDS if f in rec]
+        conn = self._get_connection()
+        try:
+            cur = conn.execute(
+                "INSERT INTO video_downloads (" + ", ".join(cols) + ") VALUES (" +
+                ", ".join("?" for _ in cols) + ")",
+                tuple(rec[c] for c in cols),
+            )
+            conn.commit()
+            return int(cur.lastrowid)
+        finally:
+            conn.close()
+
+    def list_video_downloads(self, limit: int = 100) -> list:
+        conn = self._get_connection()
+        try:
+            rows = conn.execute(
+                "SELECT * FROM video_downloads ORDER BY "
+                "CASE status WHEN 'downloading' THEN 0 WHEN 'queued' THEN 1 ELSE 2 END, "
+                "id DESC LIMIT ?", (int(limit),)
+            ).fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            conn.close()
+
+    def get_active_video_downloads(self) -> list:
+        conn = self._get_connection()
+        try:
+            rows = conn.execute(
+                "SELECT * FROM video_downloads WHERE status IN ('queued', 'downloading') ORDER BY id"
+            ).fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            conn.close()
+
+    def update_video_download(self, dl_id: int, **fields) -> None:
+        """Patch a download row; ``updated_at`` is always bumped."""
+        if not fields:
+            return
+        keys = list(fields.keys())
+        sets = ", ".join(k + " = ?" for k in keys) + ", updated_at = datetime('now')"
+        conn = self._get_connection()
+        try:
+            conn.execute("UPDATE video_downloads SET " + sets + " WHERE id = ?",
+                         tuple(fields[k] for k in keys) + (int(dl_id),))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def clear_finished_video_downloads(self) -> int:
+        conn = self._get_connection()
+        try:
+            cur = conn.execute("DELETE FROM video_downloads WHERE status IN ('completed', 'failed')")
+            conn.commit()
+            return cur.rowcount
+        finally:
+            conn.close()
+
     # ── library mapping (which server library is Movies / TV) ─────────────────
     def get_library_selection(self, server: str) -> dict:
         return {

database/video_schema.sql

@@ -516,3 +516,27 @@ CREATE VIEW IF NOT EXISTS v_calendar AS
            v.title AS title, c.title AS parent_title
       FROM channel_videos v JOIN channels c ON c.id = v.channel_id
      WHERE v.published_at IS NOT NULL;
+
+-- DOWNLOADS — every grab initiated from the video side lands here (movies/tv/youtube).
+-- The pipeline starts the download, watches it, and on completion moves the file to the
+-- per-type library folder and marks it completed. Status: queued|downloading|completed|failed.
+CREATE TABLE IF NOT EXISTS video_downloads (
+    id            INTEGER PRIMARY KEY AUTOINCREMENT,
+    kind          TEXT NOT NULL,                 -- movie | show | youtube
+    title         TEXT,                          -- human title (e.g. the movie name)
+    release_title TEXT,                          -- the release/file being grabbed
+    source        TEXT,                          -- soulseek | torrent | usenet
+    username      TEXT,                          -- slskd uploader (for the grab + status)
+    filename      TEXT,                          -- slskd remote filename (full path)
+    size_bytes    INTEGER DEFAULT 0,
+    quality_label TEXT,
+    target_dir    TEXT,                          -- destination library folder
+    dest_path     TEXT,                          -- final moved path (set on completion)
+    status        TEXT NOT NULL DEFAULT 'downloading',
+    progress      REAL DEFAULT 0,
+    error         TEXT,
+    created_at    TEXT DEFAULT (datetime('now')),
+    updated_at    TEXT DEFAULT (datetime('now')),
+    completed_at  TEXT
+);
+CREATE INDEX IF NOT EXISTS idx_video_downloads_status ON video_downloads(status);

tests/test_video_download_pipeline.py

@@ -0,0 +1,85 @@
+"""Video download pipeline — the pure seams: slskd state classification + flatten,
+file location + destination resolution, and the video.db downloads CRUD. Isolated."""
+
+from __future__ import annotations
+
+from core.video.download_pipeline import (
+    basename_of,
+    dest_path_for,
+    find_completed_file,
+    target_dir_for,
+)
+from core.video.slskd_download import (
+    classify_state,
+    find_transfer,
+    flatten_downloads,
+    progress_pct,
+)
+
+
+def test_classify_state():
+    assert classify_state("Completed, Succeeded") == "completed"
+    assert classify_state("InProgress") == "active"
+    assert classify_state("Queued, Remotely") == "active"
+    assert classify_state("Completed, Errored") == "failed"
+    assert classify_state("Completed, Cancelled") == "failed"
+    assert classify_state("Completed, TimedOut") == "failed"
+    assert classify_state("") == "active"
+
+
+def test_flatten_and_find_transfer():
+    data = [{"username": "neo", "directories": [
+        {"files": [{"filename": r"@@a\Movie\movie.mkv", "id": "t1", "state": "InProgress",
+                    "size": 100, "bytesTransferred": 25}]}]}]
+    flat = flatten_downloads(data)
+    assert len(flat) == 1 and flat[0]["username"] == "neo" and flat[0]["id"] == "t1"
+    assert progress_pct(flat[0]) == 25.0
+    assert find_transfer(flat, "neo", r"@@a\Movie\movie.mkv")["id"] == "t1"
+    assert find_transfer(flat, "neo", "other") == {}
+    assert flatten_downloads(None) == []
+
+
+def test_progress_is_100_when_completed():
+    assert progress_pct({"state": "Completed, Succeeded", "size": 100, "transferred": 0}) == 100.0
+
+
+def test_basename_handles_both_separators():
+    assert basename_of(r"@@x\The.Wire.S02\ep.mkv") == "ep.mkv"
+    assert basename_of("a/b/c/movie.mp4") == "movie.mp4"
+    assert basename_of("") == ""
+
+
+def test_find_completed_file_by_basename():
+    files = ["/dl/SomeFolder/movie.mkv", "/dl/Other/nope.mkv"]
+    assert find_completed_file("/dl", r"@@u\Remote\movie.mkv", lambda d: files) == "/dl/SomeFolder/movie.mkv"
+    assert find_completed_file("/dl", "missing.mkv", lambda d: files) is None
+
+
+def test_dest_and_target_resolution():
+    assert dest_path_for("/media/movies", "/dl/x/movie.mkv") == "/media/movies/movie.mkv"
+    paths = {"movies_path": "/m", "tv_path": "/t", "youtube_path": "/y"}
+    assert target_dir_for("movie", paths) == "/m"
+    assert target_dir_for("show", paths) == "/t"
+    assert target_dir_for("season", paths) == "/t"
+    assert target_dir_for("youtube", paths) == "/y"
+    assert target_dir_for("weird", paths) == ""
+
+
+# ── DB CRUD ───────────────────────────────────────────────────────────────────
+def test_video_downloads_crud(tmp_path):
+    from database.video_database import VideoDatabase
+    db = VideoDatabase(database_path=str(tmp_path / "video_library.db"))
+    dl_id = db.add_video_download({
+        "kind": "movie", "title": "The Matrix", "release_title": "The.Matrix.1999.1080p",
+        "source": "soulseek", "username": "neo", "filename": r"@@a\x\m.mkv",
+        "size_bytes": 8_000_000_000, "target_dir": "/media/movies", "status": "downloading",
+    })
+    assert dl_id > 0
+    active = db.get_active_video_downloads()
+    assert len(active) == 1 and active[0]["title"] == "The Matrix"
+    db.update_video_download(dl_id, status="completed", progress=100, dest_path="/media/movies/m.mkv")
+    assert db.get_active_video_downloads() == []
+    listed = db.list_video_downloads()
+    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() == []