Dymas
292bbedd6b
Add a Queue page bulk action and API route to remove all failed jobs, cover it with regression tests, and update the version, changelog, and README for the new queue workflow.
1619 lines
65 KiB
Python
1619 lines
65 KiB
Python
#!/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 (
|
|
ANIPY_CLI,
|
|
MAX_LOG_LINES,
|
|
PROJECT_ROOT,
|
|
QUEUE_PATH,
|
|
STATE_DB_PATH,
|
|
WORKER_SHUTDOWN_TIMEOUT_SECONDS,
|
|
build_job,
|
|
cli_executable_exists,
|
|
command_for_job,
|
|
debug_log,
|
|
finalize_library_files,
|
|
job_output_dir,
|
|
job_staging_dir,
|
|
load_json,
|
|
normalize_config,
|
|
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 JellyfinSyncJobs:
|
|
CONFIG_KEYS = (
|
|
"download_dir",
|
|
"jellyfin_sync_enabled",
|
|
"jellyfin_tv_dir",
|
|
"jellyfin_movie_dir",
|
|
"discord_webhook_url",
|
|
"discord_webhook_events",
|
|
)
|
|
|
|
def __init__(self, store, sync_fn, config_getter=None, start_worker=True):
|
|
self.store = store
|
|
self.sync_fn = sync_fn
|
|
self.config_getter = config_getter or (lambda: {})
|
|
self.lock = threading.RLock()
|
|
self.wakeup = threading.Event()
|
|
self.stop_event = threading.Event()
|
|
self.pending = []
|
|
self.jobs = []
|
|
self.current_job_id = None
|
|
self._init_db()
|
|
self._restore_interrupted_jobs()
|
|
self._load_jobs()
|
|
self.worker = None
|
|
if start_worker:
|
|
self.ensure_worker()
|
|
|
|
def _connect(self):
|
|
conn = sqlite3.connect(STATE_DB_PATH, timeout=30)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
def ensure_worker(self):
|
|
with self.lock:
|
|
if self.worker is not None and self.worker.is_alive():
|
|
return
|
|
self.stop_event.clear()
|
|
self.worker = threading.Thread(target=self._run, daemon=True)
|
|
self.worker.start()
|
|
|
|
def shutdown(self, wait=False):
|
|
self.stop_event.set()
|
|
self.wakeup.set()
|
|
worker = self.worker
|
|
if wait and worker is not None and worker.is_alive():
|
|
worker.join(timeout=WORKER_SHUTDOWN_TIMEOUT_SECONDS)
|
|
return worker is None or not worker.is_alive()
|
|
|
|
def _init_db(self):
|
|
STATE_DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
with self._connect() as conn:
|
|
conn.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS jellyfin_sync_jobs (
|
|
id TEXT PRIMARY KEY,
|
|
status TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
payload TEXT NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_jellyfin_sync_created_at ON jellyfin_sync_jobs(created_at)")
|
|
conn.execute("CREATE INDEX IF NOT EXISTS idx_jellyfin_sync_status ON jellyfin_sync_jobs(status)")
|
|
|
|
def _job_from_row(self, row):
|
|
payload = row["payload"] if isinstance(row, sqlite3.Row) else row[0]
|
|
return json.loads(payload)
|
|
|
|
def _upsert_job_conn(self, conn, job):
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO jellyfin_sync_jobs (id, status, created_at, updated_at, payload)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
status = excluded.status,
|
|
updated_at = excluded.updated_at,
|
|
payload = excluded.payload
|
|
""",
|
|
(
|
|
job["id"],
|
|
job["status"],
|
|
job["created_at"],
|
|
job["updated_at"],
|
|
json.dumps(job, sort_keys=True),
|
|
),
|
|
)
|
|
|
|
def _trim_history_conn(self, conn):
|
|
rows = conn.execute(
|
|
"SELECT id FROM jellyfin_sync_jobs ORDER BY created_at DESC LIMIT -1 OFFSET 10"
|
|
).fetchall()
|
|
if rows:
|
|
conn.executemany("DELETE FROM jellyfin_sync_jobs WHERE id = ?", [(row["id"],) for row in rows])
|
|
|
|
def _save_job_locked(self, job):
|
|
with self._connect() as conn:
|
|
self._upsert_job_conn(conn, job)
|
|
self._trim_history_conn(conn)
|
|
|
|
def _load_jobs(self):
|
|
with self.lock, self._connect() as conn:
|
|
rows = conn.execute(
|
|
"SELECT payload FROM jellyfin_sync_jobs ORDER BY created_at DESC LIMIT 10"
|
|
).fetchall()
|
|
self.jobs = [self._job_from_row(row) for row in rows]
|
|
|
|
def _restore_interrupted_jobs(self):
|
|
with self.lock, self._connect() as conn:
|
|
rows = conn.execute(
|
|
"SELECT payload FROM jellyfin_sync_jobs WHERE status IN ('pending', 'running')"
|
|
).fetchall()
|
|
for row in rows:
|
|
job = self._job_from_row(row)
|
|
job["status"] = "interrupted"
|
|
job["finished_at"] = now_iso()
|
|
job["updated_at"] = job["finished_at"]
|
|
job["current_title"] = None
|
|
job["current_file"] = None
|
|
job["message"] = "Interrupted by web app restart."
|
|
self._upsert_job_conn(conn, job)
|
|
self._trim_history_conn(conn)
|
|
|
|
def _copy_job(self, job):
|
|
return json.loads(json.dumps(job))
|
|
|
|
def _find_locked(self, job_id):
|
|
for job in self.jobs:
|
|
if job["id"] == job_id:
|
|
return job
|
|
return None
|
|
|
|
def _job_config(self, config):
|
|
normalized = normalize_config(config or self.config_getter() or {})
|
|
return {key: normalized.get(key) for key in self.CONFIG_KEYS}
|
|
|
|
def status(self):
|
|
with self.lock:
|
|
job = next((candidate for candidate in self.jobs if candidate["status"] in {"pending", "running"}), None)
|
|
if job is None:
|
|
job = self.jobs[0] if self.jobs else None
|
|
active = bool(job and job["status"] in {"pending", "running"})
|
|
return {"job": self._copy_job(job) if job else None, "running": active}
|
|
|
|
def start(self, config=None, source="manual"):
|
|
source_name = str(source or "manual").strip().lower() or "manual"
|
|
with self.lock:
|
|
job = next((candidate for candidate in self.jobs if candidate["status"] in {"pending", "running"}), None)
|
|
if job is not None:
|
|
return {
|
|
"message": "Jellyfin handoff already queued." if job["status"] == "pending" else "Jellyfin handoff already running.",
|
|
"job": self._copy_job(job),
|
|
"created": False,
|
|
}
|
|
now = now_iso()
|
|
job = {
|
|
"id": os.urandom(16).hex(),
|
|
"status": "pending",
|
|
"source": source_name,
|
|
"config": self._job_config(config),
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
"started_at": None,
|
|
"finished_at": None,
|
|
"total": 0,
|
|
"completed": 0,
|
|
"moved": 0,
|
|
"errors": 0,
|
|
"moved_files": 0,
|
|
"current_title": None,
|
|
"current_file": None,
|
|
"message": "Queued Jellyfin handoff.",
|
|
"items": [],
|
|
}
|
|
self.jobs.insert(0, job)
|
|
self.jobs = self.jobs[:10]
|
|
self._save_job_locked(job)
|
|
self.pending.append(job["id"])
|
|
self.wakeup.set()
|
|
return {"message": "Jellyfin handoff started.", "job": self._copy_job(job), "created": True}
|
|
|
|
def _next_pending(self):
|
|
with self.lock:
|
|
while self.pending:
|
|
job_id = self.pending.pop(0)
|
|
job = self._find_locked(job_id)
|
|
if job is not None:
|
|
self.current_job_id = job_id
|
|
return job
|
|
return None
|
|
|
|
def _run(self):
|
|
while not self.stop_event.is_set():
|
|
job = self._next_pending()
|
|
if not job:
|
|
self.wakeup.wait(2)
|
|
self.wakeup.clear()
|
|
continue
|
|
self._run_job(job)
|
|
|
|
def _run_job(self, job):
|
|
interrupted = False
|
|
result_items = []
|
|
try:
|
|
active_config = self._job_config(job.get("config") or self.config_getter() or {})
|
|
show_ids = self.store.all_show_ids()
|
|
started_at = now_iso()
|
|
with self.lock:
|
|
job["status"] = "running"
|
|
job["started_at"] = started_at
|
|
job["updated_at"] = started_at
|
|
job["total"] = len(show_ids)
|
|
job["completed"] = 0
|
|
job["moved"] = 0
|
|
job["errors"] = 0
|
|
job["moved_files"] = 0
|
|
job["current_title"] = None
|
|
job["current_file"] = None
|
|
job["items"] = []
|
|
job["message"] = f"Moving 0/{len(show_ids)} watchlist entries to Jellyfin."
|
|
self._save_job_locked(job)
|
|
|
|
for index, show_id in enumerate(show_ids, start=1):
|
|
if self.stop_event.is_set():
|
|
interrupted = True
|
|
break
|
|
try:
|
|
item = self.store.get(show_id)
|
|
except KeyError:
|
|
item = None
|
|
if item is None:
|
|
with self.lock:
|
|
job["completed"] = index
|
|
job["updated_at"] = now_iso()
|
|
job["message"] = f"Moving {index}/{len(show_ids)}: removed entry"
|
|
self._save_job_locked(job)
|
|
continue
|
|
|
|
title = str(item.get("title") or show_id)
|
|
|
|
def progress(update):
|
|
with self.lock:
|
|
job["current_title"] = title
|
|
job["current_file"] = update.get("current_file")
|
|
job["updated_at"] = now_iso()
|
|
file_total = int(update.get("file_total") or 0)
|
|
file_index = int(update.get("file_index") or 0)
|
|
if file_total > 0:
|
|
job["message"] = (
|
|
f"Moving {index}/{len(show_ids)}: {title} "
|
|
f"({file_index}/{file_total} files)"
|
|
)
|
|
else:
|
|
job["message"] = f"Moving {index}/{len(show_ids)}: {title}"
|
|
self._save_job_locked(job)
|
|
|
|
result = self.sync_fn(item, automatic=False, config=active_config, progress_fn=progress)
|
|
result_items.append(
|
|
{
|
|
"show_id": item.get("show_id"),
|
|
"title": item.get("title"),
|
|
"media_type": item.get("media_type"),
|
|
"moved": bool(result.get("moved")),
|
|
"reason": result.get("reason"),
|
|
"target": result.get("target"),
|
|
}
|
|
)
|
|
moved_files = len(result.get("moved_files") or []) if isinstance(result.get("moved_files"), list) else 0
|
|
errored = str(result.get("reason") or "").strip() in {"error", "target_missing", "source_missing", "target_incomplete"}
|
|
with self.lock:
|
|
job["completed"] = index
|
|
job["moved"] += 1 if result.get("moved") else 0
|
|
job["moved_files"] += moved_files
|
|
job["errors"] += 1 if errored else 0
|
|
job["current_title"] = title
|
|
job["current_file"] = None
|
|
job["updated_at"] = now_iso()
|
|
job["items"] = result_items[-20:]
|
|
if errored and result.get("error"):
|
|
job["last_error"] = str(result.get("error"))
|
|
outcome = "moved" if result.get("moved") else str(result.get("reason") or "skipped")
|
|
job["message"] = f"Moving {index}/{len(show_ids)}: {title} ({outcome})"
|
|
self._save_job_locked(job)
|
|
|
|
finished_at = now_iso()
|
|
with self.lock:
|
|
job["status"] = "interrupted" if interrupted else "done"
|
|
job["finished_at"] = finished_at
|
|
job["updated_at"] = finished_at
|
|
job["current_title"] = None
|
|
job["current_file"] = None
|
|
job["items"] = result_items[-20:]
|
|
if interrupted:
|
|
job["message"] = f"Jellyfin handoff interrupted after {job['completed']} entries."
|
|
elif job["errors"]:
|
|
job["message"] = (
|
|
f"Moved {job['moved']} of {job['completed']} processed watchlist entries to Jellyfin "
|
|
f"with {job['errors']} issues."
|
|
)
|
|
else:
|
|
job["message"] = (
|
|
f"Moved {job['moved']} watchlist entr{'y' if job['moved'] == 1 else 'ies'} to Jellyfin libraries."
|
|
)
|
|
self._save_job_locked(job)
|
|
except Exception as exc:
|
|
finished_at = now_iso()
|
|
with self.lock:
|
|
job["status"] = "failed"
|
|
job["finished_at"] = finished_at
|
|
job["updated_at"] = finished_at
|
|
job["current_title"] = None
|
|
job["current_file"] = None
|
|
job["message"] = f"Jellyfin handoff failed: {exc}"
|
|
job["last_error"] = str(exc)
|
|
self._save_job_locked(job)
|
|
finally:
|
|
with self.lock:
|
|
self.current_job_id = None
|
|
|
|
|
|
class DownloadQueue:
|
|
def __init__(self, config_getter, watchlist_sync_fn=None, job_prepare_fn=None, start_worker=True):
|
|
self.config_getter = config_getter
|
|
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 clear_failed(self):
|
|
with self.lock, self._connect() as conn:
|
|
cursor = conn.execute("DELETE FROM jobs WHERE status = 'failed'")
|
|
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 _command_attempts(self, job, staging_dir):
|
|
config = normalize_config(self.config_getter() or {})
|
|
attempts = [
|
|
{
|
|
"backend": "ani-cli",
|
|
"command": command_for_job(job, backend="ani-cli"),
|
|
"env": {
|
|
**os.environ.copy(),
|
|
"ANI_CLI_DOWNLOAD_DIR": str(staging_dir),
|
|
"ANI_CLI_MODE": job["mode"],
|
|
"ANI_CLI_QUALITY": job["quality"],
|
|
"TERM": os.environ.get("TERM", "xterm-256color"),
|
|
},
|
|
}
|
|
]
|
|
fallback_enabled = bool(config.get("anipy_cli_fallback_enabled"))
|
|
fallback_available = cli_executable_exists(ANIPY_CLI)
|
|
if fallback_enabled and fallback_available:
|
|
attempts.append(
|
|
{
|
|
"backend": "anipy-cli",
|
|
"command": command_for_job(job, backend="anipy-cli", download_path=staging_dir),
|
|
"env": {
|
|
**os.environ.copy(),
|
|
"TERM": os.environ.get("TERM", "xterm-256color"),
|
|
},
|
|
}
|
|
)
|
|
return attempts, fallback_enabled, fallback_available
|
|
|
|
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):
|
|
staging_dir = job_staging_dir(job)
|
|
target_dir = job_output_dir(job)
|
|
staging_dir.mkdir(parents=True, exist_ok=True)
|
|
target_dir.mkdir(parents=True, exist_ok=True)
|
|
attempts, fallback_enabled, fallback_available = self._command_attempts(job, staging_dir)
|
|
primary_command = attempts[0]["command"]
|
|
|
|
with self.lock:
|
|
job["status"] = "running"
|
|
job["started_at"] = now_iso()
|
|
job["updated_at"] = job["started_at"]
|
|
job["command"] = printable_command(primary_command)
|
|
job["staging_dir"] = str(staging_dir)
|
|
job["target_dir"] = str(target_dir)
|
|
job["download_backend"] = "ani-cli"
|
|
job["fallback_used"] = False
|
|
job["log"] = [
|
|
f"Starting: {job['command']}",
|
|
f"Staging in: {job['staging_dir']}",
|
|
f"Final folder: {job['target_dir']}",
|
|
]
|
|
if fallback_enabled and fallback_available:
|
|
job["log"].append("anipy-cli fallback is enabled and will retry automatically if ani-cli fails.")
|
|
elif fallback_enabled:
|
|
job["log"].append("anipy-cli fallback is enabled, but anipy-cli is not installed or not executable.")
|
|
job["cancel_requested"] = False
|
|
job["_dirty_log_lines"] = 0
|
|
self._save_job_locked(job)
|
|
|
|
try:
|
|
exit_code = 1
|
|
successful_backend = None
|
|
finalize_error = None
|
|
moved_files = []
|
|
watchlist_sync_error = None
|
|
synced_watchlist_item = None
|
|
cleanup_staging = False
|
|
for index, attempt in enumerate(attempts):
|
|
backend = attempt["backend"]
|
|
command = attempt["command"]
|
|
should_fallback = index + 1 < len(attempts)
|
|
with self.lock:
|
|
job["download_backend"] = backend
|
|
job["command"] = printable_command(command)
|
|
if index > 0:
|
|
job["fallback_used"] = True
|
|
job["pid"] = None
|
|
job["updated_at"] = now_iso()
|
|
if index > 0:
|
|
job.setdefault("log", []).append(f"Retrying with {backend}: {job['command']}")
|
|
self._save_job_locked(job)
|
|
try:
|
|
process = subprocess.Popen(
|
|
command,
|
|
cwd=str(PROJECT_ROOT),
|
|
env=attempt["env"],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
bufsize=1,
|
|
start_new_session=True,
|
|
)
|
|
except Exception as exc:
|
|
if should_fallback:
|
|
self._append_log(job, f"ani-cli could not start: {exc}")
|
|
self._append_log(job, "Retrying with anipy-cli fallback.")
|
|
self._cleanup_staging_dir(job)
|
|
staging_dir.mkdir(parents=True, exist_ok=True)
|
|
continue
|
|
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()
|
|
with self.lock:
|
|
job["pid"] = None
|
|
self.current_process = None
|
|
self.current_job_id = None
|
|
self.current_job = None
|
|
self._save_job_locked(job)
|
|
|
|
if job.get("cancel_requested"):
|
|
break
|
|
|
|
if backend == "ani-cli" and exit_code != 0 and should_fallback:
|
|
self._append_log(job, f"ani-cli failed with exit code {exit_code}.")
|
|
self._append_log(job, "Retrying with anipy-cli fallback.")
|
|
self._cleanup_staging_dir(job)
|
|
staging_dir.mkdir(parents=True, exist_ok=True)
|
|
continue
|
|
|
|
if exit_code == 0:
|
|
try:
|
|
moved_files = finalize_library_files(job)
|
|
except Exception as exc:
|
|
finalize_error = exc
|
|
no_files_downloaded = "No downloaded files were found in the staging folder" in str(exc)
|
|
if backend == "ani-cli" and no_files_downloaded and should_fallback:
|
|
self._append_log(
|
|
job,
|
|
"ani-cli exited without downloading any files. Retrying with anipy-cli fallback.",
|
|
)
|
|
self._cleanup_staging_dir(job)
|
|
staging_dir.mkdir(parents=True, exist_ok=True)
|
|
finalize_error = None
|
|
continue
|
|
else:
|
|
successful_backend = backend
|
|
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
|
|
break
|
|
break
|
|
|
|
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(
|
|
f"{job.get('download_backend') or 'Downloader'} 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"
|
|
if successful_backend == "anipy-cli":
|
|
job.setdefault("log", []).append("Fallback download completed with anipy-cli.")
|
|
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:
|
|
if job.get("skip_watchlist_sync"):
|
|
job.setdefault("log", []).append("Watchlist sync skipped for detached job.")
|
|
else:
|
|
job.setdefault("log", []).append("Watchlist download state synced.")
|
|
else:
|
|
job["status"] = "failed"
|
|
job.setdefault("log", []).append(
|
|
f"{job.get('download_backend') or '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)
|