753 lines
23 KiB
Python
753 lines
23 KiB
Python
# +======================================================================
|
||
# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ]
|
||
# +----------------------------------------------------------------------
|
||
# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有
|
||
# +----------------------------------------------------------------------
|
||
# | 开源协议: Apache-2.0
|
||
# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。
|
||
# +----------------------------------------------------------------------
|
||
# | 官方网站: https://www.djangoadmin.cn
|
||
# +----------------------------------------------------------------------
|
||
# | 研发团队: 小蚂蚁云团队 荣誉出品
|
||
# +======================================================================
|
||
# | 版权和免责声明:
|
||
# | 一、知识产权声明
|
||
# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、
|
||
# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权，
|
||
# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。
|
||
# | 二、使用限制与禁止条款
|
||
# | 您理解并同意，禁止将本产品用于以下任何行为：
|
||
# | 1. 任何违反中华人民共和国法律法规的用途；
|
||
# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为；
|
||
# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。
|
||
# | 三、责任豁免
|
||
# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失
|
||
# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三
|
||
# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。
|
||
# | 四、违约与法律责任
|
||
# | 如发现任何违反本声明的行为，本团队有权：
|
||
# | 1. 无条件配合公安机关及其他有权机关的调查取证工作；
|
||
# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失；
|
||
# | 3. 保留进一步采取刑事控告等法律措施的权利。
|
||
# | 五、许可范围说明
|
||
# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包
|
||
# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通
|
||
# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、
|
||
# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造
|
||
# | 成的一切经济损失及商誉损害。
|
||
# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件
|
||
# | 许可协议》附件为准。
|
||
# +======================================================================
|
||
|
||
"""
|
||
定时任务调度核心模块
|
||
=============================================================
|
||
|
||
该模块实现了定时任务的调度核心功能。
|
||
使用 APScheduler 库来管理和执行定时任务。
|
||
|
||
主要功能：
|
||
1. 任务调度管理（添加、移除、暂停、恢复）
|
||
2. URL请求发送（支持同步/异步）
|
||
3. 触发器方法映射
|
||
4. 任务立即执行
|
||
5. 调度器初始化
|
||
"""
|
||
|
||
import logging
|
||
from datetime import datetime
|
||
from typing import Dict, Any, Optional, Union
|
||
|
||
import requests
|
||
from apscheduler.schedulers.background import BackgroundScheduler
|
||
from apscheduler.triggers.cron import CronTrigger
|
||
from apscheduler.jobstores.base import JobLookupError
|
||
from django.conf import settings
|
||
from django.utils import timezone
|
||
|
||
import threading
|
||
|
||
_thread_local = threading.local()
|
||
|
||
|
||
def _get_start_time():
|
||
"""获取当前线程的开始时间"""
|
||
return getattr(_thread_local, 'job_start_time', None)
|
||
|
||
|
||
def _set_start_time(time):
|
||
"""设置当前线程的开始时间"""
|
||
_thread_local.job_start_time = time
|
||
|
||
|
||
# 延迟初始化调度器
|
||
_scheduler = None
|
||
|
||
|
||
def get_scheduler():
|
||
"""获取调度器实例（延迟初始化）"""
|
||
global _scheduler
|
||
if _scheduler is None:
|
||
from django.conf import settings
|
||
_scheduler = BackgroundScheduler(timezone=settings.TIME_ZONE)
|
||
return _scheduler
|
||
|
||
|
||
class _SchedulerProxy:
|
||
"""调度器代理类，延迟获取实际调度器"""
|
||
|
||
def __getattr__(self, name):
|
||
return getattr(get_scheduler(), name)
|
||
|
||
def __setattr__(self, name, value):
|
||
return setattr(get_scheduler(), name, value)
|
||
|
||
|
||
scheduler = _SchedulerProxy()
|
||
|
||
|
||
# =============================================================
|
||
# URL构建辅助函数
|
||
# =============================================================
|
||
|
||
def build_full_url(url: str) -> str:
|
||
"""
|
||
构建完整的URL
|
||
|
||
将相对路径转换为完整URL，如果已经是完整URL则直接返回。
|
||
|
||
Args:
|
||
url: 原始URL（可能是相对路径或完整URL）
|
||
|
||
Returns:
|
||
str: 完整URL
|
||
|
||
Example:
|
||
>>> build_full_url('/api/job/test/sms')
|
||
'http://localhost:8000/api/job/test/sms'
|
||
>>> build_full_url('https://api.example.com/webhook')
|
||
'https://api.example.com/webhook'
|
||
"""
|
||
if not url:
|
||
return url
|
||
|
||
# 如果已经是完整URL（包含http://或https://），直接返回
|
||
if url.startswith(('http://', 'https://')):
|
||
return url
|
||
|
||
# 如果是相对路径，加上域名
|
||
if url.startswith('/'):
|
||
domain = getattr(settings, 'SITE_DOMAIN', 'http://localhost:8000')
|
||
return f"{domain}{url}"
|
||
|
||
return url
|
||
|
||
|
||
# =============================================================
|
||
# 任务执行核心函数
|
||
# =============================================================
|
||
|
||
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]:
|
||
"""
|
||
执行任务（统一入口）
|
||
|
||
支持两种执行方式：
|
||
1. URL方式：发送HTTP请求到指定URL
|
||
2. 触发器方式：调用内部注册的方法
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
url: 任务URL（可以是相对路径或完整URL）
|
||
is_sync: 是否同步执行（1-同步，0-异步）
|
||
job_trigger: 任务触发器（内部方法标识）
|
||
params: 额外参数
|
||
|
||
Returns:
|
||
Dict[str, Any]: 执行结果，包含success状态和具体数据
|
||
"""
|
||
try:
|
||
# 记录开始时间 - 使用线程局部存储
|
||
start_time = timezone.now()
|
||
_set_start_time(start_time)
|
||
|
||
logging.info(f"执行定时任务 [ID: {job_id}, URL: {url}, Trigger: {job_trigger}, Sync: {is_sync}]")
|
||
|
||
result = None
|
||
|
||
# 方式1：URL方式（优先级高）
|
||
if url:
|
||
result = _execute_by_url(job_id, url, is_sync, params)
|
||
|
||
# 方式2：触发器方式
|
||
elif job_trigger:
|
||
result = _execute_by_trigger(job_trigger, job_id, params)
|
||
|
||
else:
|
||
logging.warning(f"任务既没有URL也没有触发器，无法执行 [ID: {job_id}]")
|
||
result = {
|
||
'success': False,
|
||
'error': '任务既没有URL也没有触发器',
|
||
'job_id': job_id
|
||
}
|
||
|
||
# 记录执行日志
|
||
status = 'success' if result.get('success', False) else 'failed'
|
||
error_msg = result.get('error', '') if status == 'failed' else ''
|
||
status_code = result.get('status_code') if url else None
|
||
|
||
if is_sync == 1 or not url:
|
||
_record_execution_log(job_id, status_code, status, error_msg)
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
logging.error(f"执行任务异常 [ID: {job_id}, 错误: {str(e)}]", exc_info=True)
|
||
_record_execution_log(job_id, None, 'failed', str(e))
|
||
return {
|
||
'success': False,
|
||
'error': str(e),
|
||
'job_id': job_id
|
||
}
|
||
finally:
|
||
_set_start_time(None)
|
||
|
||
|
||
def _execute_by_url(
|
||
job_id: int,
|
||
url: str,
|
||
is_sync: int,
|
||
params: Optional[Dict[str, Any]] = None
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
通过URL方式执行任务
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
url: 任务URL
|
||
is_sync: 是否同步执行
|
||
params: 额外参数
|
||
|
||
Returns:
|
||
Dict[str, Any]: 执行结果
|
||
"""
|
||
try:
|
||
full_url = build_full_url(url)
|
||
logging.info(f"完整URL: {full_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)
|
||
|
||
except Exception as e:
|
||
logging.error(f"URL执行失败 [ID: {job_id}, URL: {url}, 错误: {str(e)}]")
|
||
return {
|
||
'success': False,
|
||
'error': str(e),
|
||
'job_id': job_id,
|
||
'method': 'url'
|
||
}
|
||
|
||
|
||
def _send_sync_request(url: str, data: Dict[str, Any], job_id: int) -> Dict[str, Any]:
|
||
"""
|
||
发送同步HTTP请求
|
||
|
||
Args:
|
||
url: 请求URL
|
||
data: 请求数据
|
||
job_id: 任务ID
|
||
|
||
Returns:
|
||
Dict[str, Any]: 请求结果
|
||
"""
|
||
try:
|
||
response = requests.post(
|
||
url,
|
||
json=data,
|
||
timeout=30,
|
||
headers={'Content-Type': 'application/json'}
|
||
)
|
||
response.raise_for_status()
|
||
|
||
try:
|
||
result_data = response.json()
|
||
except:
|
||
result_data = {'text': response.text[:200]}
|
||
|
||
logging.info(f"同步任务执行完成 [ID: {job_id}, 状态码: {response.status_code}]")
|
||
|
||
return {
|
||
'success': True,
|
||
'method': 'url_sync',
|
||
'status_code': response.status_code,
|
||
'data': result_data,
|
||
'job_id': job_id
|
||
}
|
||
|
||
except requests.exceptions.MissingSchema as e:
|
||
error_msg = f"URL格式错误: {url}"
|
||
logging.error(error_msg)
|
||
return {'success': False, 'error': error_msg, 'job_id': job_id}
|
||
|
||
except requests.exceptions.ConnectionError as e:
|
||
error_msg = f"连接失败: {url}"
|
||
logging.error(error_msg)
|
||
return {'success': False, 'error': error_msg, 'job_id': job_id}
|
||
|
||
except requests.exceptions.Timeout as e:
|
||
error_msg = "请求超时"
|
||
logging.error(error_msg)
|
||
return {'success': False, 'error': error_msg, 'job_id': job_id}
|
||
|
||
except requests.exceptions.HTTPError as e:
|
||
error_msg = f"HTTP错误: {e.response.status_code if e.response else '未知'}"
|
||
logging.error(error_msg)
|
||
return {'success': False, 'error': error_msg, 'job_id': job_id}
|
||
|
||
|
||
def _send_async_request(url: str, data: Dict[str, Any], job_id: int) -> Dict[str, Any]:
|
||
"""
|
||
发送异步HTTP请求（使用线程池）
|
||
|
||
Args:
|
||
url: 请求URL
|
||
data: 请求数据
|
||
job_id: 任务ID
|
||
|
||
Returns:
|
||
Dict[str, Any]: 触发结果
|
||
"""
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
|
||
executor = getattr(settings, 'JOB_THREAD_POOL', ThreadPoolExecutor(max_workers=10))
|
||
|
||
def async_request():
|
||
"""异步请求任务"""
|
||
start_time = timezone.now()
|
||
_set_start_time(start_time)
|
||
try:
|
||
resp = requests.post(
|
||
url,
|
||
json=data,
|
||
timeout=30,
|
||
headers={'Content-Type': 'application/json'}
|
||
)
|
||
logging.info(f"异步任务执行完成 [ID: {job_id}, 状态码: {resp.status_code}]")
|
||
_record_execution_log(job_id, resp.status_code, 'success')
|
||
except Exception as e:
|
||
logging.error(f"异步任务执行异常 [ID: {job_id}, 错误: {str(e)}]")
|
||
_record_execution_log(job_id, None, 'failed', str(e))
|
||
finally:
|
||
_set_start_time(None)
|
||
|
||
executor.submit(async_request)
|
||
logging.info(f"异步任务已触发 [ID: {job_id}]")
|
||
|
||
return {
|
||
'success': True,
|
||
'method': 'url_async',
|
||
'status': 'triggered',
|
||
'job_id': job_id
|
||
}
|
||
|
||
|
||
def _record_execution_log(job_id: int, status_code: Optional[int], status: str, error: str = ''):
|
||
"""
|
||
记录任务执行日志
|
||
|
||
将任务的执行情况记录到数据库中的 JobLog 表。
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
status_code: HTTP状态码（仅对HTTP请求有效）
|
||
status: 执行状态（'success' 或 'failed'）
|
||
error: 错误信息（失败时记录）
|
||
"""
|
||
try:
|
||
from django.utils import timezone
|
||
from application.job.models import Job
|
||
from application.job_log.models import JobLog
|
||
|
||
end_time = timezone.now()
|
||
|
||
job = Job.objects.filter(id=job_id, is_delete=0).first()
|
||
if not job:
|
||
logging.error(f"记录日志失败：任务不存在 [ID: {job_id}]")
|
||
return
|
||
|
||
start_time = _get_start_time()
|
||
consume_time = 0
|
||
|
||
if start_time and isinstance(start_time, datetime):
|
||
consume_time = int((end_time - start_time).total_seconds() * 1000)
|
||
|
||
job_message = f"任务执行{'成功' if status == 'success' else '失败'}"
|
||
if status_code:
|
||
job_message += f"，HTTP状态码: {status_code}"
|
||
|
||
job_log = JobLog.objects.create(
|
||
job_id=job_id,
|
||
job_name=job.job_name,
|
||
job_group=job.job_group,
|
||
job_trigger=job.job_trigger,
|
||
cron_expression=job.cron_expression,
|
||
status=0 if status == 'success' else 1,
|
||
job_message=job_message,
|
||
start_time=start_time,
|
||
end_time=end_time,
|
||
consume_time=consume_time,
|
||
exception_info=error if error else None,
|
||
create_user='system',
|
||
update_user='system',
|
||
)
|
||
|
||
logging.info(f"任务执行日志已记录 [ID: {job_log.id}, 任务ID: {job_id}, 状态: {status}]")
|
||
|
||
except ImportError as e:
|
||
logging.error(f"记录执行日志失败：模型导入错误 {str(e)}")
|
||
except Exception as e:
|
||
logging.error(f"记录执行日志异常: {str(e)}", exc_info=True)
|
||
|
||
|
||
def _execute_by_trigger(trigger: str, job_id: int, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
"""
|
||
通过触发器方式执行任务
|
||
|
||
根据 trigger 标识调用对应的内部方法。
|
||
|
||
Args:
|
||
trigger: 触发器标识
|
||
job_id: 任务ID
|
||
params: 执行参数
|
||
|
||
Returns:
|
||
Dict[str, Any]: 执行结果
|
||
"""
|
||
try:
|
||
logging.info(f"通过触发器执行任务 [ID: {job_id}, Trigger: {trigger}]")
|
||
|
||
from application.job.executor import job_executor
|
||
|
||
result = job_executor.execute(trigger, job_id=job_id, **(params or {}))
|
||
|
||
return {
|
||
'success': True,
|
||
'method': 'trigger',
|
||
'trigger': trigger,
|
||
'data': result,
|
||
'job_id': job_id
|
||
}
|
||
|
||
except ImportError:
|
||
error_msg = "任务执行器未实现"
|
||
logging.error(error_msg)
|
||
return {'success': False, 'error': error_msg, 'job_id': job_id}
|
||
|
||
except KeyError as e:
|
||
error_msg = f"未找到触发器对应的方法: {trigger}"
|
||
logging.error(error_msg)
|
||
return {'success': False, 'error': error_msg, 'job_id': job_id}
|
||
|
||
except Exception as e:
|
||
error_msg = f"触发器执行失败: {str(e)}"
|
||
logging.error(error_msg, exc_info=True)
|
||
return {'success': False, 'error': error_msg, 'job_id': job_id}
|
||
|
||
|
||
# =============================================================
|
||
# 调度器管理函数
|
||
# =============================================================
|
||
|
||
def add_job_to_scheduler(job_id: int) -> bool:
|
||
"""
|
||
添加任务到调度器
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
|
||
Returns:
|
||
bool: 是否添加成功
|
||
"""
|
||
try:
|
||
from application.job.models import Job
|
||
|
||
job = Job.objects.filter(id=job_id, is_delete=0).first()
|
||
|
||
if not job:
|
||
logging.warning(f"任务不存在 [ID: {job_id}]")
|
||
return False
|
||
|
||
if job.status != 1:
|
||
logging.info(f"任务状态不可用，跳过调度 [ID: {job.id}, 状态: {job.status}]")
|
||
return False
|
||
|
||
if not job.cron_expression:
|
||
logging.warning(f"任务Cron表达式为空 [ID: {job.id}]")
|
||
return False
|
||
|
||
scheduler_job_id = f"job_{job.id}"
|
||
|
||
try:
|
||
scheduler.remove_job(scheduler_job_id)
|
||
logging.info(f"已移除现有任务 [ID: {job.id}]")
|
||
except JobLookupError:
|
||
pass
|
||
|
||
if job.url:
|
||
func = execute_job
|
||
args = [job.id, job.url, job.is_sync, job.job_trigger, {}]
|
||
elif job.job_trigger:
|
||
func = execute_job
|
||
args = [job.id, None, job.is_sync, job.job_trigger, {}]
|
||
else:
|
||
logging.warning(f"任务既没有URL也没有触发器 [ID: {job.id}]")
|
||
return False
|
||
|
||
scheduler.add_job(
|
||
func=func,
|
||
trigger=CronTrigger.from_crontab(job.cron_expression),
|
||
args=args,
|
||
id=scheduler_job_id,
|
||
name=job.job_name,
|
||
replace_existing=True,
|
||
misfire_grace_time=30,
|
||
coalesce=True
|
||
)
|
||
|
||
logging.info(f"任务已添加到调度器 [ID: {job.id}, 名称: {job.job_name}, Cron: {job.cron_expression}]")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logging.error(f"添加任务到调度器异常 [ID: {job_id}, 错误: {str(e)}]", exc_info=True)
|
||
return False
|
||
|
||
|
||
def remove_job_from_scheduler(job_id: int) -> bool:
|
||
"""
|
||
从调度器移除任务
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
|
||
Returns:
|
||
bool: 是否移除成功
|
||
"""
|
||
try:
|
||
scheduler_job_id = f"job_{job_id}"
|
||
scheduler.remove_job(scheduler_job_id)
|
||
logging.info(f"任务已从调度器移除 [ID: {job_id}]")
|
||
return True
|
||
|
||
except JobLookupError:
|
||
logging.info(f"任务不存在于调度器中 [ID: {job_id}]")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logging.error(f"从调度器移除任务异常 [ID: {job_id}, 错误: {str(e)}]")
|
||
return False
|
||
|
||
|
||
def pause_job_in_scheduler(job_id: int) -> bool:
|
||
"""
|
||
暂停调度器中的任务
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
|
||
Returns:
|
||
bool: 是否暂停成功
|
||
"""
|
||
try:
|
||
scheduler_job_id = f"job_{job_id}"
|
||
scheduler.pause_job(scheduler_job_id)
|
||
logging.info(f"任务已暂停 [ID: {job_id}]")
|
||
return True
|
||
|
||
except JobLookupError:
|
||
logging.warning(f"任务不存在于调度器中 [ID: {job_id}]")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logging.error(f"暂停任务异常 [ID: {job_id}, 错误: {str(e)}]")
|
||
return False
|
||
|
||
|
||
def resume_job_in_scheduler(job_id: int) -> bool:
|
||
"""
|
||
恢复调度器中的任务
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
|
||
Returns:
|
||
bool: 是否恢复成功
|
||
"""
|
||
try:
|
||
scheduler_job_id = f"job_{job_id}"
|
||
scheduler.resume_job(scheduler_job_id)
|
||
logging.info(f"任务已恢复 [ID: {job_id}]")
|
||
return True
|
||
|
||
except JobLookupError:
|
||
logging.warning(f"任务不存在于调度器中 [ID: {job_id}]")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logging.error(f"恢复任务异常 [ID: {job_id}, 错误: {str(e)}]")
|
||
return False
|
||
|
||
|
||
# =============================================================
|
||
# 任务立即执行函数
|
||
# =============================================================
|
||
|
||
def run_job_once(job_id: int, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
"""
|
||
立即执行一次任务（不依赖调度器）
|
||
|
||
Args:
|
||
job_id: 任务ID
|
||
params: 执行参数（可选）
|
||
|
||
Returns:
|
||
Dict[str, Any]: 执行结果
|
||
"""
|
||
try:
|
||
from application.job.models import Job
|
||
|
||
job = Job.objects.filter(id=job_id, is_delete=0).first()
|
||
|
||
if not job:
|
||
error_msg = f"任务不存在 [ID: {job_id}]"
|
||
logging.error(error_msg)
|
||
return {'success': False, 'error': error_msg}
|
||
|
||
logging.info(f"立即执行任务 [ID: {job_id}, 名称: {job.job_name}]")
|
||
|
||
result = execute_job(
|
||
job_id=job.id,
|
||
url=job.url,
|
||
is_sync=1,
|
||
job_trigger=job.job_trigger,
|
||
params=params or {}
|
||
)
|
||
|
||
return result
|
||
|
||
except Exception as e:
|
||
logging.error(f"立即执行任务异常 [ID: {job_id}, 错误: {str(e)}]", exc_info=True)
|
||
return {'success': False, 'error': str(e)}
|
||
|
||
|
||
# =============================================================
|
||
# 调度器初始化和关闭函数
|
||
# =============================================================
|
||
|
||
def init_scheduler() -> int:
|
||
"""
|
||
初始化调度器，加载所有运行中的任务
|
||
|
||
在 Django 启动时调用，加载所有状态为运行中的任务到调度器。
|
||
|
||
Returns:
|
||
int: 成功加载的任务数量
|
||
"""
|
||
try:
|
||
from application.job.models import Job
|
||
|
||
if not scheduler.running:
|
||
scheduler.start()
|
||
logging.info("调度器已启动")
|
||
|
||
jobs = Job.objects.filter(
|
||
status=1,
|
||
is_delete=0
|
||
).exclude(
|
||
cron_expression__isnull=True
|
||
).exclude(
|
||
cron_expression=''
|
||
)
|
||
|
||
logging.info(f"找到 {len(jobs)} 个待加载的任务")
|
||
|
||
success_count = 0
|
||
for job in jobs:
|
||
if add_job_to_scheduler(job.id):
|
||
success_count += 1
|
||
|
||
logging.info(f"调度器初始化完成，成功加载 {success_count}/{len(jobs)} 个任务")
|
||
|
||
_print_scheduler_status()
|
||
|
||
return success_count
|
||
|
||
except Exception as e:
|
||
logging.error(f"初始化调度器异常: {str(e)}", exc_info=True)
|
||
return 0
|
||
|
||
|
||
def _print_scheduler_status():
|
||
"""打印调度器当前状态（用于调试）"""
|
||
try:
|
||
jobs = scheduler.get_jobs()
|
||
logging.info(f"调度器当前状态: 运行中={scheduler.running}, 任务数={len(jobs)}")
|
||
|
||
for job in jobs:
|
||
next_run = job.next_run_time
|
||
next_run_str = next_run.strftime('%Y-%m-%d %H:%M:%S') if next_run else '未知'
|
||
logging.info(f" - 任务 [{job.id}] {job.name}, 下次执行: {next_run_str}")
|
||
|
||
except Exception as e:
|
||
logging.error(f"获取调度器状态失败: {str(e)}")
|
||
|
||
|
||
def shutdown_scheduler(wait: bool = True) -> bool:
|
||
"""
|
||
关闭调度器
|
||
|
||
Args:
|
||
wait: 是否等待正在执行的任务完成
|
||
|
||
Returns:
|
||
bool: 是否成功关闭
|
||
"""
|
||
try:
|
||
if scheduler.running:
|
||
scheduler.shutdown(wait=wait)
|
||
logging.info(f"调度器已关闭 (wait={wait})")
|
||
return True
|
||
except Exception as e:
|
||
logging.error(f"关闭调度器异常: {str(e)}")
|
||
return False
|
||
|
||
|
||
def reload_scheduler() -> int:
|
||
"""
|
||
重新加载调度器（清除所有任务后重新加载）
|
||
|
||
Returns:
|
||
int: 成功加载的任务数量
|
||
"""
|
||
try:
|
||
scheduler.remove_all_jobs()
|
||
logging.info("已移除所有调度任务")
|
||
|
||
return init_scheduler()
|
||
|
||
except Exception as e:
|
||
logging.error(f"重新加载调度器异常: {str(e)}")
|
||
return 0
|