|
|
import json
|
|
|
from collections.abc import Callable
|
|
|
from datetime import datetime
|
|
|
from typing import Any
|
|
|
|
|
|
from apscheduler.events import (
|
|
|
EVENT_ALL,
|
|
|
EVENT_ALL_JOBS_REMOVED,
|
|
|
EVENT_EXECUTOR_ADDED,
|
|
|
EVENT_EXECUTOR_REMOVED,
|
|
|
EVENT_JOB_ADDED,
|
|
|
EVENT_JOB_ERROR,
|
|
|
EVENT_JOB_EXECUTED,
|
|
|
EVENT_JOB_MAX_INSTANCES,
|
|
|
EVENT_JOB_MISSED,
|
|
|
EVENT_JOB_MODIFIED,
|
|
|
EVENT_JOB_REMOVED,
|
|
|
EVENT_JOB_SUBMITTED,
|
|
|
EVENT_JOBSTORE_ADDED,
|
|
|
EVENT_JOBSTORE_REMOVED,
|
|
|
EVENT_SCHEDULER_PAUSED,
|
|
|
EVENT_SCHEDULER_RESUMED,
|
|
|
EVENT_SCHEDULER_SHUTDOWN,
|
|
|
EVENT_SCHEDULER_START,
|
|
|
EVENT_SCHEDULER_STARTED,
|
|
|
JobEvent,
|
|
|
JobExecutionEvent,
|
|
|
JobSubmissionEvent,
|
|
|
SchedulerEvent,
|
|
|
)
|
|
|
from apscheduler.executors.asyncio import AsyncIOExecutor
|
|
|
from apscheduler.executors.pool import ProcessPoolExecutor, ThreadPoolExecutor
|
|
|
from apscheduler.job import Job
|
|
|
from apscheduler.jobstores.base import ConflictingIdError
|
|
|
from apscheduler.jobstores.memory import MemoryJobStore
|
|
|
from apscheduler.jobstores.redis import RedisJobStore
|
|
|
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
|
|
|
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
|
from apscheduler.triggers.date import DateTrigger
|
|
|
from apscheduler.triggers.interval import IntervalTrigger
|
|
|
from redis.asyncio import Redis
|
|
|
|
|
|
from app.config.setting import settings
|
|
|
from app.core.database import engine
|
|
|
from app.core.logger import logger
|
|
|
from app.plugin.module_task.cronjob.node.model import NodeModel
|
|
|
from app.utils.cron_util import CronUtil
|
|
|
|
|
|
# 任务状态常量（与 JobModel.status 注释保持一致：0:待执行 1:执行中 2:成功 3:失败 4:超时 5:已取消）
|
|
|
JOB_STATUS_PENDING = 0
|
|
|
JOB_STATUS_RUNNING = 1
|
|
|
JOB_STATUS_SUCCESS = 2
|
|
|
JOB_STATUS_FAILED = 3
|
|
|
JOB_STATUS_TIMEOUT = 4
|
|
|
JOB_STATUS_CANCELLED = 5
|
|
|
|
|
|
# 调度器状态常量（0:已停止 1:运行中 2:已暂停）
|
|
|
SCHEDULER_STATUS_STOPPED = 0
|
|
|
SCHEDULER_STATUS_RUNNING = 1
|
|
|
SCHEDULER_STATUS_PAUSED = 2
|
|
|
|
|
|
scheduler = AsyncIOScheduler()
|
|
|
scheduler.configure(
|
|
|
jobstores={
|
|
|
# 多 Worker 部署时必须使用 RedisJobStore，保证任务只被一个 Worker 执行
|
|
|
"default": RedisJobStore(
|
|
|
host=settings.REDIS_HOST,
|
|
|
port=int(settings.REDIS_PORT),
|
|
|
username=settings.REDIS_USER or None,
|
|
|
password=settings.REDIS_PASSWORD or None,
|
|
|
db=int(settings.REDIS_DB_NAME),
|
|
|
),
|
|
|
"sqlalchemy": SQLAlchemyJobStore(url=settings.DB_URI, engine=engine),
|
|
|
"memory": MemoryJobStore(),
|
|
|
},
|
|
|
executors={
|
|
|
"default": AsyncIOExecutor(),
|
|
|
"threadpool": ThreadPoolExecutor(max_workers=10),
|
|
|
"processpool": ProcessPoolExecutor(max_workers=1),
|
|
|
},
|
|
|
job_defaults={
|
|
|
"coalesce": True,
|
|
|
"max_instances": 5,
|
|
|
},
|
|
|
timezone="Asia/Shanghai",
|
|
|
)
|
|
|
|
|
|
|
|
|
class SchedulerUtil:
|
|
|
"""
|
|
|
定时任务相关方法
|
|
|
"""
|
|
|
|
|
|
redis_instance: Redis | None = None
|
|
|
# 临时存储 job_name，用于在 EVENT_JOB_SUBMITTED 时获取
|
|
|
# 格式可以是: str (任务名称) 或 tuple[str, str] (原任务ID, 任务名称)
|
|
|
_job_name_cache: dict[str, str | tuple[str, str]] = {}
|
|
|
|
|
|
@classmethod
|
|
|
def scheduler_event_listener(cls, event: JobEvent | JobExecutionEvent) -> None:
|
|
|
"""
|
|
|
监听任务执行事件，记录执行日志；每次执行新建日志行，保留历史。
|
|
|
|
|
|
参数:
|
|
|
- event (JobEvent | JobExecutionEvent): APScheduler 事件对象。
|
|
|
|
|
|
返回:
|
|
|
- None
|
|
|
"""
|
|
|
try:
|
|
|
# 事件处理器映射
|
|
|
event_handlers: dict[int, Callable] = {
|
|
|
# 调度器事件
|
|
|
EVENT_SCHEDULER_STARTED: cls._handle_scheduler_started,
|
|
|
EVENT_SCHEDULER_START: cls._handle_scheduler_started,
|
|
|
EVENT_SCHEDULER_SHUTDOWN: cls._handle_scheduler_shutdown,
|
|
|
EVENT_SCHEDULER_PAUSED: cls._handle_scheduler_paused,
|
|
|
EVENT_SCHEDULER_RESUMED: cls._handle_scheduler_resumed,
|
|
|
# 执行器事件
|
|
|
EVENT_EXECUTOR_ADDED: cls._handle_executor_added,
|
|
|
EVENT_EXECUTOR_REMOVED: cls._handle_executor_removed,
|
|
|
# JobStore 事件
|
|
|
EVENT_JOBSTORE_ADDED: cls._handle_jobstore_added,
|
|
|
EVENT_JOBSTORE_REMOVED: cls._handle_jobstore_removed,
|
|
|
EVENT_ALL_JOBS_REMOVED: cls._handle_all_jobs_removed,
|
|
|
# 任务事件
|
|
|
EVENT_JOB_ADDED: cls._handle_job_added,
|
|
|
EVENT_JOB_REMOVED: cls._handle_job_removed,
|
|
|
EVENT_JOB_MODIFIED: cls._handle_job_modified,
|
|
|
EVENT_JOB_SUBMITTED: cls._handle_job_submitted,
|
|
|
EVENT_JOB_EXECUTED: cls._handle_job_executed,
|
|
|
EVENT_JOB_ERROR: cls._handle_job_error,
|
|
|
EVENT_JOB_MISSED: cls._handle_job_missed,
|
|
|
EVENT_JOB_MAX_INSTANCES: cls._handle_job_max_instances,
|
|
|
}
|
|
|
|
|
|
# 处理事件
|
|
|
if event.code in event_handlers:
|
|
|
handler = event_handlers[event.code]
|
|
|
handler(event)
|
|
|
else:
|
|
|
# 处理其他事件
|
|
|
cls._handle_other_event(event)
|
|
|
|
|
|
except Exception as e:
|
|
|
logger.error(f"处理任务执行事件失败: {e!s}", exc_info=True)
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_job_submitted(cls, event: JobSubmissionEvent) -> None:
|
|
|
"""
|
|
|
处理任务提交事件
|
|
|
"""
|
|
|
job_id = str(event.job_id)
|
|
|
job = cls.get_job(job_id=job_id)
|
|
|
|
|
|
if job:
|
|
|
logger.info(f"任务 {job_id} ({job.name}) 已提交执行")
|
|
|
|
|
|
trigger_type = cls._get_trigger_type(job_id)
|
|
|
|
|
|
# 周期性任务（cron/interval）：更新 pending 状态为 running
|
|
|
if trigger_type in ("cron", "interval"):
|
|
|
cls._update_job_log(
|
|
|
job_id=job_id,
|
|
|
status=JOB_STATUS_RUNNING,
|
|
|
)
|
|
|
else:
|
|
|
# 一次性任务（manual/date）：创建新的 running 状态日志
|
|
|
cls._create_job_log(
|
|
|
job_id=job_id,
|
|
|
job_name=job.name,
|
|
|
trigger_type=trigger_type,
|
|
|
status=JOB_STATUS_RUNNING,
|
|
|
)
|
|
|
else:
|
|
|
# 任务可能已经被移除（一次性任务执行完毕后自动移除）
|
|
|
# 尝试从缓存获取 job_name 和原任务 ID
|
|
|
cached_value = cls._job_name_cache.pop(job_id, None)
|
|
|
# 处理新的缓存格式 (原任务ID, 任务名称) 或旧的格式 任务名称
|
|
|
if isinstance(cached_value, tuple):
|
|
|
original_job_id, job_name = cached_value
|
|
|
else:
|
|
|
original_job_id, job_name = job_id, cached_value
|
|
|
|
|
|
logger.info(f"任务 {job_id} 提交执行，但未找到任务信息（可能已被移除），尝试创建日志")
|
|
|
result = cls._create_job_log(
|
|
|
job_id=original_job_id,
|
|
|
job_name=job_name,
|
|
|
trigger_type="manual",
|
|
|
status=JOB_STATUS_RUNNING,
|
|
|
)
|
|
|
if result:
|
|
|
logger.info(f"任务 {original_job_id} 日志创建成功，id={result}")
|
|
|
else:
|
|
|
logger.error(f"任务 {original_job_id} 日志创建失败")
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_job_executed(cls, event: JobExecutionEvent) -> None:
|
|
|
"""
|
|
|
处理任务执行成功事件
|
|
|
"""
|
|
|
job_id = str(event.job_id)
|
|
|
retval = getattr(event, "retval", None)
|
|
|
scheduled_run_time = getattr(event, "scheduled_run_time", None)
|
|
|
|
|
|
logger.info(f"任务 {job_id} 执行成功")
|
|
|
if retval:
|
|
|
logger.debug(f"任务 {job_id} 返回值: {retval}")
|
|
|
if scheduled_run_time:
|
|
|
logger.debug(f"任务 {job_id} 计划执行时间: {scheduled_run_time}")
|
|
|
|
|
|
# 更新执行日志
|
|
|
cls._update_latest_job_log(
|
|
|
job_id=job_id,
|
|
|
status=JOB_STATUS_SUCCESS,
|
|
|
result=str(retval) if retval else None,
|
|
|
)
|
|
|
|
|
|
# 为周期性任务创建新的 pending 状态日志，等待下次执行
|
|
|
job = cls.get_job(job_id=job_id)
|
|
|
if job:
|
|
|
trigger_type = cls._get_trigger_type(job_id)
|
|
|
if trigger_type in ("cron", "interval") and job.next_run_time:
|
|
|
cls._create_job_log(
|
|
|
job_id=job_id,
|
|
|
job_name=job.name,
|
|
|
trigger_type=trigger_type,
|
|
|
status=JOB_STATUS_PENDING,
|
|
|
)
|
|
|
logger.debug(f"任务 {job_id} 已创建新的 pending 状态日志，等待下次执行")
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_job_error(cls, event: JobExecutionEvent) -> None:
|
|
|
"""
|
|
|
处理任务执行失败事件
|
|
|
"""
|
|
|
job_id = str(event.job_id)
|
|
|
exception = getattr(event, "exception", None)
|
|
|
traceback = getattr(event, "traceback", None)
|
|
|
scheduled_run_time = getattr(event, "scheduled_run_time", None)
|
|
|
|
|
|
logger.error(f"任务 {job_id} 执行失败: {exception!s}")
|
|
|
if traceback:
|
|
|
logger.debug(f"任务 {job_id} 错误堆栈: {traceback}")
|
|
|
if scheduled_run_time:
|
|
|
logger.debug(f"任务 {job_id} 计划执行时间: {scheduled_run_time}")
|
|
|
|
|
|
# 更新执行日志
|
|
|
cls._update_latest_job_log(
|
|
|
job_id=job_id,
|
|
|
status=JOB_STATUS_FAILED,
|
|
|
result="failed",
|
|
|
error=str(exception) if exception else "未知错误",
|
|
|
)
|
|
|
|
|
|
# 为周期性任务创建新的 pending 状态日志，等待下次执行
|
|
|
job = cls.get_job(job_id=job_id)
|
|
|
if job:
|
|
|
trigger_type = cls._get_trigger_type(job_id)
|
|
|
if trigger_type in ("cron", "interval") and job.next_run_time:
|
|
|
cls._create_job_log(
|
|
|
job_id=job_id,
|
|
|
job_name=job.name,
|
|
|
trigger_type=trigger_type,
|
|
|
status=JOB_STATUS_PENDING,
|
|
|
)
|
|
|
logger.debug(f"任务 {job_id} 已创建新的 pending 状态日志，等待下次执行")
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_job_missed(cls, event: JobEvent) -> None:
|
|
|
"""
|
|
|
处理任务错过执行时间事件
|
|
|
"""
|
|
|
job_id = str(event.job_id)
|
|
|
job = cls.get_job(job_id=job_id)
|
|
|
|
|
|
logger.warning(f"任务 {job_id} 错过执行时间")
|
|
|
if job:
|
|
|
logger.debug(f"任务 {job_id} ({job.name}) 错过执行")
|
|
|
|
|
|
# 更新执行日志
|
|
|
cls._update_latest_job_log(
|
|
|
job_id=job_id,
|
|
|
status=JOB_STATUS_TIMEOUT,
|
|
|
result="timeout",
|
|
|
error="任务错过执行时间",
|
|
|
)
|
|
|
|
|
|
# 为周期性任务创建新的 pending 状态日志，等待下次执行
|
|
|
if job:
|
|
|
trigger_type = cls._get_trigger_type(job_id)
|
|
|
if trigger_type in ("cron", "interval") and job.next_run_time:
|
|
|
cls._create_job_log(
|
|
|
job_id=job_id,
|
|
|
job_name=job.name,
|
|
|
trigger_type=trigger_type,
|
|
|
status=JOB_STATUS_PENDING,
|
|
|
)
|
|
|
logger.debug(f"任务 {job_id} 已创建新的 pending 状态日志，等待下次执行")
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_job_removed(cls, event: JobEvent) -> None:
|
|
|
"""
|
|
|
处理任务被移除事件
|
|
|
|
|
|
注意：APScheduler 对于一次性任务（DateTrigger）会先触发 JOB_REMOVED，
|
|
|
然后再触发 JOB_SUBMITTED 和 JOB_EXECUTED。因此：
|
|
|
- 对于一次性任务，不应该在 JOB_REMOVED 时创建或更新日志
|
|
|
- 只有周期性任务在移除时才需要更新日志状态
|
|
|
"""
|
|
|
job_id = str(event.job_id)
|
|
|
jobstore = getattr(event, "jobstore", "unknown")
|
|
|
|
|
|
logger.info(f"任务 {job_id} 从 {jobstore} 存储器中移除")
|
|
|
|
|
|
# 检查是否是一次性任务（DateTrigger）
|
|
|
# 如果任务已经不存在，说明可能是一次性任务执行后被自动移除
|
|
|
# 这种情况下不需要更新日志，因为 JOB_EXECUTED 会处理
|
|
|
job = cls.get_job(job_id=job_id)
|
|
|
if job is None:
|
|
|
# 任务已经被移除，可能是一次性任务
|
|
|
# 不需要在这里创建日志，JOB_SUBMITTED 和 JOB_EXECUTED 会处理
|
|
|
logger.debug(f"任务 {job_id} 已从调度器中移除（可能是一次性任务），跳过日志更新")
|
|
|
return
|
|
|
|
|
|
# 任务还存在，说明是周期性任务被手动移除
|
|
|
# 更新执行日志
|
|
|
cls._update_job_log_on_removed(job_id=job_id)
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_job_added(cls, event: JobEvent) -> None:
|
|
|
"""
|
|
|
处理任务添加事件
|
|
|
"""
|
|
|
job_id = str(event.job_id)
|
|
|
jobstore = event.jobstore
|
|
|
job = cls.get_job(job_id=job_id)
|
|
|
|
|
|
if job:
|
|
|
logger.info(f"任务 {job_id} ({job.name}) 已添加到 {jobstore} 存储器")
|
|
|
|
|
|
# 为周期性任务（cron/interval）创建初始的 pending 状态日志
|
|
|
trigger_type = cls._get_trigger_type(job_id)
|
|
|
if trigger_type in ("cron", "interval"):
|
|
|
# 清理旧的 pending 日志，避免重启导致 pending 日志累积
|
|
|
cls._cleanup_obsolete_pending_logs(job_id)
|
|
|
cls._create_job_log(
|
|
|
job_id=job_id,
|
|
|
job_name=job.name,
|
|
|
trigger_type=trigger_type,
|
|
|
status=JOB_STATUS_PENDING,
|
|
|
)
|
|
|
logger.info(f"任务 {job_id} 已创建初始 pending 状态日志")
|
|
|
else:
|
|
|
logger.info(f"任务 {job_id} 已添加到 {jobstore} 存储器")
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_job_modified(cls, event: JobEvent) -> None:
|
|
|
"""
|
|
|
处理任务修改事件
|
|
|
"""
|
|
|
job_id = str(event.job_id)
|
|
|
jobstore = event.jobstore
|
|
|
job = cls.get_job(job_id=job_id)
|
|
|
|
|
|
if job:
|
|
|
logger.info(f"任务 {job_id} ({job.name}) 已在 {jobstore} 存储器中修改")
|
|
|
else:
|
|
|
logger.info(f"任务 {job_id} 已在 {jobstore} 存储器中修改")
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_scheduler_started(cls, event: SchedulerEvent) -> None:
|
|
|
"""
|
|
|
处理调度器启动事件
|
|
|
"""
|
|
|
logger.info("调度器已启动")
|
|
|
cls._update_scheduler_status(SCHEDULER_STATUS_RUNNING)
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_scheduler_shutdown(cls, event: SchedulerEvent) -> None:
|
|
|
"""
|
|
|
处理调度器关闭事件
|
|
|
"""
|
|
|
logger.info("调度器已关闭")
|
|
|
cls._update_scheduler_status(SCHEDULER_STATUS_STOPPED)
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_scheduler_paused(cls, event: SchedulerEvent) -> None:
|
|
|
"""
|
|
|
处理调度器暂停事件
|
|
|
"""
|
|
|
logger.info("调度器已暂停")
|
|
|
cls._update_scheduler_status(SCHEDULER_STATUS_PAUSED)
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_scheduler_resumed(cls, event: SchedulerEvent) -> None:
|
|
|
"""
|
|
|
处理调度器恢复事件
|
|
|
"""
|
|
|
logger.info("调度器已恢复运行")
|
|
|
cls._update_scheduler_status(SCHEDULER_STATUS_RUNNING)
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_executor_added(cls, event: SchedulerEvent) -> None:
|
|
|
"""
|
|
|
处理执行器添加事件
|
|
|
"""
|
|
|
alias = event.alias
|
|
|
if alias:
|
|
|
logger.info(f"执行器 {alias} 已添加到调度器")
|
|
|
cls._update_executor_info(alias, "added")
|
|
|
else:
|
|
|
logger.warning("执行器添加事件，但别名为空")
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_executor_removed(cls, event: SchedulerEvent) -> None:
|
|
|
"""
|
|
|
处理执行器移除事件
|
|
|
"""
|
|
|
alias = event.alias
|
|
|
if alias:
|
|
|
logger.info(f"执行器 {alias} 已从调度器中移除")
|
|
|
cls._update_executor_info(alias, "removed")
|
|
|
else:
|
|
|
logger.warning("执行器移除事件，但别名为空")
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_jobstore_added(cls, event: SchedulerEvent) -> None:
|
|
|
"""
|
|
|
处理 JobStore 添加事件
|
|
|
"""
|
|
|
alias = event.alias
|
|
|
if alias:
|
|
|
logger.info(f"JobStore {alias} 已添加到调度器")
|
|
|
cls._update_jobstore_info(alias, "added")
|
|
|
else:
|
|
|
logger.warning("JobStore 添加事件，但别名为空")
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_jobstore_removed(cls, event: SchedulerEvent) -> None:
|
|
|
"""
|
|
|
处理 JobStore 移除事件
|
|
|
"""
|
|
|
alias = event.alias
|
|
|
if alias:
|
|
|
logger.info(f"JobStore {alias} 已从调度器中移除")
|
|
|
cls._update_jobstore_info(alias, "removed")
|
|
|
else:
|
|
|
logger.warning("JobStore 移除事件，但别名为空")
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_all_jobs_removed(cls, event: SchedulerEvent) -> None:
|
|
|
"""
|
|
|
处理所有任务移除事件
|
|
|
注意：清空调度器任务不应该清空执行日志，而是将所有 pending 状态的日志更新为 cancelled
|
|
|
"""
|
|
|
logger.info("所有任务已从调度器中移除")
|
|
|
cls._cancel_all_pending_job_logs()
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_job_max_instances(cls, event: JobEvent) -> None:
|
|
|
"""
|
|
|
处理任务达到最大实例数事件
|
|
|
"""
|
|
|
job_id = str(event.job_id)
|
|
|
logger.warning(f"任务 {job_id} 已达到最大实例数限制，无法启动新实例")
|
|
|
|
|
|
@classmethod
|
|
|
def _handle_other_event(
|
|
|
cls, event: SchedulerEvent | JobEvent | JobExecutionEvent | JobSubmissionEvent
|
|
|
) -> None:
|
|
|
"""
|
|
|
处理其他事件
|
|
|
"""
|
|
|
event_code = event.code
|
|
|
event_type = type(event).__name__
|
|
|
logger.debug(f"收到未处理的事件: {event_type} (code: {event_code})")
|
|
|
|
|
|
@classmethod
|
|
|
def _update_scheduler_status(cls, status: int) -> None:
|
|
|
"""
|
|
|
仅记录调度器状态到内存（不再写入 sys_param 表）。
|
|
|
前端通过 /scheduler/status 接口直接从 SchedulerUtil 读取，避免污染配置表。
|
|
|
|
|
|
参数:
|
|
|
- status (int): 调度器状态 (0: stopped / 1: running / 2: paused)
|
|
|
"""
|
|
|
cls._last_scheduler_status = status
|
|
|
logger.info(f"调度器状态变更: {status}")
|
|
|
|
|
|
@classmethod
|
|
|
def _update_executor_info(cls, alias: str | None, action: str) -> None:
|
|
|
"""
|
|
|
仅记录执行器状态到内存（不再写入 sys_param 表）。
|
|
|
|
|
|
参数:
|
|
|
- alias (str | None): 执行器别名
|
|
|
- action (str): 操作 (added/removed)
|
|
|
"""
|
|
|
if not alias:
|
|
|
logger.warning("执行器别名为空，跳过更新")
|
|
|
return
|
|
|
logger.debug(f"执行器 {alias} {action}")
|
|
|
|
|
|
@classmethod
|
|
|
def _update_jobstore_info(cls, alias: str | None, action: str) -> None:
|
|
|
"""
|
|
|
仅记录 JobStore 状态到内存（不再写入 sys_param 表）。
|
|
|
|
|
|
参数:
|
|
|
- alias (str | None): JobStore 别名
|
|
|
- action (str): 操作 (added/removed)
|
|
|
"""
|
|
|
if not alias:
|
|
|
logger.warning("JobStore 别名为空，跳过更新")
|
|
|
return
|
|
|
logger.debug(f"JobStore {alias} {action}")
|
|
|
|
|
|
@classmethod
|
|
|
def _clear_all_job_logs(cls) -> None:
|
|
|
"""
|
|
|
清空所有任务日志（仅用于手动清空，不建议使用）
|
|
|
"""
|
|
|
try:
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
from app.plugin.module_task.cronjob.job.model import JobModel
|
|
|
|
|
|
with Session(engine) as session:
|
|
|
session.query(JobModel).delete()
|
|
|
session.commit()
|
|
|
logger.info("所有任务日志已清空")
|
|
|
except Exception as e:
|
|
|
logger.error(f"清空任务日志失败: {e!s}", exc_info=True)
|
|
|
|
|
|
@classmethod
|
|
|
def _cancel_all_pending_job_logs(cls) -> None:
|
|
|
"""
|
|
|
将所有 pending 状态的执行日志更新为 cancelled
|
|
|
用于清空调度器任务时，不删除日志而是更新状态
|
|
|
"""
|
|
|
try:
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
from app.plugin.module_task.cronjob.job.model import JobModel
|
|
|
|
|
|
with Session(engine) as session:
|
|
|
session.query(JobModel).filter(JobModel.status == JOB_STATUS_PENDING).update({
|
|
|
"status": JOB_STATUS_CANCELLED
|
|
|
})
|
|
|
session.commit()
|
|
|
logger.info("所有待执行任务日志已标记为已取消")
|
|
|
except Exception as e:
|
|
|
logger.error(f"取消待执行任务日志失败: {e!s}", exc_info=True)
|
|
|
|
|
|
@classmethod
|
|
|
def _get_trigger_type(cls, job_id: str) -> str:
|
|
|
"""
|
|
|
获取任务的触发类型
|
|
|
"""
|
|
|
job = cls.get_job(job_id=job_id)
|
|
|
if not job:
|
|
|
return "manual"
|
|
|
trigger = job.trigger
|
|
|
if isinstance(trigger, CronTrigger):
|
|
|
return "cron"
|
|
|
elif isinstance(trigger, IntervalTrigger):
|
|
|
return "interval"
|
|
|
elif isinstance(trigger, DateTrigger):
|
|
|
if trigger.run_date:
|
|
|
now = datetime.now(trigger.run_date.tzinfo)
|
|
|
diff = abs((trigger.run_date - now).total_seconds())
|
|
|
if diff < 60:
|
|
|
return "manual"
|
|
|
return "date"
|
|
|
return "manual"
|
|
|
|
|
|
@classmethod
|
|
|
async def init_scheduler(cls, redis: Redis | None = None) -> None:
|
|
|
"""
|
|
|
应用启动时初始化定时任务调度器。
|
|
|
|
|
|
参数:
|
|
|
- redis (Redis | None): 可选 Redis 实例，供任务侧使用。
|
|
|
|
|
|
返回:
|
|
|
- None
|
|
|
"""
|
|
|
if redis:
|
|
|
cls.redis_instance = redis
|
|
|
scheduler.start()
|
|
|
scheduler.add_listener(cls.scheduler_event_listener, EVENT_ALL)
|
|
|
scheduler.resume()
|
|
|
cls._register_system_jobs()
|
|
|
|
|
|
@classmethod
|
|
|
def _register_system_jobs(cls) -> None:
|
|
|
"""
|
|
|
注册系统级定时任务（租户到期/续费提醒/归档清理/订单取消/日志清理）。
|
|
|
在 init_scheduler 末尾自动调用。
|
|
|
"""
|
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
|
from apscheduler.triggers.interval import IntervalTrigger
|
|
|
|
|
|
from app.api.v1.module_platform.order.service import OrderService
|
|
|
from app.api.v1.module_platform.tenant.service import TenantService
|
|
|
|
|
|
# 租户到期检查（每小时）
|
|
|
scheduler.add_job(
|
|
|
TenantService.check_tenant_expiry,
|
|
|
trigger=IntervalTrigger(hours=1),
|
|
|
id="system_tenant_expiry_check",
|
|
|
name="租户到期检查",
|
|
|
replace_existing=True,
|
|
|
)
|
|
|
# 宽限期续费提醒（每天 9:00）
|
|
|
scheduler.add_job(
|
|
|
TenantService.send_grace_reminders,
|
|
|
trigger=CronTrigger(hour=9, minute=0),
|
|
|
id="system_grace_reminder",
|
|
|
name="宽限期续费提醒",
|
|
|
replace_existing=True,
|
|
|
)
|
|
|
# 过期租户归档清理（每月 1 号 2:00）
|
|
|
scheduler.add_job(
|
|
|
TenantService.clean_expired_tenants,
|
|
|
trigger=CronTrigger(day=1, hour=2, minute=0),
|
|
|
id="system_clean_expired",
|
|
|
name="过期租户归档清理",
|
|
|
replace_existing=True,
|
|
|
)
|
|
|
# 超时订单取消（每 5 分钟）
|
|
|
scheduler.add_job(
|
|
|
OrderService.cancel_expired_orders,
|
|
|
trigger=IntervalTrigger(minutes=5),
|
|
|
id="system_cancel_expired_orders",
|
|
|
name="超时订单取消",
|
|
|
replace_existing=True,
|
|
|
)
|
|
|
# 操作日志清理（每周日 3:00）
|
|
|
from app.api.v1.module_system.log.service import OperationLogService
|
|
|
|
|
|
scheduler.add_job(
|
|
|
OperationLogService.cleanup_operation_log,
|
|
|
trigger=CronTrigger(day_of_week="sun", hour=3, minute=0),
|
|
|
id="system_cleanup_operation_log",
|
|
|
name="操作日志清理",
|
|
|
replace_existing=True,
|
|
|
)
|
|
|
logger.info("✅ 6 个系统周期任务已注册（租户到期/续费提醒/归档清理/订单取消/日志清理）")
|
|
|
|
|
|
@classmethod
|
|
|
def _task_wrapper(cls, job_id: str | int, code_block: str | None, *args, **kwargs):
|
|
|
"""
|
|
|
任务执行包装器，执行自定义代码块（同步版本，用于 ThreadPoolExecutor）
|
|
|
|
|
|
支持完整的 Python 语法，包括 import 语句
|
|
|
"""
|
|
|
import types
|
|
|
|
|
|
def run_sync_handler():
|
|
|
"""
|
|
|
在独立模块命名空间中执行代码块并调用 handler。
|
|
|
|
|
|
返回:
|
|
|
- Any: handler 返回值；无代码块时为 None。
|
|
|
"""
|
|
|
if not code_block:
|
|
|
return None
|
|
|
|
|
|
# 创建一个新的模块作为执行环境
|
|
|
module = types.ModuleType(f"node_task_{job_id}")
|
|
|
module.__dict__["__builtins__"] = __builtins__
|
|
|
|
|
|
# 在模块环境中执行代码
|
|
|
exec(code_block, module.__dict__)
|
|
|
|
|
|
# 获取 handler 函数
|
|
|
handler = module.__dict__.get("handler")
|
|
|
if handler and callable(handler):
|
|
|
return handler(*args, **kwargs)
|
|
|
raise ValueError("代码块必须定义 handler(*args, **kwargs) 函数")
|
|
|
|
|
|
try:
|
|
|
result = run_sync_handler()
|
|
|
return result
|
|
|
except Exception as e:
|
|
|
logger.error(f"任务 {job_id} 执行失败: {e!s}")
|
|
|
raise
|
|
|
|
|
|
@classmethod
|
|
|
def _get_job_state(cls, job) -> str | None:
|
|
|
"""
|
|
|
获取任务状态（解析为可读的JSON格式）
|
|
|
"""
|
|
|
import json
|
|
|
import pickle
|
|
|
|
|
|
if not job:
|
|
|
return None
|
|
|
|
|
|
state = job.__getstate__()
|
|
|
|
|
|
def serialize_value(obj):
|
|
|
"""
|
|
|
将 job state 中的嵌套对象转为可 JSON 化的 Python 结构。
|
|
|
|
|
|
参数:
|
|
|
- obj (Any): 任意嵌套对象。
|
|
|
|
|
|
返回:
|
|
|
- Any: 标量、dict、list 或简化后的描述。
|
|
|
"""
|
|
|
if obj is None:
|
|
|
return None
|
|
|
if isinstance(obj, (str, int, float, bool)):
|
|
|
return obj
|
|
|
if isinstance(obj, bytes):
|
|
|
try:
|
|
|
decoded = pickle.loads(obj)
|
|
|
return serialize_value(decoded)
|
|
|
except Exception:
|
|
|
return obj.decode("utf-8", errors="replace")
|
|
|
if isinstance(obj, dict):
|
|
|
return {k: serialize_value(v) for k, v in obj.items()}
|
|
|
if isinstance(obj, (list, tuple)):
|
|
|
return [serialize_value(item) for item in obj]
|
|
|
if hasattr(obj, "__dict__"):
|
|
|
obj_dict = {}
|
|
|
for k, v in obj.__dict__.items():
|
|
|
if not k.startswith("_"):
|
|
|
obj_dict[k] = serialize_value(v)
|
|
|
return {"__class__": obj.__class__.__name__, **obj_dict}
|
|
|
try:
|
|
|
return str(obj)
|
|
|
except Exception:
|
|
|
return f"<{type(obj).__name__}>"
|
|
|
|
|
|
parsed_state = serialize_value(state)
|
|
|
return json.dumps(parsed_state, ensure_ascii=False, indent=2)
|
|
|
|
|
|
@classmethod
|
|
|
def get_job_state_from_blob(cls, blob_data: bytes) -> Any:
|
|
|
"""
|
|
|
从 BLOB 数据反序列化任务状态
|
|
|
|
|
|
参数:
|
|
|
- blob_data: apscheduler_jobs 表中的 job_state 字段（BLOB 类型）
|
|
|
|
|
|
返回:
|
|
|
- 反序列化后的任务状态
|
|
|
"""
|
|
|
import pickle
|
|
|
|
|
|
if not blob_data:
|
|
|
return None
|
|
|
|
|
|
def serialize_value(obj: Any) -> Any:
|
|
|
"""
|
|
|
递归反序列化 BLOB 中的嵌套结构为可 JSON 化数据。
|
|
|
|
|
|
参数:
|
|
|
- obj (Any): 节点对象。
|
|
|
|
|
|
返回:
|
|
|
- Any: 标量、dict、list 或字符串化结果。
|
|
|
"""
|
|
|
if obj is None:
|
|
|
return None
|
|
|
if isinstance(obj, (str, int, float, bool)):
|
|
|
return obj
|
|
|
if isinstance(obj, bytes):
|
|
|
try:
|
|
|
decoded = pickle.loads(obj)
|
|
|
return serialize_value(decoded)
|
|
|
except Exception:
|
|
|
return obj.decode("utf-8", errors="replace")
|
|
|
if isinstance(obj, dict):
|
|
|
return {k: serialize_value(v) for k, v in obj.items()}
|
|
|
if isinstance(obj, (list, tuple)):
|
|
|
return [serialize_value(item) for item in obj]
|
|
|
if hasattr(obj, "__dict__"):
|
|
|
obj_dict = {}
|
|
|
for k, v in obj.__dict__.items():
|
|
|
if not k.startswith("_"):
|
|
|
obj_dict[k] = serialize_value(v)
|
|
|
return {"__class__": obj.__class__.__name__, **obj_dict}
|
|
|
try:
|
|
|
return str(obj)
|
|
|
except Exception:
|
|
|
return f"<{type(obj).__name__}>"
|
|
|
|
|
|
try:
|
|
|
state = pickle.loads(blob_data)
|
|
|
return serialize_value(state)
|
|
|
except Exception as e:
|
|
|
return {"error": str(e), "raw_data": str(blob_data[:200])}
|
|
|
|
|
|
@classmethod
|
|
|
def _cleanup_obsolete_pending_logs(cls, job_id: str) -> None:
|
|
|
"""清理该 job_id 的所有旧的 pending 日志，避免重启累积。"""
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
from app.plugin.module_task.cronjob.job.model import JobModel
|
|
|
|
|
|
try:
|
|
|
with Session(engine) as session:
|
|
|
deleted = (
|
|
|
session.query(JobModel)
|
|
|
.filter(JobModel.job_id == job_id, JobModel.status == JOB_STATUS_PENDING)
|
|
|
.delete(synchronize_session=False)
|
|
|
)
|
|
|
session.commit()
|
|
|
if deleted:
|
|
|
logger.info(f"清理了任务 {job_id} 的 {deleted} 条旧 pending 日志")
|
|
|
except Exception as e:
|
|
|
logger.error(f"清理任务 {job_id} 旧 pending 日志失败: {e}")
|
|
|
|
|
|
@classmethod
|
|
|
def _create_job_log(
|
|
|
cls,
|
|
|
job_id: str,
|
|
|
job_name: str | None = None,
|
|
|
trigger_type: str = "manual",
|
|
|
status: int = JOB_STATUS_RUNNING,
|
|
|
) -> int | None:
|
|
|
"""
|
|
|
创建执行日志
|
|
|
"""
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
from app.plugin.module_task.cronjob.job.model import JobModel
|
|
|
|
|
|
try:
|
|
|
job = cls.get_job(job_id=job_id)
|
|
|
next_run_time = str(job.next_run_time) if job and job.next_run_time else None
|
|
|
job_state = cls._get_job_state(job) if job else None
|
|
|
# 如果没有传入 job_name，尝试从 job 获取
|
|
|
if not job_name and job:
|
|
|
job_name = job.name
|
|
|
|
|
|
with Session(engine) as session:
|
|
|
job_log = JobModel(
|
|
|
job_id=job_id,
|
|
|
job_name=job_name,
|
|
|
trigger_type=trigger_type,
|
|
|
status=status,
|
|
|
next_run_time=next_run_time,
|
|
|
job_state=job_state,
|
|
|
)
|
|
|
session.add(job_log)
|
|
|
session.commit()
|
|
|
logger.info(f"执行日志创建成功: job_id={job_id}, id={job_log.id}")
|
|
|
return job_log.id
|
|
|
except Exception as e:
|
|
|
logger.error(f"创建执行日志失败: job_id={job_id}, error={e}", exc_info=True)
|
|
|
return None
|
|
|
|
|
|
@classmethod
|
|
|
def _update_job_log(
|
|
|
cls, job_id: str, status: int, result: str | None = None, error: str | None = None
|
|
|
) -> None:
|
|
|
"""
|
|
|
更新执行日志（更新该 job_id 最新的 pending 状态日志）
|
|
|
用于周期性任务在提交执行时将 pending 更新为 running
|
|
|
"""
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
from app.plugin.module_task.cronjob.job.model import JobModel
|
|
|
|
|
|
job = cls.get_job(job_id=job_id)
|
|
|
next_run_time = str(job.next_run_time) if job and job.next_run_time else None
|
|
|
job_state = cls._get_job_state(job) if job else None
|
|
|
|
|
|
with Session(engine) as session:
|
|
|
job_log = (
|
|
|
session
|
|
|
.query(JobModel)
|
|
|
.filter(JobModel.job_id == job_id, JobModel.status == JOB_STATUS_PENDING)
|
|
|
.order_by(JobModel.created_time.desc())
|
|
|
.first()
|
|
|
)
|
|
|
if job_log:
|
|
|
job_log.status = status
|
|
|
if next_run_time:
|
|
|
job_log.next_run_time = next_run_time
|
|
|
if job_state:
|
|
|
job_log.job_state = job_state
|
|
|
if result:
|
|
|
job_log.result = result
|
|
|
if error:
|
|
|
job_log.error = error
|
|
|
session.commit()
|
|
|
else:
|
|
|
logger.warning(f"未找到任务 {job_id} 的待执行日志记录")
|
|
|
|
|
|
@classmethod
|
|
|
def _update_latest_job_log(
|
|
|
cls, job_id: str, status: int, result: str | None = None, error: str | None = None
|
|
|
) -> None:
|
|
|
"""
|
|
|
更新最新的执行日志（更新该 job_id 最新的一条日志）
|
|
|
用于每次执行完成后更新状态
|
|
|
"""
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
from app.plugin.module_task.cronjob.job.model import JobModel
|
|
|
|
|
|
try:
|
|
|
job = cls.get_job(job_id=job_id)
|
|
|
next_run_time = str(job.next_run_time) if job and job.next_run_time else None
|
|
|
job_state = cls._get_job_state(job) if job else None
|
|
|
|
|
|
with Session(engine) as session:
|
|
|
# 首先尝试更新 running 状态的日志
|
|
|
job_log = (
|
|
|
session
|
|
|
.query(JobModel)
|
|
|
.filter(JobModel.job_id == job_id, JobModel.status == JOB_STATUS_RUNNING)
|
|
|
.order_by(JobModel.created_time.desc())
|
|
|
.first()
|
|
|
)
|
|
|
if job_log:
|
|
|
job_log.status = status
|
|
|
if next_run_time:
|
|
|
job_log.next_run_time = next_run_time
|
|
|
if job_state:
|
|
|
job_log.job_state = job_state
|
|
|
if result:
|
|
|
job_log.result = result
|
|
|
if error:
|
|
|
job_log.error = error
|
|
|
session.commit()
|
|
|
logger.info(f"执行日志更新成功: job_id={job_id}, id={job_log.id}, status={status}")
|
|
|
return
|
|
|
|
|
|
# 没有找到 running 状态的日志，尝试更新 cancelled 状态的日志
|
|
|
# 这种情况发生在 EVENT_JOB_REMOVED 先于 EVENT_JOB_SUBMITTED 触发时
|
|
|
job_log = (
|
|
|
session
|
|
|
.query(JobModel)
|
|
|
.filter(JobModel.job_id == job_id, JobModel.status == JOB_STATUS_CANCELLED)
|
|
|
.order_by(JobModel.created_time.desc())
|
|
|
.first()
|
|
|
)
|
|
|
if job_log:
|
|
|
job_log.status = status
|
|
|
if next_run_time:
|
|
|
job_log.next_run_time = next_run_time
|
|
|
if job_state:
|
|
|
job_log.job_state = job_state
|
|
|
if result:
|
|
|
job_log.result = result
|
|
|
if error:
|
|
|
job_log.error = error
|
|
|
session.commit()
|
|
|
logger.info(f"执行日志更新成功: job_id={job_id}, id={job_log.id}, status={status}")
|
|
|
return
|
|
|
|
|
|
# 创建新的日志记录
|
|
|
logger.debug(f"未找到任务 {job_id} 的日志记录，创建新日志")
|
|
|
trigger_type = cls._get_trigger_type(job_id) if job else "manual"
|
|
|
new_log = JobModel(
|
|
|
job_id=job_id,
|
|
|
job_name=job.name if job else None,
|
|
|
trigger_type=trigger_type,
|
|
|
status=status,
|
|
|
next_run_time=next_run_time,
|
|
|
job_state=job_state,
|
|
|
result=result,
|
|
|
error=error,
|
|
|
)
|
|
|
session.add(new_log)
|
|
|
session.commit()
|
|
|
logger.info(f"执行日志创建成功: job_id={job_id}, id={new_log.id}, status={status}")
|
|
|
except Exception as e:
|
|
|
logger.error(
|
|
|
f"更新执行日志失败: job_id={job_id}, status={status}, error={e}", exc_info=True
|
|
|
)
|
|
|
|
|
|
@classmethod
|
|
|
def _update_job_log_on_removed(cls, job_id: str) -> None:
|
|
|
"""
|
|
|
任务被移除时，更新最新的 pending 或 running 状态日志为 cancelled
|
|
|
|
|
|
事件触发顺序分析：
|
|
|
1. 一次性任务（manual/date）：
|
|
|
- EVENT_JOB_SUBMITTED -> 创建日志（status=running）
|
|
|
- EVENT_JOB_EXECUTED/ERROR -> 更新日志（status=success/failed）
|
|
|
- EVENT_JOB_REMOVED -> 日志已更新，不会被标记为 cancelled
|
|
|
|
|
|
2. 周期性任务（cron/interval）：
|
|
|
- EVENT_JOB_SUBMITTED -> 创建日志（status=running）
|
|
|
- EVENT_JOB_EXECUTED/ERROR -> 更新日志（status=success/failed）
|
|
|
- 下次执行 -> EVENT_JOB_SUBMITTED -> 创建新日志（status=running）
|
|
|
- EVENT_JOB_REMOVED -> 将 pending/running 标记为 cancelled
|
|
|
|
|
|
3. 特殊情况：
|
|
|
- 一次性任务在执行前被删除：running -> cancelled
|
|
|
- 周期性任务在 pending 状态被删除：pending -> cancelled
|
|
|
|
|
|
注意：
|
|
|
- 只有当任务还在 pending 或 running 状态时才更新为 cancelled
|
|
|
- 如果任务已经执行完成（success/failed/timeout），则不需要更新
|
|
|
- 如果任务已经标记为 cancelled，则不需要更新
|
|
|
"""
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
from app.plugin.module_task.cronjob.job.model import JobModel
|
|
|
|
|
|
with Session(engine) as session:
|
|
|
job_log = (
|
|
|
session
|
|
|
.query(JobModel)
|
|
|
.filter(JobModel.job_id == job_id, JobModel.status.in_([JOB_STATUS_PENDING, JOB_STATUS_RUNNING]))
|
|
|
.order_by(JobModel.created_time.desc())
|
|
|
.first()
|
|
|
)
|
|
|
if job_log:
|
|
|
job_log.status = JOB_STATUS_CANCELLED
|
|
|
session.commit()
|
|
|
logger.info(f"任务 {job_id} 的执行日志已标记为已取消")
|
|
|
|
|
|
@classmethod
|
|
|
def get_job_status(cls, job_id: str | int) -> int:
|
|
|
"""
|
|
|
获取单个任务的当前状态。
|
|
|
|
|
|
参数:
|
|
|
- job_id (str | int): 调度器任务 ID。
|
|
|
|
|
|
返回:
|
|
|
- int: 0=运行中 1=暂停中 2=已停止 3=未知。
|
|
|
"""
|
|
|
job = cls.get_job(job_id=str(job_id))
|
|
|
if not job:
|
|
|
return 3
|
|
|
|
|
|
# 判断是否暂停：next_run_time 为 None 表示任务已暂停
|
|
|
if job.next_run_time is None:
|
|
|
return 1
|
|
|
|
|
|
if scheduler.state == 0:
|
|
|
return 2
|
|
|
|
|
|
return 0
|
|
|
|
|
|
@classmethod
|
|
|
def add_and_run_job_now(cls, job_info: NodeModel) -> Job:
|
|
|
"""
|
|
|
立即执行任务（加入调度器并尽快触发一次）。
|
|
|
|
|
|
参数:
|
|
|
- job_info (NodeModel): 节点/任务配置。
|
|
|
|
|
|
返回:
|
|
|
- Job: APScheduler Job 对象。
|
|
|
"""
|
|
|
# 使用稍微延迟的时间，确保事件监听器能够捕获事件
|
|
|
from datetime import timedelta
|
|
|
|
|
|
trigger = DateTrigger(run_date=datetime.now() + timedelta(seconds=0.1))
|
|
|
job = cls._add_job_with_trigger(job_info, trigger)
|
|
|
# 注意：不需要手动创建执行日志，EVENT_JOB_SUBMITTED 事件会自动创建
|
|
|
return job
|
|
|
|
|
|
@classmethod
|
|
|
def add_cron_job(
|
|
|
cls,
|
|
|
job_info: NodeModel,
|
|
|
trigger_args: str | None = None,
|
|
|
start_date: str | None = None,
|
|
|
end_date: str | None = None,
|
|
|
) -> Job:
|
|
|
"""
|
|
|
创建 Cron 定时任务。
|
|
|
|
|
|
参数:
|
|
|
- job_info (NodeModel): 任务信息。
|
|
|
- trigger_args (str | None): Cron 表达式，默认取节点配置。
|
|
|
- start_date (str | None): 开始时间。
|
|
|
- end_date (str | None): 结束时间。
|
|
|
|
|
|
返回:
|
|
|
- Job: 已注册的 APScheduler Job。
|
|
|
"""
|
|
|
cron_expr = trigger_args or job_info.trigger_args
|
|
|
if not cron_expr:
|
|
|
raise ValueError("Cron触发器缺少参数")
|
|
|
|
|
|
fields = cron_expr.strip().split()
|
|
|
if len(fields) not in (6, 7):
|
|
|
raise ValueError("无效的 Cron 表达式")
|
|
|
if not CronUtil.validate_cron_expression(cron_expr):
|
|
|
raise ValueError(f"Cron表达式不正确: {cron_expr}")
|
|
|
|
|
|
parsed_fields = [field if field != "?" else "*" for field in fields]
|
|
|
if len(fields) == 6:
|
|
|
parsed_fields.append("*")
|
|
|
|
|
|
second, minute, hour, day, month, day_of_week, year = tuple(parsed_fields)
|
|
|
|
|
|
if (
|
|
|
second == "*"
|
|
|
and minute == "*"
|
|
|
and hour == "*"
|
|
|
and day == "*"
|
|
|
and month == "*"
|
|
|
and day_of_week in ("*", "?")
|
|
|
):
|
|
|
raise ValueError(
|
|
|
"Cron表达式不允许每秒执行，请至少指定秒数（如：0 * * * * ? * 表示每分钟执行）"
|
|
|
)
|
|
|
|
|
|
trigger = CronTrigger(
|
|
|
second=second,
|
|
|
minute=minute,
|
|
|
hour=hour,
|
|
|
day=day,
|
|
|
month=month,
|
|
|
day_of_week=day_of_week,
|
|
|
year=year,
|
|
|
start_date=start_date or job_info.start_date,
|
|
|
end_date=end_date or job_info.end_date,
|
|
|
timezone="Asia/Shanghai",
|
|
|
)
|
|
|
return cls._add_job_with_trigger(job_info, trigger)
|
|
|
|
|
|
@classmethod
|
|
|
def add_interval_job(
|
|
|
cls,
|
|
|
job_info: NodeModel,
|
|
|
trigger_args: str | None = None,
|
|
|
start_date: str | None = None,
|
|
|
end_date: str | None = None,
|
|
|
) -> Job:
|
|
|
"""
|
|
|
创建间隔执行任务。
|
|
|
|
|
|
参数:
|
|
|
- job_info (NodeModel): 任务信息。
|
|
|
- trigger_args (str | None): 间隔参数「秒 分 时 天 周」，默认取节点配置。
|
|
|
- start_date (str | None): 开始时间。
|
|
|
- end_date (str | None): 结束时间。
|
|
|
|
|
|
返回:
|
|
|
- Job: 已注册的 APScheduler Job。
|
|
|
"""
|
|
|
interval_args = trigger_args or job_info.trigger_args
|
|
|
if not interval_args:
|
|
|
raise ValueError("interval触发器缺少参数")
|
|
|
|
|
|
fields = interval_args.strip().split()
|
|
|
if len(fields) != 5:
|
|
|
raise ValueError("无效的 interval 表达式，格式: 秒 分 时 天 周")
|
|
|
|
|
|
second, minute, hour, day, week = tuple(
|
|
|
int(field) if field != "*" else 0 for field in fields
|
|
|
)
|
|
|
trigger = IntervalTrigger(
|
|
|
weeks=week,
|
|
|
days=day,
|
|
|
hours=hour,
|
|
|
minutes=minute,
|
|
|
seconds=second,
|
|
|
start_date=start_date or job_info.start_date,
|
|
|
end_date=end_date or job_info.end_date,
|
|
|
timezone="Asia/Shanghai",
|
|
|
)
|
|
|
return cls._add_job_with_trigger(job_info, trigger)
|
|
|
|
|
|
@classmethod
|
|
|
def add_date_job(cls, job_info: NodeModel, run_date: str | None = None) -> Job:
|
|
|
"""
|
|
|
创建指定时刻执行一次的任务。
|
|
|
|
|
|
参数:
|
|
|
- job_info (NodeModel): 任务信息。
|
|
|
- run_date (str | None): 执行时间字符串，默认取节点 trigger 配置。
|
|
|
|
|
|
返回:
|
|
|
- Job: 已注册的 APScheduler Job。
|
|
|
"""
|
|
|
date_str = run_date or job_info.trigger_args
|
|
|
if not date_str:
|
|
|
raise ValueError("date触发器缺少执行时间参数")
|
|
|
|
|
|
trigger = DateTrigger(run_date=date_str, timezone="Asia/Shanghai")
|
|
|
return cls._add_job_with_trigger(job_info, trigger)
|
|
|
|
|
|
@classmethod
|
|
|
def _add_job_with_trigger(cls, job_info: NodeModel, trigger) -> Job:
|
|
|
"""
|
|
|
添加任务到调度器
|
|
|
"""
|
|
|
code_block = job_info.func
|
|
|
if not code_block or not code_block.strip():
|
|
|
raise ValueError("任务代码块不能为空")
|
|
|
|
|
|
jobstore = job_info.jobstore or "sqlalchemy"
|
|
|
executor = job_info.executor or "threadpool"
|
|
|
|
|
|
job_args = []
|
|
|
if job_info.args:
|
|
|
args_str = str(job_info.args).strip()
|
|
|
if args_str:
|
|
|
job_args = [arg.strip() for arg in args_str.split(",") if arg.strip()]
|
|
|
|
|
|
job_kwargs = {}
|
|
|
if job_info.kwargs:
|
|
|
kwargs_str = str(job_info.kwargs).strip()
|
|
|
if kwargs_str:
|
|
|
try:
|
|
|
job_kwargs = json.loads(kwargs_str)
|
|
|
except json.JSONDecodeError:
|
|
|
raise ValueError(f"关键字参数JSON格式无效: {kwargs_str}")
|
|
|
|
|
|
# 缓存 job_name，用于 EVENT_JOB_SUBMITTED 时获取
|
|
|
cls._job_name_cache[str(job_info.id)] = job_info.name or ""
|
|
|
|
|
|
try:
|
|
|
job = scheduler.add_job(
|
|
|
func=cls._task_wrapper,
|
|
|
trigger=trigger,
|
|
|
args=[str(job_info.id), code_block, *job_args],
|
|
|
kwargs=job_kwargs,
|
|
|
id=str(job_info.id),
|
|
|
name=job_info.name,
|
|
|
coalesce=job_info.coalesce,
|
|
|
max_instances=1,
|
|
|
jobstore=jobstore,
|
|
|
executor=executor,
|
|
|
)
|
|
|
logger.info(f"任务 {job_info.id} 添加到 {jobstore} 存储器成功")
|
|
|
return job
|
|
|
except ConflictingIdError:
|
|
|
scheduler.remove_job(job_id=str(job_info.id), jobstore=jobstore)
|
|
|
job = scheduler.add_job(
|
|
|
func=cls._task_wrapper,
|
|
|
trigger=trigger,
|
|
|
args=[str(job_info.id), code_block, *job_args],
|
|
|
kwargs=job_kwargs,
|
|
|
id=str(job_info.id),
|
|
|
name=job_info.name,
|
|
|
coalesce=job_info.coalesce,
|
|
|
max_instances=1,
|
|
|
jobstore=jobstore,
|
|
|
executor=executor,
|
|
|
)
|
|
|
logger.info(f"任务 {job_info.id} 已存在，已移除旧任务并重新添加")
|
|
|
return job
|
|
|
|
|
|
@classmethod
|
|
|
def start(cls, paused: bool = False) -> None:
|
|
|
"""
|
|
|
启动全局调度器。
|
|
|
|
|
|
参数:
|
|
|
- paused (bool): 是否以暂停状态启动。
|
|
|
|
|
|
返回:
|
|
|
- None
|
|
|
"""
|
|
|
scheduler.start(paused=paused)
|
|
|
|
|
|
@classmethod
|
|
|
async def shutdown(cls, wait: bool = False):
|
|
|
"""
|
|
|
关闭调度器。
|
|
|
|
|
|
参数:
|
|
|
- wait (bool): 是否等待当前任务结束。
|
|
|
|
|
|
返回:
|
|
|
- 与 APScheduler shutdown 返回值一致。
|
|
|
"""
|
|
|
return scheduler.shutdown(wait=wait)
|
|
|
|
|
|
@classmethod
|
|
|
def configure(
|
|
|
cls, gconfig: dict | None = None, prefix: str = "apscheduler.", **options
|
|
|
) -> None:
|
|
|
"""
|
|
|
透传配置底层 APScheduler。
|
|
|
|
|
|
参数:
|
|
|
- gconfig (dict | None): 全局配置字典。
|
|
|
- prefix (str): 配置键前缀。
|
|
|
- **options: 其它 configure 关键字参数。
|
|
|
|
|
|
返回:
|
|
|
- None
|
|
|
"""
|
|
|
scheduler.configure(gconfig or {}, prefix, **options)
|
|
|
|
|
|
@classmethod
|
|
|
def pause(cls) -> None:
|
|
|
"""
|
|
|
暂停调度器。
|
|
|
|
|
|
返回:
|
|
|
- None
|
|
|
"""
|
|
|
scheduler.pause()
|
|
|
|
|
|
@classmethod
|
|
|
def resume(cls) -> None:
|
|
|
"""
|
|
|
恢复调度器。
|
|
|
|
|
|
返回:
|
|
|
- None
|
|
|
"""
|
|
|
scheduler.resume()
|
|
|
|
|
|
@classmethod
|
|
|
def is_running(cls) -> bool:
|
|
|
"""
|
|
|
调度器是否处于运行态。
|
|
|
|
|
|
返回:
|
|
|
- bool: 是否 running。
|
|
|
"""
|
|
|
return scheduler.running
|
|
|
|
|
|
@classmethod
|
|
|
def get_scheduler_state(cls) -> str:
|
|
|
"""
|
|
|
将调度器内部 state 码映射为中文状态。
|
|
|
|
|
|
返回:
|
|
|
- str: 停止 / 运行中 / 暂停 / 未知。
|
|
|
"""
|
|
|
if scheduler.state == 0:
|
|
|
return "停止"
|
|
|
if scheduler.state == 1:
|
|
|
return "运行中"
|
|
|
if scheduler.state == 2:
|
|
|
return "暂停"
|
|
|
return "未知"
|
|
|
|
|
|
@classmethod
|
|
|
def get_job(cls, job_id: str | int, jobstore: str | None = None) -> Job | None:
|
|
|
"""
|
|
|
按 ID 获取单个任务。
|
|
|
|
|
|
参数:
|
|
|
- job_id (str | int): 任务 ID。
|
|
|
- jobstore (str | None): 存储器别名。
|
|
|
|
|
|
返回:
|
|
|
- Job | None: 任务对象或不存在。
|
|
|
"""
|
|
|
return scheduler.get_job(str(job_id), jobstore)
|
|
|
|
|
|
@classmethod
|
|
|
def get_jobs(cls, jobstore: str | None = None) -> list[Job]:
|
|
|
"""
|
|
|
列出指定存储器中的任务。
|
|
|
|
|
|
参数:
|
|
|
- jobstore (str | None): 存储器别名，None 表示默认存储。
|
|
|
|
|
|
返回:
|
|
|
- list[Job]: 任务列表。
|
|
|
"""
|
|
|
return scheduler.get_jobs(jobstore)
|
|
|
|
|
|
@classmethod
|
|
|
def get_all_jobs(cls) -> list[Job]:
|
|
|
"""
|
|
|
列出所有存储器中的任务。
|
|
|
|
|
|
返回:
|
|
|
- list[Job]: 任务列表。
|
|
|
"""
|
|
|
return scheduler.get_jobs()
|
|
|
|
|
|
@classmethod
|
|
|
def remove_job(cls, job_id: str | int, jobstore: str | None = None) -> None:
|
|
|
"""
|
|
|
从调度器移除指定任务。
|
|
|
|
|
|
参数:
|
|
|
- job_id (str | int): 任务 ID。
|
|
|
- jobstore (str | None): 存储器别名。
|
|
|
|
|
|
返回:
|
|
|
- None
|
|
|
"""
|
|
|
scheduler.remove_job(str(job_id), jobstore)
|
|
|
|
|
|
@classmethod
|
|
|
def clear_jobs(cls) -> None:
|
|
|
"""
|
|
|
移除所有存储器中的全部任务。
|
|
|
|
|
|
返回:
|
|
|
- None
|
|
|
"""
|
|
|
scheduler.remove_all_jobs()
|
|
|
|
|
|
@classmethod
|
|
|
def print_jobs(cls, jobstore: str | None = None) -> str:
|
|
|
"""
|
|
|
打印调度器任务信息
|
|
|
|
|
|
参数:
|
|
|
- jobstore: 存储器别名，None 表示所有存储器
|
|
|
|
|
|
返回:
|
|
|
- str: 格式化的任务信息
|
|
|
"""
|
|
|
import io
|
|
|
|
|
|
output = io.StringIO()
|
|
|
scheduler.print_jobs(jobstore=jobstore, out=output)
|
|
|
return output.getvalue()
|
|
|
|
|
|
@classmethod
|
|
|
def sync_jobs_to_db(cls) -> int:
|
|
|
"""
|
|
|
将调度器中的任务同步到数据库
|
|
|
|
|
|
返回:
|
|
|
- int: 同步的任务数量
|
|
|
"""
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
from app.plugin.module_task.cronjob.job.model import JobModel
|
|
|
|
|
|
jobs = cls.get_all_jobs()
|
|
|
sync_count = 0
|
|
|
|
|
|
with Session(engine) as session:
|
|
|
for job in jobs:
|
|
|
existing_log = (
|
|
|
session
|
|
|
.query(JobModel)
|
|
|
.filter(JobModel.job_id == str(job.id), JobModel.status == JOB_STATUS_PENDING)
|
|
|
.first()
|
|
|
)
|
|
|
if not existing_log:
|
|
|
job_log = JobModel(
|
|
|
job_id=str(job.id),
|
|
|
job_name=job.name,
|
|
|
trigger_type=cls._get_trigger_type(str(job.id)),
|
|
|
status=JOB_STATUS_PENDING,
|
|
|
next_run_time=str(job.next_run_time) if job.next_run_time else None,
|
|
|
job_state=cls._get_job_state(job),
|
|
|
)
|
|
|
session.add(job_log)
|
|
|
sync_count += 1
|
|
|
session.commit()
|
|
|
|
|
|
return sync_count
|
|
|
|
|
|
@classmethod
|
|
|
def pause_job(cls, job_id: str | int, jobstore: str | None = None) -> Job | None:
|
|
|
"""
|
|
|
暂停单个任务。
|
|
|
|
|
|
参数:
|
|
|
- job_id (str | int): 任务 ID。
|
|
|
- jobstore (str | None): 存储器别名。
|
|
|
|
|
|
返回:
|
|
|
- Job | None: 暂停后的 Job 或 None。
|
|
|
"""
|
|
|
return scheduler.pause_job(str(job_id), jobstore)
|
|
|
|
|
|
@classmethod
|
|
|
def resume_job(cls, job_id: str | int, jobstore: str | None = None) -> Job | None:
|
|
|
"""
|
|
|
恢复单个任务。
|
|
|
|
|
|
参数:
|
|
|
- job_id (str | int): 任务 ID。
|
|
|
- jobstore (str | None): 存储器别名。
|
|
|
|
|
|
返回:
|
|
|
- Job | None: 恢复后的 Job 或 None。
|
|
|
"""
|
|
|
return scheduler.resume_job(str(job_id), jobstore)
|
|
|
|
|
|
@classmethod
|
|
|
def modify_job(cls, job_id: str | int, jobstore: str | None = None, **changes) -> Job | None:
|
|
|
"""
|
|
|
修改已存在任务的属性。
|
|
|
|
|
|
参数:
|
|
|
- job_id (str | int): 任务 ID。
|
|
|
- jobstore (str | None): 存储器别名。
|
|
|
- **changes: 传给 modify_job 的变更字段。
|
|
|
|
|
|
返回:
|
|
|
- Job | None: 修改后的 Job 或 None。
|
|
|
"""
|
|
|
return scheduler.modify_job(str(job_id), jobstore, **changes)
|
|
|
|
|
|
@classmethod
|
|
|
def run_job_now(cls, job_id: str | int, jobstore: str | None = None) -> Job | None:
|
|
|
"""
|
|
|
立即执行任务（通过临时 Job，不修改原任务 trigger）。
|
|
|
|
|
|
参数:
|
|
|
- job_id (str | int): 原任务 ID。
|
|
|
- jobstore (str | None): 存储器别名。
|
|
|
|
|
|
返回:
|
|
|
- Job | None: 临时任务对象；原任务不存在时为 None。
|
|
|
|
|
|
注意:
|
|
|
- 不改变原任务的触发器配置，仅追加一次性执行。
|
|
|
"""
|
|
|
job = cls.get_job(job_id=job_id, jobstore=jobstore)
|
|
|
if not job:
|
|
|
return None
|
|
|
|
|
|
# 创建一个新的临时任务 ID
|
|
|
temp_job_id = f"{job_id}_run_now_{datetime.now().timestamp()}"
|
|
|
|
|
|
# 缓存 job_name 和原任务 ID，用于 EVENT_JOB_SUBMITTED 时获取
|
|
|
# 格式: (原任务ID, 任务名称)
|
|
|
cls._job_name_cache[temp_job_id] = (str(job_id), f"{job.name}(立即执行)")
|
|
|
|
|
|
# 创建临时任务，延迟 0.1 秒执行，确保事件监听器能够捕获事件
|
|
|
from datetime import timedelta
|
|
|
|
|
|
trigger = DateTrigger(
|
|
|
run_date=datetime.now() + timedelta(seconds=0.1), timezone="Asia/Shanghai"
|
|
|
)
|
|
|
temp_job = scheduler.add_job(
|
|
|
func=job.func,
|
|
|
trigger=trigger,
|
|
|
args=job.args,
|
|
|
kwargs=job.kwargs,
|
|
|
id=temp_job_id,
|
|
|
name=f"{job.name}(立即执行)",
|
|
|
jobstore=jobstore or "default",
|
|
|
executor=job.executor,
|
|
|
max_instances=1,
|
|
|
)
|
|
|
|
|
|
logger.info(f"任务 {job_id} 已触发立即执行，临时任务 ID: {temp_job_id}")
|
|
|
return temp_job
|