add: 持久化所有任务记录

webgui/api/server.py

@@ -223,10 +223,27 @@ def init_db() -> None:
               FOREIGN KEY(schedule_id) REFERENCES job_schedules(id) ON DELETE CASCADE
             );
 
+            CREATE TABLE IF NOT EXISTS one_time_jobs (
+              id INTEGER PRIMARY KEY AUTOINCREMENT,
+              action TEXT NOT NULL,
+              src TEXT NOT NULL,
+              dst TEXT NOT NULL,
+              jobid INTEGER,
+              status TEXT NOT NULL,
+              error TEXT,
+              started_at TEXT,
+              finished_at TEXT,
+              status_json TEXT,
+              created_at TEXT NOT NULL,
+              updated_at TEXT NOT NULL
+            );
+
             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);
+            CREATE INDEX IF NOT EXISTS idx_one_time_jobs_updated
+              ON one_time_jobs(updated_at DESC);
             """
         )
         ensure_column(conn, "job_schedules", "schedule_kind", "TEXT NOT NULL DEFAULT 'daily'")
@@ -278,6 +295,32 @@ def row_to_job(row: sqlite3.Row) -> dict[str, Any]:
     }
 
 
+def row_to_one_time_job(row: sqlite3.Row) -> dict[str, Any]:
+    snapshot = None
+    if row["status_json"]:
+        try:
+            snapshot = json.loads(row["status_json"])
+        except json.JSONDecodeError:
+            snapshot = None
+    return {
+        "id": row["id"],
+        "action": row["action"],
+        "src": row["src"],
+        "dst": row["dst"],
+        "jobid": row["jobid"],
+        "lastJobid": row["jobid"],
+        "currentJobid": row["jobid"] if row["status"] == "running" else None,
+        "status": row["status"],
+        "error": row["error"],
+        "startedAt": row["started_at"],
+        "finishedAt": row["finished_at"],
+        "running": row["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()
@@ -293,6 +336,21 @@ def list_jobs() -> list[dict[str, Any]]:
     return [row_to_job(row) for row in rows]
 
 
+def fetch_one_time_job(job_id: int) -> dict[str, Any]:
+    with db_lock, connect() as conn:
+        row = conn.execute("SELECT * FROM one_time_jobs WHERE id=?", (job_id,)).fetchone()
+    if row is None:
+        raise ApiError(404, "job not found")
+    return row_to_one_time_job(row)
+
+
+def list_one_time_jobs() -> list[dict[str, Any]]:
+    refresh_one_time_jobs()
+    with db_lock, connect() as conn:
+        rows = conn.execute("SELECT * FROM one_time_jobs ORDER BY updated_at DESC, id DESC").fetchall()
+    return [row_to_one_time_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()
@@ -355,6 +413,30 @@ def create_job(payload: dict[str, Any]) -> dict[str, Any]:
     return fetch_job(schedule_id)
 
 
+def create_one_time_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()
+    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")
+
+    now = iso(utcnow())
+    with db_lock, connect() as conn:
+        cur = conn.execute(
+            """
+            INSERT INTO one_time_jobs
+              (action, src, dst, status, created_at, updated_at)
+            VALUES (?, ?, ?, 'starting', ?, ?)
+            """,
+            (action, src, dst, now, now),
+        )
+        job_id = int(cur.lastrowid)
+    start_one_time_job(job_id)
+    return fetch_one_time_job(job_id)
+
+
 def parse_optional_int(value: Any) -> int | None:
     if value in (None, ""):
         return None
@@ -429,6 +511,52 @@ def stop_job(schedule_id: int) -> dict[str, Any]:
     return fetch_job(schedule_id)
 
 
+def manual_run_one_time_job(job_id: int) -> dict[str, Any]:
+    with db_lock, connect() as conn:
+        row = conn.execute("SELECT * FROM one_time_jobs WHERE id=?", (job_id,)).fetchone()
+    if row is None:
+        raise ApiError(404, "job not found")
+    if row["status"] == "running":
+        raise ApiError(409, "job is already running")
+    start_one_time_job(job_id)
+    return fetch_one_time_job(job_id)
+
+
+def stop_one_time_job(job_id: int) -> dict[str, Any]:
+    with db_lock, connect() as conn:
+        row = conn.execute("SELECT * FROM one_time_jobs WHERE id=?", (job_id,)).fetchone()
+    if row is None:
+        raise ApiError(404, "job not found")
+    jobid = row["jobid"]
+    if row["status"] != "running" or jobid is None:
+        return row_to_one_time_job(row)
+
+    rc_post("job/stop", {"jobid": jobid})
+    now = iso(utcnow())
+    snapshot = json.dumps({"finished": True, "error": "stopped", "jobid": jobid}, separators=(",", ":"))
+    with db_lock, connect() as conn:
+        conn.execute(
+            """
+            UPDATE one_time_jobs
+            SET status='stopped',
+                error=NULL,
+                finished_at=?,
+                status_json=?,
+                updated_at=?
+            WHERE id=?
+            """,
+            (now, snapshot, now, job_id),
+        )
+    return fetch_one_time_job(job_id)
+
+
+def delete_one_time_job(job_id: int) -> None:
+    with db_lock, connect() as conn:
+        cur = conn.execute("DELETE FROM one_time_jobs WHERE id=?", (job_id,))
+        if cur.rowcount == 0:
+            raise ApiError(404, "job not found")
+
+
 def rc_post(
     path: str,
     payload: dict[str, Any],
@@ -546,6 +674,66 @@ def start_schedule(row: sqlite3.Row, trigger: str) -> None:
         )
 
 
+def start_one_time_job(job_id: int) -> None:
+    with db_lock, connect() as conn:
+        row = conn.execute("SELECT * FROM one_time_jobs WHERE id=?", (job_id,)).fetchone()
+    if row is None:
+        raise ApiError(404, "job not found")
+
+    now = iso(utcnow())
+    action = row["action"]
+    payload: dict[str, Any] = {
+        "srcFs": row["src"],
+        "dstFs": row["dst"],
+        "_async": True,
+        "_group": f"webgui/once/{job_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)
+        snapshot = json.dumps({"finished": True, "error": error}, separators=(",", ":"))
+        with db_lock, connect() as conn:
+            conn.execute(
+                """
+                UPDATE one_time_jobs
+                SET jobid=NULL,
+                    status='failed',
+                    error=?,
+                    started_at=?,
+                    finished_at=?,
+                    status_json=?,
+                    updated_at=?
+                WHERE id=?
+                """,
+                (error, now, now, snapshot, now, job_id),
+            )
+        return
+
+    snapshot = json.dumps({"finished": False, "jobid": jobid}, separators=(",", ":"))
+    with db_lock, connect() as conn:
+        conn.execute(
+            """
+            UPDATE one_time_jobs
+            SET jobid=?,
+                status='running',
+                error=NULL,
+                started_at=?,
+                finished_at=NULL,
+                status_json=?,
+                updated_at=?
+            WHERE id=?
+            """,
+            (jobid, now, snapshot, now, job_id),
+        )
+
+
 def status_name(status: dict[str, Any]) -> str:
     if not status.get("finished"):
         return "running"
@@ -570,6 +758,19 @@ def refresh_running_jobs() -> None:
         apply_status(row, status)
 
 
+def refresh_one_time_jobs() -> None:
+    with db_lock, connect() as conn:
+        rows = conn.execute("SELECT * FROM one_time_jobs WHERE status='running' AND jobid IS NOT NULL").fetchall()
+
+    for row in rows:
+        jobid = int(row["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_one_time_status(row, status)
+
+
 def apply_status(row: sqlite3.Row, status: dict[str, Any]) -> None:
     schedule_id = int(row["id"])
     jobid = int(row["current_jobid"])
@@ -628,6 +829,44 @@ def apply_status(row: sqlite3.Row, status: dict[str, Any]) -> None:
         )
 
 
+def apply_one_time_status(row: sqlite3.Row, status: dict[str, Any]) -> None:
+    job_id = int(row["id"])
+    jobid = int(row["jobid"])
+    now = iso(utcnow())
+    snapshot = json.dumps(status, separators=(",", ":"), ensure_ascii=False)
+    name = status_name(status)
+
+    if name == "running":
+        with db_lock, connect() as conn:
+            conn.execute(
+                """
+                UPDATE one_time_jobs
+                SET status='running',
+                    error=NULL,
+                    status_json=?,
+                    updated_at=?
+                WHERE id=? AND jobid=? AND status='running'
+                """,
+                (snapshot, now, job_id, jobid),
+            )
+        return
+
+    error = status.get("error")
+    with db_lock, connect() as conn:
+        conn.execute(
+            """
+            UPDATE one_time_jobs
+            SET status=?,
+                error=?,
+                finished_at=?,
+                status_json=?,
+                updated_at=?
+            WHERE id=? AND jobid=? AND status='running'
+            """,
+            (name, error, now, snapshot, now, job_id, jobid),
+        )
+
+
 def run_due_jobs() -> None:
     with scheduler_lock:
         now = iso(utcnow())
@@ -647,6 +886,7 @@ def run_due_jobs() -> None:
 
 def scheduler_tick() -> None:
     refresh_running_jobs()
+    refresh_one_time_jobs()
     run_due_jobs()
 
 
@@ -703,6 +943,14 @@ class Handler(BaseHTTPRequestHandler):
                 self.send_json(201, {"job": create_job(self.read_json())})
                 return
 
+            if parts == ["api", "one-time-jobs"] and method == "GET":
+                self.send_json(200, {"jobs": list_one_time_jobs()})
+                return
+
+            if parts == ["api", "one-time-jobs"] and method == "POST":
+                self.send_json(201, {"job": create_one_time_job(self.read_json())})
+                return
+
             if len(parts) == 3 and parts[:2] == ["api", "jobs"]:
                 schedule_id = parse_id(parts[2])
                 if method == "DELETE":
@@ -710,6 +958,13 @@ class Handler(BaseHTTPRequestHandler):
                     self.send_json(200, {"ok": True})
                     return
 
+            if len(parts) == 3 and parts[:2] == ["api", "one-time-jobs"]:
+                job_id = parse_id(parts[2])
+                if method == "DELETE":
+                    delete_one_time_job(job_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":
@@ -719,6 +974,15 @@ class Handler(BaseHTTPRequestHandler):
                     self.send_json(200, {"job": stop_job(schedule_id)})
                     return
 
+            if len(parts) == 4 and parts[:2] == ["api", "one-time-jobs"] and method == "POST":
+                job_id = parse_id(parts[2])
+                if parts[3] == "run":
+                    self.send_json(200, {"job": manual_run_one_time_job(job_id)})
+                    return
+                if parts[3] == "stop":
+                    self.send_json(200, {"job": stop_one_time_job(job_id)})
+                    return
+
             raise ApiError(404, "not found")
         except ApiError as exc:
             self.send_json(exc.status, {"error": exc.message})

webgui/api/server_test.py

@@ -1,7 +1,10 @@
 import unittest
+import tempfile
+import os
 from io import BytesIO
 from datetime import datetime, timezone
 
+import server
 from server import ApiError, Handler, MAX_BODY_BYTES, compute_next_run_at
 
 
@@ -112,5 +115,62 @@ class HandlerSecurityTest(unittest.TestCase):
         self.assertEqual(Handler.read_json(handler), {"ok": True})
 
 
+class OneTimeJobTest(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.old_db_path = server.DB_PATH
+        self.old_rc_post = server.rc_post
+        server.DB_PATH = os.path.join(self.tmp.name, "jobs.sqlite")
+        self.next_jobid = 100
+
+        def fake_rc_post(path, payload, timeout=30, allow_error_body=False):
+            if path.startswith("sync/"):
+                self.next_jobid += 1
+                return {"jobid": self.next_jobid}
+            if path == "job/status":
+                return {"finished": False, "jobid": payload["jobid"], "progress": {"bytes": 1, "totalBytes": 2}}
+            if path == "job/stop":
+                return {}
+            raise AssertionError(path)
+
+        server.rc_post = fake_rc_post
+        server.init_db()
+
+    def tearDown(self):
+        server.rc_post = self.old_rc_post
+        server.DB_PATH = self.old_db_path
+        self.tmp.cleanup()
+
+    def test_create_and_list_one_time_job(self):
+        job = server.create_one_time_job({"action": "copy", "src": "/tmp/a", "dst": "/tmp/b"})
+
+        self.assertEqual(job["id"], 1)
+        self.assertEqual(job["jobid"], 101)
+        self.assertEqual(job["status"], "running")
+        self.assertEqual(job["src"], "/tmp/a")
+
+        jobs = server.list_one_time_jobs()
+        self.assertEqual(len(jobs), 1)
+        self.assertEqual(jobs[0]["id"], 1)
+        self.assertEqual(jobs[0]["status"], "running")
+
+    def test_stop_one_time_job_persists_stopped_status(self):
+        job = server.create_one_time_job({"action": "copy", "src": "/tmp/a", "dst": "/tmp/b"})
+        stopped = server.stop_one_time_job(job["id"])
+
+        self.assertEqual(stopped["status"], "stopped")
+        self.assertFalse(stopped["running"])
+        self.assertIsNotNone(stopped["statusSnapshot"])
+
+    def test_run_one_time_job_again_reuses_record_with_new_jobid(self):
+        job = server.create_one_time_job({"action": "copy", "src": "/tmp/a", "dst": "/tmp/b"})
+        server.stop_one_time_job(job["id"])
+        restarted = server.manual_run_one_time_job(job["id"])
+
+        self.assertEqual(restarted["id"], job["id"])
+        self.assertEqual(restarted["jobid"], 102)
+        self.assertEqual(restarted["status"], "running")
+
+
 if __name__ == "__main__":
     unittest.main()