#!/usr/bin/env python3 """Queue and background job services for ani-cli-web.""" import json import os import shutil import signal import sqlite3 import subprocess import threading import time import re from app_support import ( MAX_LOG_LINES, PROJECT_ROOT, QUEUE_PATH, STATE_DB_PATH, WORKER_SHUTDOWN_TIMEOUT_SECONDS, build_job, command_for_job, debug_log, finalize_library_files, job_output_dir, job_staging_dir, load_json, now_iso, printable_command, send_discord_webhook_event, strip_control, ) LOG_PERSIST_INTERVAL_SECONDS = 0.75 LOG_PERSIST_LINE_BATCH = 8 JOB_STRUCTURED_COLUMNS = ( "show_id", "title", "anime_name", "media_type", "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, config_getter=None, start_worker=True): self.store = store 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.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_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 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 _refresh_delay_seconds(self): config = self.config_getter() or {} try: delay = int(str(config.get("watchlist_refresh_delay_seconds", 5)).strip() or "5") except (TypeError, ValueError): delay = 5 return max(0, min(300, delay)) 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 _stdout_log(self, message): print(f"[watchlist-refresh] {message}", flush=True) 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, 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: if source_name == "scheduled": self._stdout_log( f"Scheduled refresh skipped because another refresh is already {job['status']} (job={job['id']})." ) 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", "source": source_name, "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() if source_name == "scheduled": self._stdout_log(f"Scheduled refresh queued (job={job['id']}).") 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): interrupted = False source_name = str(job.get("source") or "manual").strip().lower() or "manual" new_episode_items = [] refresh_error_items = [] try: show_ids = self.store.all_show_ids(categories=("watching", "planned")) delay_seconds = self._refresh_delay_seconds() 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) if source_name == "scheduled": self._stdout_log(f"Scheduled refresh started (job={job['id']}, total={len(show_ids)}).") for index, show_id in enumerate(show_ids, start=1): if self.stop_event.is_set(): interrupted = True break title = show_id errored = False skipped = False error_message = "" try: try: item = self.store.refresh(show_id, include_thumbnail=False, refresh_source=source_name) except TypeError as exc: if "refresh_source" not in str(exc): raise 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 "" if item.get("had_new_episodes"): new_episode_items.append( { "show_id": item.get("show_id"), "title": title, "sub_count": item.get("sub_count"), "dub_count": item.get("dub_count"), "sub_delta": item.get("sub_delta"), "dub_delta": item.get("dub_delta"), "expected_count": item.get("expected_count"), "airing_status": item.get("airing_status"), "thumbnail_path": item.get("thumbnail_path"), } ) except Exception as exc: errored = True error_message = str(exc) debug_log("watchlist.refresh_all.item_error", show_id=show_id, error=exc) with self.lock: job["completed"] = index job["errors"] += 1 if errored else 0 job["current_title"] = title job["updated_at"] = now_iso() if errored and error_message: job["last_error"] = error_message refresh_error_items.append({"show_id": show_id, "title": title, "error": error_message}) if 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) if index < len(show_ids) and delay_seconds > 0: if self.stop_event.wait(delay_seconds): interrupted = True break 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 if interrupted: job["message"] = f"Watchlist refresh interrupted after {job['completed']} entries." elif 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) if source_name == "scheduled": if interrupted: self._stdout_log( f"Scheduled refresh interrupted (job={job['id']}, completed={job['completed']}/{job['total']})." ) elif job["errors"]: self._stdout_log( f"Scheduled refresh finished with errors (job={job['id']}, completed={job['completed']}/{job['total']}, errors={job['errors']})." ) else: self._stdout_log( f"Scheduled refresh finished successfully (job={job['id']}, completed={job['completed']}/{job['total']})." ) config = self.config_getter() or {} if interrupted: send_discord_webhook_event( config, "watchlist_refresh_interrupted", { "source": source_name, "completed": job.get("completed"), "total": job.get("total"), "message": job.get("message"), }, ) else: if new_episode_items: send_discord_webhook_event( config, "watchlist_refresh_new_episodes", { "source": source_name, "job_id": job.get("id"), "items": new_episode_items, }, ) if job.get("errors"): send_discord_webhook_event( config, "watchlist_refresh_failed", { "source": source_name, "completed": job.get("completed"), "total": job.get("total"), "errors": job.get("errors"), "items": refresh_error_items, "error": job.get("last_error"), }, ) 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}" job["last_error"] = str(exc) self._save_job_locked(job) if source_name == "scheduled": self._stdout_log(f"Scheduled refresh failed (job={job['id']}): {exc}") send_discord_webhook_event( self.config_getter() or {}, "watchlist_refresh_failed", { "source": source_name, "completed": job.get("completed"), "total": job.get("total"), "errors": max(1, int(job.get("errors") or 0)), "items": refresh_error_items, "error": str(exc), }, ) finally: with self.lock: self.current_executor = None self.current_job_id = None class WatchlistAutoRefreshScheduler: def __init__(self, config_getter, refresh_start_fn, start_worker=True): self.config_getter = config_getter self.refresh_start_fn = refresh_start_fn self.lock = threading.RLock() self.wakeup = threading.Event() self.stop_event = threading.Event() self.worker = None self.last_enabled = None self.last_interval_seconds = None self.last_attempt_monotonic = None if start_worker: self.ensure_worker() 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 notify_config_changed(self): self.wakeup.set() def _config_state(self): config = self.config_getter() or {} enabled = bool(config.get("watchlist_auto_refresh_enabled")) try: minutes = int(str(config.get("watchlist_auto_refresh_minutes", 60)).strip() or "60") except (TypeError, ValueError): minutes = 60 minutes = max(1, min(7 * 24 * 60, minutes)) return enabled, minutes * 60 def tick(self, now_monotonic=None): now = time.monotonic() if now_monotonic is None else float(now_monotonic) enabled, interval_seconds = self._config_state() with self.lock: changed = enabled != self.last_enabled or interval_seconds != self.last_interval_seconds if changed: self.last_enabled = enabled self.last_interval_seconds = interval_seconds self.last_attempt_monotonic = now if not enabled: return None if self.last_attempt_monotonic is None: self.last_attempt_monotonic = now return interval_seconds remaining = interval_seconds - (now - self.last_attempt_monotonic) if remaining > 0: return remaining self.last_attempt_monotonic = now try: result = self.refresh_start_fn(source="scheduled") debug_log("watchlist.auto_refresh.tick", result=result) except Exception as exc: debug_log("watchlist.auto_refresh.error", error=exc) send_discord_webhook_event( self.config_getter() or {}, "runtime_error", { "source": "watchlist.auto_refresh", "error": str(exc), }, ) return interval_seconds def _run(self): while not self.stop_event.is_set(): wait_seconds = self.tick() self.wakeup.wait(timeout=wait_seconds) self.wakeup.clear() 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_SHUTDOWN_TIMEOUT_SECONDS) if worker.is_alive(): self._signal_active_process(signal.SIGKILL) worker.join(timeout=WORKER_SHUTDOWN_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, media_type 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"), ("media_type", "ALTER TABLE jobs ADD COLUMN media_type 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, media_type, 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, media_type = excluded.media_type, 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("media_type"), 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 get(self, job_id): with self.lock: return self._find(job_id) 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 _expand_episode_spec(self, episode_spec, available_episodes): spec = str(episode_spec or "").strip() values = [str(value).strip() for value in available_episodes or [] if str(value).strip()] if not spec or not values: return [] expanded = [] seen = set() parts = [part for part in re.split(r"\s+", spec) if part] index_map = {value: index for index, value in enumerate(values)} for part in parts: if "-" in part: start, end = [piece.strip() for piece in part.split("-", 1)] if start in index_map and end in index_map: start_index = index_map[start] end_index = index_map[end] if start_index > end_index: start_index, end_index = end_index, start_index for value in values[start_index : end_index + 1]: if value not in seen: seen.add(value) expanded.append(value) continue if part in index_map and part not in seen: seen.add(part) expanded.append(part) return expanded def covered_episodes(self, show_id, mode, available_episodes): normalized_show_id = str(show_id or "").strip() normalized_mode = str(mode or "").strip().lower() if not normalized_show_id or not normalized_mode: return set() with self.lock, self._connect() as conn: rows = conn.execute( """ SELECT episodes FROM jobs WHERE show_id = ? AND mode = ? AND status IN ('pending', 'running', 'done') ORDER BY created_at ASC """, (normalized_show_id, normalized_mode), ).fetchall() covered = set() for row in rows: covered.update(self._expand_episode_spec((row["episodes"] if isinstance(row, sqlite3.Row) else row[0]), available_episodes)) return covered def detach_show(self, show_id): normalized_show_id = str(show_id or "").strip() if not normalized_show_id: return {"ok": True, "count": 0} count = 0 with self.lock, self._connect() as conn: rows = conn.execute("SELECT * FROM jobs WHERE show_id = ?", (normalized_show_id,)).fetchall() for row in rows: job = self._job_from_row(row) job["show_id"] = "" job["skip_watchlist_sync"] = True job["updated_at"] = now_iso() self._upsert_job_conn(conn, job) count += 1 if self.current_job is not None and str(self.current_job.get("show_id") or "").strip() == normalized_show_id: self.current_job["show_id"] = "" self.current_job["skip_watchlist_sync"] = True self.current_job["updated_at"] = now_iso() return {"ok": True, "count": count} def retry(self, job_id): with self.lock: job = self._find(job_id) if job["status"] == "running": raise ValueError("Running jobs cannot be retried") if job["status"] not in {"failed", "canceled"}: raise ValueError("Only failed or canceled jobs can 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["cancel_requested"] = False 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 _cleanup_staging_dir(self, job): staging_dir = job_staging_dir(job) if not staging_dir.exists(): return False try: shutil.rmtree(staging_dir) debug_log("queue.cleanup.staging_removed", job_id=job.get("id"), staging_dir=staging_dir) return True except OSError as exc: debug_log("queue.cleanup.staging_failed", job_id=job.get("id"), staging_dir=staging_dir, error=exc) return False def _fail_job_startup(self, job, exc): message = f"Could not start download: {exc}" with self.lock: job["status"] = "failed" job["exit_code"] = 1 job["pid"] = None job["finished_at"] = now_iso() job["updated_at"] = job["finished_at"] job.setdefault("log", []).append(message) self.current_process = None self.current_job_id = None self.current_job = None self._save_job_locked(job) self._cleanup_staging_dir(job) def _fail_job_runtime(self, job, exc, process=None): if process is not None: try: os.killpg(os.getpgid(process.pid), signal.SIGTERM) except OSError: try: process.terminate() except OSError: pass message = f"Download worker failed unexpectedly: {exc}" with self.lock: job["status"] = "failed" job["exit_code"] = 1 job["pid"] = None job["finished_at"] = now_iso() job["updated_at"] = job["finished_at"] job.setdefault("log", []).append(message) self.current_process = None self.current_job_id = None self.current_job = None self._save_job_locked(job) self._cleanup_staging_dir(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) try: process = subprocess.Popen( command, cwd=str(PROJECT_ROOT), env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, start_new_session=True, ) except Exception as exc: self._fail_job_startup(job, exc) return 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) try: 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 synced_watchlist_item = None cleanup_staging = False 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: synced_watchlist_item = 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" message = str(finalize_error).strip() if "No downloaded files were found in the staging folder" in message: job.setdefault("log", []).append( "ani-cli exited without downloading any files. The selected result or episodes may not be downloadable." ) else: job.setdefault("log", []).append(f"Library rename failed: {finalize_error}") cleanup_staging = True 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}.") cleanup_staging = True if canceled: cleanup_staging = True self.current_process = None self.current_job_id = None self.current_job = None self._save_job_locked(job) if cleanup_staging: self._cleanup_staging_dir(job) elif exit_code == 0 and not finalize_error and not canceled: download_title = ( str((synced_watchlist_item or {}).get("title") or "").strip() or str(job.get("title") or "").strip() or str(job.get("query") or "").strip() or "Unknown anime" ) try: send_discord_webhook_event( self.config_getter() or {}, "download_completed", { "show_id": (synced_watchlist_item or {}).get("show_id") or job.get("show_id"), "title": download_title, "mode": job.get("mode"), "episodes": job.get("episodes"), "category": (synced_watchlist_item or {}).get("category") or "finished", "thumbnail_path": (synced_watchlist_item or {}).get("thumbnail_path"), "moved_files": moved_files, }, ) except Exception as exc: debug_log("discord_webhook.download_completed_failed", title=download_title, error=exc) except Exception as exc: self._fail_job_runtime(job, exc, process=process)