From ceccb4ee650458c25497207c51bd3306bf2eb560 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Fri, 19 Jun 2026 14:53:18 -0700 Subject: [PATCH] video downloads: monitor thread + grab/active/clear endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 54 +++++++++++++ core/video/download_monitor.py | 109 ++++++++++++++++++++++++++ tests/test_video_api.py | 38 +++++++++ tests/test_video_download_pipeline.py | 61 ++++++++++++++ 4 files changed, 262 insertions(+) create mode 100644 core/video/download_monitor.py diff --git a/api/video/downloads.py b/api/video/downloads.py index 392d3491..359ccefb 100644 --- a/api/video/downloads.py +++ b/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. diff --git a/core/video/download_monitor.py b/core/video/download_monitor.py new file mode 100644 index 00000000..801393ee --- /dev/null +++ b/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"] diff --git a/tests/test_video_api.py b/tests/test_video_api.py index 4a850148..f3a53f28 100644 --- a/tests/test_video_api.py +++ b/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 diff --git a/tests/test_video_download_pipeline.py b/tests/test_video_download_pipeline.py index f70ff860..f0e6519c 100644 --- a/tests/test_video_download_pipeline.py +++ b/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"]