danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

video automations: auto-download movie + episode wishlist (soulseek)

Browse source 
the soulseek counterpart of the youtube drain — finally makes the people/airing
scans pay off. two automations (Auto-Download Movie / Episode Wishlist, hourly):
for each wished+released item, do a bounded blocking slskd search, pick the top
ACCEPTED release per the quality profile, and enqueue it exactly like a manual
grab (same add_video_download shape → the monitor finishes + organises it).

same standard as youtube: processes the WHOLE eligible wishlist (no total cap)
but searches a few at a time (max_concurrent, default 3, via a thread pool). a
busy guard skips the next hourly tick while a drain is still working. movies gate
on status='wanted' (skips monitored); episodes are all-wished. items already
downloading are skipped. quiet skip if the library folder isn't set.

reuses the real infra (build_query/slskd_search/_evaluate_hits/start_download/
download_monitor) — pure pick/select/record seam-tested; db queries added. 377
automation+video tests green.
This commit is contained in:
BoulderBadgeDad 2026-06-25 22:08:02 -07:00
parent fb9459fb5c
commit b404420c76
8 changed files with 505 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
core/automation/blocks.py
View file

@ -274,6 +274,16 @@ ACTIONS: list[dict] = [
]},
{"type": "video_scan_watchlist_playlists", "label": "Scan Watchlist Playlists", "icon": "list", "scope": "video",
"description": "For every YouTube playlist you follow, wishlist its videos you don't have yet (and anything later added to it) — mirrors the whole list. Each playlist becomes its own show in the library (playlist-as-show). Shorts excluded. Pair with a schedule.", "available": True},
{"type": "video_download_movie_wishlist", "label": "Download Movie Wishlist", "icon": "download", "scope": "video",
"description": "Auto-grab your wished, released MOVIES from Soulseek: searches each, picks the best release per your quality profile, and downloads it (the monitor finishes + organises it). Skips unreleased ('monitored') ones and any already downloading. Needs the Movie library folder + slskd set. Pair with an hourly schedule.", "available": True,
"config_fields": [
{"key": "max_concurrent", "type": "number", "label": "Max simultaneous searches", "default": 3, "min": 1}
]},
{"type": "video_download_episode_wishlist", "label": "Download Episode Wishlist", "icon": "download", "scope": "video",
"description": "Auto-grab your wished EPISODES from Soulseek: searches each, picks the best release per your quality profile, and downloads it. Fed by the 'Wishlist Today's Airings' scan. Needs the TV library folder + slskd set. Pair with an hourly schedule.", "available": True,
"config_fields": [
{"key": "max_concurrent", "type": "number", "label": "Max simultaneous searches", "default": 3, "min": 1}
]},
{"type": "video_process_youtube_wishlist", "label": "Download YouTube Wishlist", "icon": "download", "scope": "video",
"description": "Download wished YouTube videos (yt-dlp), organised as a Plex 'TV by date' show (channel/year/date). Queues the WHOLE wishlist — the setting only limits how many download at the same time; each finished one starts the next, so it all drains. A completed download leaves the wishlist. Needs the YouTube library folder set on Settings → Downloads.", "available": True,
"config_fields": [

13
core/automation/handlers/registration.py
View file

@ -41,6 +41,7 @@ from core.automation.handlers.video_scan_watchlist_people import auto_video_scan
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_watchlist_playlists import auto_video_scan_watchlist_playlists
from core.automation.handlers.video_process_wishlist import auto_video_process_wishlist, is_running
from core.automation.handlers.video_scan_library import (
auto_video_scan_library, auto_video_scan_server, auto_video_update_database,
)
@ -254,6 +255,18 @@ def register_all(deps: AutomationDeps) -> None:
'video_process_youtube_wishlist',
lambda config: auto_video_process_youtube_wishlist(config, deps),
)
# Soulseek drain: auto-grab wished movies / episodes (search → pick best → download).
# The guard skips an hourly tick while a drain is still working (searches are slow).
engine.register_action_handler(
'video_download_movie_wishlist',
lambda config: auto_video_process_wishlist(config, deps, media_type='movie'),
lambda: is_running('movie'),
)
engine.register_action_handler(
'video_download_episode_wishlist',
lambda config: auto_video_process_wishlist(config, deps, media_type='episode'),
lambda: is_running('episode'),
)
# Progress + history callbacks: the engine invokes these around
# each handler run. Lift the closures from

250
core/automation/handlers/video_process_wishlist.py Normal file
View file

@ -0,0 +1,250 @@
"""Automation handlers: ``video_download_movie_wishlist`` / ``video_download_episode_wishlist``.
The Soulseek counterpart of the YouTube wishlist drain — the piece that finally makes the
people/airing scans pay off. For wished, RELEASED movies (and aired episodes) it searches
Soulseek, picks the best release per the quality profile, and enqueues the download; the
existing ``download_monitor`` finishes + organises + archives it, exactly like a manual grab.
Shape mirrors the YouTube drain (Boulder: "same standard"): it processes the WHOLE eligible
wishlist (no total cap), but the slow part — each item needs a ~20s blocking Soulseek search
— runs only a FEW at a time (``max_concurrent``). A ``guard`` keeps the next hourly tick from
overlapping a run that's still working, so it can't pile up.
Movies are gated on ``status='wanted'`` (released; skips 'monitored'/unreleased). Episodes
are all-wished (the airing scan only adds aired ones). Items already downloading are skipped
so re-runs never double-grab. The pick is the top ACCEPTED release — the ranker already
encodes the quality profile's accept/reject/score, so no extra rules here.
Shared automation side (may import ``core.video`` / ``api.video``); owns its own progress.
The search + enqueue are injected seams, so selection/pick/record are pure + unit-tested.
"""
from __future__ import annotations
import json
import threading
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Callable, Dict, Iterable, List, Optional
from core.automation.deps import AutomationDeps
# ── pure helpers ──────────────────────────────────────────────────────────────
def pick_best(candidates: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""The best ACCEPTED release from a ranked candidate list. ``_evaluate_hits`` already
sorts best-first (accepted, score, availability), so the first accepted is the pick."""
for c in candidates or []:
if c.get("accepted"):
return c
return None
def item_key(item: Dict[str, Any], media_type: str) -> tuple:
"""Stable identity for de-duping a wished item against active downloads."""
if media_type == "movie":
return ("movie", str(item.get("tmdb_id")))
return ("episode", str(item.get("show_tmdb_id")),
int(item.get("season_number") or 0), int(item.get("episode_number") or 0))
def active_download_keys(active: Iterable[Dict[str, Any]]) -> set:
"""Identity keys for the movie/episode downloads already in flight, so we don't
re-grab them. Episodes read season/episode out of the row's ``search_ctx``."""
keys = set()
for d in active or []:
kind = str(d.get("kind") or "").lower()
if kind == "movie":
keys.add(("movie", str(d.get("media_id"))))
elif kind == "episode":
ctx = d.get("search_ctx")
if isinstance(ctx, str):
try:
ctx = json.loads(ctx)
except (ValueError, TypeError):
ctx = {}
ctx = ctx if isinstance(ctx, dict) else {}
keys.add(("episode", str(d.get("media_id")),
int(ctx.get("season") or 0), int(ctx.get("episode") or 0)))
return keys
def search_context(item: Dict[str, Any], media_type: str) -> Dict[str, Any]:
"""The ``search_ctx`` the download row carries (drives the monitor's requery)."""
if media_type == "movie":
return {"scope": "movie", "title": item.get("title"), "year": item.get("year")}
return {"scope": "episode", "title": item.get("show_title"),
"season": item.get("season_number"), "episode": item.get("episode_number"),
"year": (str(item.get("air_date") or "")[:4] or None)}
def build_download_record(item: Dict[str, Any], best: Dict[str, Any], candidates: List[Dict[str, Any]],
*, media_type: str, target_dir: str, query: Any) -> Dict[str, Any]:
"""The ``add_video_download`` row for a chosen release — identical shape to a manual
grab, so the monitor finishes it the same way (other accepted hits become the retry
pool)."""
ctx = search_context(item, media_type)
rest = [c for c in (candidates or []) if c.get("filename") != best.get("filename")]
media_id = str(item.get("tmdb_id") if media_type == "movie" else item.get("show_tmdb_id"))
return {
"kind": media_type, "title": ctx["title"],
"release_title": best.get("title") or best.get("filename"),
"source": "soulseek", "username": best.get("username"), "filename": best.get("filename"),
"size_bytes": int(best.get("size_bytes") or 0), "quality_label": best.get("quality_label"),
"target_dir": target_dir, "status": "downloading",
"media_id": media_id, "media_source": "tmdb", "year": ctx.get("year"),
"poster_url": item.get("poster_url"),
"candidates": json.dumps(rest), "search_ctx": json.dumps(ctx),
"tried_queries": json.dumps([query] if query else []),
"tried_files": json.dumps([best.get("filename")]), "attempts": 0,
}
# ── production seams ──────────────────────────────────────────────────────────
def _default_fetch_items(media_type: str) -> List[Dict[str, Any]]:
from api.video import get_video_db
db = get_video_db()
return db.movie_wishlist_to_download() if media_type == "movie" else db.episode_wishlist_to_download()
def _default_active_keys(media_type: str) -> set:
from api.video import get_video_db
return active_download_keys(get_video_db().get_active_video_downloads())
def _default_target_dir(media_type: str) -> str:
from api.video import get_video_db
db = get_video_db()
if media_type == "movie":
return db.get_setting("movies_path") or db.get_setting("transfer_path") or ""
return db.get_setting("tv_path") or ""
def _default_search(item: Dict[str, Any], media_type: str) -> List[Dict[str, Any]]:
"""A bounded blocking Soulseek search → ranked candidates (same path the retry worker +
manual search use). Returns [] if slskd isn't configured / nothing came back."""
from api.video.downloads import _evaluate_hits
from core.video.download_monitor import _search_for_retry
from core.video.quality_profile import load as load_profile
from core.video.slskd_search import build_query
from api.video import get_video_db
ctx = search_context(item, media_type)
query = build_query(ctx["scope"], ctx["title"], year=ctx.get("year"),
season=ctx.get("season"), episode=ctx.get("episode"))
res = _search_for_retry(query) or {}
profile = load_profile(get_video_db())
return _evaluate_hits(res.get("hits") or [], profile, ctx["scope"],
ctx.get("season"), ctx.get("episode"))
def _default_enqueue(item: Dict[str, Any], best: Dict[str, Any], candidates: List[Dict[str, Any]],
media_type: str, target_dir: str) -> bool:
"""Start the slskd transfer + write the download row (exactly like the manual flow),
then ensure the monitor is running. Returns True if slskd accepted it."""
from api.video import get_video_db
from core.video.download_monitor import ensure_started
from core.video.slskd_download import start_download
from core.video.slskd_search import build_query
started = start_download(best.get("username"), best.get("filename"), best.get("size_bytes") or 0)
if not started.get("ok"):
return False
ctx = search_context(item, media_type)
query = build_query(ctx["scope"], ctx["title"], year=ctx.get("year"),
season=ctx.get("season"), episode=ctx.get("episode"))
get_video_db().add_video_download(
build_download_record(item, best, candidates, media_type=media_type,
target_dir=target_dir, query=query))
ensure_started(get_video_db)
return True
# ── guard: keep an in-progress drain from overlapping the next tick ───────────
_running: Dict[str, bool] = {"movie": False, "episode": False}
def is_running(media_type: str) -> bool:
return bool(_running.get(media_type))
def auto_video_process_wishlist(
config: Dict[str, Any],
deps: AutomationDeps,
*,
media_type: str = "movie",
fetch_items: Optional[Callable[[str], List[Dict[str, Any]]]] = None,
active_keys: Optional[Callable[[str], Iterable]] = None,
target_dir: Optional[Callable[[str], str]] = None,
search: Optional[Callable[[Dict[str, Any], str], List[Dict[str, Any]]]] = None,
enqueue: Optional[Callable[..., bool]] = None,
) -> Dict[str, Any]:
"""Auto-grab the wished movies (or episodes): search Soulseek, pick the best release,
enqueue. Processes the whole eligible wishlist, a few searches at a time.
Returns ``{'status': 'completed', 'searched': int, 'grabbed': int, ...}``."""
fetch_items = fetch_items or _default_fetch_items
active_keys = active_keys or _default_active_keys
target_dir = target_dir or _default_target_dir
search = search or _default_search
enqueue = enqueue or _default_enqueue
automation_id = config.get('_automation_id')
concurrency = max(1, int(config.get('max_concurrent', 3) or 3))
label = 'movie' if media_type == 'movie' else 'episode'
_running[media_type] = True
try:
root = target_dir(media_type)
if not root:
where = 'Movie' if media_type == 'movie' else 'TV'
deps.update_progress(automation_id, status='finished', progress=100, phase='Complete',
log_line='%s library folder not set — skipping (Settings → Downloads)' % where,
log_type='info')
return {'status': 'completed', 'searched': 0, 'grabbed': 0,
'skipped': 'no_folder', '_manages_own_progress': True}
deps.update_progress(automation_id, phase='Checking the wishlist…', progress=5,
log_line='Looking for wished %ss to grab' % label, log_type='info')
items = fetch_items(media_type) or []
active = set(active_keys(media_type) or set())
todo = [it for it in items if item_key(it, media_type) not in active]
if not todo:
deps.update_progress(automation_id, status='finished', progress=100, phase='Complete',
log_line='Nothing new to grab (%d already in flight)' % len(active),
log_type='info')
return {'status': 'completed', 'searched': 0, 'grabbed': 0, '_manages_own_progress': True}
grabbed = [0]
searched = [0]
total = len(todo)
lock = threading.Lock()
def _one(it):
cands = search(it, media_type) or []
best = pick_best(cands)
ok = bool(best) and bool(enqueue(it, best, cands, media_type, root))
name = it.get('title') or it.get('show_title') or '?'
if media_type == 'episode':
name = "%s S%02dE%02d" % (name, int(it.get('season_number') or 0),
int(it.get('episode_number') or 0))
with lock:
searched[0] += 1
if ok:
grabbed[0] += 1
deps.update_progress(
automation_id, phase='Searching + grabbing…',
progress=10 + int(85 * searched[0] / max(total, 1)),
log_line=("Grabbed '%s'" % name) if ok else ("No acceptable release for '%s'" % name),
log_type='success' if ok else 'info')
with ThreadPoolExecutor(max_workers=concurrency) as ex:
list(ex.map(_one, todo))
done = ('Grabbed %d %s(s) of %d searched' % (grabbed[0], label, searched[0])) if grabbed[0] \
else ('Searched %d %s(s) — no acceptable releases found' % (searched[0], label))
deps.update_progress(automation_id, status='finished', progress=100, phase='Complete',
log_line=done, log_type='success' if grabbed[0] else 'info')
return {'status': 'completed', 'searched': searched[0], 'grabbed': grabbed[0],
'_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}
finally:
_running[media_type] = False

21
core/automation_engine.py
View file

@ -279,6 +279,27 @@ SYSTEM_AUTOMATIONS = [
'initial_delay': 1500,
'owned_by': 'video',
},
# Soulseek drains: auto-grab wished movies / episodes hourly (search → pick best →
# download). A guard skips the tick while a drain is still working. Skip quietly until
# the library folder + slskd are set.
{
'name': 'Auto-Download Movie Wishlist',
'trigger_type': 'schedule',
'trigger_config': {'interval': 1, 'unit': 'hours'},
'action_type': 'video_download_movie_wishlist',
'action_config': {'max_concurrent': 3},
'initial_delay': 1620,
'owned_by': 'video',
},
{
'name': 'Auto-Download Episode Wishlist',
'trigger_type': 'schedule',
'trigger_config': {'interval': 1, 'unit': 'hours'},
'action_type': 'video_download_episode_wishlist',
'action_config': {'max_concurrent': 3},
'initial_delay': 1740,
'owned_by': 'video',
},
]

26
database/video_database.py
View file

@ -2465,6 +2465,32 @@ class VideoDatabase:
finally:
conn.close()
def movie_wishlist_to_download(self) -> list:
"""Wished movies ready to grab: only ``status='wanted'`` (released) — skips
'monitored' (unreleased) and anything the engine already advanced
(searching/downloading/downloaded/failed). Newest year first."""
conn = self._get_connection()
try:
return [dict(r) for r in conn.execute(
"SELECT tmdb_id, title, year, poster_url FROM video_wishlist "
"WHERE kind='movie' AND status='wanted' AND tmdb_id IS NOT NULL "
"ORDER BY year DESC, id DESC")]
finally:
conn.close()
def episode_wishlist_to_download(self) -> list:
"""Wished episodes ready to grab (all of them — the airing scan only wishlists
episodes that have aired). Newest air date first."""
conn = self._get_connection()
try:
return [dict(r) for r in conn.execute(
"SELECT tmdb_id AS show_tmdb_id, title AS show_title, season_number, "
"episode_number, episode_title, air_date, poster_url, library_id "
"FROM video_wishlist WHERE kind='episode' AND tmdb_id IS NOT NULL "
"ORDER BY air_date DESC, id DESC")]
finally:
conn.close()
def add_episodes_to_wishlist(self, show_tmdb_id, show_title, episodes, *,
poster_url=None, library_id=None, server_source=None) -> int:
"""Wish for one or more episodes of a show (the show's tmdb id keys them).

4
tests/automation/test_handler_registration.py
View file

@ -64,6 +64,8 @@ EXPECTED_ACTION_NAMES = frozenset({
'video_scan_watchlist_channels',
'video_scan_watchlist_playlists',
'video_process_youtube_wishlist',
'video_download_movie_wishlist',
'video_download_episode_wishlist',
'video_clean_search_history',
'video_clean_completed_downloads',
'video_full_cleanup',
@ -81,6 +83,8 @@ EXPECTED_GUARDED_ACTIONS = frozenset({
'deep_scan_library',
'run_duplicate_cleaner',
'start_quality_scan',
'video_download_movie_wishlist',
'video_download_episode_wishlist',
})

178
tests/test_video_process_wishlist.py Normal file
View file

@ -0,0 +1,178 @@
"""Soulseek auto-grab for the movie + episode wishlist: search → pick best accepted release
→ enqueue (the monitor finishes it). Pure pick/select/record + the handler with the slow
slskd search + enqueue injected, plus the DB queries that feed it.
"""
from __future__ import annotations
import json
import pytest
from core.automation.handlers.video_process_wishlist import (
active_download_keys,
auto_video_process_wishlist,
build_download_record,
item_key,
pick_best,
)
class _Deps:
def __init__(self):
self.progress = []
def update_progress(self, automation_id, **kw):
self.progress.append(kw)
def _cand(fn, *, accepted=True, score=10, user="u"):
return {"filename": fn, "title": fn, "username": user, "size_bytes": 1000,
"quality_label": "WEBDL-1080p", "accepted": accepted, "score": score}
# ── pure ──────────────────────────────────────────────────────────────────────
def test_pick_best_takes_first_accepted():
cands = [_cand("a", accepted=False), _cand("b", accepted=True), _cand("c", accepted=True)]
assert pick_best(cands)["filename"] == "b"
def test_pick_best_none_when_all_rejected():
assert pick_best([_cand("a", accepted=False), _cand("b", accepted=False)]) is None
assert pick_best([]) is None
def test_item_key_movie_and_episode():
assert item_key({"tmdb_id": 5}, "movie") == ("movie", "5")
assert item_key({"show_tmdb_id": 9, "season_number": 1, "episode_number": 3}, "episode") \
== ("episode", "9", 1, 3)
def test_active_download_keys_reads_ctx_for_episodes():
active = [
{"kind": "movie", "media_id": "5"},
{"kind": "episode", "media_id": "9", "search_ctx": json.dumps({"season": 1, "episode": 3})},
]
keys = active_download_keys(active)
assert ("movie", "5") in keys and ("episode", "9", 1, 3) in keys
def test_build_record_movie_shape():
item = {"tmdb_id": 5, "title": "The Matrix", "year": "1999", "poster_url": "/p.jpg"}
best = _cand("Matrix.1999.1080p.mkv")
rest = [best, _cand("other.mkv")]
rec = build_download_record(item, best, rest, media_type="movie", target_dir="/movies", query="matrix 1999")
assert rec["kind"] == "movie" and rec["title"] == "The Matrix" and rec["media_id"] == "5"
assert rec["source"] == "soulseek" and rec["status"] == "downloading"
assert rec["target_dir"] == "/movies" and rec["filename"] == "Matrix.1999.1080p.mkv"
assert json.loads(rec["search_ctx"]) == {"scope": "movie", "title": "The Matrix", "year": "1999"}
assert [c["filename"] for c in json.loads(rec["candidates"])] == ["other.mkv"] # best excluded
def test_build_record_episode_shape():
item = {"show_tmdb_id": 9, "show_title": "Breaking Bad", "season_number": 1,
"episode_number": 3, "air_date": "2008-02-10"}
best = _cand("BrBa.S01E03.mkv")
rec = build_download_record(item, best, [best], media_type="episode", target_dir="/tv", query="q")
assert rec["kind"] == "episode" and rec["media_id"] == "9" and rec["year"] == "2008"
assert json.loads(rec["search_ctx"]) == {"scope": "episode", "title": "Breaking Bad",
"season": 1, "episode": 3, "year": "2008"}
# ── handler ───────────────────────────────────────────────────────────────────
def _run(items, *, active=None, root="/movies", media_type="movie", searches=None):
enq = []
seen = []
def search(item, mt):
seen.append(item_key(item, mt))
if searches is None:
return [_cand("rel-" + str(item.get("tmdb_id") or item.get("show_tmdb_id")))]
return searches.get(item_key(item, mt), [])
def enqueue(item, best, cands, mt, target):
enq.append((item_key(item, mt), best["filename"], target))
return True
deps = _Deps()
res = auto_video_process_wishlist(
{"_automation_id": "a", "max_concurrent": 2}, deps, media_type=media_type,
fetch_items=lambda mt: items, active_keys=lambda mt: set(active or set()),
target_dir=lambda mt: root, search=search, enqueue=enqueue)
return res, enq, seen, deps
def test_grabs_each_wished_movie():
items = [{"tmdb_id": 1, "title": "A", "year": "2020"}, {"tmdb_id": 2, "title": "B", "year": "2021"}]
res, enq, seen, _ = _run(items)
assert res["status"] == "completed" and res["grabbed"] == 2 and res["searched"] == 2
assert {e[0] for e in enq} == {("movie", "1"), ("movie", "2")}
assert enq[0][2] == "/movies"
def test_skips_items_already_downloading():
items = [{"tmdb_id": 1, "title": "A"}, {"tmdb_id": 2, "title": "B"}]
res, enq, seen, _ = _run(items, active={("movie", "1")})
assert {e[0] for e in enq} == {("movie", "2")} # 1 already in flight
assert ("movie", "1") not in seen # not even searched
def test_no_acceptable_release_grabs_nothing():
items = [{"tmdb_id": 1, "title": "A"}]
res, enq, _, _ = _run(items, searches={("movie", "1"): [_cand("junk", accepted=False)]})
assert res["searched"] == 1 and res["grabbed"] == 0 and enq == []
def test_missing_library_folder_is_a_quiet_skip():
res, enq, _, deps = _run([{"tmdb_id": 1, "title": "A"}], root="")
assert res["status"] == "completed" and res.get("skipped") == "no_folder"
assert enq == [] and not any(p.get("status") == "error" for p in deps.progress)
def test_nothing_wished_is_a_clean_noop():
res, enq, _, _ = _run([])
assert res["status"] == "completed" and res["grabbed"] == 0 and enq == []
def test_episode_mode_keys_and_grabs():
items = [{"show_tmdb_id": 9, "show_title": "BrBa", "season_number": 1, "episode_number": 3}]
res, enq, seen, _ = _run(items, root="/tv", media_type="episode")
assert res["grabbed"] == 1 and enq[0][0] == ("episode", "9", 1, 3) and enq[0][2] == "/tv"
def test_top_level_error_is_caught_and_clears_guard():
from core.automation.handlers.video_process_wishlist import is_running
def boom(mt):
raise RuntimeError("db down")
res = auto_video_process_wishlist({"_automation_id": "x"}, _Deps(), media_type="movie",
target_dir=lambda mt: "/m", fetch_items=boom)
assert res["status"] == "error" and "db down" in res["error"]
assert is_running("movie") is False # guard released even on error
# ── DB queries ────────────────────────────────────────────────────────────────
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_movie_wishlist_to_download_only_wanted(db):
db.add_movie_to_wishlist(1, "Released", year="2020", status="wanted")
db.add_movie_to_wishlist(2, "Upcoming", year="2027", status="monitored")
rows = db.movie_wishlist_to_download()
assert [r["tmdb_id"] for r in rows] == [1] # monitored is skipped
def test_episode_wishlist_to_download_shape(db):
db.add_episodes_to_wishlist(9, "Breaking Bad", [
{"season_number": 1, "episode_number": 1, "air_date": "2008-01-20"},
{"season_number": 1, "episode_number": 2, "air_date": "2008-01-27"}])
rows = db.episode_wishlist_to_download()
assert len(rows) == 2
top = rows[0]
assert top["show_tmdb_id"] == 9 and top["show_title"] == "Breaking Bad"
assert top["season_number"] == 1 and top["episode_number"] == 2 # newest air date first

3
webui/static/stats-automations.js
View file

@ -1778,6 +1778,7 @@ const _autoIcons = {
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_scan_watchlist_playlists: '\uD83C\uDFB5', video_process_youtube_wishlist: '\u2B07\uFE0F',
video_download_movie_wishlist: '\uD83C\uDFAC', video_download_episode_wishlist: '\uD83D\uDCFA',
};
// --- Inspiration Templates ---
@ -3376,6 +3377,8 @@ function _autoFormatAction(type) {
video_scan_watchlist_channels: 'Scan Watchlist Channels',
video_scan_watchlist_playlists: 'Scan Watchlist Playlists',
video_process_youtube_wishlist: 'Download YouTube Wishlist',
video_download_movie_wishlist: 'Download Movie Wishlist',
video_download_episode_wishlist: 'Download Episode Wishlist',
};
return labels[type] || type || 'Unknown';
}