youtube fulfillment: drain wishlist -> download queue

new 'Download YouTube Wishlist' automation: pushes a polite batch of wished
youtube videos into the shared video_downloads queue + spawns the yt-dlp worker
per video. skips in-flight ones (no double-grab); big backlogs drain over
several scheduled runs (batch_size, default 3). needs the youtube library folder
set.

- download_monitor: SKIP source='youtube' rows (owned by their worker thread, no
  slskd transfer to match) — surgical, slskd path untouched.
- db.youtube_wishlist_to_download(): flat newest-first list of wished videos with
  channel/title/date/thumb for organising.

block + registration + icon/label + drift test. all seam-tested.

core/automation/blocks.py

@@ -272,6 +272,11 @@ ACTIONS: list[dict] = [
      "config_fields": [
          {"key": "backfill_count", "type": "number", "label": "Always keep the last N videos from each channel", "default": 10, "min": 0}
      ]},
+    {"type": "video_process_youtube_wishlist", "label": "Download YouTube Wishlist", "icon": "download", "scope": "video",
+     "description": "Grab wished YouTube videos: pushes a batch into the download queue (yt-dlp), organised as a Plex 'TV by date' show (channel/year/date). A completed download leaves the wishlist. Needs the YouTube library folder set on Settings → Downloads. Pair with a schedule to drain continuously.", "available": True,
+     "config_fields": [
+         {"key": "batch_size", "type": "number", "label": "Max downloads to queue per run", "default": 3, "min": 0}
+     ]},
     # Video twins of the music maintenance actions. Distinct action_type (the
     # system seeder keys on action_type, so a shared key would collide with the
     # music row) but the SAME shared handler — the cleanup operates on the common

core/automation/handlers/registration.py

@@ -39,6 +39,7 @@ from core.automation.handlers.search_and_download import auto_search_and_downloa
 from core.automation.handlers.video_auto_wishlist_airing import auto_video_add_airing_episodes
 from core.automation.handlers.video_scan_watchlist_people import auto_video_scan_watchlist_people
 from core.automation.handlers.video_scan_watchlist_channels import auto_video_scan_watchlist_channels
+from core.automation.handlers.video_process_youtube_wishlist import auto_video_process_youtube_wishlist
 from core.automation.handlers.video_scan_library import (
     auto_video_scan_library, auto_video_scan_server, auto_video_update_database,
 )
@@ -241,6 +242,11 @@ def register_all(deps: AutomationDeps) -> None:
         'video_scan_watchlist_channels',
         lambda config: auto_video_scan_watchlist_channels(config, deps),
     )
+    # Drain side: enqueue wished YouTube videos into the download queue (yt-dlp worker).
+    engine.register_action_handler(
+        'video_process_youtube_wishlist',
+        lambda config: auto_video_process_youtube_wishlist(config, deps),
+    )
 
     # Progress + history callbacks: the engine invokes these around
     # each handler run. Lift the closures from

core/automation/handlers/video_process_youtube_wishlist.py

