251 lines
8.3 KiB
Python
251 lines
8.3 KiB
Python
"""异步盘后管道任务注册表 — 每个 job 独立 JSON 文件。
|
|
|
|
设计:
|
|
- job_store/ 文件夹,每个 job 一个 {id}.json,最多保留 max_jobs 个文件
|
|
- running/pending 状态的 job 仅存内存(高频读写)
|
|
- succeeded/failed 后写入独立文件并从内存释放
|
|
- 列表查询 = 内存中的活跃 job + 磁盘文件扫描,按时间排序
|
|
- 单个查询 = 内存优先,没有则读磁盘
|
|
- 创建新 job 前检查文件数量,>= max_jobs 时删除最老的文件
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import threading
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
JobStatus = Literal["pending", "running", "succeeded", "failed"]
|
|
|
|
|
|
def _default_store_dir() -> Path:
|
|
from app.config import settings
|
|
return settings.data_dir / "job_store"
|
|
|
|
|
|
_STORE_DIR = _default_store_dir()
|
|
|
|
|
|
class JobStore:
|
|
def __init__(self, max_jobs: int = 50, store_dir: Path = _STORE_DIR) -> None:
|
|
self._max_jobs = max_jobs
|
|
self._store_dir = store_dir
|
|
self._active_jobs: dict[str, dict[str, Any]] = {} # running/pending
|
|
self._active_id: str | None = None
|
|
self._lock = threading.Lock()
|
|
self._store_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# ===== persistence =====
|
|
|
|
def _write_file(self, job: dict[str, Any]) -> None:
|
|
"""将终态 job 写入独立 JSON 文件。"""
|
|
path = self._store_dir / f"{job['id']}.json"
|
|
try:
|
|
path.write_text(
|
|
json.dumps(job, ensure_ascii=False, indent=None),
|
|
encoding="utf-8",
|
|
)
|
|
except Exception:
|
|
logger.warning("failed to write job file %s", path)
|
|
|
|
def _read_file(self, job_id: str) -> dict[str, Any] | None:
|
|
"""从磁盘读取单个 job 文件。"""
|
|
path = self._store_dir / f"{job_id}.json"
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
return json.loads(path.read_text("utf-8"))
|
|
except Exception:
|
|
logger.warning("failed to read job file %s", path)
|
|
return None
|
|
|
|
def _delete_oldest(self) -> None:
|
|
"""删除最老的 job 文件,保持文件数量 < max_jobs。"""
|
|
try:
|
|
files = sorted(self._store_dir.glob("*.json"), key=lambda f: f.stat().st_mtime)
|
|
except Exception:
|
|
return
|
|
while len(files) >= self._max_jobs:
|
|
oldest = files.pop(0)
|
|
try:
|
|
oldest.unlink()
|
|
except Exception:
|
|
logger.warning("failed to delete old job file %s", oldest)
|
|
|
|
def _job_files_sorted(self) -> list[dict[str, Any]]:
|
|
"""扫描磁盘上所有 job 文件,按 started_at 从新到旧排序。"""
|
|
jobs: list[dict[str, Any]] = []
|
|
for f in self._store_dir.glob("*.json"):
|
|
try:
|
|
jobs.append(json.loads(f.read_text("utf-8")))
|
|
except Exception:
|
|
continue
|
|
jobs.sort(key=lambda j: j.get("started_at") or "", reverse=True)
|
|
return jobs
|
|
|
|
# ===== lifecycle =====
|
|
|
|
def create(self) -> str:
|
|
with self._lock:
|
|
if self._active_id and self._active_jobs.get(self._active_id, {}).get("status") == "running":
|
|
return self._active_id
|
|
|
|
job_id = uuid.uuid4().hex[:10]
|
|
self._active_jobs[job_id] = {
|
|
"id": job_id,
|
|
"status": "pending",
|
|
"stage": "init",
|
|
"progress": 0,
|
|
"stage_pct": 0,
|
|
"log": [],
|
|
"started_at": None,
|
|
"finished_at": None,
|
|
"duration_s": None,
|
|
"result": None,
|
|
"error": None,
|
|
}
|
|
self._active_id = job_id
|
|
return job_id
|
|
|
|
def start(self, job_id: str) -> None:
|
|
with self._lock:
|
|
j = self._active_jobs.get(job_id)
|
|
if not j:
|
|
return
|
|
j["status"] = "running"
|
|
j["started_at"] = datetime.utcnow().isoformat(timespec="seconds") + "Z"
|
|
|
|
def succeed(self, job_id: str, result: Any) -> None:
|
|
with self._lock:
|
|
j = self._active_jobs.pop(job_id, None)
|
|
if not j:
|
|
return
|
|
j["status"] = "succeeded"
|
|
j["finished_at"] = datetime.utcnow().isoformat(timespec="seconds") + "Z"
|
|
j["progress"] = 100
|
|
j["result"] = result
|
|
j["duration_s"] = _duration_s(j)
|
|
if self._active_id == job_id:
|
|
self._active_id = None
|
|
self._delete_oldest()
|
|
self._write_file(j)
|
|
|
|
def fail(self, job_id: str, error: str) -> None:
|
|
with self._lock:
|
|
j = self._active_jobs.pop(job_id, None)
|
|
if not j:
|
|
return
|
|
j["status"] = "failed"
|
|
j["finished_at"] = datetime.utcnow().isoformat(timespec="seconds") + "Z"
|
|
j["error"] = error
|
|
j["duration_s"] = _duration_s(j)
|
|
if self._active_id == job_id:
|
|
self._active_id = None
|
|
self._delete_oldest()
|
|
self._write_file(j)
|
|
|
|
# ===== progress =====
|
|
|
|
def progress(self, job_id: str, stage: str, pct: int, msg: str,
|
|
stage_pct: int | None = None, skip_log: bool = False) -> None:
|
|
with self._lock:
|
|
j = self._active_jobs.get(job_id)
|
|
if not j:
|
|
return
|
|
j["stage"] = stage
|
|
j["progress"] = max(0, min(100, int(pct)))
|
|
if stage_pct is not None:
|
|
j["stage_pct"] = max(0, min(100, int(stage_pct)))
|
|
elif j["stage"] != stage:
|
|
j["stage_pct"] = 0
|
|
entry = {
|
|
"ts": datetime.utcnow().isoformat(timespec="seconds") + "Z",
|
|
"stage": stage,
|
|
"msg": msg,
|
|
}
|
|
if skip_log:
|
|
entry["_skip"] = True
|
|
if skip_log and j["log"] and j["log"][-1].get("stage") == stage and j["log"][-1].get("_skip"):
|
|
j["log"][-1] = entry
|
|
else:
|
|
j["log"].append(entry)
|
|
if len(j["log"]) > 200:
|
|
j["log"] = j["log"][-200:]
|
|
|
|
# ===== query =====
|
|
|
|
def get(self, job_id: str) -> dict[str, Any] | None:
|
|
# 内存中的活跃 job 优先
|
|
j = self._active_jobs.get(job_id)
|
|
if j:
|
|
return j
|
|
# 否则从磁盘读
|
|
return self._read_file(job_id)
|
|
|
|
def list_recent(self, limit: int = 20) -> list[dict[str, Any]]:
|
|
# 合并: 内存中的活跃 job + 磁盘文件
|
|
all_jobs: list[dict[str, Any]] = list(self._active_jobs.values())
|
|
all_jobs.extend(self._job_files_sorted())
|
|
# 按 started_at 从新到旧排序,去重(理论上不会有重复)
|
|
seen: set[str] = set()
|
|
result: list[dict[str, Any]] = []
|
|
for j in sorted(all_jobs, key=lambda x: x.get("started_at") or "", reverse=True):
|
|
jid = j["id"]
|
|
if jid in seen:
|
|
continue
|
|
seen.add(jid)
|
|
result.append(_summary(j))
|
|
if len(result) >= limit:
|
|
break
|
|
return result
|
|
|
|
def active_id(self) -> str | None:
|
|
return self._active_id
|
|
|
|
def clear(self) -> None:
|
|
"""清空所有任务（内存 + 磁盘文件）。"""
|
|
with self._lock:
|
|
self._active_jobs.clear()
|
|
self._active_id = None
|
|
for f in self._store_dir.glob("*.json"):
|
|
try:
|
|
f.unlink()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _summary(j: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"id": j["id"],
|
|
"status": j["status"],
|
|
"stage": j["stage"],
|
|
"progress": j["progress"],
|
|
"stage_pct": j.get("stage_pct", 0),
|
|
"started_at": j["started_at"],
|
|
"finished_at": j["finished_at"],
|
|
"duration_s": j["duration_s"],
|
|
"result": j["result"],
|
|
"error": j["error"],
|
|
}
|
|
|
|
|
|
def _duration_s(j: dict[str, Any]) -> float | None:
|
|
if not j.get("started_at") or not j.get("finished_at"):
|
|
return None
|
|
try:
|
|
s = datetime.fromisoformat(j["started_at"])
|
|
e = datetime.fromisoformat(j["finished_at"])
|
|
return round((e - s).total_seconds(), 2)
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
|
|
|
|
# 进程内单例
|
|
job_store = JobStore()
|