diff --git a/.gitignore b/.gitignore index cf9cb63..0f8ce84 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,7 @@ config/rclone/rclone.conf # Docker build artifacts / runtime .docker/ +webgui/data/ # Editor / OS cruft .idea/ diff --git a/docker-compose.yml b/docker-compose.yml index 2c3ad60..f32f146 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -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 /: (--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: diff --git a/webgui/api/server.py b/webgui/api/server.py new file mode 100644 index 0000000..b092114 --- /dev/null +++ b/webgui/api/server.py @@ -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() diff --git a/webgui/web/assets/js/jobs_api.js b/webgui/web/assets/js/jobs_api.js new file mode 100644 index 0000000..14744f6 --- /dev/null +++ b/webgui/web/assets/js/jobs_api.js @@ -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; +}