@@ -0,0 +1,145 @@
+"""Automation handler: ``video_process_youtube_wishlist`` action.
+
+The drain side of the YouTube fulfillment lane. The watchlist-channels scan keeps the
+wishlist fed with new uploads; THIS takes wished YouTube videos and pushes them into the
+shared ``video_downloads`` queue, spawning the yt-dlp worker per video. The worker
+(``core.video.youtube_download``) downloads → organises (channel/year/date) → archives to
+history → removes the video from the wishlist, so a completed grab leaves the wishlist and
+won't be re-queued.
+
+Polite by default: only a small BATCH is enqueued per run (a big first-time backlog drains
+over several scheduled runs rather than spawning hundreds of yt-dlp processes at once), and
+videos already in flight (an active ``source='youtube'`` download) are skipped so re-runs
+never double-grab.
+
+Shared automation side (may import ``core.video`` / ``api.video``); owns its own progress.
+All I/O is injected as seams, so the selection + batching is a pure unit-testable function;
+production lazily binds the real DB + the worker spawn.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Dict, Iterable, List, Optional
+
+from core.automation.deps import AutomationDeps
+
+
+def select_to_enqueue(wanted: List[Dict[str, Any]], active_ids: Iterable, batch_size: int) -> List[Dict[str, Any]]:
+    """Which wished videos to enqueue now: skip ones already in flight, cap at the batch
+    size (0 = no cap). Pure."""
+    active = {str(x) for x in (active_ids or ()) if x}
+    out: List[Dict[str, Any]] = []
+    for v in wanted or []:
+        vid = v.get("video_id")
+        if not vid or str(vid) in active:
+            continue
+        out.append(v)
+        if batch_size and len(out) >= batch_size:
+            break
+    return out
+
+
+# ── production seams ──────────────────────────────────────────────────────────
+def _default_youtube_root() -> str:
+    from api.video import get_video_db
+    return get_video_db().get_setting("youtube_path") or ""
+
+
+def _default_fetch_wanted() -> List[Dict[str, Any]]:
+    from api.video import get_video_db
+    return get_video_db().youtube_wishlist_to_download()
+
+
+def _default_active_ids() -> List[Any]:
+    from api.video import get_video_db
+    return [d.get("media_id") for d in get_video_db().get_active_video_downloads()
+            if d.get("source") == "youtube" and d.get("media_id")]
+
+
+def _default_enqueue(video: Dict[str, Any], root: str) -> Any:
+    """Create the download row + spawn the yt-dlp worker thread. Returns the row id."""
+    import json
+    import threading
+    from api.video import get_video_db
+    from core.video.sources import resolve_video_server
+    from core.video.youtube_download import run_youtube_download
+    db = get_video_db()
+    dl_id = db.add_video_download({
+        "kind": "youtube", "source": "youtube", "media_source": "youtube",
+        "title": video.get("video_title") or video.get("channel_title"),
+        "media_id": video.get("video_id"), "target_dir": root, "status": "downloading",
+        "year": video.get("published_at"), "poster_url": video.get("thumbnail_url"),
+        "search_ctx": json.dumps({"channel": video.get("channel_title"),
+                                  "video_title": video.get("video_title"),
+                                  "published_at": video.get("published_at"),
+                                  "server_source": resolve_video_server()}),
+    })
+    threading.Thread(target=run_youtube_download, args=(dl_id, get_video_db),
+                     daemon=True, name="yt-dl-%s" % dl_id).start()
+    return dl_id
+
+
+def auto_video_process_youtube_wishlist(
+    config: Dict[str, Any],
+    deps: AutomationDeps,
+    *,
+    youtube_root: Optional[Callable[[], str]] = None,
+    fetch_wanted: Optional[Callable[[], List[Dict[str, Any]]]] = None,
+    active_ids: Optional[Callable[[], Iterable]] = None,
+    enqueue: Optional[Callable[[Dict[str, Any], str], Any]] = None,
+) -> Dict[str, Any]:
+    """Enqueue a batch of wished YouTube videos for download.
+
+    Returns ``{'status': 'completed', 'queued': int, 'remaining': int, ...}``."""
+    youtube_root = youtube_root or _default_youtube_root
+    fetch_wanted = fetch_wanted or _default_fetch_wanted
+    active_ids = active_ids or _default_active_ids
+    enqueue = enqueue or _default_enqueue
+    automation_id = config.get('_automation_id')
+    batch_size = max(0, int(config.get('batch_size', 3) or 0))
+
+    try:
+        root = youtube_root()
+        if not root:
+            msg = 'Set the YouTube library folder on Settings → Downloads first'
+            deps.update_progress(automation_id, status='error', phase='Error',
+                                 log_line=msg, log_type='error')
+            return {'status': 'error', 'error': msg, '_manages_own_progress': True}
+
+        deps.update_progress(automation_id, phase='Checking the YouTube wishlist…', progress=10,
+                             log_line='Looking for new videos to download', log_type='info')
+        wanted = fetch_wanted() or []
+        active = list(active_ids() or [])
+        picks = select_to_enqueue(wanted, active, batch_size)
+        if not picks:
+            note = ('All %d wished video(s) are already downloading' % len(active)) if active \
+                else 'No wished YouTube videos to download'
+            deps.update_progress(automation_id, status='finished', progress=100, phase='Complete',
+                                 log_line=note, log_type='info')
+            return {'status': 'completed', 'queued': 0, 'remaining': max(0, len(wanted) - len(active)),
+                    '_manages_own_progress': True}
+
+        queued = 0
+        for i, v in enumerate(picks):
+            deps.update_progress(automation_id, phase='Queueing downloads…',
+                                 progress=20 + int(70 * i / max(len(picks), 1)),
+                                 log_line="Queued '%s'" % (v.get('video_title') or v.get('video_id')),
+                                 log_type='info')
+            try:
+                if enqueue(v, root) is not None:
+                    queued += 1
+            except Exception:   # noqa: BLE001 - one bad enqueue shouldn't stop the batch
+                deps.update_progress(automation_id, log_type='warning',
+                                     log_line="Couldn't queue '%s'" % (v.get('video_title') or v.get('video_id')))
+
+        remaining = max(0, len(wanted) - len(active) - queued)
+        done = 'Queued %d YouTube download(s)' % queued
+        if remaining:
+            done += ' · %d more waiting (drains over the next runs)' % remaining
+        deps.update_progress(automation_id, status='finished', progress=100, phase='Complete',
+                             log_line=done, log_type='success')
+        return {'status': 'completed', 'queued': queued, 'remaining': remaining,
+                '_manages_own_progress': True}
+    except Exception as e:  # noqa: BLE001
+        deps.update_progress(automation_id, status='error', phase='Error', log_line=str(e), log_type='error')
+        return {'status': 'error', 'error': str(e), '_manages_own_progress': True}

core/video/download_monitor.py

@@ -352,8 +352,10 @@ def _spawn_requery(dl_id) -> None:
 
 
 def _tick(db) -> None:
-    # 'searching' rows are owned by their requery thread — skip them here.
-    active = [d for d in db.get_active_video_downloads() if d.get("status") != "searching"]
+    # 'searching' rows are owned by their requery thread; 'youtube' rows are owned by
+    # their yt-dlp worker thread (no slskd transfer to match) — skip both here.
+    active = [d for d in db.get_active_video_downloads()
+              if d.get("status") != "searching" and d.get("source") != "youtube"]
     if not active:
         _misses.clear()
         return

database/video_database.py

@@ -3026,6 +3026,24 @@ class VideoDatabase:
         finally:
             conn.close()
 
+    def youtube_wishlist_to_download(self, limit: int = 0) -> list:
+        """Flat list of wished YouTube videos for the fulfillment worker to grab, newest
+        upload first. Each carries what organising + the download row need: the video id,
+        its channel, the video title, thumbnail, and upload date. (A completed download
+        removes its wishlist row, so this list naturally shrinks as the queue drains.)"""
+        conn = self._get_connection()
+        try:
+            sql = ("SELECT source_id AS video_id, parent_source_id AS channel_id, "
+                   "title AS channel_title, episode_title AS video_title, "
+                   "still_url AS thumbnail_url, air_date AS published_at "
+                   "FROM video_wishlist WHERE kind='video' AND source='youtube' "
+                   "AND source_id IS NOT NULL ORDER BY air_date DESC, id DESC")
+            if limit and int(limit) > 0:
+                sql += " LIMIT %d" % int(limit)
+            return [dict(r) for r in conn.execute(sql)]
+        finally:
+            conn.close()
+
     def mark_channel_dates_enriched(self, channel_id, date_count=0, method="innertube") -> None:
         """Record that the background enricher swept this channel (skip re-sweeps).
         ``method`` tags which source produced the dates; legacy rows have NULL and

