add: 添加 SQLite 任务持久化
This commit is contained in:
@@ -3,6 +3,7 @@ config/rclone/rclone.conf
|
||||
|
||||
# Docker build artifacts / runtime
|
||||
.docker/
|
||||
webgui/data/
|
||||
|
||||
# Editor / OS cruft
|
||||
.idea/
|
||||
|
||||
+25
-4
@@ -1,12 +1,13 @@
|
||||
# Docker Compose stack for the rclone webgui.
|
||||
#
|
||||
# A single rclone rcd container does all three jobs on one port:
|
||||
# The rclone rcd container does three jobs on one port:
|
||||
# - Serves the static frontend from webgui/web/ (--rc-files)
|
||||
# - Serves RC API endpoints at POST /* (built-in)
|
||||
# - Serves remote files at GET /<remote>:<path> (--rc-serve)
|
||||
#
|
||||
# Same-origin from the browser, so no CORS / no nginx / no double ports.
|
||||
# For TLS or custom headers, put a reverse proxy of your choice in front.
|
||||
# The jobs-api sidecar stores recurring task definitions in SQLite and starts
|
||||
# scheduled runs through rclone RC. The browser calls it on port 5581.
|
||||
# For TLS, auth, or custom headers, put a reverse proxy of your choice in front.
|
||||
#
|
||||
# rclone config lives in ./config/rclone/rclone.conf (bind-mounted).
|
||||
# If it doesn't exist yet, create your remotes with:
|
||||
@@ -16,7 +17,8 @@
|
||||
# Usage:
|
||||
# docker compose up -d # start
|
||||
# open http://localhost:5580
|
||||
# docker compose logs -f rclone # tail logs
|
||||
# docker compose logs -f rclone # tail rclone logs
|
||||
# docker compose logs -f jobs-api # tail scheduler/API logs
|
||||
# docker compose down # stop
|
||||
|
||||
services:
|
||||
@@ -51,6 +53,25 @@ services:
|
||||
- "5580:8080"
|
||||
restart: unless-stopped
|
||||
|
||||
jobs-api:
|
||||
image: python:3.12-alpine
|
||||
container_name: rclone-jobs-api
|
||||
command: ["python", "/app/server.py"]
|
||||
depends_on:
|
||||
- rclone
|
||||
volumes:
|
||||
- ./webgui/api:/app:ro
|
||||
- ./webgui/data:/data
|
||||
environment:
|
||||
- JOBS_DB=/data/jobs.sqlite
|
||||
- JOBS_API_HOST=0.0.0.0
|
||||
- JOBS_API_PORT=8081
|
||||
- JOBS_SCHEDULER_INTERVAL=15
|
||||
- RCLONE_RC_URL=http://rclone:8080
|
||||
ports:
|
||||
- "5581:8081"
|
||||
restart: unless-stopped
|
||||
|
||||
volumes:
|
||||
rclone-cache:
|
||||
rclone-data:
|
||||
|
||||
@@ -0,0 +1,764 @@
|
||||
#!/usr/bin/env python3
|
||||
"""SQLite-backed recurring job API for the rclone WebGUI.
|
||||
|
||||
This service intentionally uses only Python's standard library. It stores
|
||||
recurring transfer definitions and the latest rclone job status snapshots in
|
||||
SQLite, then starts due jobs through rclone's RC API.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import threading
|
||||
import traceback
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from calendar import monthrange
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from typing import Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
|
||||
DB_PATH = os.environ.get("JOBS_DB", "/data/jobs.sqlite")
|
||||
RCLONE_RC_URL = os.environ.get("RCLONE_RC_URL", "http://rclone:8080").rstrip("/")
|
||||
HOST = os.environ.get("JOBS_API_HOST", "0.0.0.0")
|
||||
PORT = int(os.environ.get("JOBS_API_PORT", "8081"))
|
||||
SCHEDULER_INTERVAL_SECONDS = int(os.environ.get("JOBS_SCHEDULER_INTERVAL", "15"))
|
||||
VALID_ACTIONS = {"copy", "sync", "move"}
|
||||
VALID_SCHEDULE_KINDS = {"daily", "weekly", "monthly"}
|
||||
|
||||
db_lock = threading.RLock()
|
||||
scheduler_lock = threading.RLock()
|
||||
stop_event = threading.Event()
|
||||
|
||||
|
||||
class ApiError(Exception):
|
||||
def __init__(self, status: int, message: str):
|
||||
super().__init__(message)
|
||||
self.status = status
|
||||
self.message = message
|
||||
|
||||
|
||||
class RcloneError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def utcnow() -> datetime:
|
||||
return datetime.now(timezone.utc).replace(microsecond=0)
|
||||
|
||||
|
||||
def iso(dt: datetime) -> str:
|
||||
return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
||||
|
||||
|
||||
def parse_time(value: Any) -> datetime:
|
||||
if not isinstance(value, str) or not value.strip():
|
||||
raise ValueError("expected ISO timestamp")
|
||||
raw = value.strip()
|
||||
if raw.endswith("Z"):
|
||||
raw = raw[:-1] + "+00:00"
|
||||
dt = datetime.fromisoformat(raw)
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=timezone.utc)
|
||||
return dt.astimezone(timezone.utc).replace(microsecond=0)
|
||||
|
||||
|
||||
def parse_schedule_time(value: Any) -> tuple[int, int]:
|
||||
if not isinstance(value, str):
|
||||
raise ValueError("expected HH:MM")
|
||||
parts = value.strip().split(":")
|
||||
if len(parts) != 2:
|
||||
raise ValueError("expected HH:MM")
|
||||
hour = int(parts[0])
|
||||
minute = int(parts[1])
|
||||
if hour < 0 or hour > 23 or minute < 0 or minute > 59:
|
||||
raise ValueError("expected HH:MM")
|
||||
return hour, minute
|
||||
|
||||
|
||||
def parse_timezone_offset(value: Any) -> int:
|
||||
try:
|
||||
offset = int(value)
|
||||
except (TypeError, ValueError):
|
||||
return 0
|
||||
# Keep this intentionally bounded to real-world UTC offsets.
|
||||
if offset < -14 * 60 or offset > 14 * 60:
|
||||
raise ValueError("invalid timezone offset")
|
||||
return offset
|
||||
|
||||
|
||||
def add_months(year: int, month: int, months: int) -> tuple[int, int]:
|
||||
index = (year * 12 + (month - 1)) + months
|
||||
return index // 12, index % 12 + 1
|
||||
|
||||
|
||||
def local_to_utc(local_dt: datetime, offset_minutes: int) -> datetime:
|
||||
utc = local_dt - timedelta(minutes=offset_minutes)
|
||||
return utc.replace(tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def compute_next_run_at(
|
||||
kind: str,
|
||||
schedule_time: str,
|
||||
offset_minutes: int,
|
||||
after_dt: datetime,
|
||||
weekday: int | None = None,
|
||||
monthday: int | None = None,
|
||||
) -> str:
|
||||
hour, minute = parse_schedule_time(schedule_time)
|
||||
local_after = (after_dt.astimezone(timezone.utc) + timedelta(minutes=offset_minutes)).replace(tzinfo=None)
|
||||
|
||||
if kind == "daily":
|
||||
candidate = local_after.replace(hour=hour, minute=minute, second=0, microsecond=0)
|
||||
if candidate <= local_after:
|
||||
candidate += timedelta(days=1)
|
||||
return iso(local_to_utc(candidate, offset_minutes))
|
||||
|
||||
if kind == "weekly":
|
||||
if weekday is None or weekday < 1 or weekday > 7:
|
||||
raise ValueError("scheduleWeekday must be 1-7")
|
||||
target_weekday = weekday - 1
|
||||
days_ahead = target_weekday - local_after.weekday()
|
||||
candidate = (local_after + timedelta(days=days_ahead)).replace(
|
||||
hour=hour,
|
||||
minute=minute,
|
||||
second=0,
|
||||
microsecond=0,
|
||||
)
|
||||
if days_ahead < 0 or candidate <= local_after:
|
||||
candidate += timedelta(days=7)
|
||||
return iso(local_to_utc(candidate, offset_minutes))
|
||||
|
||||
if kind == "monthly":
|
||||
if monthday is None or monthday < 1 or monthday > 31:
|
||||
raise ValueError("scheduleMonthday must be 1-31")
|
||||
for offset_months in range(0, 36):
|
||||
year, month = add_months(local_after.year, local_after.month, offset_months)
|
||||
if monthday > monthrange(year, month)[1]:
|
||||
continue
|
||||
candidate = datetime(year, month, monthday, hour, minute)
|
||||
if candidate > local_after:
|
||||
return iso(local_to_utc(candidate, offset_minutes))
|
||||
raise ValueError("could not compute next monthly run")
|
||||
|
||||
raise ValueError("invalid schedule kind")
|
||||
|
||||
|
||||
def compute_row_next_run_at(row: sqlite3.Row, after_dt: datetime) -> str:
|
||||
return compute_next_run_at(
|
||||
row["schedule_kind"],
|
||||
row["schedule_time"],
|
||||
int(row["timezone_offset_minutes"] or 0),
|
||||
after_dt,
|
||||
row["schedule_weekday"],
|
||||
row["schedule_monthday"],
|
||||
)
|
||||
|
||||
|
||||
def connect() -> sqlite3.Connection:
|
||||
conn = sqlite3.connect(DB_PATH, timeout=30)
|
||||
conn.row_factory = sqlite3.Row
|
||||
conn.execute("PRAGMA foreign_keys=ON")
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
return conn
|
||||
|
||||
|
||||
def init_db() -> None:
|
||||
db_dir = os.path.dirname(DB_PATH)
|
||||
if db_dir:
|
||||
os.makedirs(db_dir, exist_ok=True)
|
||||
with db_lock, connect() as conn:
|
||||
conn.executescript(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS job_schedules (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
action TEXT NOT NULL,
|
||||
src TEXT NOT NULL,
|
||||
dst TEXT NOT NULL,
|
||||
interval_seconds INTEGER NOT NULL,
|
||||
schedule_kind TEXT NOT NULL DEFAULT 'daily',
|
||||
schedule_time TEXT NOT NULL DEFAULT '00:00',
|
||||
schedule_weekday INTEGER,
|
||||
schedule_monthday INTEGER,
|
||||
timezone_offset_minutes INTEGER NOT NULL DEFAULT 0,
|
||||
next_run_at TEXT NOT NULL,
|
||||
enabled INTEGER NOT NULL DEFAULT 1,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL,
|
||||
last_jobid INTEGER,
|
||||
current_jobid INTEGER,
|
||||
last_status TEXT,
|
||||
last_error TEXT,
|
||||
last_started_at TEXT,
|
||||
last_finished_at TEXT,
|
||||
last_status_json TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS job_runs (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
schedule_id INTEGER NOT NULL,
|
||||
jobid INTEGER,
|
||||
trigger TEXT NOT NULL,
|
||||
status TEXT NOT NULL,
|
||||
error TEXT,
|
||||
started_at TEXT,
|
||||
finished_at TEXT,
|
||||
status_json TEXT,
|
||||
created_at TEXT NOT NULL,
|
||||
FOREIGN KEY(schedule_id) REFERENCES job_schedules(id) ON DELETE CASCADE
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_job_schedules_due
|
||||
ON job_schedules(enabled, current_jobid, next_run_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_job_runs_schedule
|
||||
ON job_runs(schedule_id, created_at DESC);
|
||||
"""
|
||||
)
|
||||
ensure_column(conn, "job_schedules", "schedule_kind", "TEXT NOT NULL DEFAULT 'daily'")
|
||||
ensure_column(conn, "job_schedules", "schedule_time", "TEXT NOT NULL DEFAULT '00:00'")
|
||||
ensure_column(conn, "job_schedules", "schedule_weekday", "INTEGER")
|
||||
ensure_column(conn, "job_schedules", "schedule_monthday", "INTEGER")
|
||||
ensure_column(conn, "job_schedules", "timezone_offset_minutes", "INTEGER NOT NULL DEFAULT 0")
|
||||
|
||||
|
||||
def ensure_column(conn: sqlite3.Connection, table: str, column: str, definition: str) -> None:
|
||||
cols = {row["name"] for row in conn.execute(f"PRAGMA table_info({table})")}
|
||||
if column not in cols:
|
||||
conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} {definition}")
|
||||
|
||||
|
||||
def row_to_job(row: sqlite3.Row) -> dict[str, Any]:
|
||||
snapshot = None
|
||||
if row["last_status_json"]:
|
||||
try:
|
||||
snapshot = json.loads(row["last_status_json"])
|
||||
except json.JSONDecodeError:
|
||||
snapshot = None
|
||||
status = row["last_status"]
|
||||
if not status:
|
||||
status = "scheduled"
|
||||
return {
|
||||
"id": row["id"],
|
||||
"action": row["action"],
|
||||
"src": row["src"],
|
||||
"dst": row["dst"],
|
||||
"intervalSeconds": row["interval_seconds"],
|
||||
"scheduleKind": row["schedule_kind"],
|
||||
"scheduleTime": row["schedule_time"],
|
||||
"scheduleWeekday": row["schedule_weekday"],
|
||||
"scheduleMonthday": row["schedule_monthday"],
|
||||
"timezoneOffsetMinutes": row["timezone_offset_minutes"],
|
||||
"nextRunAt": row["next_run_at"],
|
||||
"enabled": bool(row["enabled"]),
|
||||
"lastJobid": row["last_jobid"],
|
||||
"currentJobid": row["current_jobid"],
|
||||
"status": status,
|
||||
"error": row["last_error"],
|
||||
"startedAt": row["last_started_at"],
|
||||
"finishedAt": row["last_finished_at"],
|
||||
"running": row["current_jobid"] is not None and status == "running",
|
||||
"statusSnapshot": snapshot,
|
||||
"createdAt": row["created_at"],
|
||||
"updatedAt": row["updated_at"],
|
||||
}
|
||||
|
||||
|
||||
def fetch_job(schedule_id: int) -> dict[str, Any]:
|
||||
with db_lock, connect() as conn:
|
||||
row = conn.execute("SELECT * FROM job_schedules WHERE id=?", (schedule_id,)).fetchone()
|
||||
if row is None:
|
||||
raise ApiError(404, "job not found")
|
||||
return row_to_job(row)
|
||||
|
||||
|
||||
def list_jobs() -> list[dict[str, Any]]:
|
||||
refresh_running_jobs()
|
||||
with db_lock, connect() as conn:
|
||||
rows = conn.execute("SELECT * FROM job_schedules ORDER BY id DESC").fetchall()
|
||||
return [row_to_job(row) for row in rows]
|
||||
|
||||
|
||||
def create_job(payload: dict[str, Any]) -> dict[str, Any]:
|
||||
action = str(payload.get("action", "")).strip()
|
||||
src = str(payload.get("src", "")).strip()
|
||||
dst = str(payload.get("dst", "")).strip()
|
||||
schedule_kind = str(payload.get("scheduleKind", "")).strip()
|
||||
schedule_time = str(payload.get("scheduleTime", "")).strip()
|
||||
|
||||
if action not in VALID_ACTIONS:
|
||||
raise ApiError(400, "invalid action")
|
||||
if not src or not dst:
|
||||
raise ApiError(400, "src and dst are required")
|
||||
if schedule_kind not in VALID_SCHEDULE_KINDS:
|
||||
raise ApiError(400, "scheduleKind must be daily, weekly, or monthly")
|
||||
try:
|
||||
parse_schedule_time(schedule_time)
|
||||
timezone_offset_minutes = parse_timezone_offset(payload.get("timezoneOffsetMinutes"))
|
||||
schedule_weekday = parse_optional_int(payload.get("scheduleWeekday"))
|
||||
schedule_monthday = parse_optional_int(payload.get("scheduleMonthday"))
|
||||
next_run_at = compute_next_run_at(
|
||||
schedule_kind,
|
||||
schedule_time,
|
||||
timezone_offset_minutes,
|
||||
utcnow(),
|
||||
schedule_weekday,
|
||||
schedule_monthday,
|
||||
)
|
||||
except (TypeError, ValueError):
|
||||
raise ApiError(400, "invalid fixed schedule") from None
|
||||
|
||||
interval_seconds = legacy_interval_seconds(schedule_kind)
|
||||
|
||||
now = iso(utcnow())
|
||||
with db_lock, connect() as conn:
|
||||
cur = conn.execute(
|
||||
"""
|
||||
INSERT INTO job_schedules
|
||||
(action, src, dst, interval_seconds, schedule_kind, schedule_time,
|
||||
schedule_weekday, schedule_monthday, timezone_offset_minutes,
|
||||
next_run_at, enabled, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?)
|
||||
""",
|
||||
(
|
||||
action,
|
||||
src,
|
||||
dst,
|
||||
interval_seconds,
|
||||
schedule_kind,
|
||||
schedule_time,
|
||||
schedule_weekday,
|
||||
schedule_monthday,
|
||||
timezone_offset_minutes,
|
||||
next_run_at,
|
||||
now,
|
||||
now,
|
||||
),
|
||||
)
|
||||
schedule_id = int(cur.lastrowid)
|
||||
|
||||
run_due_jobs()
|
||||
return fetch_job(schedule_id)
|
||||
|
||||
|
||||
def parse_optional_int(value: Any) -> int | None:
|
||||
if value in (None, ""):
|
||||
return None
|
||||
return int(value)
|
||||
|
||||
|
||||
def legacy_interval_seconds(schedule_kind: str) -> int:
|
||||
if schedule_kind == "weekly":
|
||||
return 7 * 24 * 60 * 60
|
||||
if schedule_kind == "monthly":
|
||||
return 30 * 24 * 60 * 60
|
||||
return 24 * 60 * 60
|
||||
|
||||
|
||||
def delete_job(schedule_id: int) -> None:
|
||||
with db_lock, connect() as conn:
|
||||
cur = conn.execute("DELETE FROM job_schedules WHERE id=?", (schedule_id,))
|
||||
if cur.rowcount == 0:
|
||||
raise ApiError(404, "job not found")
|
||||
|
||||
|
||||
def manual_run(schedule_id: int) -> dict[str, Any]:
|
||||
with scheduler_lock:
|
||||
refresh_running_jobs()
|
||||
with db_lock, connect() as conn:
|
||||
row = conn.execute("SELECT * FROM job_schedules WHERE id=?", (schedule_id,)).fetchone()
|
||||
if row is None:
|
||||
raise ApiError(404, "job not found")
|
||||
if row["current_jobid"] is not None:
|
||||
raise ApiError(409, "job is already running")
|
||||
start_schedule(row, "manual")
|
||||
return fetch_job(schedule_id)
|
||||
|
||||
|
||||
def stop_job(schedule_id: int) -> dict[str, Any]:
|
||||
with db_lock, connect() as conn:
|
||||
row = conn.execute("SELECT * FROM job_schedules WHERE id=?", (schedule_id,)).fetchone()
|
||||
if row is None:
|
||||
raise ApiError(404, "job not found")
|
||||
jobid = row["current_jobid"]
|
||||
if jobid is None:
|
||||
return row_to_job(row)
|
||||
|
||||
rc_post("job/stop", {"jobid": jobid})
|
||||
now_dt = utcnow()
|
||||
now = iso(now_dt)
|
||||
next_run_at = compute_row_next_run_at(row, now_dt)
|
||||
snapshot = json.dumps({"finished": True, "error": "stopped", "jobid": jobid}, separators=(",", ":"))
|
||||
with db_lock, connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE job_schedules
|
||||
SET current_jobid=NULL,
|
||||
last_status='stopped',
|
||||
last_error=NULL,
|
||||
last_finished_at=?,
|
||||
last_status_json=?,
|
||||
next_run_at=?,
|
||||
updated_at=?
|
||||
WHERE id=?
|
||||
""",
|
||||
(now, snapshot, next_run_at, now, schedule_id),
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE job_runs
|
||||
SET status='stopped', finished_at=?, status_json=?
|
||||
WHERE schedule_id=? AND jobid=? AND status='running'
|
||||
""",
|
||||
(now, snapshot, schedule_id, jobid),
|
||||
)
|
||||
return fetch_job(schedule_id)
|
||||
|
||||
|
||||
def rc_post(
|
||||
path: str,
|
||||
payload: dict[str, Any],
|
||||
timeout: int = 30,
|
||||
allow_error_body: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
data = json.dumps(payload).encode("utf-8")
|
||||
req = urllib.request.Request(
|
||||
f"{RCLONE_RC_URL}/{path.lstrip('/')}",
|
||||
data=data,
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout) as res:
|
||||
raw = res.read()
|
||||
except urllib.error.HTTPError as exc:
|
||||
raw = exc.read()
|
||||
message = raw.decode("utf-8", "replace") if raw else str(exc)
|
||||
try:
|
||||
body = json.loads(message)
|
||||
message = body.get("error") or body.get("message") or message
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
raise RcloneError(f"{path}: {message}") from exc
|
||||
except urllib.error.URLError as exc:
|
||||
raise RcloneError(f"{path}: {exc.reason}") from exc
|
||||
|
||||
if not raw:
|
||||
return {}
|
||||
try:
|
||||
body = json.loads(raw.decode("utf-8"))
|
||||
except json.JSONDecodeError as exc:
|
||||
raise RcloneError(f"{path}: invalid JSON response") from exc
|
||||
if body.get("error") and not allow_error_body:
|
||||
raise RcloneError(f"{path}: {body['error']}")
|
||||
return body
|
||||
|
||||
|
||||
def start_schedule(row: sqlite3.Row, trigger: str) -> None:
|
||||
now_dt = utcnow()
|
||||
now = iso(now_dt)
|
||||
schedule_id = int(row["id"])
|
||||
action = row["action"]
|
||||
payload: dict[str, Any] = {
|
||||
"srcFs": row["src"],
|
||||
"dstFs": row["dst"],
|
||||
"_async": True,
|
||||
"_group": f"webgui/recurring/{schedule_id}/{action}",
|
||||
}
|
||||
if action == "move":
|
||||
payload["deleteEmptySrcDirs"] = True
|
||||
|
||||
try:
|
||||
res = rc_post(f"sync/{action}", payload)
|
||||
jobid = res.get("jobid")
|
||||
if jobid is None:
|
||||
raise RcloneError(f"sync/{action}: missing jobid")
|
||||
except Exception as exc:
|
||||
error = str(exc)
|
||||
next_run_at = compute_row_next_run_at(row, now_dt)
|
||||
snapshot = json.dumps({"finished": True, "error": error}, separators=(",", ":"))
|
||||
with db_lock, connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO job_runs
|
||||
(schedule_id, jobid, trigger, status, error, started_at, finished_at, status_json, created_at)
|
||||
VALUES (?, NULL, ?, 'failed', ?, ?, ?, ?, ?)
|
||||
""",
|
||||
(schedule_id, trigger, error, now, now, snapshot, now),
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE job_schedules
|
||||
SET current_jobid=NULL,
|
||||
last_status='failed',
|
||||
last_error=?,
|
||||
last_started_at=?,
|
||||
last_finished_at=?,
|
||||
last_status_json=?,
|
||||
next_run_at=?,
|
||||
updated_at=?
|
||||
WHERE id=?
|
||||
""",
|
||||
(error, now, now, snapshot, next_run_at, now, schedule_id),
|
||||
)
|
||||
return
|
||||
|
||||
snapshot = json.dumps({"finished": False, "jobid": jobid}, separators=(",", ":"))
|
||||
next_run_at = compute_row_next_run_at(row, now_dt)
|
||||
with db_lock, connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO job_runs
|
||||
(schedule_id, jobid, trigger, status, started_at, status_json, created_at)
|
||||
VALUES (?, ?, ?, 'running', ?, ?, ?)
|
||||
""",
|
||||
(schedule_id, jobid, trigger, now, snapshot, now),
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE job_schedules
|
||||
SET current_jobid=?,
|
||||
last_jobid=?,
|
||||
last_status='running',
|
||||
last_error=NULL,
|
||||
last_started_at=?,
|
||||
last_finished_at=NULL,
|
||||
last_status_json=?,
|
||||
next_run_at=?,
|
||||
updated_at=?
|
||||
WHERE id=?
|
||||
""",
|
||||
(jobid, jobid, now, snapshot, next_run_at, now, schedule_id),
|
||||
)
|
||||
|
||||
|
||||
def status_name(status: dict[str, Any]) -> str:
|
||||
if not status.get("finished"):
|
||||
return "running"
|
||||
if status.get("error"):
|
||||
return "failed"
|
||||
if status.get("success"):
|
||||
return "done"
|
||||
return "finished"
|
||||
|
||||
|
||||
def refresh_running_jobs() -> None:
|
||||
with db_lock, connect() as conn:
|
||||
rows = conn.execute("SELECT * FROM job_schedules WHERE current_jobid IS NOT NULL").fetchall()
|
||||
|
||||
for row in rows:
|
||||
schedule_id = int(row["id"])
|
||||
jobid = int(row["current_jobid"])
|
||||
try:
|
||||
status = rc_post("job/status", {"jobid": jobid}, timeout=15, allow_error_body=True)
|
||||
except Exception as exc:
|
||||
status = {"finished": True, "error": str(exc), "jobid": jobid}
|
||||
apply_status(row, status)
|
||||
|
||||
|
||||
def apply_status(row: sqlite3.Row, status: dict[str, Any]) -> None:
|
||||
schedule_id = int(row["id"])
|
||||
jobid = int(row["current_jobid"])
|
||||
now_dt = utcnow()
|
||||
now = iso(now_dt)
|
||||
snapshot = json.dumps(status, separators=(",", ":"), ensure_ascii=False)
|
||||
name = status_name(status)
|
||||
|
||||
if name == "running":
|
||||
with db_lock, connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE job_schedules
|
||||
SET last_status='running',
|
||||
last_error=NULL,
|
||||
last_status_json=?,
|
||||
updated_at=?
|
||||
WHERE id=? AND current_jobid=?
|
||||
""",
|
||||
(snapshot, now, schedule_id, jobid),
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE job_runs
|
||||
SET status='running', status_json=?
|
||||
WHERE schedule_id=? AND jobid=? AND status='running'
|
||||
""",
|
||||
(snapshot, schedule_id, jobid),
|
||||
)
|
||||
return
|
||||
|
||||
error = status.get("error")
|
||||
next_run_at = compute_row_next_run_at(row, now_dt)
|
||||
with db_lock, connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE job_schedules
|
||||
SET current_jobid=NULL,
|
||||
last_status=?,
|
||||
last_error=?,
|
||||
last_finished_at=?,
|
||||
last_status_json=?,
|
||||
next_run_at=?,
|
||||
updated_at=?
|
||||
WHERE id=? AND current_jobid=?
|
||||
""",
|
||||
(name, error, now, snapshot, next_run_at, now, schedule_id, jobid),
|
||||
)
|
||||
conn.execute(
|
||||
"""
|
||||
UPDATE job_runs
|
||||
SET status=?, error=?, finished_at=?, status_json=?
|
||||
WHERE schedule_id=? AND jobid=? AND status='running'
|
||||
""",
|
||||
(name, error, now, snapshot, schedule_id, jobid),
|
||||
)
|
||||
|
||||
|
||||
def run_due_jobs() -> None:
|
||||
with scheduler_lock:
|
||||
now = iso(utcnow())
|
||||
with db_lock, connect() as conn:
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT * FROM job_schedules
|
||||
WHERE enabled=1 AND current_jobid IS NULL AND next_run_at <= ?
|
||||
ORDER BY next_run_at ASC, id ASC
|
||||
""",
|
||||
(now,),
|
||||
).fetchall()
|
||||
|
||||
for row in rows:
|
||||
start_schedule(row, "scheduled")
|
||||
|
||||
|
||||
def scheduler_tick() -> None:
|
||||
refresh_running_jobs()
|
||||
run_due_jobs()
|
||||
|
||||
|
||||
def scheduler_loop() -> None:
|
||||
while not stop_event.wait(SCHEDULER_INTERVAL_SECONDS):
|
||||
try:
|
||||
scheduler_tick()
|
||||
except Exception:
|
||||
traceback.print_exc()
|
||||
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
protocol_version = "HTTP/1.1"
|
||||
|
||||
def do_OPTIONS(self) -> None:
|
||||
self.send_json(204, None)
|
||||
|
||||
def do_GET(self) -> None:
|
||||
self.handle_request("GET")
|
||||
|
||||
def do_POST(self) -> None:
|
||||
self.handle_request("POST")
|
||||
|
||||
def do_DELETE(self) -> None:
|
||||
self.handle_request("DELETE")
|
||||
|
||||
def handle_request(self, method: str) -> None:
|
||||
try:
|
||||
parsed = urlparse(self.path)
|
||||
path = parsed.path.rstrip("/") or "/"
|
||||
parts = [p for p in path.split("/") if p]
|
||||
|
||||
if method == "GET" and path == "/health":
|
||||
self.send_json(200, {"ok": True})
|
||||
return
|
||||
|
||||
if parts == ["api", "jobs"] and method == "GET":
|
||||
self.send_json(200, {"jobs": list_jobs()})
|
||||
return
|
||||
|
||||
if parts == ["api", "jobs"] and method == "POST":
|
||||
self.send_json(201, {"job": create_job(self.read_json())})
|
||||
return
|
||||
|
||||
if len(parts) == 3 and parts[:2] == ["api", "jobs"]:
|
||||
schedule_id = parse_id(parts[2])
|
||||
if method == "DELETE":
|
||||
delete_job(schedule_id)
|
||||
self.send_json(200, {"ok": True})
|
||||
return
|
||||
|
||||
if len(parts) == 4 and parts[:2] == ["api", "jobs"] and method == "POST":
|
||||
schedule_id = parse_id(parts[2])
|
||||
if parts[3] == "run":
|
||||
self.send_json(200, {"job": manual_run(schedule_id)})
|
||||
return
|
||||
if parts[3] == "stop":
|
||||
self.send_json(200, {"job": stop_job(schedule_id)})
|
||||
return
|
||||
|
||||
raise ApiError(404, "not found")
|
||||
except ApiError as exc:
|
||||
self.send_json(exc.status, {"error": exc.message})
|
||||
except RcloneError as exc:
|
||||
self.send_json(502, {"error": str(exc)})
|
||||
except Exception as exc:
|
||||
traceback.print_exc()
|
||||
self.send_json(500, {"error": str(exc)})
|
||||
|
||||
def read_json(self) -> dict[str, Any]:
|
||||
length = int(self.headers.get("Content-Length") or "0")
|
||||
if length == 0:
|
||||
return {}
|
||||
raw = self.rfile.read(length)
|
||||
try:
|
||||
data = json.loads(raw.decode("utf-8"))
|
||||
except json.JSONDecodeError:
|
||||
raise ApiError(400, "invalid JSON") from None
|
||||
if not isinstance(data, dict):
|
||||
raise ApiError(400, "expected JSON object")
|
||||
return data
|
||||
|
||||
def send_json(self, status: int, body: Any) -> None:
|
||||
raw = b"" if body is None else json.dumps(body, ensure_ascii=False).encode("utf-8")
|
||||
self.send_response(status)
|
||||
self.send_header("Access-Control-Allow-Origin", "*")
|
||||
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
||||
self.send_header("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS")
|
||||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||||
self.send_header("Content-Length", str(len(raw)))
|
||||
self.end_headers()
|
||||
if raw:
|
||||
self.wfile.write(raw)
|
||||
|
||||
def log_message(self, fmt: str, *args: Any) -> None:
|
||||
print(f"{self.address_string()} - {fmt % args}")
|
||||
|
||||
|
||||
def parse_id(value: str) -> int:
|
||||
try:
|
||||
result = int(value)
|
||||
except ValueError:
|
||||
raise ApiError(404, "not found") from None
|
||||
if result <= 0:
|
||||
raise ApiError(404, "not found")
|
||||
return result
|
||||
|
||||
|
||||
def main() -> None:
|
||||
init_db()
|
||||
thread = threading.Thread(target=scheduler_loop, name="job-scheduler", daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
scheduler_tick()
|
||||
except Exception:
|
||||
traceback.print_exc()
|
||||
server = ThreadingHTTPServer((HOST, PORT), Handler)
|
||||
print(f"jobs api listening on {HOST}:{PORT}, db={DB_PATH}, rclone={RCLONE_RC_URL}")
|
||||
try:
|
||||
server.serve_forever()
|
||||
finally:
|
||||
stop_event.set()
|
||||
server.server_close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,66 @@
|
||||
// jobs_api.js — client for the SQLite-backed recurring jobs sidecar.
|
||||
|
||||
const params = new URLSearchParams(location.search);
|
||||
const QUERY_API = params.get("jobsApi");
|
||||
const DEFAULT_PORT = "5581";
|
||||
const API_BASE = QUERY_API
|
||||
? QUERY_API.replace(/\/$/, "")
|
||||
: `${location.protocol}//${location.hostname || "127.0.0.1"}:${DEFAULT_PORT}`;
|
||||
|
||||
export function jobsApiURL() {
|
||||
return API_BASE;
|
||||
}
|
||||
|
||||
export async function listRecurringJobs() {
|
||||
const res = await request("/api/jobs");
|
||||
return (res && res.jobs) || [];
|
||||
}
|
||||
|
||||
export async function createRecurringJob(job) {
|
||||
const res = await request("/api/jobs", {
|
||||
method: "POST",
|
||||
body: job,
|
||||
});
|
||||
return res && res.job;
|
||||
}
|
||||
|
||||
export async function runRecurringJobNow(id) {
|
||||
const res = await request(`/api/jobs/${encodeURIComponent(id)}/run`, {
|
||||
method: "POST",
|
||||
});
|
||||
return res && res.job;
|
||||
}
|
||||
|
||||
export async function stopRecurringJob(id) {
|
||||
const res = await request(`/api/jobs/${encodeURIComponent(id)}/stop`, {
|
||||
method: "POST",
|
||||
});
|
||||
return res && res.job;
|
||||
}
|
||||
|
||||
export async function deleteRecurringJob(id) {
|
||||
return request(`/api/jobs/${encodeURIComponent(id)}`, {
|
||||
method: "DELETE",
|
||||
});
|
||||
}
|
||||
|
||||
async function request(path, { method = "GET", body } = {}) {
|
||||
const headers = { "Content-Type": "application/json" };
|
||||
const res = await fetch(`${API_BASE}${path}`, {
|
||||
method,
|
||||
headers,
|
||||
body: body == null ? undefined : JSON.stringify(body),
|
||||
});
|
||||
let data = {};
|
||||
const ct = res.headers.get("Content-Type") || "";
|
||||
if (ct.includes("application/json")) {
|
||||
data = await res.json();
|
||||
} else {
|
||||
const text = await res.text();
|
||||
data = text ? { error: text } : {};
|
||||
}
|
||||
if (!res.ok) {
|
||||
throw new Error(data.error || `HTTP ${res.status}`);
|
||||
}
|
||||
return data;
|
||||
}
|
||||
Reference in New Issue
Block a user