Release ani-cli-web 0.26.0
This commit is contained in:
Dymas
+840
@@ -0,0 +1,840 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user