733 lines
26 KiB
Python
733 lines
26 KiB
Python
# +======================================================================
|
||
# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ]
|
||
# +----------------------------------------------------------------------
|
||
# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有
|
||
# +----------------------------------------------------------------------
|
||
# | 开源协议: Apache-2.0
|
||
# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。
|
||
# +----------------------------------------------------------------------
|
||
# | 官方网站: https://www.djangoadmin.cn
|
||
# +----------------------------------------------------------------------
|
||
# | 研发团队: 小蚂蚁云团队 荣誉出品
|
||
# +======================================================================
|
||
# | 版权和免责声明:
|
||
# | 一、知识产权声明
|
||
# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、
|
||
# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权，
|
||
# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。
|
||
# | 二、使用限制与禁止条款
|
||
# | 您理解并同意，禁止将本产品用于以下任何行为：
|
||
# | 1. 任何违反中华人民共和国法律法规的用途；
|
||
# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为；
|
||
# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。
|
||
# | 三、责任豁免
|
||
# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失
|
||
# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三
|
||
# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。
|
||
# | 四、违约与法律责任
|
||
# | 如发现任何违反本声明的行为，本团队有权：
|
||
# | 1. 无条件配合公安机关及其他有权机关的调查取证工作；
|
||
# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失；
|
||
# | 3. 保留进一步采取刑事控告等法律措施的权利。
|
||
# | 五、许可范围说明
|
||
# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包
|
||
# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通
|
||
# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、
|
||
# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造
|
||
# | 成的一切经济损失及商誉损害。
|
||
# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件
|
||
# | 许可协议》附件为准。
|
||
# +======================================================================
|
||
|
||
"""
|
||
定时任务调度核心模块
|
||
=============================================================
|
||
|
||
该模块实现了定时任务的调度核心功能。
|
||
使用APScheduler库来管理和执行定时任务。
|
||
"""
|
||
|
||
import logging
|
||
import sys
|
||
import threading
|
||
import uuid
|
||
from datetime import datetime
|
||
from typing import Dict, Any, Optional
|
||
|
||
import requests
|
||
from apscheduler.schedulers.background import BackgroundScheduler
|
||
from apscheduler.triggers.cron import CronTrigger
|
||
from apscheduler.jobstores.base import JobLookupError
|
||
|
||
from config import INTERNAL_API_KEY, FLASK_SITE_DOMAIN
|
||
from src.common.utils.model_utils import model_to_dict
|
||
from src.core.redis_client import redis
|
||
|
||
# ============================================================
|
||
# 全局变量和日志配置
|
||
# ============================================================
|
||
|
||
# 线程局部存储（用于线程间数据隔离）
|
||
_thread_local = threading.local()
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 调度器实例（延迟初始化）
|
||
_scheduler = None
|
||
|
||
# 调度线程内可用的 Flask app（init_scheduler 在 app 上下文内捕获，供写任务日志用）。
|
||
# APScheduler 的 cron 回调运行在后台线程，没有请求上下文，写库前需手动推 app 上下文。
|
||
_SCHEDULER_APP = None
|
||
|
||
|
||
# ============================================================
|
||
# 调度器实例获取函数
|
||
# ============================================================
|
||
|
||
|
||
def get_scheduler():
|
||
"""
|
||
获取调度器实例（延迟初始化）
|
||
|
||
使用延迟初始化模式，确保调度器只被创建一次。
|
||
避免在模块导入时立即创建，提高应用启动速度。
|
||
|
||
Returns:
|
||
BackgroundScheduler: APScheduler 后台调度器实例
|
||
"""
|
||
global _scheduler
|
||
if _scheduler is None:
|
||
# 创建后台调度器，设置时区为北京时间
|
||
_scheduler = BackgroundScheduler(timezone="Asia/Shanghai")
|
||
return _scheduler
|
||
|
||
|
||
# 创建全局调度器实例
|
||
scheduler = get_scheduler()
|
||
|
||
# ============================================================
|
||
# 分布式任务锁（防重复执行）
|
||
# ============================================================
|
||
|
||
#: 锁最长持有时间（秒）。作为进程崩溃时的兜底，超时自动释放，
|
||
#: 避免长期占用导致任务被永久跳过。
|
||
_JOB_LOCK_TTL_SECONDS = 300
|
||
|
||
|
||
class _JobExecutionLock:
|
||
"""
|
||
基于 Redis 的分布式任务锁。
|
||
|
||
背景：调度器是进程内的 BackgroundScheduler，gunicorn 多 worker / 多实例
|
||
部署时每个进程都会注册并触发同一 cron 任务。通过 Redis SET NX EX 原子操作
|
||
保证同一 job_id 在同一触发周期内只有一个实例真正执行。
|
||
|
||
acquire() 失败（Redis 不可用）时降级放行，避免锁服务故障导致任务停摆。
|
||
"""
|
||
|
||
def __init__(self, job_id: int):
|
||
self.key = f"scheduler:job_lock:{job_id}"
|
||
self.token = uuid.uuid4().hex
|
||
|
||
def acquire(self) -> bool:
|
||
try:
|
||
# 原子 SET key token NX EX ttl：并发下仅一个实例能成功
|
||
return bool(redis.client.set(self.key, self.token, nx=True, ex=_JOB_LOCK_TTL_SECONDS))
|
||
except Exception as e:
|
||
logger.error(f"获取调度任务锁异常 [ID: {self.key}]: {str(e)}")
|
||
return True
|
||
|
||
def release(self):
|
||
try:
|
||
# 仅删除自己持有的锁，避免误删其他实例刚获取的新锁
|
||
if redis.client.get(self.key) == self.token:
|
||
redis.client.delete(self.key)
|
||
except Exception as e:
|
||
logger.error(f"释放调度任务锁异常 [ID: {self.key}]: {str(e)}")
|
||
|
||
|
||
# ============================================================
|
||
# URL 构建工具函数
|
||
# ============================================================
|
||
|
||
|
||
def build_full_url(url: str) -> str:
|
||
"""
|
||
构建完整的URL
|
||
|
||
将相对路径拼接为完整的带域名的URL，
|
||
如果已经是完整URL则直接返回。
|
||
|
||
Args:
|
||
url: 原始URL（相对路径或完整URL）
|
||
|
||
Returns:
|
||
str: 完整URL
|
||
|
||
Examples:
|
||
>>> build_full_url('/job/sendSms')
|
||
'http://127.0.0.1:8023/job/sendSms'
|
||
>>> build_full_url('https://api.example.com/test')
|
||
'https://api.example.com/test'
|
||
"""
|
||
if not url:
|
||
return url
|
||
|
||
# 已经是完整URL，直接返回
|
||
if url.startswith(("http://", "https://")):
|
||
return url
|
||
|
||
# 相对路径，拼接站点域名
|
||
if url.startswith("/"):
|
||
# 内部 API 调用统一挂到 /api/v1（P6 API 版本化）
|
||
if not url.startswith("/api/"):
|
||
url = f"/api/v1{url}"
|
||
url = f"{FLASK_SITE_DOMAIN}{url}"
|
||
|
||
return url
|
||
|
||
|
||
# ============================================================
|
||
# 任务执行日志（每次实际执行后写入 flask_job_log）
|
||
# ============================================================
|
||
|
||
|
||
def _extract_log_fields(result) -> tuple:
|
||
"""从执行结果提取 (success, message, error) 三元组，兼容 URL/触发器两种返回形态。
|
||
|
||
- URL 任务：result = {"success": bool, "status_code":..., "data":...}
|
||
- 触发器任务：result = {"success": True, "data": {"success": bool, "message": ...}}
|
||
（真正成败在 data 里，由 executor 返回）
|
||
"""
|
||
if not isinstance(result, dict):
|
||
return True, str(result), None
|
||
success = result.get("success")
|
||
data = result.get("data")
|
||
if isinstance(data, dict) and "success" in data:
|
||
success = data.get("success")
|
||
message = result.get("message") or (data.get("message") if isinstance(data, dict) else None) or ""
|
||
error = result.get("error")
|
||
if not message and data is not None:
|
||
message = str(data)
|
||
return bool(success), str(message or ""), (str(error) if error else None)
|
||
|
||
|
||
def _write_job_log(job_id, success, message, error, start, end):
|
||
"""在 app 上下文内写任务日志（尽力而为，写失败不影响任务执行结果）。"""
|
||
if _SCHEDULER_APP is None:
|
||
logger.error("调度器 app 未绑定，跳过任务日志写入 [ID: %s]", job_id)
|
||
return
|
||
try:
|
||
with _SCHEDULER_APP.app_context():
|
||
from src.modules.job_log.service import write_job_log
|
||
write_job_log(
|
||
job_id, success=success, message=message, error=error,
|
||
start_time=start, end_time=end,
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"写入任务日志异常 [ID: {job_id}]: {e}")
|
||
|
||
|
||
def _execute_and_log(job_id, func, *args, **kwargs):
|
||
"""执行任务并自动记录执行日志（开始/结束时间、耗时、结果、异常）。
|
||
|
||
无论成功失败（含异常）都会写一条 flask_job_log；返回执行结果给调用方。
|
||
"""
|
||
start = datetime.now()
|
||
try:
|
||
result = func(*args, **kwargs)
|
||
success, message, error = _extract_log_fields(result)
|
||
except Exception as e:
|
||
result = {"success": False, "error": str(e)}
|
||
success, message, error = False, str(e), str(e)
|
||
logger.error(f"执行任务异常 [ID: {job_id}]: {e}")
|
||
end = datetime.now()
|
||
_write_job_log(job_id, success, message, error, start, end)
|
||
elapsed = (end - start).total_seconds()
|
||
status = "成功" if success else "失败"
|
||
print(f"[JobScheduler] 任务 [ID: {job_id}] 执行{status}，耗时 {elapsed:.2f}s，结果: {message or error}")
|
||
return result
|
||
|
||
|
||
# ============================================================
|
||
# 任务执行核心函数
|
||
# ============================================================
|
||
|
||
|
||
def execute_job(
|
||
job_id: int,
|
||
url: Optional[str] = None,
|
||
is_sync: int = 0,
|
||
job_trigger: Optional[str] = None,
|
||
params: Optional[Dict[str, Any]] = None,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
执行任务（调度器调用的入口函数）
|
||
|
||
这是 APScheduler 调用的核心函数，根据任务配置分发到不同的执行方式。
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
url: 任务URL（HTTP请求方式）
|
||
is_sync: 是否同步执行（1-同步，0-异步）
|
||
job_trigger: 任务触发器（内部调用方式）
|
||
params: 额外参数
|
||
|
||
Returns:
|
||
Dict: 执行结果，包含 success 和 data/error 字段
|
||
"""
|
||
logger.debug(f"execute_job id={job_id} url={url} sync={is_sync} trigger={job_trigger} params={params}")
|
||
sys.stdout.flush()
|
||
|
||
logger.info(f"执行定时任务 [ID: {job_id}, URL: {url}, Trigger: {job_trigger}]")
|
||
print(f"[JobScheduler] 开始执行任务 [ID: {job_id}, URL: {url}, Trigger: {job_trigger}, 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}]")
|
||
|
||
# 根据任务类型分发执行（同步路径持有分布式锁，防多实例重复执行）
|
||
if url:
|
||
if is_sync == 1:
|
||
return _run_with_job_lock(job_id, _execute_by_url, job_id, url, is_sync, params)
|
||
# 异步路径：父线程立即返回，锁由子线程获取（见 _send_async_request）
|
||
return _execute_by_url(job_id, url, is_sync, params)
|
||
|
||
if job_trigger:
|
||
return _run_with_job_lock(job_id, _execute_by_trigger, job_trigger, job_id, params)
|
||
|
||
# 无效的任务配置
|
||
return {"success": False, "error": "任务既没有URL也没有触发器"}
|
||
|
||
|
||
def _run_with_job_lock(job_id: int, func, *args, **kwargs) -> Dict[str, Any]:
|
||
"""
|
||
在分布式锁保护下执行任务函数，保证同一任务只被一个调度器实例执行。
|
||
|
||
调用方负责传入实际执行函数；锁的获取/释放、异常兜底在此统一处理。
|
||
"""
|
||
lock = _JobExecutionLock(job_id)
|
||
if not lock.acquire():
|
||
logger.info(f"定时任务 [ID: {job_id}] 已被其他调度器实例执行，本次跳过")
|
||
print(f"[JobScheduler] 任务 [ID: {job_id}] 已被其他实例占用，本次跳过")
|
||
return {"success": False, "error": "任务已被其他调度器实例执行"}
|
||
|
||
try:
|
||
return func(*args, **kwargs)
|
||
except Exception as e:
|
||
logger.error(f"执行任务异常 [ID: {job_id}]: {str(e)}")
|
||
return {"success": False, "error": str(e)}
|
||
finally:
|
||
lock.release()
|
||
|
||
|
||
# ============================================================
|
||
# URL 方式任务执行
|
||
# ============================================================
|
||
|
||
|
||
def _execute_by_url(job_id: int, url: str, is_sync: int, params: Optional[Dict] = None) -> Dict:
|
||
"""
|
||
通过URL方式执行任务
|
||
|
||
构建请求数据并发送HTTP请求。
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
url: 目标URL
|
||
is_sync: 1-同步等待响应，0-异步不等待
|
||
params: 额外参数
|
||
|
||
Returns:
|
||
Dict: 执行结果
|
||
"""
|
||
# 构建完整URL
|
||
full_url = build_full_url(url)
|
||
|
||
# 构建请求数据
|
||
request_data = {"job_id": job_id, "timestamp": datetime.now().isoformat(), "params": params or {}} # 请求时间戳
|
||
|
||
# 根据同步/异步选择发送方式
|
||
if is_sync == 1:
|
||
return _send_sync_request(full_url, request_data, job_id)
|
||
else:
|
||
return _send_async_request(full_url, request_data, job_id)
|
||
|
||
|
||
def _send_sync_request(url: str, data: Dict, job_id: int) -> Dict:
|
||
"""
|
||
发送同步HTTP请求（等待响应）
|
||
|
||
同步发送POST请求，等待目标服务返回响应后继续执行。
|
||
|
||
Args:
|
||
url: 目标URL
|
||
data: 请求数据（JSON格式）
|
||
job_id: 任务ID（用于日志）
|
||
|
||
Returns:
|
||
Dict: 包含响应状态码和数据的执行结果
|
||
"""
|
||
def _do() -> Dict:
|
||
logger.info(f"发送同步请求 [ID: {job_id}, URL: {url}]")
|
||
|
||
# 准备请求头
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json",
|
||
"User-Agent": "Scheduler/1.0",
|
||
# 添加定时任务专用密钥，用于跳过权限验证
|
||
"X-Scheduler-Key": INTERNAL_API_KEY,
|
||
}
|
||
|
||
# 发送POST请求
|
||
response = requests.post(url, json=data, timeout=30, headers=headers) # 30秒超时
|
||
response.raise_for_status() # 非200状态码抛出异常
|
||
|
||
# 解析响应数据
|
||
try:
|
||
result_data = response.json()
|
||
except Exception:
|
||
# JSON解析失败时返回原始文本（限制长度）
|
||
result_data = {"text": response.text[:200]}
|
||
|
||
return {"success": True, "status_code": response.status_code, "data": result_data}
|
||
|
||
# 同步执行（真实干活的地方）：成功/失败/异常都写任务日志
|
||
return _execute_and_log(job_id, _do)
|
||
|
||
|
||
def _send_async_request(url: str, data: Dict, job_id: int) -> Dict:
|
||
"""
|
||
发送异步HTTP请求（不等待响应）
|
||
|
||
异步发送POST请求，在独立线程中执行，不等待响应结果。
|
||
|
||
Args:
|
||
url: 目标URL
|
||
data: 请求数据（JSON格式）
|
||
job_id: 任务ID（用于日志）
|
||
|
||
Returns:
|
||
Dict: 立即返回触发状态
|
||
"""
|
||
import threading
|
||
|
||
def async_request():
|
||
"""异步请求执行函数（在独立线程中运行）"""
|
||
# 多实例下每个 worker 都会触发该线程，必须加锁保证只执行一次
|
||
lock = _JobExecutionLock(job_id)
|
||
if not lock.acquire():
|
||
logger.info(f"异步任务 [ID: {job_id}] 已被其他调度器实例执行，本次跳过")
|
||
print(f"[JobScheduler] 异步任务 [ID: {job_id}] 已被其他实例占用，本次跳过")
|
||
return
|
||
try:
|
||
def _do() -> Dict:
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"X-Scheduler-Key": INTERNAL_API_KEY,
|
||
}
|
||
resp = requests.post(url, json=data, timeout=30, headers=headers)
|
||
logger.info(f"异步任务完成 [ID: {job_id}, 状态码: {resp.status_code}]")
|
||
print(f"[JobScheduler] 异步任务完成 [ID: {job_id}, 状态码: {resp.status_code}]")
|
||
return {"success": True, "status_code": resp.status_code,
|
||
"message": f"异步任务完成，状态码 {resp.status_code}"}
|
||
|
||
# 异步真实干活的地方：成功/失败/异常都写任务日志
|
||
_execute_and_log(job_id, _do)
|
||
except Exception as e:
|
||
logger.error(f"异步任务异常 [ID: {job_id}]: {str(e)}")
|
||
finally:
|
||
lock.release()
|
||
|
||
# 创建并启动独立线程执行异步请求
|
||
thread = threading.Thread(target=async_request)
|
||
thread.start()
|
||
print(f"[JobScheduler] 异步任务已触发 [ID: {job_id}, URL: {url}]")
|
||
|
||
return {"success": True, "status": "triggered"}
|
||
|
||
|
||
def _execute_by_trigger(trigger: str, job_id: int, params: Optional[Dict] = None) -> Dict:
|
||
"""
|
||
通过触发器方式执行任务
|
||
|
||
调用内部任务执行器执行业务逻辑。
|
||
|
||
Args:
|
||
trigger: 触发器标识（任务类型）
|
||
job_id: 任务ID
|
||
params: 额外参数
|
||
|
||
Returns:
|
||
Dict: 执行结果
|
||
"""
|
||
def _do() -> Dict:
|
||
from src.modules.job.executor import job_executor
|
||
|
||
result = job_executor.execute(trigger, job_id=job_id, **(params or {}))
|
||
return {"success": True, "trigger": trigger, "data": result}
|
||
|
||
# 触发器任务真实干活的地方：成功/失败/异常都写任务日志
|
||
return _execute_and_log(job_id, _do)
|
||
|
||
|
||
# ============================================================
|
||
# 调度器任务管理函数
|
||
# ============================================================
|
||
|
||
|
||
def add_job_to_scheduler(job_dict: dict) -> bool:
|
||
"""
|
||
添加任务到调度器
|
||
|
||
将任务配置添加到APScheduler中，设置Cron触发规则。
|
||
|
||
Args:
|
||
job_dict: 任务参数字典，包含以下字段：
|
||
- id: 任务ID
|
||
- status: 任务状态（1-启用）
|
||
- cronExpression: Cron表达式
|
||
- url: 请求URL
|
||
- isSync: 是否同步
|
||
- jobTrigger: 触发器标识
|
||
- jobName: 任务名称
|
||
|
||
Returns:
|
||
bool: 添加成功返回True，否则返回False
|
||
"""
|
||
try:
|
||
# ============================================================
|
||
# 1. 从字典中提取任务参数
|
||
# ============================================================
|
||
job_id = job_dict.get("id")
|
||
job_status = job_dict.get("status")
|
||
job_cron = job_dict.get("cronExpression")
|
||
job_url = job_dict.get("url")
|
||
job_sync = job_dict.get("isSync")
|
||
job_trigger = job_dict.get("jobTrigger")
|
||
job_name = job_dict.get("jobName")
|
||
|
||
logger.debug(f"add_job id={job_id} status={job_status} cron='{job_cron}'")
|
||
|
||
# ============================================================
|
||
# 2. 参数校验
|
||
# ============================================================
|
||
# 检查任务状态是否启用
|
||
if job_status != 1:
|
||
logger.debug(f"跳过添加任务: 状态不是启用 (id={job_id}, status={job_status})")
|
||
return False
|
||
|
||
# 检查Cron表达式是否为空
|
||
if not job_cron:
|
||
logger.warning(f"跳过添加任务: Cron表达式为空 (id={job_id})")
|
||
return False
|
||
|
||
# ============================================================
|
||
# 3. 验证Cron表达式并计算下次执行时间
|
||
# ============================================================
|
||
try:
|
||
trigger = CronTrigger.from_crontab(job_cron)
|
||
from datetime import datetime
|
||
|
||
next_run = trigger.get_next_fire_time(None, datetime.now())
|
||
logger.debug(f"Cron表达式有效 (id={job_id}, 下次执行: {next_run})")
|
||
|
||
if next_run is None:
|
||
logger.warning(f"无法计算下次执行时间，Cron表达式可能永远不会触发 (id={job_id})")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"Cron表达式无效 (id={job_id}): {str(e)}")
|
||
return False
|
||
|
||
# ============================================================
|
||
# 4. 添加任务到调度器
|
||
# ============================================================
|
||
scheduler_job_id = f"job_{job_id}"
|
||
|
||
# 移除已存在的同名任务（避免重复）
|
||
try:
|
||
scheduler.remove_job(scheduler_job_id)
|
||
except JobLookupError:
|
||
pass
|
||
|
||
# 添加新任务
|
||
scheduler.add_job(
|
||
func=execute_job, # 执行函数
|
||
trigger=trigger, # Cron触发器
|
||
args=[job_id, job_url, job_sync, job_trigger, {}], # 函数参数
|
||
id=scheduler_job_id, # 任务唯一标识
|
||
name=job_name, # 任务显示名称
|
||
replace_existing=True, # 替换已存在的任务
|
||
misfire_grace_time=30, # 错过执行的宽限时间（秒）
|
||
coalesce=True, # 合并错过的执行
|
||
)
|
||
|
||
logger.info(f"任务已添加到调度器 [ID: {job_id}, cron: {job_cron}]")
|
||
print(f"[JobScheduler] 任务已添加到调度器 [ID: {job_id}, 名称: {job_name}, cron: {job_cron}, 下次执行: {next_run}]")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"添加任务到调度器异常: {str(e)}")
|
||
return False
|
||
|
||
|
||
def remove_job_from_scheduler(job_id: int) -> bool:
|
||
"""
|
||
从调度器移除任务
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
|
||
Returns:
|
||
bool: 移除成功返回True，否则返回False
|
||
"""
|
||
try:
|
||
scheduler.remove_job(f"job_{job_id}")
|
||
return True
|
||
except JobLookupError:
|
||
# 任务不存在，视为成功
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"移除任务异常: {str(e)}")
|
||
return False
|
||
|
||
|
||
def pause_job_in_scheduler(job_id: int) -> bool:
|
||
"""
|
||
暂停调度器中的任务
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
|
||
Returns:
|
||
bool: 暂停成功返回True，否则返回False
|
||
"""
|
||
try:
|
||
scheduler.pause_job(f"job_{job_id}")
|
||
return True
|
||
except JobLookupError:
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"暂停任务异常: {str(e)}")
|
||
return False
|
||
|
||
|
||
def resume_job_in_scheduler(job_id: int) -> bool:
|
||
"""
|
||
恢复调度器中的任务
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
|
||
Returns:
|
||
bool: 恢复成功返回True，否则返回False
|
||
"""
|
||
try:
|
||
scheduler.resume_job(f"job_{job_id}")
|
||
return True
|
||
except JobLookupError:
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"恢复任务异常: {str(e)}")
|
||
return False
|
||
|
||
|
||
def run_job_once(job_id: int, params: Optional[Dict] = None) -> Dict:
|
||
"""
|
||
立即执行一次任务（不依赖调度器）
|
||
|
||
忽略Cron表达式，立即执行指定任务一次。
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
params: 执行参数
|
||
|
||
Returns:
|
||
Dict: 执行结果
|
||
"""
|
||
try:
|
||
from src.modules.job.models import Job
|
||
|
||
job = Job.query.filter(Job.id == job_id, Job.is_delete == 0).first()
|
||
|
||
if not job:
|
||
return {"success": False, "error": "任务不存在"}
|
||
|
||
# 直接执行任务（同步方式）
|
||
return execute_job(job.id, job.url, 1, job.job_trigger, params or {})
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
|
||
# ============================================================
|
||
# 调度器初始化和重载函数
|
||
# ============================================================
|
||
|
||
|
||
def init_scheduler() -> int:
|
||
"""
|
||
初始化调度器
|
||
|
||
启动调度器并从数据库加载所有启用的任务。
|
||
|
||
Returns:
|
||
int: 成功加载的任务数量
|
||
"""
|
||
global _SCHEDULER_APP
|
||
# 捕获 Flask app（供调度线程写任务日志时推 app 上下文；Scheduler.init_app 在 app 上下文内调用）
|
||
try:
|
||
from flask import current_app
|
||
_SCHEDULER_APP = current_app._get_current_object()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
from src.modules.job.models import Job
|
||
|
||
# 启动调度器（如果尚未启动）
|
||
if not scheduler.running:
|
||
scheduler.start()
|
||
logger.info("调度器已启动")
|
||
print(f"[JobScheduler] 调度器已启动 [{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}]")
|
||
|
||
# 查询所有启用的任务
|
||
jobs = Job.query.filter(Job.status == 1, Job.is_delete == 0).all()
|
||
|
||
# 使用 model_to_dict 将每个 job 对象转换为字典
|
||
# 避免 SQLAlchemy Session 分离问题
|
||
job_dicts = []
|
||
for job in jobs:
|
||
job_dict = dict(model_to_dict(job))
|
||
job_dicts.append(job_dict)
|
||
|
||
# 使用字典数据添加任务到调度器
|
||
success_count = 0
|
||
for job_dict in job_dicts:
|
||
if add_job_to_scheduler(job_dict):
|
||
success_count += 1
|
||
|
||
logger.info(f"调度器初始化完成，加载 {success_count}/{len(job_dicts)} 个任务")
|
||
print(f"[JobScheduler] 调度器初始化完成，加载 {success_count}/{len(job_dicts)} 个任务")
|
||
return success_count
|
||
|
||
except Exception as e:
|
||
logger.error(f"初始化调度器异常: {str(e)}")
|
||
return 0
|
||
|
||
|
||
def reload_scheduler() -> int:
|
||
"""
|
||
重新加载调度器
|
||
|
||
清空所有现有任务，然后重新从数据库加载。
|
||
|
||
Returns:
|
||
int: 成功加载的任务数量
|
||
"""
|
||
try:
|
||
# 清空所有现有任务
|
||
scheduler.remove_all_jobs()
|
||
print(f"[JobScheduler] 开始重新加载所有定时任务 [{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}]")
|
||
# 重新初始化
|
||
return init_scheduler()
|
||
except Exception as e:
|
||
logger.error(f"重新加载调度器异常: {str(e)}")
|
||
return 0
|