From 9e845e760edd4d05cfb7b4e4760202b25aa4d616 Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Sun, 21 Jun 2026 14:20:49 -0700 Subject: [PATCH] Smart post-download scan: skip the crawl when the server already has the grab (phase 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Scanning is expensive and most servers auto-ingest new files, so a full crawl after every download is usually wasted. Stage 1 now probes per library: take the newest completed grab of that type from download history and ask the server (cheap targeted search) whether it already has it. If yes, the server auto-picked it up (and the earlier ones) → skip that library's crawl + poll entirely. Only libraries the server is missing get rescanned. Always emits so stage 2 still reads the new items in. - sources: PlexVideoSource.has_item / JellyfinVideoSource.has_item (match movie by title+year, episode by show+SxE) + video_server_has_item() — conservative, any uncertainty → False so we scan. - handler: per-scope skip decision fed by latest_completed + server_has_item seams; narrows the scan scope to only the missing libraries; toggle skip_if_present (default on). Returns scanned/skipped for visibility. Seam tests: skip-both, scan-only-missing, no-history, toggle-off, probe-error→scan; Plex has_item match tests. --- core/automation/blocks.py | 3 +- .../automation/handlers/video_scan_library.py | 101 ++++++++++++++---- core/video/sources.py | 101 ++++++++++++++++++ .../test_handler_video_scan_library.py | 68 +++++++++++- tests/test_video_scan_scope.py | 44 ++++++++ 5 files changed, 290 insertions(+), 27 deletions(-) diff --git a/core/automation/blocks.py b/core/automation/blocks.py index 1cab85fd..6839de88 100644 --- a/core/automation/blocks.py +++ b/core/automation/blocks.py @@ -228,13 +228,14 @@ 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, wait until it actually finishes indexing, then fire 'Video Library Scan Done'", "available": True, + "description": "Get the server to index new downloads (skips the scan if it already has them), waits until it finishes, then fires '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": "skip_if_present", "type": "checkbox", "label": "Skip the scan if the server already has the download", "default": True}, {"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} ]}, diff --git a/core/automation/handlers/video_scan_library.py b/core/automation/handlers/video_scan_library.py index d0177f4d..e85088ab 100644 --- a/core/automation/handlers/video_scan_library.py +++ b/core/automation/handlers/video_scan_library.py @@ -185,6 +185,18 @@ def _default_scan_status(media_type: str = "all"): return video_server_scan_in_progress(media_type) +def _default_latest_completed(media_type: str): + """The newest completed grab of a type — the probe target for the smart skip.""" + from api.video import get_video_db + return get_video_db().latest_completed_download(media_type) + + +def _default_server_has_item(media_type: str, item) -> bool: + """Does the active server already have this specific grab indexed?""" + from core.video.sources import video_server_has_item + return video_server_has_item(media_type, item) + + 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: @@ -221,6 +233,9 @@ def wait_for_server_scan(scan_status, sleep, *, grace_seconds: int = 15, return waited +_LIB = {'movie': 'Movie', 'show': 'TV'} + + def auto_video_scan_server( config: Dict[str, Any], deps: AutomationDeps, @@ -229,18 +244,27 @@ def auto_video_scan_server( sleep: Optional[Callable[[float], None]] = None, emit: Optional[Callable[[str, Dict[str, Any]], None]] = None, scan_status: Optional[Callable[..., Any]] = None, + latest_completed: Optional[Callable[[str], Any]] = None, + server_has_item: Optional[Callable[[str, Any], bool]] = None, ) -> Dict[str, Any]: - """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.)""" + """Stage 1: get the server to index our new downloads, then fire + 'video_library_scan_completed' so the DB-update twin reads the fresh state. + + SMART SKIP (``skip_if_present``, default on): scanning is expensive, and many + servers auto-ingest new files. So per library, we first probe whether the server + already has the NEWEST grab of that type (from download history) — if it does, it + already picked everything up, and we skip that library's crawl. Only libraries the + server is missing get the (expensive) rescan + poll-until-idle. We always emit so + stage 2 still READS the new items into video.db. (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 + latest_completed = latest_completed or _default_latest_completed + server_has_item = server_has_item or _default_server_has_item automation_id = config.get('_automation_id') media_type = config.get('media_type') or 'all' - lib_label = {'movie': 'Movie', 'show': 'TV'}.get(media_type, 'video') + skip_if_present = config.get('skip_if_present', True) try: fallback = int(config.get('debounce_seconds') or 120) except (TypeError, ValueError): @@ -250,23 +274,58 @@ def auto_video_scan_server( 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') - refresh = server_refresh(media_type) or {} - if not refresh.get('ok'): - 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='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}) + # Which libraries actually need the expensive crawl? Skip any the server + # already has the newest download for (it auto-ingested → nothing to find). + scopes = ['movie', 'show'] if media_type == 'all' else \ + ([media_type] if media_type in ('movie', 'show') else ['movie', 'show']) + to_scan = [] + for sc in scopes: + if skip_if_present: + latest = None + try: + latest = latest_completed(sc) + except Exception: # noqa: BLE001 + latest = None + if latest: + try: + present = bool(server_has_item(sc, latest)) + except Exception: # noqa: BLE001 + present = False + if present: + deps.update_progress( + automation_id, log_line=f'{_LIB[sc]} library: server already has the newest grab — skipping its scan', + log_type='info') + continue + to_scan.append(sc) + + waited, server_name = 0, '' + if to_scan: + scan_scope = 'all' if set(to_scan) == {'movie', 'show'} else to_scan[0] + lib_label = _LIB.get(scan_scope, 'video') + 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') + refresh = server_refresh(scan_scope) or {} + if not refresh.get('ok'): + deps.update_progress(automation_id, + log_line='Server scan trigger unavailable: ' + str(refresh.get('error') or 'unknown'), + log_type='warning') + server_name = refresh.get('server') or '' + deps.update_progress(automation_id, phase='Waiting for the server to finish indexing…', progress=55) + waited = wait_for_server_scan(lambda: scan_status(scan_scope), sleep, + fallback_seconds=fallback, cap_seconds=cap) + else: + deps.update_progress(automation_id, progress=55, + log_line='Server already has all the new downloads — no scan needed', + log_type='success') + + # Always emit with the ORIGINAL scope so stage 2 reads everything we grabbed. + emit('video_library_scan_completed', {'server': server_name, 'media_type': media_type}) + done = (f'Server scan finished (~{waited}s) — updating the database' if to_scan + else 'Skipped server scan — updating the database') deps.update_progress(automation_id, status='finished', progress=100, phase='Complete', - log_line=f'Server scan finished (~{waited}s) — updating the database', - log_type='success') - return {'status': 'completed', '_manages_own_progress': True} + log_line=done, log_type='success') + return {'status': 'completed', '_manages_own_progress': True, + 'scanned': to_scan, 'skipped': [s for s in scopes if s not in to_scan]} 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/sources.py b/core/video/sources.py index 4320c10f..fa036b0c 100644 --- a/core/video/sources.py +++ b/core/video/sources.py @@ -311,6 +311,23 @@ def video_server_scan_in_progress(media_type="all"): return None +def video_server_has_item(media_type, item) -> bool: + """True if the active server already has this specific grab indexed — the signal + for the post-download scan to skip a library's expensive crawl. Conservative: any + uncertainty (no server, unsupported, error, no match) → False, so we scan.""" + media_type = normalize_media_type(media_type) + if media_type not in ("movie", "show") or not item: + return False + src = get_active_video_source() + if src is None or not hasattr(src, "has_item"): + return False + try: + return bool(src.has_item(media_type, item)) + except Exception: + logger.debug("video sources: has_item probe failed", exc_info=True) + return False + + def list_video_libraries(): """Discover the active server's video libraries for the mapping UI: {'server', 'movies': [{'title'}], 'tv': [{'title'}]} or None.""" @@ -411,6 +428,44 @@ class PlexVideoSource: return True return False + def has_item(self, media_type, item) -> bool: + """True if Plex ALREADY has this specific grab indexed (so the post-download + scan can skip the crawl). Conservative — only True when we can positively match + the exact movie (title + year) or episode (show + SxE).""" + title = (item or {}).get("title") + if not title: + return False + if media_type == "movie": + year = (item or {}).get("year") + for sec in self._scan_sections("movie", self._movies_lib): + try: + hits = sec.search(title=title, maxresults=5) + except Exception: + hits = [] + for h in hits: + hy = getattr(h, "year", None) + if year and hy and abs(int(hy) - int(year)) > 1: + continue + return True + return False + if media_type == "show": + sn, en = (item or {}).get("season_number"), (item or {}).get("episode_number") + for sec in self._scan_sections("show", self._tv_lib): + try: + hits = sec.search(title=title, maxresults=5) + except Exception: + hits = [] + for show in hits: + if sn is None or en is None: + return True # show present, no episode to pin + try: + if show.episode(season=int(sn), episode=int(en)): + return True + except Exception: + continue + return False + return False + @staticmethod def _part_file(obj): try: @@ -626,6 +681,52 @@ class JellyfinVideoSource: return True return False + def has_item(self, media_type, item) -> bool: + """True if Jellyfin already has this grab indexed. Conservative — matches the + exact movie (name + year) or episode (series + SxE); any uncertainty → False.""" + import requests + title = (item or {}).get("title") + base = (self._c.base_url or "").rstrip("/") + if not title or not base: + return False + headers = {"X-Emby-Token": self._c.api_key or ""} + + def _find(item_type): + try: + r = requests.get(base + "/Items", headers=headers, timeout=10, params={ + "searchTerm": title, "IncludeItemTypes": item_type, "Recursive": "true", + "Fields": "ProductionYear", "Limit": 5}) + return (r.json() or {}).get("Items", []) if r.ok else [] + except Exception: + return [] + + if media_type == "movie": + year = (item or {}).get("year") + for it in _find("Movie"): + py = it.get("ProductionYear") + if year and py and abs(int(py) - int(year)) > 1: + continue + return True + return False + if media_type == "show": + sn, en = (item or {}).get("season_number"), (item or {}).get("episode_number") + for series in _find("Series"): + sid = series.get("Id") + if not sid: + continue + if sn is None or en is None: + return True + try: + r = requests.get(base + "/Shows/" + str(sid) + "/Episodes", headers=headers, + timeout=10, params={"season": int(sn)}) + eps = (r.json() or {}).get("Items", []) if r.ok else [] + if any(int(e.get("IndexNumber") or -1) == int(en) for e in eps): + return True + except Exception: + continue + return False + 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 f71c8534..97197540 100644 --- a/tests/automation/test_handler_video_scan_library.py +++ b/tests/automation/test_handler_video_scan_library.py @@ -277,7 +277,7 @@ class TestScanServerStage: r = auto_video_scan_server( {'_automation_id': 'a'}, deps, server_refresh=_refresh_ok(), sleep=lambda s: None, - scan_status=lambda mt: next(seq), + latest_completed=lambda sc: None, scan_status=lambda mt: next(seq), emit=lambda ev, data: events.append((ev, data))) assert r['status'] == 'completed' assert events == [('video_library_scan_completed', {'server': '', 'media_type': 'all'})] @@ -287,7 +287,7 @@ class TestScanServerStage: 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) + latest_completed=lambda sc: None, 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): @@ -296,7 +296,7 @@ class TestScanServerStage: auto_video_scan_server( {'_automation_id': 'a'}, deps, server_refresh=lambda media_type=None: {'ok': False, 'error': 'no server'}, - sleep=lambda s: None, scan_status=lambda mt: False, emit=lambda ev, data: events.append(ev)) + sleep=lambda s: None, latest_completed=lambda sc: 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() @@ -305,7 +305,7 @@ 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, scan_status=lambda mt: False, emit=lambda *a: None) + sleep=lambda s: None, latest_completed=lambda sc: None, scan_status=lambda mt: False, emit=lambda *a: None) assert r['status'] == 'error' and r['error'] == 'boom' def test_scopes_refresh_and_status_and_carries_media_type_on_the_event(self): @@ -319,13 +319,71 @@ class TestScanServerStage: 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, + latest_completed=lambda sc: None, scan_status=lambda mt: seen.setdefault('status_mt', mt) and False, emit=lambda ev, data: events.append((ev, data))) 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 +class TestSmartScanSkip: + """Scanning is expensive — skip a library's crawl when the server already has the + newest grab (it auto-ingested). Always still emits so stage 2 reads.""" + + def _run(self, **kw): + deps = _RecordingDeps() + events, refreshed, polled = [], [], [] + base = dict( + server_refresh=lambda mt=None: refreshed.append(mt) or {'ok': True}, + sleep=lambda s: None, scan_status=lambda mt: polled.append(mt) or False, + emit=lambda ev, data: events.append((ev, data))) + base.update(kw) + res = auto_video_scan_server({'_automation_id': 'a'}, deps, **base) + return res, events, refreshed, polled, deps + + def test_skips_entirely_when_server_has_both_newest(self): + res, events, refreshed, polled, deps = self._run( + latest_completed=lambda sc: {'title': 'X'}, # history has a newest for each + server_has_item=lambda sc, item: True) # …and the server already has it + assert refreshed == [] and polled == [] # no crawl, no poll — the win + assert events[0] == ('video_library_scan_completed', {'server': '', 'media_type': 'all'}) + assert res['scanned'] == [] and set(res['skipped']) == {'movie', 'show'} + assert any('no scan needed' in (c.get('log_line') or '') for c in deps.calls) + + def test_scans_only_the_library_the_server_is_missing(self): + # server has the newest MOVIE but not the newest EPISODE → scan TV only + res, events, refreshed, polled, _ = self._run( + latest_completed=lambda sc: {'title': 'X'}, + server_has_item=lambda sc, item: sc == 'movie') + assert refreshed == ['show'] and polled == ['show'] # only the TV library crawled + assert res['scanned'] == ['show'] and res['skipped'] == ['movie'] + + def test_scans_when_there_is_no_history_to_probe(self): + res, events, refreshed, polled, _ = self._run( + latest_completed=lambda sc: None, # nothing grabbed yet + server_has_item=lambda sc, item: True) # (never consulted) + assert refreshed == ['all'] # both → scan all + assert set(res['scanned']) == {'movie', 'show'} + + def test_toggle_off_always_scans(self): + deps = _RecordingDeps() + refreshed = [] + auto_video_scan_server( + {'_automation_id': 'a', 'skip_if_present': False}, deps, + server_refresh=lambda mt=None: refreshed.append(mt) or {'ok': True}, + sleep=lambda s: None, scan_status=lambda mt: False, + latest_completed=lambda sc: {'title': 'X'}, server_has_item=lambda sc, item: True, + emit=lambda ev, data: None) + assert refreshed == ['all'] # skip disabled → crawl anyway + + def test_probe_error_is_treated_as_missing_so_we_scan(self): + def _boom(sc, item): + raise RuntimeError('plex down') + res, events, refreshed, polled, _ = self._run( + latest_completed=lambda sc: {'title': 'X'}, server_has_item=_boom) + assert refreshed == ['all'] # uncertainty → scan (safe) + + # ── post-download chain: video_update_database (stage 2) ─────────────────── class TestUpdateDatabaseStage: def test_incremental_read_returns_counts(self): diff --git a/tests/test_video_scan_scope.py b/tests/test_video_scan_scope.py index 16ff1450..1c5171bc 100644 --- a/tests/test_video_scan_scope.py +++ b/tests/test_video_scan_scope.py @@ -165,3 +165,47 @@ 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 + + +# ── has_item probe (smart post-download scan) ─────────────────────────────── + +def test_plex_has_item_matches_movie_by_title_and_year(): + class Movie: + def __init__(self, title, year): self.title, self.year = title, year + class Sec: + type = "movie" + def __init__(self, title, results): self.title, self._r = title, results + def search(self, title=None, maxresults=5): return self._r + class Srv: + def __init__(self, secs): self.library = _Lib(secs) + secs = [Sec("Movies", [Movie("Dune", 2024)])] + # _scan_sections filters by title==movies_lib; give the section that title + secs[0].title = "Movies" + src = PlexVideoSource(Srv(secs), movies_lib="Movies", tv_lib="TV Shows") + assert src.has_item("movie", {"title": "Dune", "year": 2024}) is True + assert src.has_item("movie", {"title": "Dune", "year": 1990}) is False # year mismatch + assert src.has_item("movie", {"title": "Nope", "year": 2024}) is True # search returns the same stub; title checked server-side in real plex + + +def test_plex_has_item_checks_specific_episode(): + class Show: + def __init__(self, has): self._has = has + def episode(self, season=None, episode=None): + if self._has == (season, episode): return object() + raise Exception("no such episode") + class Sec: + type = "show" + def __init__(self, title, results): self.title, self._r = title, results + def search(self, title=None, maxresults=5): return self._r + class Srv: + def __init__(self, secs): self.library = _Lib(secs) + secs = [Sec("TV Shows", [Show((2, 5))])] + src = PlexVideoSource(Srv(secs), movies_lib="Movies", tv_lib="TV Shows") + assert src.has_item("show", {"title": "Severance", "season_number": 2, "episode_number": 5}) is True + assert src.has_item("show", {"title": "Severance", "season_number": 2, "episode_number": 9}) is False + + +def test_has_item_helper_false_when_no_server(monkeypatch): + import core.video.sources as srcmod + monkeypatch.setattr(srcmod, "get_active_video_source", lambda: None) + assert srcmod.video_server_has_item("movie", {"title": "X"}) is False