From 9f687c061ab1f8d3902ddb5d269769c0b927bb54 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Fri, 19 Jun 2026 14:50:27 -0700 Subject: [PATCH] video downloads pipeline: data + pure logic foundation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 54 +++++++++++++ core/video/slskd_download.py | 104 ++++++++++++++++++++++++++ database/video_database.py | 67 ++++++++++++++++- database/video_schema.sql | 24 ++++++ tests/test_video_download_pipeline.py | 85 +++++++++++++++++++++ 5 files changed, 333 insertions(+), 1 deletion(-) create mode 100644 core/video/download_pipeline.py create mode 100644 core/video/slskd_download.py create mode 100644 tests/test_video_download_pipeline.py diff --git a/core/video/download_pipeline.py b/core/video/download_pipeline.py new file mode 100644 index 00000000..ba587bc7 --- /dev/null +++ b/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"] diff --git a/core/video/slskd_download.py b/core/video/slskd_download.py new file mode 100644 index 00000000..0f29e83c --- /dev/null +++ b/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"] diff --git a/database/video_database.py b/database/video_database.py index 5ce0c0db..44055ef1 100644 --- a/database/video_database.py +++ b/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 { diff --git a/database/video_schema.sql b/database/video_schema.sql index 295b6325..67751393 100644 --- a/database/video_schema.sql +++ b/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); diff --git a/tests/test_video_download_pipeline.py b/tests/test_video_download_pipeline.py new file mode 100644 index 00000000..f70ff860 --- /dev/null +++ b/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() == []