diff --git a/core/automation/blocks.py b/core/automation/blocks.py index 306bca2f..dbbfb2f4 100644 --- a/core/automation/blocks.py +++ b/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 diff --git a/core/automation/handlers/registration.py b/core/automation/handlers/registration.py index 1cd30406..a48bbdbe 100644 --- a/core/automation/handlers/registration.py +++ b/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 diff --git a/core/automation/handlers/video_process_youtube_wishlist.py b/core/automation/handlers/video_process_youtube_wishlist.py new file mode 100644 index 00000000..9ab79730 --- /dev/null +++ b/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} diff --git a/core/video/download_monitor.py b/core/video/download_monitor.py index 46038570..0b83c6cd 100644 --- a/core/video/download_monitor.py +++ b/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 diff --git a/database/video_database.py b/database/video_database.py index 725bdf2a..0d0382aa 100644 --- a/database/video_database.py +++ b/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 diff --git a/tests/automation/test_handler_registration.py b/tests/automation/test_handler_registration.py index 5a4873be..b926e72c 100644 --- a/tests/automation/test_handler_registration.py +++ b/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', diff --git a/tests/test_video_process_youtube_wishlist.py b/tests/test_video_process_youtube_wishlist.py new file mode 100644 index 00000000..edecdc12 --- /dev/null +++ b/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 diff --git a/webui/static/stats-automations.js b/webui/static/stats-automations.js index 8d5733ef..a38d74b7 100644 --- a/webui/static/stats-automations.js +++ b/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'; }