feat(schedule): move system jobs to DB-driven config and dashboard management
This commit is contained in:
liuwei
@@ -1,10 +1,21 @@
|
||||
import asyncio
|
||||
import inspect
|
||||
import threading
|
||||
from collections import deque
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Callable, Awaitable, List, Dict, Optional, Any
|
||||
from typing import Callable, Awaitable, List, Dict, Optional, Any, Tuple
|
||||
|
||||
|
||||
class AsyncJob:
|
||||
"""异步定时任务中心。
|
||||
|
||||
设计目标：
|
||||
1. 保持装饰器用法兼容（原有插件和 main.py 不用重写）
|
||||
2. 支持运行中动态增删改任务
|
||||
3. 提供任务可观测信息（状态、下次运行、执行日志）
|
||||
4. 支持后台手动触发、启停、修改调度策略
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self._jobs: Dict[str, Dict[str, Any]] = {}
|
||||
self._running_tasks: Dict[str, asyncio.Task] = {}
|
||||
@@ -35,34 +46,242 @@ class AsyncJob:
|
||||
return value
|
||||
return None
|
||||
|
||||
def _register(self, func: Callable, wrapper: Callable[[], Awaitable], trigger: str):
|
||||
@staticmethod
|
||||
def _safe_iso(dt: Optional[datetime]) -> Optional[str]:
|
||||
return dt.isoformat(timespec="seconds") if dt else None
|
||||
|
||||
@staticmethod
|
||||
def _normalize_time_str(time_str: str) -> str:
|
||||
text = str(time_str or "").strip()
|
||||
if not text:
|
||||
raise ValueError("时间不能为空")
|
||||
return datetime.strptime(text, "%H:%M").strftime("%H:%M")
|
||||
|
||||
def _normalize_trigger_config(self, trigger_type: str, trigger_config: Dict[str, Any]) -> Dict[str, Any]:
|
||||
if trigger_type == "every_seconds":
|
||||
seconds = int(trigger_config.get("seconds", 0))
|
||||
if seconds <= 0:
|
||||
raise ValueError("seconds 必须大于 0")
|
||||
return {"seconds": seconds}
|
||||
|
||||
if trigger_type == "at_times":
|
||||
time_list = trigger_config.get("time_list", [])
|
||||
if not isinstance(time_list, list) or not time_list:
|
||||
raise ValueError("time_list 不能为空")
|
||||
normalized = sorted(set(self._normalize_time_str(t) for t in time_list))
|
||||
return {"time_list": normalized}
|
||||
|
||||
if trigger_type == "every_weekday_time":
|
||||
weekday = int(trigger_config.get("weekday", -1))
|
||||
if weekday < 0 or weekday > 6:
|
||||
raise ValueError("weekday 必须在 0-6")
|
||||
return {"weekday": weekday, "time_str": self._normalize_time_str(trigger_config.get("time_str", ""))}
|
||||
|
||||
if trigger_type == "every_week_time":
|
||||
weekday = int(trigger_config.get("weekday", -1))
|
||||
if weekday < 0 or weekday > 6:
|
||||
raise ValueError("weekday 必须在 0-6")
|
||||
return {"weekday": weekday, "time_str": self._normalize_time_str(trigger_config.get("time_str", ""))}
|
||||
|
||||
if trigger_type == "every_month_last_day_time":
|
||||
return {"time_str": self._normalize_time_str(trigger_config.get("time_str", ""))}
|
||||
|
||||
raise ValueError(f"不支持的触发器类型: {trigger_type}")
|
||||
|
||||
def _format_trigger(self, trigger_type: str, trigger_config: Dict[str, Any]) -> str:
|
||||
if trigger_type == "every_seconds":
|
||||
return f"每 {trigger_config['seconds']} 秒"
|
||||
if trigger_type == "at_times":
|
||||
return "每天 " + ", ".join(trigger_config.get("time_list", []))
|
||||
if trigger_type == "every_weekday_time":
|
||||
return f"每周{trigger_config['weekday']} {trigger_config['time_str']}"
|
||||
if trigger_type == "every_week_time":
|
||||
return f"每周{trigger_config['weekday']} {trigger_config['time_str']}"
|
||||
if trigger_type == "every_month_last_day_time":
|
||||
return f"每月最后一天 {trigger_config['time_str']}"
|
||||
return trigger_type
|
||||
|
||||
def _compute_next_run(self, trigger_type: str, trigger_config: Dict[str, Any], now: Optional[datetime] = None) -> datetime:
|
||||
current = now or datetime.now()
|
||||
|
||||
if trigger_type == "every_seconds":
|
||||
return current + timedelta(seconds=int(trigger_config["seconds"]))
|
||||
|
||||
if trigger_type == "at_times":
|
||||
parsed_times = [datetime.strptime(t, "%H:%M").time() for t in trigger_config.get("time_list", [])]
|
||||
targets = []
|
||||
for t in parsed_times:
|
||||
target = datetime.combine(current.date(), t)
|
||||
if target <= current:
|
||||
target += timedelta(days=1)
|
||||
targets.append(target)
|
||||
return min(targets)
|
||||
|
||||
if trigger_type in ("every_weekday_time", "every_week_time"):
|
||||
weekday = int(trigger_config["weekday"])
|
||||
target_time = datetime.strptime(trigger_config["time_str"], "%H:%M").time()
|
||||
days_ahead = (weekday - current.weekday() + 7) % 7
|
||||
target_date = current.date() + timedelta(days=days_ahead)
|
||||
target_dt = datetime.combine(target_date, target_time)
|
||||
if target_dt <= current:
|
||||
target_dt += timedelta(days=7)
|
||||
return target_dt
|
||||
|
||||
if trigger_type == "every_month_last_day_time":
|
||||
target_time = datetime.strptime(trigger_config["time_str"], "%H:%M").time()
|
||||
if current.month == 12:
|
||||
next_month = datetime(current.year + 1, 1, 1)
|
||||
else:
|
||||
next_month = datetime(current.year, current.month + 1, 1)
|
||||
last_day = next_month - timedelta(days=1)
|
||||
target_dt = datetime.combine(last_day.date(), target_time)
|
||||
if target_dt <= current:
|
||||
if current.month == 12:
|
||||
next_month = datetime(current.year + 1, 2, 1)
|
||||
elif current.month == 11:
|
||||
next_month = datetime(current.year + 1, 1, 1)
|
||||
else:
|
||||
next_month = datetime(current.year, current.month + 2, 1)
|
||||
last_day = next_month - timedelta(days=1)
|
||||
target_dt = datetime.combine(last_day.date(), target_time)
|
||||
return target_dt
|
||||
|
||||
raise ValueError(f"未知触发器类型: {trigger_type}")
|
||||
|
||||
def _append_log(self, job: Dict[str, Any], level: str, message: str):
|
||||
logs: deque = job["logs"]
|
||||
logs.append(
|
||||
{
|
||||
"time": self._safe_iso(datetime.now()),
|
||||
"level": level,
|
||||
"message": message,
|
||||
}
|
||||
)
|
||||
|
||||
def _register(
|
||||
self,
|
||||
func: Callable,
|
||||
trigger_type: str,
|
||||
trigger_config: Dict[str, Any],
|
||||
job_name: Optional[str] = None,
|
||||
description: str = "",
|
||||
job_key: Optional[str] = None,
|
||||
):
|
||||
owner = self._infer_owner(func)
|
||||
normalized_config = self._normalize_trigger_config(trigger_type, trigger_config)
|
||||
job_id = self._next_job_id()
|
||||
display_name = (job_name or getattr(func, "__name__", None) or job_id).strip()
|
||||
owner_name = owner.__class__.__name__ if owner is not None else "system"
|
||||
created_at = datetime.now()
|
||||
|
||||
with self._lock:
|
||||
self._jobs[job_id] = {
|
||||
"id": job_id,
|
||||
"job_key": (job_key or "").strip(),
|
||||
"name": display_name,
|
||||
"description": description or "",
|
||||
"func": func,
|
||||
"wrapper": wrapper,
|
||||
"trigger": trigger,
|
||||
"trigger_type": trigger_type,
|
||||
"trigger_config": normalized_config,
|
||||
"trigger_text": self._format_trigger(trigger_type, normalized_config),
|
||||
"owner_id": id(owner) if owner is not None else None,
|
||||
"owner_name": owner.__class__.__name__ if owner is not None else None,
|
||||
"owner_name": owner_name,
|
||||
"enabled": True,
|
||||
"running": False,
|
||||
"last_run_at": None,
|
||||
"last_status": "never",
|
||||
"last_error": "",
|
||||
"last_duration_ms": None,
|
||||
"next_run_at": None,
|
||||
"run_count": 0,
|
||||
"success_count": 0,
|
||||
"fail_count": 0,
|
||||
"created_at": created_at,
|
||||
"updated_at": created_at,
|
||||
"logs": deque(maxlen=200),
|
||||
}
|
||||
self._append_log(self._jobs[job_id], "info", f"任务已注册: {display_name}")
|
||||
if self._running and self._loop and self._loop.is_running():
|
||||
self._loop.call_soon_threadsafe(self._start_job_in_loop, job_id)
|
||||
return job_id
|
||||
|
||||
async def _execute_job(self, job_id: str, reason: str = "schedule") -> Tuple[bool, str]:
|
||||
job = self._jobs.get(job_id)
|
||||
if not job:
|
||||
return False, "任务不存在"
|
||||
|
||||
if job.get("running"):
|
||||
return False, "任务正在执行中"
|
||||
|
||||
func = job["func"]
|
||||
started_at = datetime.now()
|
||||
job["running"] = True
|
||||
job["last_run_at"] = started_at
|
||||
self._append_log(job, "info", f"开始执行，触发来源: {reason}")
|
||||
|
||||
try:
|
||||
result = func()
|
||||
if inspect.isawaitable(result):
|
||||
await result
|
||||
job["last_status"] = "success"
|
||||
job["last_error"] = ""
|
||||
job["success_count"] += 1
|
||||
self._append_log(job, "success", "执行成功")
|
||||
return True, "执行成功"
|
||||
except asyncio.CancelledError:
|
||||
job["last_status"] = "cancelled"
|
||||
job["last_error"] = "任务被取消"
|
||||
self._append_log(job, "warning", "任务被取消")
|
||||
raise
|
||||
except Exception as e:
|
||||
job["last_status"] = "failed"
|
||||
job["last_error"] = str(e)
|
||||
job["fail_count"] += 1
|
||||
self._append_log(job, "error", f"执行失败: {e}")
|
||||
return False, str(e)
|
||||
finally:
|
||||
finished_at = datetime.now()
|
||||
job["run_count"] += 1
|
||||
job["running"] = False
|
||||
job["last_duration_ms"] = int((finished_at - started_at).total_seconds() * 1000)
|
||||
job["updated_at"] = finished_at
|
||||
|
||||
async def _job_loop(self, job_id: str):
|
||||
while True:
|
||||
job = self._jobs.get(job_id)
|
||||
if not job:
|
||||
return
|
||||
|
||||
if not job.get("enabled", True):
|
||||
job["next_run_at"] = None
|
||||
await asyncio.sleep(1)
|
||||
continue
|
||||
|
||||
try:
|
||||
next_run = self._compute_next_run(job["trigger_type"], job["trigger_config"])
|
||||
except Exception as e:
|
||||
job["last_status"] = "invalid_schedule"
|
||||
job["last_error"] = str(e)
|
||||
job["next_run_at"] = None
|
||||
self._append_log(job, "error", f"调度配置非法: {e}")
|
||||
await asyncio.sleep(5)
|
||||
continue
|
||||
|
||||
job["next_run_at"] = next_run
|
||||
wait_seconds = max((next_run - datetime.now()).total_seconds(), 0)
|
||||
await asyncio.sleep(wait_seconds)
|
||||
|
||||
# 睡眠结束后再检查一次，避免刚被禁用/删除还执行
|
||||
job = self._jobs.get(job_id)
|
||||
if not job or not job.get("enabled", True):
|
||||
continue
|
||||
|
||||
await self._execute_job(job_id, reason="schedule")
|
||||
|
||||
def _start_job_in_loop(self, job_id: str):
|
||||
job = self._jobs.get(job_id)
|
||||
if not job or job_id in self._running_tasks:
|
||||
if job_id in self._running_tasks:
|
||||
return
|
||||
|
||||
async def runner():
|
||||
try:
|
||||
await job["wrapper"]()
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[AsyncJob] 任务异常退出: {job_id}, trigger={job.get('trigger')}, error={e}")
|
||||
|
||||
task = asyncio.create_task(runner(), name=f"async_job:{job_id}")
|
||||
task = asyncio.create_task(self._job_loop(job_id), name=f"async_job:{job_id}")
|
||||
self._running_tasks[job_id] = task
|
||||
task.add_done_callback(lambda _task, _id=job_id: self._running_tasks.pop(_id, None))
|
||||
|
||||
@@ -71,6 +290,11 @@ class AsyncJob:
|
||||
if task:
|
||||
task.cancel()
|
||||
|
||||
def _restart_job_in_loop(self, job_id: str):
|
||||
self._cancel_job_in_loop(job_id)
|
||||
if job_id in self._jobs:
|
||||
self._start_job_in_loop(job_id)
|
||||
|
||||
def remove_job(self, job_id: str) -> bool:
|
||||
with self._lock:
|
||||
existed = job_id in self._jobs or job_id in self._running_tasks
|
||||
@@ -97,161 +321,233 @@ class AsyncJob:
|
||||
removed += 1
|
||||
return removed
|
||||
|
||||
def every_seconds(self, seconds: int):
|
||||
def decorator(func: Callable):
|
||||
async def wrapper():
|
||||
while True:
|
||||
try:
|
||||
await func()
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[AsyncJob] every_seconds 任务执行异常: {e}")
|
||||
await asyncio.sleep(seconds)
|
||||
def get_jobs_snapshot(self) -> List[Dict[str, Any]]:
|
||||
with self._lock:
|
||||
snapshots = []
|
||||
for job in self._jobs.values():
|
||||
snapshots.append(
|
||||
{
|
||||
"id": job["id"],
|
||||
"job_key": job.get("job_key", ""),
|
||||
"name": job["name"],
|
||||
"description": job.get("description", ""),
|
||||
"trigger_type": job["trigger_type"],
|
||||
"trigger_config": dict(job["trigger_config"]),
|
||||
"trigger_text": job.get("trigger_text", ""),
|
||||
"owner_name": job.get("owner_name", "system"),
|
||||
"enabled": job.get("enabled", True),
|
||||
"running": job.get("running", False),
|
||||
"last_run_at": self._safe_iso(job.get("last_run_at")),
|
||||
"last_status": job.get("last_status", "never"),
|
||||
"last_error": job.get("last_error", ""),
|
||||
"last_duration_ms": job.get("last_duration_ms"),
|
||||
"next_run_at": self._safe_iso(job.get("next_run_at")),
|
||||
"run_count": job.get("run_count", 0),
|
||||
"success_count": job.get("success_count", 0),
|
||||
"fail_count": job.get("fail_count", 0),
|
||||
"created_at": self._safe_iso(job.get("created_at")),
|
||||
"updated_at": self._safe_iso(job.get("updated_at")),
|
||||
}
|
||||
)
|
||||
snapshots.sort(key=lambda item: item["name"])
|
||||
return snapshots
|
||||
|
||||
self._register(func, wrapper, f"every_seconds({seconds})")
|
||||
def get_job_logs(self, job_id: str, limit: int = 100) -> List[Dict[str, Any]]:
|
||||
with self._lock:
|
||||
job = self._jobs.get(job_id)
|
||||
if not job:
|
||||
return []
|
||||
logs: deque = job.get("logs", deque())
|
||||
data = list(logs)
|
||||
if limit > 0:
|
||||
data = data[-limit:]
|
||||
return data
|
||||
|
||||
def trigger_job_now(self, job_id: str, operator: str = "dashboard") -> Tuple[bool, str]:
|
||||
with self._lock:
|
||||
job = self._jobs.get(job_id)
|
||||
loop = self._loop
|
||||
running = self._running
|
||||
|
||||
if not job:
|
||||
return False, "任务不存在"
|
||||
if job.get("running"):
|
||||
return False, "任务正在执行中"
|
||||
if not running or not loop or not loop.is_running():
|
||||
return False, "任务调度器未运行"
|
||||
|
||||
def _trigger():
|
||||
asyncio.create_task(self._execute_job(job_id, reason=f"manual:{operator}"))
|
||||
|
||||
loop.call_soon_threadsafe(_trigger)
|
||||
return True, "任务已触发"
|
||||
|
||||
def get_job_id_by_key(self, job_key: str) -> Optional[str]:
|
||||
key = str(job_key or "").strip()
|
||||
if not key:
|
||||
return None
|
||||
with self._lock:
|
||||
for job_id, job in self._jobs.items():
|
||||
if job.get("job_key") == key:
|
||||
return job_id
|
||||
return None
|
||||
|
||||
def register_callable(
|
||||
self,
|
||||
func: Callable,
|
||||
trigger_type: str,
|
||||
trigger_config: Dict[str, Any],
|
||||
job_name: Optional[str] = None,
|
||||
description: str = "",
|
||||
job_key: Optional[str] = None,
|
||||
) -> str:
|
||||
"""运行时注册任务（非装饰器方式）。"""
|
||||
return self._register(
|
||||
func=func,
|
||||
trigger_type=trigger_type,
|
||||
trigger_config=trigger_config,
|
||||
job_name=job_name,
|
||||
description=description,
|
||||
job_key=job_key,
|
||||
)
|
||||
|
||||
def set_job_enabled(self, job_id: str, enabled: bool) -> Tuple[bool, str]:
|
||||
with self._lock:
|
||||
job = self._jobs.get(job_id)
|
||||
loop = self._loop
|
||||
running = self._running
|
||||
if not job:
|
||||
return False, "任务不存在"
|
||||
job["enabled"] = bool(enabled)
|
||||
job["updated_at"] = datetime.now()
|
||||
state_text = "启用" if enabled else "停用"
|
||||
self._append_log(job, "info", f"任务已{state_text}")
|
||||
|
||||
if running and loop and loop.is_running():
|
||||
loop.call_soon_threadsafe(self._restart_job_in_loop, job_id)
|
||||
return True, f"任务已{state_text}"
|
||||
|
||||
def update_job_schedule(self, job_id: str, trigger_type: str, trigger_config: Dict[str, Any]) -> Tuple[bool, str]:
|
||||
try:
|
||||
normalized = self._normalize_trigger_config(trigger_type, trigger_config)
|
||||
except Exception as e:
|
||||
return False, f"调度参数非法: {e}"
|
||||
|
||||
with self._lock:
|
||||
job = self._jobs.get(job_id)
|
||||
loop = self._loop
|
||||
running = self._running
|
||||
if not job:
|
||||
return False, "任务不存在"
|
||||
|
||||
job["trigger_type"] = trigger_type
|
||||
job["trigger_config"] = normalized
|
||||
job["trigger_text"] = self._format_trigger(trigger_type, normalized)
|
||||
job["updated_at"] = datetime.now()
|
||||
self._append_log(job, "info", f"调度已更新: {job['trigger_text']}")
|
||||
|
||||
if running and loop and loop.is_running():
|
||||
loop.call_soon_threadsafe(self._restart_job_in_loop, job_id)
|
||||
return True, "调度更新成功"
|
||||
|
||||
def every_seconds(self, seconds: int, job_name: Optional[str] = None, description: str = "", job_key: Optional[str] = None):
|
||||
def decorator(func: Callable):
|
||||
self._register(
|
||||
func=func,
|
||||
trigger_type="every_seconds",
|
||||
trigger_config={"seconds": seconds},
|
||||
job_name=job_name,
|
||||
description=description,
|
||||
job_key=job_key,
|
||||
)
|
||||
return func
|
||||
|
||||
return decorator
|
||||
|
||||
def every_minutes(self, minutes: int):
|
||||
return self.every_seconds(minutes * 60)
|
||||
def every_minutes(self, minutes: int, job_name: Optional[str] = None, description: str = "", job_key: Optional[str] = None):
|
||||
return self.every_seconds(minutes * 60, job_name=job_name, description=description, job_key=job_key)
|
||||
|
||||
def every_hours(self, hours: int):
|
||||
return self.every_seconds(hours * 3600)
|
||||
def every_hours(self, hours: int, job_name: Optional[str] = None, description: str = "", job_key: Optional[str] = None):
|
||||
return self.every_seconds(hours * 3600, job_name=job_name, description=description, job_key=job_key)
|
||||
|
||||
def at_times(self, time_list: List[str]):
|
||||
def at_times(
|
||||
self,
|
||||
time_list: List[str],
|
||||
job_name: Optional[str] = None,
|
||||
description: str = "",
|
||||
job_key: Optional[str] = None,
|
||||
):
|
||||
def decorator(func: Callable):
|
||||
parsed_times = [datetime.strptime(t, "%H:%M").time() for t in time_list]
|
||||
|
||||
async def wrapper():
|
||||
while True:
|
||||
now = datetime.now()
|
||||
targets = []
|
||||
for t in parsed_times:
|
||||
target = datetime.combine(now.date(), t)
|
||||
if target <= now:
|
||||
target += timedelta(days=1)
|
||||
targets.append(target)
|
||||
|
||||
next_target = min(targets)
|
||||
wait_seconds = (next_target - now).total_seconds()
|
||||
await asyncio.sleep(max(wait_seconds, 0))
|
||||
try:
|
||||
await func()
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[AsyncJob] at_times 任务执行异常: {e}")
|
||||
|
||||
self._register(func, wrapper, f"at_times({time_list})")
|
||||
self._register(
|
||||
func=func,
|
||||
trigger_type="at_times",
|
||||
trigger_config={"time_list": time_list},
|
||||
job_name=job_name,
|
||||
description=description,
|
||||
job_key=job_key,
|
||||
)
|
||||
return func
|
||||
|
||||
return decorator
|
||||
|
||||
def every_weekday_time(self, weekday: int, time_str: str):
|
||||
"""
|
||||
每周 weekday（0=周一） 的 time_str（如10:00）时间执行
|
||||
"""
|
||||
|
||||
def every_weekday_time(
|
||||
self,
|
||||
weekday: int,
|
||||
time_str: str,
|
||||
job_name: Optional[str] = None,
|
||||
description: str = "",
|
||||
job_key: Optional[str] = None,
|
||||
):
|
||||
def decorator(func: Callable):
|
||||
async def wrapper():
|
||||
while True:
|
||||
now = datetime.now()
|
||||
target_time = datetime.strptime(time_str, "%H:%M").time()
|
||||
|
||||
# 构造下一个执行时间
|
||||
days_ahead = (weekday - now.weekday() + 7) % 7
|
||||
target_date = now.date() + timedelta(days=days_ahead)
|
||||
target_dt = datetime.combine(target_date, target_time)
|
||||
|
||||
if target_dt <= now:
|
||||
target_dt += timedelta(days=7)
|
||||
|
||||
sleep_secs = (target_dt - now).total_seconds()
|
||||
await asyncio.sleep(sleep_secs)
|
||||
try:
|
||||
await func()
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[AsyncJob] every_weekday_time 任务执行异常: {e}")
|
||||
|
||||
self._register(func, wrapper, f"every_weekday_time({weekday}, {time_str})")
|
||||
self._register(
|
||||
func=func,
|
||||
trigger_type="every_weekday_time",
|
||||
trigger_config={"weekday": weekday, "time_str": time_str},
|
||||
job_name=job_name,
|
||||
description=description,
|
||||
job_key=job_key,
|
||||
)
|
||||
return func
|
||||
|
||||
return decorator
|
||||
|
||||
def every_week_time(self, weekday: int, time_str: str):
|
||||
"""
|
||||
每周 weekday（0=周一，6=周日） 的 time_str 时间执行
|
||||
"""
|
||||
|
||||
def every_week_time(
|
||||
self,
|
||||
weekday: int,
|
||||
time_str: str,
|
||||
job_name: Optional[str] = None,
|
||||
description: str = "",
|
||||
job_key: Optional[str] = None,
|
||||
):
|
||||
def decorator(func: Callable):
|
||||
async def wrapper():
|
||||
while True:
|
||||
now = datetime.now()
|
||||
target_time = datetime.strptime(time_str, "%H:%M").time()
|
||||
|
||||
days_ahead = (weekday - now.weekday() + 7) % 7
|
||||
target_date = now.date() + timedelta(days=days_ahead)
|
||||
target_dt = datetime.combine(target_date, target_time)
|
||||
|
||||
if target_dt <= now:
|
||||
target_dt += timedelta(days=7)
|
||||
|
||||
sleep_secs = (target_dt - now).total_seconds()
|
||||
await asyncio.sleep(sleep_secs)
|
||||
try:
|
||||
await func()
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[AsyncJob] every_week_time 任务执行异常: {e}")
|
||||
|
||||
self._register(func, wrapper, f"every_week_time({weekday}, {time_str})")
|
||||
self._register(
|
||||
func=func,
|
||||
trigger_type="every_week_time",
|
||||
trigger_config={"weekday": weekday, "time_str": time_str},
|
||||
job_name=job_name,
|
||||
description=description,
|
||||
job_key=job_key,
|
||||
)
|
||||
return func
|
||||
|
||||
return decorator
|
||||
|
||||
def every_month_last_day_time(self, time_str: str):
|
||||
"""
|
||||
每月最后一天的 time_str 时间执行
|
||||
"""
|
||||
|
||||
def every_month_last_day_time(
|
||||
self,
|
||||
time_str: str,
|
||||
job_name: Optional[str] = None,
|
||||
description: str = "",
|
||||
job_key: Optional[str] = None,
|
||||
):
|
||||
def decorator(func: Callable):
|
||||
async def wrapper():
|
||||
while True:
|
||||
now = datetime.now()
|
||||
target_time = datetime.strptime(time_str, "%H:%M").time()
|
||||
|
||||
if now.month == 12:
|
||||
next_month = datetime(now.year + 1, 1, 1)
|
||||
else:
|
||||
next_month = datetime(now.year, now.month + 1, 1)
|
||||
last_day = next_month - timedelta(days=1)
|
||||
target_dt = datetime.combine(last_day.date(), target_time)
|
||||
|
||||
if target_dt <= now:
|
||||
if now.month == 12:
|
||||
next_month = datetime(now.year + 1, 2, 1)
|
||||
elif now.month == 11:
|
||||
next_month = datetime(now.year + 1, 1, 1)
|
||||
else:
|
||||
next_month = datetime(now.year, now.month + 2, 1)
|
||||
last_day = next_month - timedelta(days=1)
|
||||
target_dt = datetime.combine(last_day.date(), target_time)
|
||||
|
||||
sleep_secs = (target_dt - now).total_seconds()
|
||||
await asyncio.sleep(sleep_secs)
|
||||
try:
|
||||
await func()
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as e:
|
||||
print(f"[AsyncJob] every_month_last_day_time 任务执行异常: {e}")
|
||||
|
||||
self._register(func, wrapper, f"every_month_last_day_time({time_str})")
|
||||
self._register(
|
||||
func=func,
|
||||
trigger_type="every_month_last_day_time",
|
||||
trigger_config={"time_str": time_str},
|
||||
job_name=job_name,
|
||||
description=description,
|
||||
job_key=job_key,
|
||||
)
|
||||
return func
|
||||
|
||||
return decorator
|
||||
|
||||
177
utils/system_jobs.py
Normal file
177
utils/system_jobs.py
Normal file
@@ -0,0 +1,177 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Awaitable, Callable, Dict, List
|
||||
|
||||
from loguru import logger
|
||||
|
||||
from db.system_job_db import SystemJobDBOperator
|
||||
from utils.decorator.async_job import async_job
|
||||
|
||||
|
||||
def get_system_job_definitions(robot) -> List[Dict[str, Any]]:
|
||||
"""系统任务定义（业务函数映射）。
|
||||
|
||||
说明：这里只维护“任务 key 与业务函数”的绑定关系；
|
||||
调度时间、启停状态全部从数据库 t_system_jobs 读取。
|
||||
"""
|
||||
return [
|
||||
{
|
||||
"job_key": "news_baidu_report_auto",
|
||||
"name": "百度新闻日报",
|
||||
"description": "每天 08:30 推送百度新闻",
|
||||
"trigger_type": "at_times",
|
||||
"trigger_config": {"time_list": ["08:30"]},
|
||||
"handler": robot.news_baidu_report_auto,
|
||||
},
|
||||
{
|
||||
"job_key": "epic_free_games",
|
||||
"name": "Epic 免费游戏推送",
|
||||
"description": "每周五 10:00 推送 Epic 免费游戏",
|
||||
"trigger_type": "every_weekday_time",
|
||||
"trigger_config": {"weekday": 4, "time_str": "10:00"},
|
||||
"handler": robot.send_epic_free_games,
|
||||
},
|
||||
{
|
||||
"job_key": "message_count_to_db",
|
||||
"name": "消息计数入库",
|
||||
"description": "每天 02:30 将 Redis 消息计数写入 SQLite",
|
||||
"trigger_type": "at_times",
|
||||
"trigger_config": {"time_list": ["02:30"]},
|
||||
"handler": robot.message_count_to_db,
|
||||
},
|
||||
{
|
||||
"job_key": "message_ranking_push",
|
||||
"name": "群消息排行推送",
|
||||
"description": "每天 09:30 生成并发送群消息排行",
|
||||
"trigger_type": "at_times",
|
||||
"trigger_config": {"time_list": ["09:30"]},
|
||||
"handler": robot.generate_and_send_ranking,
|
||||
},
|
||||
{
|
||||
"job_key": "sehuatang_pdf_push",
|
||||
"name": "涩图 PDF 推送",
|
||||
"description": "每天 15:30 生成并发送涩图 PDF",
|
||||
"trigger_type": "at_times",
|
||||
"trigger_config": {"time_list": ["15:30"]},
|
||||
"handler": robot.generate_sehuatang_pdf,
|
||||
},
|
||||
{
|
||||
"job_key": "xiuren_download",
|
||||
"name": "秀人网下载任务",
|
||||
"description": "每天 01:30 执行秀人网下载任务",
|
||||
"trigger_type": "at_times",
|
||||
"trigger_config": {"time_list": ["01:30"]},
|
||||
"handler": robot.xiu_ren_download_task,
|
||||
},
|
||||
{
|
||||
"job_key": "shenshi_r15_download",
|
||||
"name": "绅士 R15 下载任务",
|
||||
"description": "每天 02:30 执行绅士 R15 下载任务",
|
||||
"trigger_type": "at_times",
|
||||
"trigger_config": {"time_list": ["02:30"]},
|
||||
"handler": robot.shen_shi_download_task,
|
||||
},
|
||||
{
|
||||
"job_key": "login_check",
|
||||
"name": "登录状态巡检",
|
||||
"description": "每天 14:43 执行登录二次校验",
|
||||
"trigger_type": "at_times",
|
||||
"trigger_config": {"time_list": ["14:43"]},
|
||||
"handler": robot.login_twice_auto_auth,
|
||||
},
|
||||
{
|
||||
"job_key": "update_image_cache",
|
||||
"name": "图片缓存更新",
|
||||
"description": "每天 05:00 扫描并更新图片缓存",
|
||||
"trigger_type": "at_times",
|
||||
"trigger_config": {"time_list": ["05:00"]},
|
||||
"handler": _build_image_cache_handler(robot),
|
||||
},
|
||||
{
|
||||
"job_key": "process_pending_images",
|
||||
"name": "待下载图片补偿处理",
|
||||
"description": "每 5 分钟处理一次待下载图片/表情，避免数据库锁竞争",
|
||||
"trigger_type": "every_seconds",
|
||||
"trigger_config": {"seconds": 300},
|
||||
"handler": _build_process_pending_images_handler(robot),
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
def _build_image_cache_handler(robot) -> Callable[[], Awaitable[None]]:
|
||||
async def _handler():
|
||||
from plugins.xiuren_image.images_cache import ImageCacheManager
|
||||
|
||||
logger.info("开始执行图片缓存更新任务")
|
||||
manager = ImageCacheManager("/mnt/nfs_share")
|
||||
await manager.update_image_cache()
|
||||
logger.info("图片缓存更新完成")
|
||||
|
||||
return _handler
|
||||
|
||||
|
||||
def _build_process_pending_images_handler(robot) -> Callable[[], Awaitable[None]]:
|
||||
async def _handler():
|
||||
if hasattr(robot, "message_storage") and robot.message_storage:
|
||||
await robot.message_storage.process_pending_images(minutes_ago=10, batch_size=20)
|
||||
|
||||
return _handler
|
||||
|
||||
|
||||
class SystemJobLoader:
|
||||
"""系统任务加载器：从数据库读取调度配置并注册到 async_job。"""
|
||||
|
||||
def __init__(self, robot, system_job_db: SystemJobDBOperator):
|
||||
self.robot = robot
|
||||
self.db = system_job_db
|
||||
self._job_defs = {item["job_key"]: item for item in get_system_job_definitions(robot)}
|
||||
self._registered_job_ids: List[str] = []
|
||||
|
||||
def init_and_load(self):
|
||||
self.db.init_tables()
|
||||
self._seed_defaults()
|
||||
self.reload_from_db()
|
||||
|
||||
def _seed_defaults(self):
|
||||
for item in self._job_defs.values():
|
||||
existed = self.db.get_job(item["job_key"])
|
||||
if existed:
|
||||
continue
|
||||
self.db.upsert_job(
|
||||
{
|
||||
"job_key": item["job_key"],
|
||||
"name": item["name"],
|
||||
"description": item.get("description", ""),
|
||||
"trigger_type": item["trigger_type"],
|
||||
"trigger_config": item["trigger_config"],
|
||||
"enabled": True,
|
||||
}
|
||||
)
|
||||
|
||||
def reload_from_db(self):
|
||||
# 先移除当前注册任务，避免重复调度
|
||||
for job_id in self._registered_job_ids:
|
||||
async_job.remove_job(job_id)
|
||||
self._registered_job_ids = []
|
||||
|
||||
jobs = self.db.list_jobs()
|
||||
for row in jobs:
|
||||
job_key = row.get("job_key")
|
||||
if not row.get("enabled", 1):
|
||||
continue
|
||||
definition = self._job_defs.get(job_key)
|
||||
if not definition:
|
||||
logger.warning(f"系统任务 {job_key} 在代码中无处理器，已跳过注册")
|
||||
continue
|
||||
|
||||
handler = definition["handler"]
|
||||
job_id = async_job.register_callable(
|
||||
func=handler,
|
||||
trigger_type=row.get("trigger_type", definition["trigger_type"]),
|
||||
trigger_config=row.get("trigger_config", definition["trigger_config"]),
|
||||
job_name=row.get("name") or definition["name"],
|
||||
description=row.get("description") or definition.get("description", ""),
|
||||
job_key=job_key,
|
||||
)
|
||||
self._registered_job_ids.append(job_id)
|
||||
Reference in New Issue
Block a user