#!/usr/bin/env python3 """Queue and background job services for ani-cli-web.""" import concurrent.futures import json import os import signal import sqlite3 import subprocess import threading import time from app_support import ( MAX_LOG_LINES, PROJECT_ROOT, QUEUE_PATH, STATE_DB_PATH, build_job, command_for_job, finalize_library_files, job_output_dir, job_staging_dir, load_json, now_iso, printable_command, strip_control, debug_log, ) LOG_PERSIST_INTERVAL_SECONDS = 0.75 LOG_PERSIST_LINE_BATCH = 8 WATCHLIST_REFRESH_MAX_WORKERS = 4 WORKER_JOIN_TIMEOUT_SECONDS = 2.5 JOB_STRUCTURED_COLUMNS = ( "show_id", "title", "anime_name", "season", "query", "result_index", "mode", "quality", "episodes", "download_dir", "started_at", "finished_at", "exit_code", "pid", "command", "staging_dir", "target_dir", "cancel_requested", ) JOB_LOG_COLUMN = "log_payload" JOB_EXTRA_PAYLOAD_COLUMN = "payload" class WatchlistRefreshJobs: def __init__(self, store, start_worker=True): self.store = store self.lock = threading.RLock() self.wakeup = threading.Event() self.stop_event = threading.Event() self.pending = [] self.jobs = [] self.current_job_id = None self.current_executor = 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() executor = None with self.lock: executor = self.current_executor if executor is not None: executor.shutdown(wait=False, cancel_futures=True) self.wakeup.set() worker = self.worker if wait and worker is not None and worker.is_alive(): worker.join(timeout=WORKER_JOIN_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 watchlist_refresh_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_watchlist_refresh_created_at ON watchlist_refresh_jobs(created_at)") conn.execute("CREATE INDEX IF NOT EXISTS idx_watchlist_refresh_status ON watchlist_refresh_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 watchlist_refresh_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 watchlist_refresh_jobs ORDER BY created_at DESC LIMIT -1 OFFSET 10" ).fetchall() if rows: conn.executemany("DELETE FROM watchlist_refresh_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 watchlist_refresh_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 watchlist_refresh_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["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 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): 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": "Watchlist refresh already queued." if job["status"] == "pending" else "Watchlist refresh already running.", "job": self._copy_job(job) if job else None, "created": False, } now = now_iso() job = { "id": os.urandom(16).hex(), "status": "pending", "created_at": now, "updated_at": now, "started_at": None, "finished_at": None, "total": 0, "completed": 0, "errors": 0, "current_title": None, "message": "Queued watchlist refresh.", } 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": "Watchlist refresh 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): try: 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["errors"] = 0 job["current_title"] = None job["message"] = f"Refreshing 0/{len(show_ids)} watchlist entries." self._save_job_locked(job) def refresh_one(show_id): title = show_id errored = False skipped = False error_message = "" if self.stop_event.is_set(): return {"title": title, "errored": False, "skipped": True, "error": ""} try: item = self.store.refresh(show_id, include_thumbnail=False) if item is None: skipped = True else: title = item.get("title") or title errored = item.get("status") == "error" error_message = str(item.get("status_message") or "").strip() if errored else "" except Exception as exc: errored = True error_message = str(exc) debug_log("watchlist.refresh_all.item_error", show_id=show_id, error=exc) return {"title": title, "errored": errored, "skipped": skipped, "error": error_message} max_workers = max(1, min(WATCHLIST_REFRESH_MAX_WORKERS, len(show_ids) or 1)) with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: with self.lock: self.current_executor = executor futures = [executor.submit(refresh_one, show_id) for show_id in show_ids] for index, future in enumerate(concurrent.futures.as_completed(futures), start=1): result = future.result() title = result["title"] with self.lock: job["completed"] = index job["errors"] += 1 if result["errored"] else 0 job["current_title"] = title job["updated_at"] = now_iso() if result["errored"] and result["error"]: job["last_error"] = result["error"] if result["skipped"]: job["message"] = f"Refreshing {index}/{len(show_ids)}: removed entry" else: job["message"] = f"Refreshing {index}/{len(show_ids)}: {title}" self._save_job_locked(job) with self.lock: self.current_executor = None finished_at = now_iso() with self.lock: job["status"] = "done" job["finished_at"] = finished_at job["updated_at"] = finished_at job["current_title"] = None if job["errors"]: job["message"] = ( f"Refreshed {job['completed']} watchlist entries with {job['errors']} refresh errors." ) else: job["message"] = f"Refreshed {job['completed']} watchlist entries." 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["message"] = f"Watchlist refresh failed: {exc}" self._save_job_locked(job) finally: with self.lock: self.current_executor = None 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 self.watchlist_sync_fn = watchlist_sync_fn self.job_prepare_fn = job_prepare_fn self.lock = threading.RLock() self.wakeup = threading.Event() self.stop_event = threading.Event() self.current_process = None self.current_job_id = None self.current_job = None self._init_db() self._migrate_json_queue() self._restore_interrupted_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, cancel_active=False): should_cancel = bool(cancel_active or wait) self.stop_event.set() if should_cancel: self._signal_active_process(signal.SIGTERM) self.wakeup.set() worker = self.worker if wait and worker is not None and worker.is_alive(): worker.join(timeout=WORKER_JOIN_TIMEOUT_SECONDS) if worker.is_alive(): self._signal_active_process(signal.SIGKILL) worker.join(timeout=WORKER_JOIN_TIMEOUT_SECONDS) return worker is None or not worker.is_alive() def _signal_active_process(self, signum): with self.lock: process = self.current_process job = self.current_job if process is None: return False if job is not None: job["cancel_requested"] = True try: os.killpg(os.getpgid(process.pid), signum) return True except OSError: try: if signum == signal.SIGKILL: process.kill() else: process.terminate() return True except OSError: return False 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 jobs ( id TEXT PRIMARY KEY, status TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, show_id TEXT, title TEXT, anime_name TEXT, season TEXT, query TEXT, result_index INTEGER, mode TEXT, quality TEXT, episodes TEXT, download_dir TEXT, started_at TEXT, finished_at TEXT, exit_code INTEGER, pid INTEGER, command TEXT, staging_dir TEXT, target_dir TEXT, cancel_requested INTEGER NOT NULL DEFAULT 0, log_payload TEXT NOT NULL DEFAULT '[]', payload TEXT NOT NULL DEFAULT '{}' ) """ ) columns = {row["name"] for row in conn.execute("PRAGMA table_info(jobs)").fetchall()} alter_statements = ( ("show_id", "ALTER TABLE jobs ADD COLUMN show_id TEXT"), ("title", "ALTER TABLE jobs ADD COLUMN title TEXT"), ("anime_name", "ALTER TABLE jobs ADD COLUMN anime_name TEXT"), ("season", "ALTER TABLE jobs ADD COLUMN season TEXT"), ("query", "ALTER TABLE jobs ADD COLUMN query TEXT"), ("result_index", "ALTER TABLE jobs ADD COLUMN result_index INTEGER"), ("mode", "ALTER TABLE jobs ADD COLUMN mode TEXT"), ("quality", "ALTER TABLE jobs ADD COLUMN quality TEXT"), ("episodes", "ALTER TABLE jobs ADD COLUMN episodes TEXT"), ("download_dir", "ALTER TABLE jobs ADD COLUMN download_dir TEXT"), ("started_at", "ALTER TABLE jobs ADD COLUMN started_at TEXT"), ("finished_at", "ALTER TABLE jobs ADD COLUMN finished_at TEXT"), ("exit_code", "ALTER TABLE jobs ADD COLUMN exit_code INTEGER"), ("pid", "ALTER TABLE jobs ADD COLUMN pid INTEGER"), ("command", "ALTER TABLE jobs ADD COLUMN command TEXT"), ("staging_dir", "ALTER TABLE jobs ADD COLUMN staging_dir TEXT"), ("target_dir", "ALTER TABLE jobs ADD COLUMN target_dir TEXT"), ("cancel_requested", "ALTER TABLE jobs ADD COLUMN cancel_requested INTEGER NOT NULL DEFAULT 0"), (JOB_LOG_COLUMN, "ALTER TABLE jobs ADD COLUMN log_payload TEXT NOT NULL DEFAULT '[]'"), ) for column, statement in alter_statements: if column not in columns: conn.execute(statement) conn.execute("CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status)") conn.execute("CREATE INDEX IF NOT EXISTS idx_jobs_created_at ON jobs(created_at)") def _migrate_json_queue(self): if not QUEUE_PATH.exists(): return jobs = load_json(QUEUE_PATH, []) if not jobs: QUEUE_PATH.rename(QUEUE_PATH.with_suffix(".json.migrated")) return with self.lock, self._connect() as conn: existing = conn.execute("SELECT COUNT(*) AS count FROM jobs").fetchone()["count"] if existing: return for job in jobs: if isinstance(job, dict) and job.get("id"): self._upsert_job_conn(conn, job) QUEUE_PATH.rename(QUEUE_PATH.with_suffix(".json.migrated")) def _restore_interrupted_jobs(self): with self.lock, self._connect() as conn: rows = conn.execute("SELECT * FROM jobs WHERE status = 'running'").fetchall() for row in rows: job = self._job_from_row(row) job["status"] = "pending" job["pid"] = None job.setdefault("log", []).append("Restarted after web app restart.") self._upsert_job_conn(conn, job) def _job_from_row(self, row): if not isinstance(row, sqlite3.Row): return json.loads(row[0]) raw = dict(row) payload = raw.get(JOB_EXTRA_PAYLOAD_COLUMN) or "{}" try: job = json.loads(payload) except json.JSONDecodeError: job = {} if not isinstance(job, dict): job = {} log_payload = raw.get(JOB_LOG_COLUMN) or "[]" try: log_lines = json.loads(log_payload) except json.JSONDecodeError: log_lines = [] if not isinstance(log_lines, list): log_lines = [] for key in ("id", "status", "created_at", "updated_at"): value = raw.get(key) if value is not None: job[key] = value for key in JOB_STRUCTURED_COLUMNS: value = raw.get(key) if value is None: continue if key == "cancel_requested": job[key] = bool(int(value or 0)) else: job[key] = value job["log"] = [str(line) for line in log_lines][-MAX_LOG_LINES:] return job def _extra_job_payload(self, job): clean = dict(job) clean.pop("process", None) clean.pop("_last_persist_monotonic", None) clean.pop("_dirty_log_lines", None) clean.pop("log", None) for key in ("id", "status", "created_at", "updated_at"): clean.pop(key, None) for key in JOB_STRUCTURED_COLUMNS: clean.pop(key, None) return clean def _upsert_job_conn(self, conn, job): log_lines = [str(line) for line in job.get("log", [])][-MAX_LOG_LINES:] extra_payload = self._extra_job_payload(job) conn.execute( """ INSERT INTO jobs ( id, status, created_at, updated_at, show_id, title, anime_name, season, query, result_index, mode, quality, episodes, download_dir, started_at, finished_at, exit_code, pid, command, staging_dir, target_dir, cancel_requested, log_payload, payload ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET status = excluded.status, updated_at = excluded.updated_at, show_id = excluded.show_id, title = excluded.title, anime_name = excluded.anime_name, season = excluded.season, query = excluded.query, result_index = excluded.result_index, mode = excluded.mode, quality = excluded.quality, episodes = excluded.episodes, download_dir = excluded.download_dir, started_at = excluded.started_at, finished_at = excluded.finished_at, exit_code = excluded.exit_code, pid = excluded.pid, command = excluded.command, staging_dir = excluded.staging_dir, target_dir = excluded.target_dir, cancel_requested = excluded.cancel_requested, log_payload = excluded.log_payload, payload = excluded.payload """, ( job["id"], job.get("status", "pending"), job.get("created_at") or now_iso(), job.get("updated_at") or now_iso(), job.get("show_id"), job.get("title"), job.get("anime_name"), job.get("season"), job.get("query"), job.get("result_index"), job.get("mode"), job.get("quality"), job.get("episodes"), job.get("download_dir"), job.get("started_at"), job.get("finished_at"), job.get("exit_code"), job.get("pid"), job.get("command"), job.get("staging_dir"), job.get("target_dir"), 1 if job.get("cancel_requested") else 0, json.dumps(log_lines), json.dumps(extra_payload, sort_keys=True), ), ) def _save_job_locked(self, job): with self._connect() as conn: self._upsert_job_conn(conn, job) job["_last_persist_monotonic"] = time.monotonic() job["_dirty_log_lines"] = 0 def list(self, page=1, per_page=10): page = max(1, int(page or 1)) per_page = min(50, max(1, int(per_page or 10))) offset = (page - 1) * per_page with self.lock, self._connect() as conn: total = conn.execute("SELECT COUNT(*) AS count FROM jobs").fetchone()["count"] rows = conn.execute( "SELECT * FROM jobs ORDER BY created_at DESC LIMIT ? OFFSET ?", (per_page, offset), ).fetchall() pages = max(1, (total + per_page - 1) // per_page) return { "jobs": [self._job_from_row(row) for row in rows], "page": page, "per_page": per_page, "total": total, "pages": pages, } def add(self, payload): job = build_job(payload, self.config_getter()) if self.job_prepare_fn is not None: prepared = self.job_prepare_fn(dict(job)) if prepared is not None: job = prepared with self.lock: self._save_job_locked(job) self.wakeup.set() return job def retry(self, job_id): with self.lock: job = self._find(job_id) if job["status"] == "running": raise ValueError("Running jobs cannot be retried") job["status"] = "pending" job["exit_code"] = None job["pid"] = None job["started_at"] = None job["finished_at"] = None job["updated_at"] = now_iso() job["log"] = [] self._save_job_locked(job) self.wakeup.set() return job def retry_all_failed(self): count = 0 with self.lock, self._connect() as conn: rows = conn.execute("SELECT * FROM jobs WHERE status = 'failed'").fetchall() for row in rows: job = self._job_from_row(row) job["status"] = "pending" job["exit_code"] = None job["pid"] = None job["started_at"] = None job["finished_at"] = None job["updated_at"] = now_iso() job["log"] = [] self._upsert_job_conn(conn, job) count += 1 if count: self.wakeup.set() return {"ok": True, "count": count} def cancel(self, job_id): with self.lock: job = self._find(job_id) if job["status"] == "pending": job["status"] = "canceled" job["updated_at"] = now_iso() self._save_job_locked(job) return job if job["status"] != "running": return job if self.current_job_id == job_id and self.current_process: try: os.killpg(os.getpgid(self.current_process.pid), signal.SIGTERM) except OSError: self.current_process.terminate() updated_at = now_iso() job["cancel_requested"] = True job["updated_at"] = updated_at if self.current_job is not None and self.current_job.get("id") == job_id: self.current_job["cancel_requested"] = True self.current_job["updated_at"] = updated_at self._save_job_locked(job) return job raise ValueError("Could not find the running process") def remove(self, job_id): with self.lock: job = self._find(job_id) if job["status"] == "running": raise ValueError("Running jobs must be canceled before removing") with self._connect() as conn: conn.execute("DELETE FROM jobs WHERE id = ?", (job_id,)) return {"ok": True} def clear_finished(self): with self.lock, self._connect() as conn: cursor = conn.execute("DELETE FROM jobs WHERE status IN ('done', 'canceled')") return {"ok": True, "count": cursor.rowcount} def _find(self, job_id): with self._connect() as conn: row = conn.execute("SELECT * FROM jobs WHERE id = ?", (job_id,)).fetchone() if row: return self._job_from_row(row) raise KeyError("Job not found") def _next_pending(self): with self.lock: with self._connect() as conn: row = conn.execute( "SELECT * FROM jobs WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1" ).fetchone() if row: return self._job_from_row(row) return None def _append_log(self, job, line): text = strip_control(line).strip() if not text: return with self.lock: job.setdefault("log", []).append(text) job["log"] = job["log"][-MAX_LOG_LINES:] job["updated_at"] = now_iso() job["_dirty_log_lines"] = int(job.get("_dirty_log_lines") or 0) + 1 last_persist = float(job.get("_last_persist_monotonic") or 0.0) should_flush = ( job["_dirty_log_lines"] >= LOG_PERSIST_LINE_BATCH or (time.monotonic() - last_persist) >= LOG_PERSIST_INTERVAL_SECONDS ) if should_flush: self._save_job_locked(job) 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): command = command_for_job(job) staging_dir = job_staging_dir(job) target_dir = job_output_dir(job) env = os.environ.copy() env.update( { "ANI_CLI_DOWNLOAD_DIR": str(staging_dir), "ANI_CLI_MODE": job["mode"], "ANI_CLI_QUALITY": job["quality"], "TERM": env.get("TERM", "xterm-256color"), } ) staging_dir.mkdir(parents=True, exist_ok=True) target_dir.mkdir(parents=True, exist_ok=True) with self.lock: job["status"] = "running" job["started_at"] = now_iso() job["updated_at"] = job["started_at"] job["command"] = printable_command(command) job["staging_dir"] = str(staging_dir) job["target_dir"] = str(target_dir) job["log"] = [ f"Starting: {job['command']}", f"Staging in: {job['staging_dir']}", f"Final folder: {job['target_dir']}", ] job["cancel_requested"] = False job["_dirty_log_lines"] = 0 self._save_job_locked(job) process = subprocess.Popen( command, cwd=str(PROJECT_ROOT), env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, start_new_session=True, ) with self.lock: self.current_process = process self.current_job_id = job["id"] self.current_job = job job["pid"] = process.pid self._save_job_locked(job) assert process.stdout is not None for line in process.stdout: self._append_log(job, line) exit_code = process.wait() finalize_error = None moved_files = [] watchlist_sync_error = None if exit_code == 0 and not job.get("cancel_requested"): try: moved_files = finalize_library_files(job) except Exception as exc: finalize_error = exc else: if self.watchlist_sync_fn is not None: try: self.watchlist_sync_fn(job) except Exception as exc: watchlist_sync_error = exc with self.lock: canceled = job.get("cancel_requested") job["exit_code"] = 1 if finalize_error else exit_code job["pid"] = None job["finished_at"] = now_iso() job["updated_at"] = job["finished_at"] if canceled: job["status"] = "canceled" job.setdefault("log", []).append("Canceled.") elif finalize_error: job["status"] = "failed" job.setdefault("log", []).append(f"Library rename failed: {finalize_error}") elif exit_code == 0: job["status"] = "done" for path in moved_files: job.setdefault("log", []).append(f"Saved: {path}") job.setdefault("log", []).append("Download completed.") if watchlist_sync_error: job.setdefault("log", []).append(f"Watchlist sync failed: {watchlist_sync_error}") elif self.watchlist_sync_fn is not None: job.setdefault("log", []).append("Watchlist download state synced.") else: job["status"] = "failed" job.setdefault("log", []).append(f"Download failed with exit code {exit_code}.") self.current_process = None self.current_job_id = None self.current_job = None self._save_job_locked(job)