From a16afd1f9ea5f20fe23dccd40086844d0cd717e9 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Sun, 21 Jun 2026 13:44:52 -0700 Subject: [PATCH] Post-download scan: wait until Plex's scan queue is actually idle, not a fixed 2min MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A fixed debounce can't fit a big library — 8500 movies + 4500 shows scan sequentially through Plex's queue and can take 10-20 min, so the old 120s wait read the DB before Plex finished and fresh downloads showed up late. Now Stage 1 (video_scan_server) fires the rescan then POLLS the server until its scan queue goes idle, then emits the done event. - sources: PlexVideoSource.is_scanning (section.refreshing + activity feed, scoped by media_type) and JellyfinVideoSource.is_scanning (scheduled-task state), plus video_server_scan_in_progress() returning True/False/None. - handler: pure wait_for_server_scan(scan_status, sleep, …) — grace, then poll every interval until idle or a generous cap; falls back to the fixed wait only when the server can't report status (None). debounce_seconds is now that fallback; new max_wait_minutes caps the poll. Seam tests for the poll logic (idle/poll/fallback/cap/lost-status), the handler wiring, and Plex scan-status detection. --- core/automation/blocks.py | 5 +- .../automation/handlers/video_scan_library.py | 76 ++++++++++++++++--- core/video/sources.py | 55 ++++++++++++++ .../test_handler_video_scan_library.py | 71 ++++++++++++++--- tests/test_video_scan_scope.py | 39 ++++++++++ 5 files changed, 220 insertions(+), 26 deletions(-) diff --git a/core/automation/blocks.py b/core/automation/blocks.py index e3e410fc..1cab85fd 100644 --- a/core/automation/blocks.py +++ b/core/automation/blocks.py @@ -228,14 +228,15 @@ ACTIONS: list[dict] = [ # Post-download chain actions (two stages, like music's scan_library + # start_database_update). Stage 1 nudges the server; stage 2 reads it in. {"type": "video_scan_server", "label": "Scan Video Server", "icon": "refresh", "scope": "video", - "description": "Tell the media server to rescan your video sections, then fire 'Video Library Scan Done'", "available": True, + "description": "Tell the media server to rescan your video sections, wait until it actually finishes indexing, then fire 'Video Library Scan Done'", "available": True, "config_fields": [ {"key": "media_type", "type": "select", "label": "Library", "options": [{"value": "all", "label": "Movies + TV"}, {"value": "movie", "label": "Movies only"}, {"value": "show", "label": "TV only"}], "default": "all"}, - {"key": "debounce_seconds", "type": "number", "label": "Wait for indexing (sec)", "default": 120, "min": 10} + {"key": "max_wait_minutes", "type": "number", "label": "Max wait for scan (min)", "default": 60, "min": 1}, + {"key": "debounce_seconds", "type": "number", "label": "Fallback wait if status unknown (sec)", "default": 120, "min": 10} ]}, {"type": "video_update_database", "label": "Update Video Database", "icon": "database", "scope": "video", "description": "Read newly-indexed media from the server into SoulSync (incremental)", "available": True, diff --git a/core/automation/handlers/video_scan_library.py b/core/automation/handlers/video_scan_library.py index 47c96d59..d0177f4d 100644 --- a/core/automation/handlers/video_scan_library.py +++ b/core/automation/handlers/video_scan_library.py @@ -174,31 +174,81 @@ def auto_video_scan_library( # ── post-download chain (video twin of music's scan_library → start_database_update) ── # Split into two stages so the calendar/library reflect a download promptly, mirroring -# the music side: tell the server to rescan, then (after it has had time to index) read -# the new state into video.db. Stage 1 emits 'video_library_scan_completed' on a debounce -# (like music's web_scan_manager time-based completion); stage 2 listens for that. +# the music side: tell the server to rescan, WAIT until its scan queue is actually idle +# (a big library can take 10-20 min — a fixed wait would read too early), then fire +# 'video_library_scan_completed'; stage 2 listens for that and reads the fresh state. + +def _default_scan_status(media_type: str = "all"): + """Production wiring: True/False if the active server is/ isn't scanning, or None + when it can't report (so the caller falls back to a fixed wait).""" + from core.video.sources import video_server_scan_in_progress + return video_server_scan_in_progress(media_type) + + +def wait_for_server_scan(scan_status, sleep, *, grace_seconds: int = 15, + interval_seconds: int = 10, cap_seconds: int = 3600, + fallback_seconds: int = 120) -> int: + """Block until the media server's scan queue goes idle, then return ~seconds waited. + + ``scan_status()`` returns True (scanning), False (idle) or None (can't tell). After + a short grace — so the just-triggered scan has time to register as running — poll + every ``interval_seconds`` until idle or ``cap_seconds`` (a generous backstop for a + huge library). If the server can't report its state, fall back to a fixed + ``fallback_seconds`` wait — exactly the old behaviour. ``sleep`` is injected and + elapsed is counted from the sleeps, so tests never actually block.""" + grace = max(0, grace_seconds) + if grace: + sleep(grace) + waited = grace + try: + status = scan_status() + except Exception: # noqa: BLE001 - status is best-effort + status = None + if status is None: + extra = max(0, fallback_seconds - waited) + if extra: + sleep(extra) + return waited + extra + while status is True and waited < cap_seconds: + sleep(interval_seconds) + waited += interval_seconds + try: + status = scan_status() + except Exception: # noqa: BLE001 + status = None + if status is None: # lost the ability to tell — stop waiting, don't hang + break + return waited + def auto_video_scan_server( config: Dict[str, Any], deps: AutomationDeps, *, - server_refresh: Optional[Callable[[], Dict[str, Any]]] = None, + server_refresh: Optional[Callable[..., Dict[str, Any]]] = None, sleep: Optional[Callable[[float], None]] = None, emit: Optional[Callable[[str, Dict[str, Any]], None]] = None, + scan_status: Optional[Callable[..., Any]] = None, ) -> Dict[str, Any]: - """Stage 1: tell the media server to rescan the video sections, wait a debounce - for it to index, then fire 'video_library_scan_completed' so the DB-update twin - reads the fresh state. (Video twin of music's scan_library.)""" + """Stage 1: tell the media server to rescan the video sections, WAIT until its scan + queue is idle (polling the server; falls back to a fixed wait if it can't report), + then fire 'video_library_scan_completed' so the DB-update twin reads the fresh state. + (Video twin of music's scan_library.)""" server_refresh = server_refresh or _default_server_refresh sleep = sleep or time.sleep emit = emit or deps.engine.emit + scan_status = scan_status or _default_scan_status automation_id = config.get('_automation_id') media_type = config.get('media_type') or 'all' lib_label = {'movie': 'Movie', 'show': 'TV'}.get(media_type, 'video') try: - debounce = int(config.get('debounce_seconds') or 120) + fallback = int(config.get('debounce_seconds') or 120) except (TypeError, ValueError): - debounce = 120 + fallback = 120 + try: + cap = int(config.get('max_wait_minutes') or 60) * 60 + except (TypeError, ValueError): + cap = 3600 try: deps.update_progress(automation_id, phase=f'Asking media server to rescan {lib_label}…', progress=20, log_line=f'Triggering server-side {lib_label} scan', log_type='info') @@ -207,13 +257,15 @@ def auto_video_scan_server( deps.update_progress(automation_id, log_line='Server scan trigger unavailable: ' + str(refresh.get('error') or 'unknown'), log_type='warning') - deps.update_progress(automation_id, phase=f'Waiting for the server to index ({debounce}s)…', progress=55) - sleep(debounce) + deps.update_progress(automation_id, phase='Waiting for the server to finish indexing…', progress=55) + waited = wait_for_server_scan(lambda: scan_status(media_type), sleep, + fallback_seconds=fallback, cap_seconds=cap) # Carry the scope so stage 2 (Update DB) reads only the library we rescanned. emit('video_library_scan_completed', {'server': refresh.get('server') or '', 'media_type': media_type}) deps.update_progress(automation_id, status='finished', progress=100, phase='Complete', - log_line='Server scan done — updating the database', log_type='success') + log_line=f'Server scan finished (~{waited}s) — updating the database', + log_type='success') return {'status': 'completed', '_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') diff --git a/core/video/sources.py b/core/video/sources.py index 984aae6b..4320c10f 100644 --- a/core/video/sources.py +++ b/core/video/sources.py @@ -296,6 +296,21 @@ def refresh_video_server_sections(media_type="all"): return {"ok": False, "error": str(e)} +def video_server_scan_in_progress(media_type="all"): + """True if the active video server is mid-scan for the given library (or either, + for 'all'); False if idle; None if it can't be determined — no server, or an + adapter that can't report scan state. Callers fall back to a fixed wait on None.""" + media_type = normalize_media_type(media_type) + src = get_active_video_source() + if src is None or not hasattr(src, "is_scanning"): + return None + try: + return bool(src.is_scanning(media_type)) + except Exception: + logger.debug("video sources: scan-status check failed", exc_info=True) + return None + + def list_video_libraries(): """Discover the active server's video libraries for the mapping UI: {'server', 'movies': [{'title'}], 'tv': [{'title'}]} or None.""" @@ -376,6 +391,26 @@ class PlexVideoSource: logger.exception("Plex: refresh failed for section %s", getattr(s, "title", "?")) return {"ok": n > 0, "sections": n} + def is_scanning(self, media_type="all") -> bool: + """True if any SELECTED video section (scoped by media_type) is currently + being scanned by Plex. Checks the per-section refreshing flag, then the + server activity feed (real-time) — mirrors the music PlexClient check.""" + sections = [] + for kind, name in (("movie", self._movies_lib), ("show", self._tv_lib)): + if media_type != "all" and media_type != kind: + continue + sections.extend(self._scan_sections(kind, name)) + for s in sections: + if getattr(s, "refreshing", False): + return True + titles = {(getattr(s, "title", "") or "").lower() for s in sections} + for act in self._server.activities(): + if getattr(act, "type", "") in ("library.scan", "library.refresh"): + at = (getattr(act, "title", "") or "").lower() + if any(t and t in at for t in titles): + return True + return False + @staticmethod def _part_file(obj): try: @@ -571,6 +606,26 @@ class JellyfinVideoSource: logger.exception("Jellyfin: refresh failed for view %s", vid) return {"ok": n > 0, "sections": n} + def is_scanning(self, media_type="all") -> bool: + """True if Jellyfin's library-scan scheduled task is running. Jellyfin's + scan isn't per-library, so media_type is ignored (any scan counts).""" + import requests + base = (self._c.base_url or "").rstrip("/") + if not base: + return False + try: + r = requests.get(base + "/ScheduledTasks", + headers={"X-Emby-Token": self._c.api_key or ""}, timeout=10) + tasks = r.json() if r.ok else [] + except Exception: + return False + for task in tasks or []: + name = (task.get("Name") or "").lower() + if (("scan" in name or "refresh" in name or "library" in name) + and task.get("State") in ("Running", "Cancelling")): + return True + return False + def counts(self, incremental=False) -> dict: def total(view, itype): resp = self._req(f"/Users/{self.uid}/Items", { diff --git a/tests/automation/test_handler_video_scan_library.py b/tests/automation/test_handler_video_scan_library.py index 9813a3f1..f71c8534 100644 --- a/tests/automation/test_handler_video_scan_library.py +++ b/tests/automation/test_handler_video_scan_library.py @@ -229,29 +229,74 @@ class TestHandlerNeverRaises: # ── post-download chain: video_scan_server (stage 1) ─────────────────────── from core.automation.handlers.video_scan_library import ( # noqa: E402 - auto_video_scan_server, auto_video_update_database) + auto_video_scan_server, auto_video_update_database, wait_for_server_scan) + + +class TestWaitForServerScan: + """Poll the server until its scan queue is idle; fixed wait only as a fallback.""" + + def test_returns_quickly_when_already_idle(self): + slept = [] + waited = wait_for_server_scan(lambda: False, slept.append, grace_seconds=15) + assert waited == 15 and slept == [15] # just the grace, then idle + + def test_polls_until_the_scan_finishes(self): + # scanning for three polls, then idle — handles a 10-20 min big-library scan + seq = iter([True, True, True, False]) + slept = [] + waited = wait_for_server_scan(lambda: next(seq), slept.append, + grace_seconds=15, interval_seconds=10) + assert waited == 15 + 30 # grace + 3 polls + assert slept == [15, 10, 10, 10] + + def test_falls_back_to_fixed_wait_when_status_unknown(self): + slept = [] + waited = wait_for_server_scan(lambda: None, slept.append, + grace_seconds=15, fallback_seconds=120) + assert waited == 120 and slept == [15, 105] # grace + (fallback - grace) + + def test_stops_at_the_cap_on_a_runaway_scan(self): + slept = [] + waited = wait_for_server_scan(lambda: True, slept.append, + grace_seconds=0, interval_seconds=10, cap_seconds=50) + assert waited == 50 # never hangs forever + + def test_stops_if_status_becomes_unknown_mid_poll(self): + seq = iter([True, None]) + slept = [] + wait_for_server_scan(lambda: next(seq), slept.append, grace_seconds=0, interval_seconds=10) + assert slept == [10] # one poll, then bail — no hang class TestScanServerStage: - def test_refreshes_waits_then_emits_scan_done(self): + def test_waits_for_idle_then_emits_scan_done(self): deps = _RecordingDeps() events = [] - slept = [] + # scanning once, then idle + seq = iter([True, False]) r = auto_video_scan_server( - {'_automation_id': 'a', 'debounce_seconds': 90}, deps, - server_refresh=_refresh_ok(), sleep=lambda s: slept.append(s), + {'_automation_id': 'a'}, deps, + server_refresh=_refresh_ok(), sleep=lambda s: None, + scan_status=lambda mt: next(seq), emit=lambda ev, data: events.append((ev, data))) assert r['status'] == 'completed' - assert slept == [90] # waited the debounce assert events == [('video_library_scan_completed', {'server': '', 'media_type': 'all'})] - def test_default_debounce_and_server_unavailable_still_emits(self): + def test_falls_back_to_fixed_wait_when_server_cant_report(self): + slept = [] + auto_video_scan_server( + {'_automation_id': 'a', 'debounce_seconds': 90}, _RecordingDeps(), + server_refresh=_refresh_ok(), sleep=slept.append, + scan_status=lambda mt: None, emit=lambda ev, data: None) + assert sum(slept) == 90 # honoured the fallback wait + + def test_server_unavailable_still_emits(self): deps = _RecordingDeps() events = [] auto_video_scan_server( {'_automation_id': 'a'}, deps, server_refresh=lambda media_type=None: {'ok': False, 'error': 'no server'}, - sleep=lambda s: None, emit=lambda ev, data: events.append(ev)) + sleep=lambda s: None, scan_status=lambda mt: False, emit=lambda ev, data: events.append(ev)) assert events == ['video_library_scan_completed'] # fires even if refresh failed assert 'warning' in deps.log_types() @@ -260,22 +305,24 @@ class TestScanServerStage: r = auto_video_scan_server( {'_automation_id': 'a'}, deps, server_refresh=lambda media_type=None: (_ for _ in ()).throw(RuntimeError('boom')), - sleep=lambda s: None, emit=lambda *a: None) + sleep=lambda s: None, scan_status=lambda mt: False, emit=lambda *a: None) assert r['status'] == 'error' and r['error'] == 'boom' - def test_scopes_refresh_and_carries_media_type_on_the_event(self): + def test_scopes_refresh_and_status_and_carries_media_type_on_the_event(self): seen = {} events = [] def _refresh(media_type=None): - seen['mt'] = media_type + seen['refresh_mt'] = media_type return {'ok': True} auto_video_scan_server( {'_automation_id': 'a', 'media_type': 'show'}, _RecordingDeps(), server_refresh=_refresh, sleep=lambda s: None, + scan_status=lambda mt: seen.setdefault('status_mt', mt) and False, emit=lambda ev, data: events.append((ev, data))) - assert seen['mt'] == 'show' # only TV sections nudged + assert seen['refresh_mt'] == 'show' # only TV sections nudged + assert seen['status_mt'] == 'show' # polled TV scan status assert events[0][1]['media_type'] == 'show' # stage 2 inherits the scope diff --git a/tests/test_video_scan_scope.py b/tests/test_video_scan_scope.py index c2fb68f7..16ff1450 100644 --- a/tests/test_video_scan_scope.py +++ b/tests/test_video_scan_scope.py @@ -126,3 +126,42 @@ def test_refresh_sections_scopes_by_media_type(): res = src.refresh_sections("all") assert secs[0].updated and secs[1].updated and res["sections"] == 2 # both + + +# ── scan-status detection (poll-until-idle uses this) ─────────────────────── + +def test_plex_is_scanning_reads_section_refreshing_flag_scoped(): + class Sec: + def __init__(self, type_, title, refreshing=False): + self.type, self.title, self.refreshing = type_, title, refreshing + + class Srv: + def __init__(self, secs): self.library = _Lib(secs) + def activities(self): return [] + + # TV section refreshing, movie idle + secs = [Sec("movie", "Movies", False), Sec("show", "TV Shows", True)] + src = PlexVideoSource(Srv(secs), movies_lib="Movies", tv_lib="TV Shows") + assert src.is_scanning("show") is True # TV is mid-scan + assert src.is_scanning("movie") is False # movie idle → not scanning for that scope + assert src.is_scanning("all") is True # either counts + + +def test_plex_is_scanning_falls_back_to_activity_feed(): + class Sec: + def __init__(self, type_, title): self.type, self.title, self.refreshing = type_, title, False + class Act: + def __init__(self, type_, title): self.type, self.title = type_, title + class Srv: + def __init__(self, secs, acts): self.library, self._acts = _Lib(secs), acts + def activities(self): return self._acts + secs = [Sec("movie", "Movies")] + src = PlexVideoSource(Srv(secs, [Act("library.refresh", "Scanning Movies…")]), + movies_lib="Movies", tv_lib="TV Shows") + assert src.is_scanning("movie") is True # no refreshing flag, but the feed shows a scan + + +def test_scan_status_helper_is_none_when_no_server(monkeypatch): + import core.video.sources as srcmod + monkeypatch.setattr(srcmod, "get_active_video_source", lambda: None) + assert srcmod.video_server_scan_in_progress("all") is None # caller falls back to fixed wait