diff --git a/CHANGELOG.md b/CHANGELOG.md index 3469639..5a6f1c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## 0.46.0 - 2026-06-14 + +- Changed manual Jellyfin handoff to run as a persisted background job instead of a single blocking request, so the Config page can show live progress while files are being moved. +- Added a Jellyfin sync status API plus restart-safe SQLite-backed job tracking with per-entry and per-file progress details for current and recent handoff runs. +- Updated Jellyfin moves to transfer files individually into the target library, which keeps progress counts accurate while preserving the existing final folder layout. + ## 0.45.18 - 2026-06-13 - Added a `Source name` field to the watchlist auto-download editor so per-show jobs can send a more specific anime title to `ani-cli` when provider searches return multiple similarly named results. diff --git a/README.md b/README.md index fa809d6..b081605 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ Local web UI for a system-wide `ani-cli` install. -Current version: `0.45.18` +Current version: `0.46.0` ## What it does @@ -12,7 +12,7 @@ Current version: `0.45.18` - Expose a small JSON summary endpoint for Homepage dashboard widgets - Refresh watchlist episode counts manually or on a schedule - Auto-download missing episodes for `Watching` entries after later refreshes, with a per-show `Source name` override for `ani-cli` search -- Move fully completed libraries into Jellyfin TV or movie folders +- Move fully completed libraries into Jellyfin TV or movie folders with a tracked background handoff job - Send optional Discord webhook notifications ## Pages @@ -20,7 +20,7 @@ Current version: `0.45.18` - `/`: Search and queue downloads, with poster-card search results below the selection box and alternate English titles under the main name when available - `/queue`: Monitor jobs, inspect logs, retry failed jobs, remove finished jobs - `/watchlist`: Track shows, refresh them, download all available episodes, and manage auto-download settings such as `Source name`, library `Name`, season, and episode offset -- `/config`: Save defaults, schedule refreshes, configure Jellyfin and Discord, view runtime info and the changelog +- `/config`: Save defaults, schedule refreshes, configure Jellyfin and Discord, monitor Jellyfin handoff progress, and view runtime info and the changelog ## Quick start @@ -104,7 +104,8 @@ Docker notes: - Optional and configured from `/config` - TV libraries move only after the show is finished and download-complete - Movie libraries move after download completion -- Manual `Run now` scans the watchlist and moves anything already eligible +- Manual `Run now` scans the watchlist in a background job and shows live progress for processed entries, moved files, and the current destination path +- Recent Jellyfin handoff jobs survive page reloads and show their latest status after app restarts ## Homepage widget API diff --git a/VERSION b/VERSION index 349248c..3010923 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.45.18 +0.46.0 diff --git a/app.py b/app.py index 9e2d5f5..5f0d4f1 100644 --- a/app.py +++ b/app.py @@ -66,7 +66,7 @@ from app_support import ( write_json, ) from http_handler import HandlerContext, build_handler_class, server_host, server_port -from queue_jobs import DownloadQueue, WatchlistAutoRefreshScheduler, WatchlistRefreshJobs +from queue_jobs import DownloadQueue, JellyfinSyncJobs, WatchlistAutoRefreshScheduler, WatchlistRefreshJobs from title_matching import ( base_title_variants, normalize_title_key, @@ -92,6 +92,7 @@ WATCHLIST = None DOWNLOAD_QUEUE = None WATCHLIST_REFRESH = None WATCHLIST_AUTO_REFRESH = None +WATCHLIST_JELLYFIN_SYNC = None def graph_request(query, variables, timeout=25): @@ -1042,15 +1043,28 @@ def watchlist_item_download_complete(item, mode=""): return True, "ready" -def _move_tree_contents(source_dir, target_dir): +def _move_tree_contents(source_dir, target_dir, progress_fn=None): moved = [] - for source_path in sorted(source_dir.rglob("*"), key=lambda path: (len(path.parts), str(path))): - if source_path.is_dir(): - continue + file_paths = [ + source_path + for source_path in sorted(source_dir.rglob("*"), key=lambda path: (len(path.parts), str(path))) + if source_path.is_file() + ] + total_files = len(file_paths) + for index, source_path in enumerate(file_paths, start=1): relative = source_path.relative_to(source_dir) destination = target_dir / relative destination.parent.mkdir(parents=True, exist_ok=True) final_destination = app_support.unique_destination(destination) + if progress_fn is not None: + progress_fn( + { + "file_index": index, + "file_total": total_files, + "source": str(source_path), + "current_file": str(final_destination), + } + ) shutil.move(str(source_path), str(final_destination)) moved.append(str(final_destination)) for path in sorted(source_dir.rglob("*"), key=lambda path: len(path.parts), reverse=True): @@ -1085,7 +1099,7 @@ def jellyfin_target_looks_complete(item, target_dir): return len(files) >= expected_count -def move_watchlist_item_to_jellyfin(item, config): +def move_watchlist_item_to_jellyfin(item, config, progress_fn=None): source_dir = watchlist_source_library_dir(item, config) target_dir = jellyfin_target_library_dir(item, config) if target_dir is None: @@ -1101,11 +1115,8 @@ def move_watchlist_item_to_jellyfin(item, config): return {"moved": False, "reason": reason, "item": item, "source": str(source_dir), "target": str(target_dir)} return {"moved": False, "reason": "source_missing", "item": item, "source": str(source_dir), "target": str(target_dir)} target_dir.parent.mkdir(parents=True, exist_ok=True) - if not target_dir.exists(): - shutil.move(str(source_dir), str(target_dir)) - moved_files = [str(path) for path in target_dir.rglob("*") if path.is_file()] - else: - moved_files = _move_tree_contents(source_dir, target_dir) + target_dir.mkdir(parents=True, exist_ok=True) + moved_files = _move_tree_contents(source_dir, target_dir, progress_fn=progress_fn) return { "moved": True, "reason": "moved", @@ -2558,7 +2569,7 @@ def queue_watchlist_auto_download(item, refresh_source="manual"): return {"queued": True, "mode": mode, "quality": quality, "episodes": specs, "jobs": jobs} -def trigger_jellyfin_sync_for_item(item, automatic=False, config=None): +def trigger_jellyfin_sync_for_item(item, automatic=False, config=None, progress_fn=None): active_config = normalize_config(config or get_config_snapshot()) if automatic and not bool(active_config.get("jellyfin_sync_enabled")): return {"moved": False, "reason": "disabled", "item": item} @@ -2575,7 +2586,7 @@ def trigger_jellyfin_sync_for_item(item, automatic=False, config=None): debug_log("discord_webhook.jellyfin_sync_failed", show_id=item.get("show_id"), reason=result.get("reason"), error=exc) return result try: - result = move_watchlist_item_to_jellyfin(item, active_config) + result = move_watchlist_item_to_jellyfin(item, active_config, progress_fn=progress_fn) except Exception as exc: result = { "moved": False, @@ -2632,6 +2643,16 @@ def run_jellyfin_sync(config=None): } +def start_jellyfin_sync(config=None): + ensure_runtime() + return WATCHLIST_JELLYFIN_SYNC.start(config=config) + + +def get_jellyfin_sync_status(): + ensure_runtime() + return WATCHLIST_JELLYFIN_SYNC.status() + + def update_watchlist_category(show_id, category): ensure_runtime() item = WATCHLIST.update_category(show_id, category) @@ -2778,6 +2799,7 @@ def runtime_state(): "download_queue": DOWNLOAD_QUEUE, "watchlist_refresh": WATCHLIST_REFRESH, "watchlist_auto_refresh": WATCHLIST_AUTO_REFRESH, + "watchlist_jellyfin_sync": WATCHLIST_JELLYFIN_SYNC, } @@ -2800,6 +2822,12 @@ def build_runtime(start_workers=None): config_getter=lambda: get_config_snapshot(config), start_worker=worker_state, ) + watchlist_jellyfin_sync = JellyfinSyncJobs( + watchlist, + sync_fn=trigger_jellyfin_sync_for_item, + config_getter=lambda: get_config_snapshot(config), + start_worker=worker_state, + ) watchlist_auto_refresh = WatchlistAutoRefreshScheduler( config_getter=lambda: get_config_snapshot(config), refresh_start_fn=watchlist_refresh.start, @@ -2811,11 +2839,12 @@ def build_runtime(start_workers=None): "download_queue": download_queue, "watchlist_refresh": watchlist_refresh, "watchlist_auto_refresh": watchlist_auto_refresh, + "watchlist_jellyfin_sync": watchlist_jellyfin_sync, } def initialize_runtime(start_workers=None): - global CONFIG, WATCHLIST, DOWNLOAD_QUEUE, WATCHLIST_REFRESH, WATCHLIST_AUTO_REFRESH + global CONFIG, WATCHLIST, DOWNLOAD_QUEUE, WATCHLIST_REFRESH, WATCHLIST_AUTO_REFRESH, WATCHLIST_JELLYFIN_SYNC with RUNTIME_LOCK: workers_allowed = background_workers_enabled() if ( @@ -2824,6 +2853,7 @@ def initialize_runtime(start_workers=None): and DOWNLOAD_QUEUE is not None and WATCHLIST_REFRESH is not None and WATCHLIST_AUTO_REFRESH is not None + and WATCHLIST_JELLYFIN_SYNC is not None ): if start_workers and workers_allowed: WATCHLIST.worker_enabled = True @@ -2831,6 +2861,7 @@ def initialize_runtime(start_workers=None): DOWNLOAD_QUEUE.ensure_worker() WATCHLIST_REFRESH.ensure_worker() WATCHLIST_AUTO_REFRESH.ensure_worker() + WATCHLIST_JELLYFIN_SYNC.ensure_worker() return runtime_state() runtime = build_runtime(start_workers=start_workers) CONFIG = runtime["config"] @@ -2838,6 +2869,7 @@ def initialize_runtime(start_workers=None): DOWNLOAD_QUEUE = runtime["download_queue"] WATCHLIST_REFRESH = runtime["watchlist_refresh"] WATCHLIST_AUTO_REFRESH = runtime["watchlist_auto_refresh"] + WATCHLIST_JELLYFIN_SYNC = runtime["watchlist_jellyfin_sync"] return runtime @@ -2848,6 +2880,7 @@ def ensure_runtime(start_workers=False): or DOWNLOAD_QUEUE is None or WATCHLIST_REFRESH is None or WATCHLIST_AUTO_REFRESH is None + or WATCHLIST_JELLYFIN_SYNC is None ): return initialize_runtime(start_workers=start_workers) if start_workers: @@ -2864,10 +2897,11 @@ def shutdown_runtime(wait=False, cancel_active_downloads=None): and DOWNLOAD_QUEUE is None and WATCHLIST_REFRESH is None and WATCHLIST_AUTO_REFRESH is None + and WATCHLIST_JELLYFIN_SYNC is None ): return False - services = (WATCHLIST, DOWNLOAD_QUEUE, WATCHLIST_REFRESH, WATCHLIST_AUTO_REFRESH) - watchlist, download_queue, watchlist_refresh, watchlist_auto_refresh = services + services = (WATCHLIST, DOWNLOAD_QUEUE, WATCHLIST_REFRESH, WATCHLIST_AUTO_REFRESH, WATCHLIST_JELLYFIN_SYNC) + watchlist, download_queue, watchlist_refresh, watchlist_auto_refresh, watchlist_jellyfin_sync = services should_cancel_downloads = bool(wait) if cancel_active_downloads is None else bool(cancel_active_downloads) stopped = True if watchlist is not None: @@ -2878,11 +2912,13 @@ def shutdown_runtime(wait=False, cancel_active_downloads=None): stopped = watchlist_refresh.shutdown(wait=wait) and stopped if watchlist_auto_refresh is not None: stopped = watchlist_auto_refresh.shutdown(wait=wait) and stopped + if watchlist_jellyfin_sync is not None: + stopped = watchlist_jellyfin_sync.shutdown(wait=wait) and stopped return stopped def reset_runtime(wait=False, cancel_active_downloads=None): - global CONFIG, WATCHLIST, DOWNLOAD_QUEUE, WATCHLIST_REFRESH, WATCHLIST_AUTO_REFRESH + global CONFIG, WATCHLIST, DOWNLOAD_QUEUE, WATCHLIST_REFRESH, WATCHLIST_AUTO_REFRESH, WATCHLIST_JELLYFIN_SYNC stopped = shutdown_runtime(wait=wait, cancel_active_downloads=cancel_active_downloads) if not stopped: if wait: @@ -2894,6 +2930,7 @@ def reset_runtime(wait=False, cancel_active_downloads=None): DOWNLOAD_QUEUE = None WATCHLIST_REFRESH = None WATCHLIST_AUTO_REFRESH = None + WATCHLIST_JELLYFIN_SYNC = None return True @@ -2906,6 +2943,7 @@ Handler = build_handler_class( episode_list=episode_list, get_config_snapshot=get_config_snapshot, get_watchlist_homepage_summary=get_watchlist_homepage_summary, + get_jellyfin_sync_status=get_jellyfin_sync_status, get_watchlist=get_watchlist, get_watchlist_refresh_status=get_watchlist_refresh_status, http_error=HttpError, @@ -2919,6 +2957,7 @@ Handler = build_handler_class( resolve_search_thumbnail=resolve_search_thumbnail, test_discord_webhook_config=test_discord_webhook_config, thumbnail_file_path=thumbnail_file_path, + start_jellyfin_sync=start_jellyfin_sync, update_all_watchlist_statuses=update_all_watchlist_statuses, update_watchlist_auto_download=update_watchlist_auto_download, update_watchlist_category=update_watchlist_category, diff --git a/config_page.py b/config_page.py index b911009..caf044c 100644 --- a/config_page.py +++ b/config_page.py @@ -591,6 +591,7 @@ CONFIG_HTML = r"""
TV entries move only after the show is marked `Finished` and at least one downloaded language covers the expected episode count. Movie entries move after they are downloaded. The manual trigger scans the current watchlist and moves anything already eligible.
+
No recent Jellyfin handoff job.
@@ -683,7 +684,9 @@ CONFIG_HTML = r""" path: "", parent_path: "", home_path: "" - } + }, + jellyfinSyncRunning: false, + jellyfinSyncJobId: null }; const $ = (id) => document.getElementById(id); @@ -700,6 +703,26 @@ CONFIG_HTML = r""" $("notice").textContent = text || ""; } + function setJellyfinSyncButton(running) { + $("runJellyfinSyncBtn").disabled = !!running; + $("runJellyfinSyncBtn").textContent = running ? "Running..." : "Run now"; + } + + function setJellyfinSyncStatus(job) { + const el = $("jellyfinSyncStatus"); + if (!job) { + el.textContent = "No recent Jellyfin handoff job."; + return; + } + if (job.status === "running") { + const title = job.current_title ? ` · ${job.current_title}` : ""; + const file = job.current_file ? ` · ${job.current_file}` : ""; + el.textContent = `${job.completed || 0}/${job.total || 0} processed · ${job.moved || 0} moved · ${job.moved_files || 0} files${title}${file}`; + return; + } + el.textContent = job.message || "Jellyfin handoff idle."; + } + function selectedWebhookEvents() { return [...document.querySelectorAll(".webhook-event:checked")].map((input) => input.value); } @@ -772,13 +795,38 @@ CONFIG_HTML = r""" } } + async function pollJellyfinSyncStatus(notifyOnComplete = true) { + try { + const data = await api("/api/config/jellyfin/sync-status"); + const job = data.job || null; + const running = !!data.running && !!job; + const wasRunning = state.jellyfinSyncRunning; + state.jellyfinSyncRunning = running; + state.jellyfinSyncJobId = job ? job.id : null; + setJellyfinSyncButton(running); + setJellyfinSyncStatus(job); + if (running) { + setNotice(job.message || "Jellyfin handoff running..."); + } else if (job && wasRunning && notifyOnComplete) { + setNotice(job.message || "Jellyfin handoff completed."); + } + } catch (error) { + setNotice(error.message); + } + } + async function runJellyfinSync() { try { const data = await api("/api/config/jellyfin/sync", { method: "POST", body: JSON.stringify(formConfigPayload()) }); - setNotice(data.message || "Jellyfin handoff completed."); + const job = data.job || null; + state.jellyfinSyncRunning = !!job && (job.status === "pending" || job.status === "running"); + state.jellyfinSyncJobId = job ? job.id : null; + setJellyfinSyncButton(state.jellyfinSyncRunning); + setJellyfinSyncStatus(job); + setNotice(data.message || "Jellyfin handoff started."); } catch (error) { setNotice(error.message); } @@ -961,8 +1009,12 @@ CONFIG_HTML = r""" (async function init() { try { await loadConfig(); + await pollJellyfinSyncStatus(false); await loadDeps(); await loadVersion(); + window.setInterval(() => { + pollJellyfinSyncStatus(); + }, 2500); } catch (error) { setNotice(error.message); } diff --git a/http_handler.py b/http_handler.py index 509cd57..afb6eb2 100644 --- a/http_handler.py +++ b/http_handler.py @@ -52,6 +52,7 @@ class HandlerContext: ensure_runtime: object episode_list: object get_config_snapshot: object + get_jellyfin_sync_status: object get_watchlist_homepage_summary: object get_watchlist: object get_watchlist_refresh_status: object @@ -64,6 +65,7 @@ class HandlerContext: save_runtime_config: object search_anime: object resolve_search_thumbnail: object + start_jellyfin_sync: object test_discord_webhook_config: object thumbnail_file_path: object update_all_watchlist_statuses: object @@ -168,7 +170,7 @@ class Handler(BaseHTTPRequestHandler): server_version = "AniCliWeb/1.0" remote_session_cookie_name = "ani_cli_web_session" remote_session_cookie_max_age_seconds = 30 * 24 * 60 * 60 - quiet_poll_paths = {"/api/queue", "/api/watchlist/refresh-status"} + quiet_poll_paths = {"/api/queue", "/api/watchlist/refresh-status", "/api/config/jellyfin/sync-status"} def _context(self): context = getattr(self, "handler_context", None) or getattr(type(self), "handler_context", None) @@ -616,6 +618,9 @@ class Handler(BaseHTTPRequestHandler): elif parsed.path == "/api/watchlist/refresh-status": Handler._runtime(self) self.json(Handler._context(self).get_watchlist_refresh_status()) + elif parsed.path == "/api/config/jellyfin/sync-status": + Handler._runtime(self) + self.json(Handler._context(self).get_jellyfin_sync_status()) else: self.error(HTTPStatus.NOT_FOUND, "Not found") except Exception as exc: @@ -656,7 +661,7 @@ class Handler(BaseHTTPRequestHandler): ): if str(merged.get(key) or "").strip(): Handler.validate_remote_path(self, merged[key], label) - self.json(Handler._context(self).run_jellyfin_sync(merged)) + self.json(Handler._context(self).start_jellyfin_sync(merged), HTTPStatus.ACCEPTED) elif parsed.path == "/api/config/webhook/test": payload = Handler.require_json_object(self, self.body_json()) validate_discord_webhook_url(payload.get("discord_webhook_url")) diff --git a/queue_jobs.py b/queue_jobs.py index 21fb18c..7817624 100644 --- a/queue_jobs.py +++ b/queue_jobs.py @@ -25,6 +25,7 @@ from app_support import ( job_output_dir, job_staging_dir, load_json, + normalize_config, now_iso, printable_command, send_discord_webhook_event, @@ -518,6 +519,326 @@ class WatchlistAutoRefreshScheduler: self.wakeup.clear() +class JellyfinSyncJobs: + CONFIG_KEYS = ( + "download_dir", + "jellyfin_sync_enabled", + "jellyfin_tv_dir", + "jellyfin_movie_dir", + "discord_webhook_url", + "discord_webhook_events", + ) + + def __init__(self, store, sync_fn, config_getter=None, start_worker=True): + self.store = store + self.sync_fn = sync_fn + self.config_getter = config_getter or (lambda: {}) + self.lock = threading.RLock() + self.wakeup = threading.Event() + self.stop_event = threading.Event() + self.pending = [] + self.jobs = [] + self.current_job_id = None + self._init_db() + self._restore_interrupted_jobs() + self._load_jobs() + self.worker = None + if start_worker: + self.ensure_worker() + + def _connect(self): + conn = sqlite3.connect(STATE_DB_PATH, timeout=30) + conn.row_factory = sqlite3.Row + return conn + + def ensure_worker(self): + with self.lock: + if self.worker is not None and self.worker.is_alive(): + return + self.stop_event.clear() + self.worker = threading.Thread(target=self._run, daemon=True) + self.worker.start() + + def shutdown(self, wait=False): + self.stop_event.set() + self.wakeup.set() + worker = self.worker + if wait and worker is not None and worker.is_alive(): + worker.join(timeout=WORKER_SHUTDOWN_TIMEOUT_SECONDS) + return worker is None or not worker.is_alive() + + def _init_db(self): + STATE_DB_PATH.parent.mkdir(parents=True, exist_ok=True) + with self._connect() as conn: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS jellyfin_sync_jobs ( + id TEXT PRIMARY KEY, + status TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + payload TEXT NOT NULL + ) + """ + ) + conn.execute("CREATE INDEX IF NOT EXISTS idx_jellyfin_sync_created_at ON jellyfin_sync_jobs(created_at)") + conn.execute("CREATE INDEX IF NOT EXISTS idx_jellyfin_sync_status ON jellyfin_sync_jobs(status)") + + def _job_from_row(self, row): + payload = row["payload"] if isinstance(row, sqlite3.Row) else row[0] + return json.loads(payload) + + def _upsert_job_conn(self, conn, job): + conn.execute( + """ + INSERT INTO jellyfin_sync_jobs (id, status, created_at, updated_at, payload) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(id) DO UPDATE SET + status = excluded.status, + updated_at = excluded.updated_at, + payload = excluded.payload + """, + ( + job["id"], + job["status"], + job["created_at"], + job["updated_at"], + json.dumps(job, sort_keys=True), + ), + ) + + def _trim_history_conn(self, conn): + rows = conn.execute( + "SELECT id FROM jellyfin_sync_jobs ORDER BY created_at DESC LIMIT -1 OFFSET 10" + ).fetchall() + if rows: + conn.executemany("DELETE FROM jellyfin_sync_jobs WHERE id = ?", [(row["id"],) for row in rows]) + + def _save_job_locked(self, job): + with self._connect() as conn: + self._upsert_job_conn(conn, job) + self._trim_history_conn(conn) + + def _load_jobs(self): + with self.lock, self._connect() as conn: + rows = conn.execute( + "SELECT payload FROM jellyfin_sync_jobs ORDER BY created_at DESC LIMIT 10" + ).fetchall() + self.jobs = [self._job_from_row(row) for row in rows] + + def _restore_interrupted_jobs(self): + with self.lock, self._connect() as conn: + rows = conn.execute( + "SELECT payload FROM jellyfin_sync_jobs WHERE status IN ('pending', 'running')" + ).fetchall() + for row in rows: + job = self._job_from_row(row) + job["status"] = "interrupted" + job["finished_at"] = now_iso() + job["updated_at"] = job["finished_at"] + job["current_title"] = None + job["current_file"] = None + job["message"] = "Interrupted by web app restart." + self._upsert_job_conn(conn, job) + self._trim_history_conn(conn) + + def _copy_job(self, job): + return json.loads(json.dumps(job)) + + def _find_locked(self, job_id): + for job in self.jobs: + if job["id"] == job_id: + return job + return None + + def _job_config(self, config): + normalized = normalize_config(config or self.config_getter() or {}) + return {key: normalized.get(key) for key in self.CONFIG_KEYS} + + def status(self): + with self.lock: + job = next((candidate for candidate in self.jobs if candidate["status"] in {"pending", "running"}), None) + if job is None: + job = self.jobs[0] if self.jobs else None + active = bool(job and job["status"] in {"pending", "running"}) + return {"job": self._copy_job(job) if job else None, "running": active} + + def start(self, config=None, source="manual"): + source_name = str(source or "manual").strip().lower() or "manual" + with self.lock: + job = next((candidate for candidate in self.jobs if candidate["status"] in {"pending", "running"}), None) + if job is not None: + return { + "message": "Jellyfin handoff already queued." if job["status"] == "pending" else "Jellyfin handoff already running.", + "job": self._copy_job(job), + "created": False, + } + now = now_iso() + job = { + "id": os.urandom(16).hex(), + "status": "pending", + "source": source_name, + "config": self._job_config(config), + "created_at": now, + "updated_at": now, + "started_at": None, + "finished_at": None, + "total": 0, + "completed": 0, + "moved": 0, + "errors": 0, + "moved_files": 0, + "current_title": None, + "current_file": None, + "message": "Queued Jellyfin handoff.", + "items": [], + } + self.jobs.insert(0, job) + self.jobs = self.jobs[:10] + self._save_job_locked(job) + self.pending.append(job["id"]) + self.wakeup.set() + return {"message": "Jellyfin handoff started.", "job": self._copy_job(job), "created": True} + + def _next_pending(self): + with self.lock: + while self.pending: + job_id = self.pending.pop(0) + job = self._find_locked(job_id) + if job is not None: + self.current_job_id = job_id + return job + return None + + def _run(self): + while not self.stop_event.is_set(): + job = self._next_pending() + if not job: + self.wakeup.wait(2) + self.wakeup.clear() + continue + self._run_job(job) + + def _run_job(self, job): + interrupted = False + result_items = [] + try: + active_config = self._job_config(job.get("config") or self.config_getter() or {}) + show_ids = self.store.all_show_ids() + started_at = now_iso() + with self.lock: + job["status"] = "running" + job["started_at"] = started_at + job["updated_at"] = started_at + job["total"] = len(show_ids) + job["completed"] = 0 + job["moved"] = 0 + job["errors"] = 0 + job["moved_files"] = 0 + job["current_title"] = None + job["current_file"] = None + job["items"] = [] + job["message"] = f"Moving 0/{len(show_ids)} watchlist entries to Jellyfin." + self._save_job_locked(job) + + for index, show_id in enumerate(show_ids, start=1): + if self.stop_event.is_set(): + interrupted = True + break + try: + item = self.store.get(show_id) + except KeyError: + item = None + if item is None: + with self.lock: + job["completed"] = index + job["updated_at"] = now_iso() + job["message"] = f"Moving {index}/{len(show_ids)}: removed entry" + self._save_job_locked(job) + continue + + title = str(item.get("title") or show_id) + + def progress(update): + with self.lock: + job["current_title"] = title + job["current_file"] = update.get("current_file") + job["updated_at"] = now_iso() + file_total = int(update.get("file_total") or 0) + file_index = int(update.get("file_index") or 0) + if file_total > 0: + job["message"] = ( + f"Moving {index}/{len(show_ids)}: {title} " + f"({file_index}/{file_total} files)" + ) + else: + job["message"] = f"Moving {index}/{len(show_ids)}: {title}" + self._save_job_locked(job) + + result = self.sync_fn(item, automatic=False, config=active_config, progress_fn=progress) + result_items.append( + { + "show_id": item.get("show_id"), + "title": item.get("title"), + "media_type": item.get("media_type"), + "moved": bool(result.get("moved")), + "reason": result.get("reason"), + "target": result.get("target"), + } + ) + moved_files = len(result.get("moved_files") or []) if isinstance(result.get("moved_files"), list) else 0 + errored = str(result.get("reason") or "").strip() in {"error", "target_missing", "source_missing", "target_incomplete"} + with self.lock: + job["completed"] = index + job["moved"] += 1 if result.get("moved") else 0 + job["moved_files"] += moved_files + job["errors"] += 1 if errored else 0 + job["current_title"] = title + job["current_file"] = None + job["updated_at"] = now_iso() + job["items"] = result_items[-20:] + if errored and result.get("error"): + job["last_error"] = str(result.get("error")) + outcome = "moved" if result.get("moved") else str(result.get("reason") or "skipped") + job["message"] = f"Moving {index}/{len(show_ids)}: {title} ({outcome})" + self._save_job_locked(job) + + finished_at = now_iso() + with self.lock: + job["status"] = "interrupted" if interrupted else "done" + job["finished_at"] = finished_at + job["updated_at"] = finished_at + job["current_title"] = None + job["current_file"] = None + job["items"] = result_items[-20:] + if interrupted: + job["message"] = f"Jellyfin handoff interrupted after {job['completed']} entries." + elif job["errors"]: + job["message"] = ( + f"Moved {job['moved']} of {job['completed']} processed watchlist entries to Jellyfin " + f"with {job['errors']} issues." + ) + else: + job["message"] = ( + f"Moved {job['moved']} watchlist entr{'y' if job['moved'] == 1 else 'ies'} to Jellyfin libraries." + ) + self._save_job_locked(job) + except Exception as exc: + finished_at = now_iso() + with self.lock: + job["status"] = "failed" + job["finished_at"] = finished_at + job["updated_at"] = finished_at + job["current_title"] = None + job["current_file"] = None + job["message"] = f"Jellyfin handoff failed: {exc}" + job["last_error"] = str(exc) + self._save_job_locked(job) + finally: + with self.lock: + self.current_job_id = None + + class DownloadQueue: def __init__(self, config_getter, watchlist_sync_fn=None, job_prepare_fn=None, start_worker=True): self.config_getter = config_getter diff --git a/test_app.py b/test_app.py index 711ce0a..5046a1b 100644 --- a/test_app.py +++ b/test_app.py @@ -31,7 +31,11 @@ APP = importlib.util.module_from_spec(SPEC) assert SPEC.loader is not None SPEC.loader.exec_module(APP) IMPORT_RUNTIME_WAS_DEFERRED = ( - APP.CONFIG is None and APP.WATCHLIST is None and APP.DOWNLOAD_QUEUE is None and APP.WATCHLIST_REFRESH is None + APP.CONFIG is None + and APP.WATCHLIST is None + and APP.DOWNLOAD_QUEUE is None + and APP.WATCHLIST_REFRESH is None + and APP.WATCHLIST_JELLYFIN_SYNC is None ) APP.initialize_runtime(start_workers=False) APP.reset_runtime(wait=True) @@ -2586,6 +2590,7 @@ class StartupBehaviorTests(unittest.TestCase): current_queue = APP.DOWNLOAD_QUEUE current_refresh = APP.WATCHLIST_REFRESH current_auto_refresh = APP.WATCHLIST_AUTO_REFRESH + current_jellyfin_sync = APP.WATCHLIST_JELLYFIN_SYNC with mock.patch.object(APP, "shutdown_runtime", return_value=False): with self.assertRaises(RuntimeError): APP.reset_runtime(wait=True) @@ -2595,6 +2600,7 @@ class StartupBehaviorTests(unittest.TestCase): self.assertIs(APP.DOWNLOAD_QUEUE, current_queue) self.assertIs(APP.WATCHLIST_REFRESH, current_refresh) self.assertIs(APP.WATCHLIST_AUTO_REFRESH, current_auto_refresh) + self.assertIs(APP.WATCHLIST_JELLYFIN_SYNC, current_jellyfin_sync) finally: APP.reset_runtime(wait=True) APP.initialize_runtime(start_workers=False) @@ -2663,6 +2669,7 @@ class StartupBehaviorTests(unittest.TestCase): self.assertIsNone(APP.DOWNLOAD_QUEUE) self.assertIsNone(APP.WATCHLIST_REFRESH) self.assertIsNone(APP.WATCHLIST_AUTO_REFRESH) + self.assertIsNone(APP.WATCHLIST_JELLYFIN_SYNC) finally: APP.initialize_runtime(start_workers=False) @@ -2679,6 +2686,7 @@ class HandlerRouteTests(unittest.TestCase): self.assertIsNone(APP.DOWNLOAD_QUEUE.worker) self.assertIsNone(APP.WATCHLIST_REFRESH.worker) self.assertIsNone(APP.WATCHLIST_AUTO_REFRESH.worker) + self.assertIsNone(APP.WATCHLIST_JELLYFIN_SYNC.worker) finally: APP.reset_runtime(wait=True) APP.initialize_runtime(start_workers=False) @@ -3115,23 +3123,31 @@ class HandlerRouteTests(unittest.TestCase): self.assertEqual(handler.json_payload, payload) self.assertEqual(handler.json_status, HTTPStatus.OK) - def test_jellyfin_sync_post_returns_summary_payload(self): + def test_jellyfin_sync_status_route_returns_json(self): + handler = DummyHandler("/api/config/jellyfin/sync-status") + payload = {"running": True, "job": {"id": "job-1", "status": "running"}} + handler.handler_context = mock.Mock(get_jellyfin_sync_status=mock.Mock(return_value=payload)) + APP.Handler.do_GET(handler) + self.assertEqual(handler.json_payload, payload) + self.assertEqual(handler.json_status, HTTPStatus.OK) + + def test_jellyfin_sync_post_returns_job_payload(self): body = b'{"download_dir":"/tmp/example","jellyfin_sync_enabled":true,"jellyfin_tv_dir":"/tmp/jellyfin-tv","jellyfin_movie_dir":"/tmp/jellyfin-movies"}' handler = DummyHandler("/api/config/jellyfin/sync", body=body, content_length=len(body)) - payload = {"message": "Moved 1 watchlist entry to Jellyfin libraries.", "moved": 1, "total": 2, "items": []} + payload = {"message": "Jellyfin handoff started.", "created": True, "job": {"id": "job-1", "status": "pending"}} handler.handler_context = mock.Mock( get_config_snapshot=mock.Mock(return_value={}), - run_jellyfin_sync=mock.Mock(return_value=payload), + start_jellyfin_sync=mock.Mock(return_value=payload), http_error=APP.HttpError, ) APP.Handler.do_POST(handler) self.assertEqual(handler.json_payload, payload) - self.assertEqual(handler.json_status, HTTPStatus.OK) + self.assertEqual(handler.json_status, HTTPStatus.ACCEPTED) def test_jellyfin_sync_post_merges_partial_payload_with_saved_config(self): body = b'{"jellyfin_sync_enabled":true}' handler = DummyHandler("/api/config/jellyfin/sync", body=body, content_length=len(body)) - payload = {"message": "Moved 0 watchlist entries to Jellyfin libraries.", "moved": 0, "total": 0, "items": []} + payload = {"message": "Jellyfin handoff already queued.", "created": False, "job": {"id": "job-1", "status": "pending"}} handler.handler_context = mock.Mock( get_config_snapshot=mock.Mock( return_value={ @@ -3143,11 +3159,11 @@ class HandlerRouteTests(unittest.TestCase): "jellyfin_movie_dir": "/tmp/jellyfin-movies", } ), - run_jellyfin_sync=mock.Mock(return_value=payload), + start_jellyfin_sync=mock.Mock(return_value=payload), ) APP.Handler.do_POST(handler) - self.assertEqual(handler.json_status, HTTPStatus.OK) - handler.handler_context.run_jellyfin_sync.assert_called_once_with( + self.assertEqual(handler.json_status, HTTPStatus.ACCEPTED) + handler.handler_context.start_jellyfin_sync.assert_called_once_with( { "download_dir": "/tmp/example", "mode": "sub",