tests/automation/test_handler_registration.py

@@ -62,6 +62,7 @@ EXPECTED_ACTION_NAMES = frozenset({
     'video_add_airing_episodes',
     'video_scan_watchlist_people',
     'video_scan_watchlist_channels',
+    'video_process_youtube_wishlist',
     'video_clean_search_history',
     'video_clean_completed_downloads',
     'video_full_cleanup',

tests/test_video_process_youtube_wishlist.py

@@ -0,0 +1,135 @@
+"""Drain side of the YouTube fulfillment lane: enqueue wished videos into the download
+queue in polite batches, skipping in-flight ones. Pure selection + handler with all I/O
+injected, plus the DB query that feeds it.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from core.automation.handlers.video_process_youtube_wishlist import (
+    auto_video_process_youtube_wishlist,
+    select_to_enqueue,
+)
+
+
+class _Deps:
+    def __init__(self):
+        self.progress = []
+
+    def update_progress(self, automation_id, **kw):
+        self.progress.append(kw)
+
+
+def _v(vid, title="T", date="2024-01-01"):
+    return {"video_id": vid, "channel_title": "Chan", "video_title": title,
+            "thumbnail_url": "/t.jpg", "published_at": date}
+
+
+# ── pure batching ─────────────────────────────────────────────────────────────
+def test_select_skips_inflight_and_caps_batch():
+    wanted = [_v("a"), _v("b"), _v("c"), _v("d")]
+    picks = select_to_enqueue(wanted, active_ids=["b"], batch_size=2)
+    assert [p["video_id"] for p in picks] == ["a", "c"]   # b skipped (in flight), capped at 2
+
+
+def test_select_zero_batch_means_no_cap():
+    wanted = [_v("a"), _v("b"), _v("c")]
+    assert len(select_to_enqueue(wanted, [], 0)) == 3
+
+
+def test_select_drops_idless():
+    assert select_to_enqueue([{"video_title": "no id"}], [], 5) == []
+
+
+# ── handler ───────────────────────────────────────────────────────────────────
+def _run(wanted, *, active=None, root="/yt", batch=3):
+    enq = []
+
+    def enqueue(video, r):
+        enq.append((video["video_id"], r))
+        return len(enq)
+
+    deps = _Deps()
+    res = auto_video_process_youtube_wishlist(
+        {"_automation_id": "a", "batch_size": batch}, deps,
+        youtube_root=lambda: root, fetch_wanted=lambda: wanted,
+        active_ids=lambda: list(active or []), enqueue=enqueue)
+    return res, enq, deps
+
+
+def test_enqueues_a_batch_and_reports_remaining():
+    wanted = [_v("a"), _v("b"), _v("c"), _v("d"), _v("e")]
+    res, enq, _ = _run(wanted, batch=2)
+    assert res["status"] == "completed" and res["queued"] == 2
+    assert [vid for vid, _ in enq] == ["a", "b"]
+    assert res["remaining"] == 3                          # 5 wanted - 2 queued
+    assert enq[0][1] == "/yt"                             # root passed through
+
+
+def test_inflight_videos_are_not_requeued():
+    wanted = [_v("a"), _v("b")]
+    res, enq, _ = _run(wanted, active=["a", "b"], batch=5)
+    assert enq == [] and res["queued"] == 0
+
+
+def test_missing_youtube_folder_is_an_error():
+    res, enq, deps = _run([_v("a")], root="")
+    assert res["status"] == "error" and "library folder" in res["error"]
+    assert enq == [] and any(p.get("status") == "error" for p in deps.progress)
+
+
+def test_nothing_wanted_is_a_clean_noop():
+    res, enq, _ = _run([])
+    assert res["status"] == "completed" and res["queued"] == 0 and enq == []
+
+
+def test_one_bad_enqueue_does_not_stop_the_batch():
+    def enqueue(video, r):
+        if video["video_id"] == "a":
+            raise RuntimeError("queue full")
+        return 1
+
+    res = auto_video_process_youtube_wishlist(
+        {"_automation_id": "x", "batch_size": 5}, _Deps(),
+        youtube_root=lambda: "/yt", fetch_wanted=lambda: [_v("a"), _v("b")],
+        active_ids=lambda: [], enqueue=enqueue)
+    assert res["status"] == "completed" and res["queued"] == 1   # b still queued
+
+
+def test_top_level_error_is_caught():
+    def boom():
+        raise RuntimeError("db down")
+    res = auto_video_process_youtube_wishlist({"_automation_id": "x"}, _Deps(),
+                                              youtube_root=lambda: "/yt", fetch_wanted=boom)
+    assert res["status"] == "error" and "db down" in res["error"]
+
+
+# ── the DB query that feeds it ────────────────────────────────────────────────
+from database.video_database import VideoDatabase  # noqa: E402
+
+
+@pytest.fixture()
+def db(tmp_path):
+    return VideoDatabase(database_path=str(tmp_path / "video_library.db"))
+
+
+def test_youtube_wishlist_to_download_shape(db):
+    db.add_videos_to_wishlist(
+        {"youtube_id": "UC1", "title": "Cool Channel", "avatar_url": "/a.jpg"},
+        [{"youtube_id": "v1", "title": "First", "published_at": "2024-03-01", "thumbnail_url": "/1.jpg"},
+         {"youtube_id": "v2", "title": "Second", "published_at": "2024-05-01", "thumbnail_url": "/2.jpg"}])
+    rows = db.youtube_wishlist_to_download()
+    assert [r["video_id"] for r in rows] == ["v2", "v1"]      # newest upload first
+    top = rows[0]
+    assert top["channel_id"] == "UC1" and top["channel_title"] == "Cool Channel"
+    assert top["video_title"] == "Second" and top["published_at"] == "2024-05-01"
+    assert top["thumbnail_url"] == "/2.jpg"
+
+
+def test_youtube_wishlist_to_download_excludes_movies(db):
+    db.add_movie_to_wishlist(99, "A Movie")
+    db.add_videos_to_wishlist({"youtube_id": "UC1", "title": "Ch"},
+                              [{"youtube_id": "v1", "title": "Vid", "published_at": "2024-01-01"}])
+    rows = db.youtube_wishlist_to_download()
+    assert [r["video_id"] for r in rows] == ["v1"]           # the movie isn't a youtube video

webui/static/stats-automations.js

@@ -1777,6 +1777,7 @@ const _autoIcons = {
     video_scan_library: '\uD83C\uDFAC', video_scan_server: '\uD83D\uDD04', video_update_database: '\uD83D\uDDC4\uFE0F',
     video_add_airing_episodes: '\uD83D\uDCFA', video_deep_scan_movies: '\uD83C\uDFAC', video_deep_scan_tv: '\uD83D\uDCFA',
     video_scan_watchlist_people: '\uD83C\uDFAD', video_scan_watchlist_channels: '\uD83D\uDCE1',
+    video_process_youtube_wishlist: '\u2B07\uFE0F',
 };
 
 // --- Inspiration Templates ---
@@ -3373,6 +3374,7 @@ function _autoFormatAction(type) {
         video_deep_scan_movies: 'Deep Scan Movie Library', video_deep_scan_tv: 'Deep Scan TV Library',
         video_scan_watchlist_people: 'Scan Watchlist People',
         video_scan_watchlist_channels: 'Scan Watchlist Channels',
+        video_process_youtube_wishlist: 'Download YouTube Wishlist',
     };
     return labels[type] || type || 'Unknown';
 }