diff --git a/core/video/download_monitor.py b/core/video/download_monitor.py index d092bbb4..63331df8 100644 --- a/core/video/download_monitor.py +++ b/core/video/download_monitor.py @@ -102,6 +102,13 @@ def _make_organizer(db): prober = probe if settings.get("verify_with_ffprobe", True) else None def organize(dl, src): + # The file's down — flip to 'importing' so the UI shows the post-processing phase + # (move into the library + nfo/artwork sidecars + subtitles) instead of sitting on + # 'downloading' while this runs. Best-effort; the patch below is the real transition. + try: + db.update_video_download(dl["id"], status="importing", progress=100) + except Exception: # noqa: BLE001, S110 - a status blip must never wedge the import + pass patch = run_import(dl, src, fs=fs, prober=prober, settings=settings) if patch.get("status") == "completed" and patch.get("dest_path"): if settings.get("save_artwork") or settings.get("write_nfo"): diff --git a/core/video/youtube_download.py b/core/video/youtube_download.py index 296a591a..03d2ec8d 100644 --- a/core/video/youtube_download.py +++ b/core/video/youtube_download.py @@ -25,6 +25,7 @@ from __future__ import annotations import json import logging import os +import shutil import threading import time from typing import Any, Callable, Dict, Optional @@ -137,6 +138,12 @@ def download_one(video_id: Any, dest_dir: str, dest_stem: str, profile: Any, con return {"ok": True, "dest_path": dest_path, "error": None} +def _default_move(src: str, dest: str) -> None: + """Move a finished staged file into the library, creating the target folders.""" + os.makedirs(os.path.dirname(dest) or ".", exist_ok=True) + shutil.move(src, dest) + + def process_youtube_download( dl: Dict[str, Any], *, @@ -146,12 +153,19 @@ def process_youtube_download( update_row: Callable[..., Any], archive: Callable[[Dict[str, Any], Dict[str, Any]], Any], clear_wishlist: Callable[[Any], Any], + stage_dir: Optional[str] = None, + move: Callable[[str, str], Any] = _default_move, progress_hook: Optional[Callable] = None, cookie_opts: Optional[dict] = None, now: Optional[Callable[[], str]] = None, ) -> Dict[str, Any]: """Fulfil one queued YouTube download. PURE — all I/O injected. + Pipeline (same shape as the movie/TV lane): download into ``stage_dir`` (the shared + download folder) → flip to 'importing' → MOVE into the organised library path → completed. + When ``stage_dir`` is None it downloads straight into the library (legacy fallback, e.g. + no download folder configured), skipping the move. + On success: row → completed (with dest_path), snapshot to history, and remove the video from the wishlist (it's in the library now; history is the permanent record). On failure: row → failed + a history snapshot; the wishlist row is LEFT so a later scan/run @@ -159,37 +173,53 @@ def process_youtube_download( now = now or (lambda: "") settings = settings if isinstance(settings, dict) else {} container = format_selection(profile)["merge_output_format"] - dest = plan_destination(dl, settings, container) + dest = plan_destination(dl, settings, container) # the FINAL organised library path stem, cont = _stem_and_container(dest, container) + # Download target: the staging folder when set, else straight to the library. Either way + # do NOT write the organised DIR back to target_dir — it's the youtube ROOT, and + # plan_destination re-derives the channel/season folders under it; clobbering it re-nests + # on a re-run (the orphan reaper re-queues an interrupted download). + dl_dir = stage_dir if stage_dir else dest.get("dir") + update_row(dl.get("id"), status="downloading", progress=0, filename=dest.get("filename")) - # Record the organised filename for the Downloads page. CRITICAL: do NOT write the - # organised DIR back to target_dir — target_dir is the youtube ROOT, and plan_destination - # re-derives the channel/season folders under it. Clobbering it would re-nest on any - # re-run (e.g. the orphan reaper re-queues an interrupted download) → - # Channel/Season/Channel/Season. Leave target_dir = root so re-processing is idempotent. - update_row(dl.get("id"), status="downloading", progress=0, - filename=dest.get("filename")) - - res = download(dl.get("media_id"), dest.get("dir"), stem, profile, cont, + res = download(dl.get("media_id"), dl_dir, stem, profile, cont, progress_hook=progress_hook, cookie_opts=cookie_opts) - if res.get("ok"): - dest_path = res.get("dest_path") or dest.get("path") + if not res.get("ok"): + err = res.get("error") or "Download failed" completed = now() - update_row(dl.get("id"), status="completed", progress=100, - dest_path=dest_path, completed_at=completed) - archive(dl, {"status": "completed", "dest_path": dest_path, "completed_at": completed}) - try: - clear_wishlist(dl.get("media_id")) - except Exception: # noqa: BLE001 - unwish is best-effort; the file is already in place - logger.exception("youtube download %s: unwish failed", dl.get("id")) - return {"status": "completed", "dest_path": dest_path} + update_row(dl.get("id"), status="failed", error=err, completed_at=completed) + archive(dl, {"status": "failed", "error": err, "completed_at": completed}) + return {"status": "failed", "error": err} + + staged_path = res.get("dest_path") or os.path.join(dl_dir or "", stem + "." + cont) + final_path = dest.get("path") or staged_path + + # Staged build → post-process into the library (the visible 'importing' phase). + if stage_dir and staged_path and staged_path != final_path: + update_row(dl.get("id"), status="importing", progress=100, filename=dest.get("filename")) + try: + move(staged_path, final_path) + except Exception as e: # noqa: BLE001 - downloaded fine but couldn't be placed + err = "Import failed: " + str(e) + completed = now() + update_row(dl.get("id"), status="import_failed", error=err, completed_at=completed) + archive(dl, {"status": "import_failed", "error": err, "completed_at": completed}) + logger.exception("youtube download %s: import move failed", dl.get("id")) + return {"status": "import_failed", "error": err} + dest_path = final_path + else: + dest_path = staged_path or final_path - err = res.get("error") or "Download failed" completed = now() - update_row(dl.get("id"), status="failed", error=err, completed_at=completed) - archive(dl, {"status": "failed", "error": err, "completed_at": completed}) - return {"status": "failed", "error": err} + update_row(dl.get("id"), status="completed", progress=100, + dest_path=dest_path, completed_at=completed) + archive(dl, {"status": "completed", "dest_path": dest_path, "completed_at": completed}) + try: + clear_wishlist(dl.get("media_id")) + except Exception: # noqa: BLE001 - unwish is best-effort; the file is already in place + logger.exception("youtube download %s: unwish failed", dl.get("id")) + return {"status": "completed", "dest_path": dest_path} # ── concurrency + pacing (the music side's lesson: cap concurrency AND space starts) ── @@ -295,10 +325,18 @@ def run_youtube_download(dl_id: Any, db_provider: Callable) -> None: try: _pace(_DELAY_SECONDS) # space fetch starts to avoid yt-dlp 429s + # Stage into the shared download folder (a 'youtube' subfolder), then transfer to the + # library — same pipeline as movies/TV. Falls back to straight-to-library if no + # download folder is configured. + from config.settings import config_manager + dl_root = str(config_manager.get("soulseek.download_path", "") or "").strip() + stage_dir = os.path.join(dl_root, "youtube") if dl_root else None + process_youtube_download( dl, profile=profile, settings=settings, update_row=db.update_video_download, archive=_archive, clear_wishlist=lambda vid: db.remove_youtube_from_wishlist("video", vid), + stage_dir=stage_dir, progress_hook=_progress, cookie_opts=cookie_opts, now=_now) finally: _active_worker_ids.discard(dl_id) # worker done — no longer protects this row diff --git a/database/video_database.py b/database/video_database.py index 6655b0b3..23d55726 100644 --- a/database/video_database.py +++ b/database/video_database.py @@ -1422,19 +1422,21 @@ class VideoDatabase: conn = self._get_connection() try: rows = conn.execute( - "SELECT * FROM video_downloads WHERE status IN ('queued', 'downloading', 'searching') ORDER BY id" + "SELECT * FROM video_downloads WHERE status IN " + "('queued', 'downloading', 'importing', 'searching') ORDER BY id" ).fetchall() return [dict(r) for r in rows] finally: conn.close() def count_active_youtube_downloads(self) -> int: - """How many YouTube downloads are ACTIVELY fetching right now (not queued) — - the concurrency gauge the wishlist pump checks before starting more.""" + """How many YouTube downloads are ACTIVELY in flight right now (fetching OR importing, + not queued) — the concurrency gauge the wishlist pump checks before starting more. + Counting 'importing' too keeps the worker's slot held while it moves into the library.""" conn = self._get_connection() try: r = conn.execute("SELECT COUNT(*) AS n FROM video_downloads " - "WHERE source='youtube' AND status='downloading'").fetchone() + "WHERE source='youtube' AND status IN ('downloading', 'importing')").fetchone() return int(r["n"]) if r else 0 finally: conn.close() diff --git a/tests/test_youtube_download.py b/tests/test_youtube_download.py index 0089e436..c64754f8 100644 --- a/tests/test_youtube_download.py +++ b/tests/test_youtube_download.py @@ -137,6 +137,43 @@ def test_process_failure_archives_but_keeps_the_wish(): assert calls["unwish"] == [] # wish kept so it can retry later +def test_process_stages_in_download_folder_then_imports_to_library(): + # the consistent pipeline: download → staging folder → 'importing' → move → library + calls, update_row, archive, clear = _recorder() + moves = [] + staged = "/downloads/youtube/Chan - 2024-03-15 - T.mp4" + res = ytd.process_youtube_download( + _dl(), profile=default_profile(), settings={}, + download=lambda vid, d, *a, **k: ({"ok": True, "dest_path": staged} + if d == "/downloads/youtube" else {"ok": False, "error": "wrong dir"}), + update_row=update_row, archive=archive, clear_wishlist=clear, + stage_dir="/downloads/youtube", move=lambda s, d: moves.append((s, d)), now=lambda: "t") + assert res["status"] == "completed" + statuses = [kw.get("status") for _, kw in calls["rows"]] + assert statuses == ["downloading", "importing", "completed"] # the visible phases + final = "/yt/Chan/Season 2024/Chan - 2024-03-15 - T.mp4" + assert moves == [(staged, final)] # staged → organised library + assert res["dest_path"] == final and calls["unwish"] == ["vid1"] + + +def test_process_import_failure_is_terminal_and_keeps_the_wish(): + calls, update_row, archive, clear = _recorder() + + def boom(_s, _d): + raise OSError("disk full") + + res = ytd.process_youtube_download( + _dl(), profile=default_profile(), settings={}, + download=lambda *a, **k: {"ok": True, "dest_path": "/downloads/youtube/x.mp4"}, + update_row=update_row, archive=archive, clear_wishlist=clear, + stage_dir="/downloads/youtube", move=boom, now=lambda: "t") + assert res["status"] == "import_failed" + statuses = [kw.get("status") for _, kw in calls["rows"]] + assert statuses == ["downloading", "importing", "import_failed"] + assert calls["archive"][-1]["status"] == "import_failed" + assert calls["unwish"] == [] # not unwished → can retry + + def test_requeue_orphaned_youtube_recovers_only_dead_downloads(): """After a restart no worker threads survive, so any 'downloading' YouTube row is an orphan → back to 'queued'. A row whose worker is still alive (in _active_worker_ids) and diff --git a/webui/static/video/video-downloads-page.js b/webui/static/video/video-downloads-page.js index 7005990c..854144f9 100644 --- a/webui/static/video/video-downloads-page.js +++ b/webui/static/video/video-downloads-page.js @@ -36,12 +36,14 @@ downloading: { label: 'Downloading', cls: 'active' }, queued: { label: 'Queued', cls: 'queued' }, searching: { label: 'Searching', cls: 'active' }, // retrying — finding another release + importing: { label: 'Importing', cls: 'active' }, // post-processing → moving into library completed: { label: 'Completed', cls: 'completed' }, failed: { label: 'Failed', cls: 'failed' }, + import_failed: { label: 'Import failed', cls: 'failed' }, cancelled: { label: 'Cancelled', cls: 'cancelled' } }; - function isActive(s) { return s === 'downloading' || s === 'queued' || s === 'searching'; } - function isFail(s) { return s === 'failed' || s === 'cancelled'; } + function isActive(s) { return s === 'downloading' || s === 'queued' || s === 'searching' || s === 'importing'; } + function isFail(s) { return s === 'failed' || s === 'cancelled' || s === 'import_failed'; } function matches(s) { return _filter === 'all' || (_filter === 'active' && isActive(s)) || (_filter === 'completed' && s === 'completed') || (_filter === 'failed' && isFail(s));