video downloads: consistent stage→import→library pipeline + 'Importing' status

both lanes now expose the post-processing phase instead of jumping downloading→completed:
- YouTube now STAGES in the shared download folder (a 'youtube' subdir) then transfers to
  the library — same as movies/TV — so no partial .part files land in the Plex folder.
  process_youtube_download gains stage_dir + move seams: download → 'importing' → move →
  completed (import_failed if the move dies; wish kept for retry). falls back to
  straight-to-library when no download folder is set.
- movies/TV: the organizer flips the row to 'importing' before the (slow) move + sidecars +
  subtitles, so that phase is visible.
- 'importing' is now an active status (get_active_video_downloads + youtube concurrency count
  + the downloads page status map), so the row stays on the page + holds its worker slot.
tested (staging + import-failure paths).

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"):

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

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()

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

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));