886 lines
34 KiB
Python
886 lines
34 KiB
Python
#!/usr/bin/env python3
|
|
|
|
"""Queue and background job services for ani-cli-web."""
|
|
|
|
import concurrent.futures
|
|
import json
|
|
import os
|
|
import shutil
|
|
import signal
|
|
import sqlite3
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
|
|
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,
|
|
strip_control,
|
|
)
|
|
|
|
|
|
LOG_PERSIST_INTERVAL_SECONDS = 0.75
|
|
LOG_PERSIST_LINE_BATCH = 8
|
|
WATCHLIST_REFRESH_MAX_WORKERS = 4
|
|
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_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 _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):
|
|
interrupted = False
|
|
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):
|
|
if self.stop_event.is_set():
|
|
interrupted = True
|
|
executor.shutdown(wait=False, cancel_futures=True)
|
|
break
|
|
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"] = "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)
|
|
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_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,
|
|
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 _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 _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)
|
|
|
|
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
|
|
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:
|
|
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}.")
|
|
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)
|