919 lines
32 KiB
Python
919 lines
32 KiB
Python
# +======================================================================
|
|
# | DjangoAdmin 企业数字化开发平台 [ 赋能开发者，助力企业发展 ]
|
|
# +----------------------------------------------------------------------
|
|
# | Copyright (c) 2021-2026 小蚂蚁云团队 版权所有
|
|
# +----------------------------------------------------------------------
|
|
# | 开源协议: Apache-2.0
|
|
# | 重要声明: 未经书面许可，不得移除或修改本框架相关版权标识。
|
|
# +----------------------------------------------------------------------
|
|
# | 官方网站: https://www.djangoadmin.cn
|
|
# +----------------------------------------------------------------------
|
|
# | 研发团队: 小蚂蚁云团队 荣誉出品
|
|
# +======================================================================
|
|
# | 版权和免责声明:
|
|
# | 一、知识产权声明
|
|
# | 本团队对该软件框架产品（以下简称“本产品”）拥有完整知识产权，包括但不限于商标权、专利权、
|
|
# | 著作权、商业秘密等。前述权利均受中华人民共和国及相关国际法律法规保护。未经本团队书面授权，
|
|
# | 任何个人或单位不得基于本产品本身申请任何形式的知识产权。
|
|
# | 二、使用限制与禁止条款
|
|
# | 您理解并同意，禁止将本产品用于以下任何行为：
|
|
# | 1. 任何违反中华人民共和国法律法规的用途；
|
|
# | 2. 任何侵害他人合法权益（包括但不限于人身权、财产权、知识产权等）的行为；
|
|
# | 3. 任何危害国家安全、公共利益及社会道德风尚的行为。
|
|
# | 三、责任豁免
|
|
# | 因使用本产品而产生的任何损失或法律纠纷（包括但不限于直接损失、间接损失、附带损失、衍生损失
|
|
# | 以及因违约、侵权等引发的责任），本团队均不承担任何法律责任。若因违规使用本产品导致任何第三
|
|
# | 方提起索赔或行政机关进行调查，您应独立承担全部责任。
|
|
# | 四、违约与法律责任
|
|
# | 如发现任何违反本声明的行为，本团队有权：
|
|
# | 1. 无条件配合公安机关及其他有权机关的调查取证工作；
|
|
# | 2. 依法追究相关方的民事责任，并要求赔偿全部损失；
|
|
# | 3. 保留进一步采取刑事控告等法律措施的权利。
|
|
# | 五、许可范围说明
|
|
# | 1、被授权方应视本产品源代码为核心机密信息，负有最高标准的保密与保管义务。禁止以任何形式（包
|
|
# | 括但不限于直接或间接的分享、复制、传播、分发、转售、出租、开源、反向工程、反编译、反汇编或通
|
|
# | 过网络服务提供给任何第三方）泄露、披露或处置源代码的全部或部分内容。任何未经授权的复制、使用、
|
|
# | 传播行为均构成对本团队知识产权的严重侵权，本团队有权依法追究其全部法律责任，并要求赔偿因此造
|
|
# | 成的一切经济损失及商誉损害。
|
|
# | 2、本产品仅限于公司或个人在法律允许范围内，用于内部且合规的软件产品研发。具体权利义务以《软件
|
|
# | 许可协议》附件为准。
|
|
# +======================================================================
|
|
|
|
"""
|
|
定时任务管理服务模块
|
|
=============================================================
|
|
|
|
该模块实现了定时任务管理相关的所有业务逻辑。
|
|
包括任务的分页查询、详情查询、添加、更新、删除、状态设置等功能。
|
|
|
|
主要功能：
|
|
1. 分页查询：支持按任务名称、状态等条件筛选
|
|
2. 详情查询：根据ID获取任务详细信息
|
|
3. 任务管理：添加、更新、删除（支持批量）
|
|
4. 状态管理：启用、暂停、恢复任务
|
|
5. 立即执行：手动触发任务执行
|
|
6. 调度器同步：自动同步任务状态到调度器
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, Any, Optional, List, Tuple
|
|
from django.core.paginator import Paginator, PageNotAnInteger, InvalidPage, EmptyPage
|
|
from django.db.models import QuerySet
|
|
from django.http import HttpRequest
|
|
|
|
from application.job import forms
|
|
from application.job.models import Job
|
|
from application.job.scheduler import (
|
|
add_job_to_scheduler,
|
|
remove_job_from_scheduler,
|
|
pause_job_in_scheduler,
|
|
resume_job_in_scheduler,
|
|
run_job_once as scheduler_run_job_once,
|
|
reload_scheduler
|
|
)
|
|
from constant.constants import PAGE_SIZE
|
|
from utils import R, regular
|
|
from utils.common import format_datetime, parse_request_body
|
|
from utils.security import get_username
|
|
|
|
# 配置日志器
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# =============================================================
|
|
# 分页查询服务
|
|
# =============================================================
|
|
|
|
def get_job_page(request: HttpRequest) -> R:
|
|
"""
|
|
查询定时任务分页数据
|
|
|
|
支持按任务名称模糊搜索、状态筛选，按创建时间倒序排列。
|
|
|
|
Args:
|
|
request: HttpRequest对象，包含查询参数
|
|
- pageNo: 当前页码（可选，默认为1）
|
|
- pageSize: 每页记录数（可选，默认为PAGE_SIZE）
|
|
- jobName: 任务名称（可选，模糊查询）
|
|
- status: 任务状态（可选，精确匹配）
|
|
|
|
Returns:
|
|
R: 包含分页数据的JSON响应，格式为：
|
|
{
|
|
"records": 任务列表,
|
|
"total": 总记录数,
|
|
"size": 每页大小,
|
|
"current": 当前页码,
|
|
"pages": 总页数
|
|
}
|
|
"""
|
|
try:
|
|
# 获取分页参数
|
|
page_no = int(request.GET.get('pageNo', 1))
|
|
page_size = int(request.GET.get('pageSize', PAGE_SIZE))
|
|
|
|
# 参数校验
|
|
if page_no < 1:
|
|
page_no = 1
|
|
if page_size < 1 or page_size > 100:
|
|
page_size = PAGE_SIZE
|
|
|
|
# 构建基础查询（只查询未删除的记录）
|
|
query = Job.objects.filter(is_delete=0)
|
|
|
|
# 应用筛选条件
|
|
query = _apply_filters(query, request)
|
|
|
|
# 排序（按创建时间倒序）
|
|
query = query.order_by("-create_time")
|
|
|
|
# 分页查询
|
|
paginator = Paginator(query, page_size)
|
|
try:
|
|
page_list = paginator.page(page_no)
|
|
except (PageNotAnInteger, InvalidPage, EmptyPage):
|
|
# 页码异常时返回第一页
|
|
page_list = paginator.page(1)
|
|
page_no = 1
|
|
|
|
# 构建返回数据
|
|
page_data = _build_page_data(page_list, paginator, page_no, page_size)
|
|
|
|
logger.info(f"查询任务分页成功，共{len(page_list)}条记录，总记录数：{paginator.count}")
|
|
return R.ok(data=page_data)
|
|
|
|
except ValueError as e:
|
|
logger.error(f"分页参数格式错误: {str(e)}")
|
|
return R.failed(msg="请求参数格式错误")
|
|
except Exception as e:
|
|
logger.error(f"查询定时任务分页数据异常: {str(e)}", exc_info=True)
|
|
return R.failed(msg="查询失败，请稍后重试")
|
|
|
|
|
|
def _apply_filters(queryset: QuerySet, request: HttpRequest) -> QuerySet:
|
|
"""
|
|
应用查询筛选条件
|
|
|
|
Args:
|
|
queryset: 初始查询集
|
|
request: HttpRequest对象
|
|
|
|
Returns:
|
|
QuerySet: 应用筛选后的查询集
|
|
"""
|
|
# 任务名称模糊筛选（使用icontains实现不区分大小写）
|
|
job_name = request.GET.get('jobName')
|
|
if job_name:
|
|
queryset = queryset.filter(job_name__icontains=job_name)
|
|
|
|
# 任务状态筛选
|
|
status = request.GET.get('status')
|
|
if status and status.isdigit():
|
|
status_val = int(status)
|
|
# 验证状态值是否有效
|
|
if status_val in dict(Job.STATUS_CHOICES):
|
|
queryset = queryset.filter(status=status_val)
|
|
|
|
# 任务分组筛选
|
|
job_group = request.GET.get('jobGroup')
|
|
if job_group:
|
|
queryset = queryset.filter(job_group=job_group)
|
|
|
|
# 是否同步筛选
|
|
is_sync = request.GET.get('isSync')
|
|
if is_sync and is_sync.isdigit():
|
|
queryset = queryset.filter(is_sync=int(is_sync))
|
|
|
|
return queryset
|
|
|
|
|
|
def _build_page_data(page_list, paginator, page_no: int, page_size: int) -> Dict[str, Any]:
|
|
"""
|
|
构建分页返回数据
|
|
|
|
Args:
|
|
page_list: 当前页数据列表
|
|
paginator: 分页器对象
|
|
page_no: 当前页码
|
|
page_size: 每页大小
|
|
|
|
Returns:
|
|
dict: 分页数据，包含records、total、size、current、pages字段
|
|
"""
|
|
records = []
|
|
for item in page_list:
|
|
# 获取执行策略文本
|
|
policy_text = dict(Job.POLICY_CHOICES).get(item.execute_policy, '')
|
|
# 获取状态文本
|
|
status_text = dict(Job.STATUS_CHOICES).get(item.status, '')
|
|
|
|
records.append({
|
|
'id': item.id,
|
|
'jobName': item.job_name,
|
|
'jobAlias': item.job_alias,
|
|
'jobGroup': item.job_group,
|
|
'jobTrigger': item.job_trigger,
|
|
'status': item.status,
|
|
'statusText': status_text,
|
|
'cronExpression': item.cron_expression,
|
|
'executePolicy': item.execute_policy,
|
|
'executePolicyText': policy_text,
|
|
'isSync': item.is_sync,
|
|
'url': item.url,
|
|
'note': item.note,
|
|
'createUser': item.create_user,
|
|
'createTime': format_datetime(item.create_time),
|
|
'updateUser': item.update_user,
|
|
'updateTime': format_datetime(item.update_time),
|
|
})
|
|
|
|
return {
|
|
'records': records,
|
|
'total': paginator.count,
|
|
'size': page_size,
|
|
'current': page_no,
|
|
'pages': paginator.num_pages
|
|
}
|
|
|
|
|
|
# =============================================================
|
|
# 详情查询服务
|
|
# =============================================================
|
|
|
|
def get_job_detail(job_id: int) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
根据ID查询定时任务详情
|
|
|
|
Args:
|
|
job_id: 任务ID
|
|
|
|
Returns:
|
|
dict/None: 任务详情或None（任务不存在时返回None）
|
|
"""
|
|
try:
|
|
# 参数校验
|
|
if not job_id or job_id <= 0:
|
|
logger.warning(f"无效的任务ID: {job_id}")
|
|
return None
|
|
|
|
# 查询任务
|
|
job = Job.objects.filter(is_delete=0, id=job_id).first()
|
|
if not job:
|
|
logger.info(f"任务不存在，ID: {job_id}")
|
|
return None
|
|
|
|
# 构建返回数据
|
|
result = {
|
|
'id': job.id,
|
|
'jobName': job.job_name,
|
|
'jobAlias': job.job_alias,
|
|
'jobGroup': job.job_group,
|
|
'jobTrigger': job.job_trigger,
|
|
'status': job.status,
|
|
'cronExpression': job.cron_expression,
|
|
'executePolicy': job.execute_policy,
|
|
'isSync': job.is_sync,
|
|
'url': job.url,
|
|
'note': job.note,
|
|
}
|
|
|
|
logger.info(f"查询任务详情成功，ID: {job_id}")
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"查询定时任务详情异常，ID: {job_id}, 错误: {str(e)}", exc_info=True)
|
|
return None
|
|
|
|
|
|
# =============================================================
|
|
# 添加任务服务
|
|
# =============================================================
|
|
|
|
def add_job(request: HttpRequest) -> R:
|
|
"""
|
|
添加定时任务
|
|
|
|
创建新的定时任务记录，如果状态为运行中且Cron表达式有效，
|
|
会自动添加到调度器中。
|
|
|
|
Args:
|
|
request: HttpRequest对象，包含任务信息的JSON数据
|
|
|
|
Returns:
|
|
R: 操作结果
|
|
- 成功: R.ok(msg="创建成功")
|
|
- 失败: R.failed(msg="错误信息")
|
|
|
|
请求参数示例:
|
|
{
|
|
"jobName": "数据同步任务", # 任务名称（必填）
|
|
"jobAlias": "同步", # 任务别名（可选）
|
|
"jobGroup": "DEFAULT", # 任务分组（可选，默认DEFAULT）
|
|
"jobTrigger": "sync_user", # 任务触发器（必填）
|
|
"status": 1, # 状态：0-未发布, 1-运行中, 2-暂停, 3-已完成
|
|
"cronExpression": "0 2 * * *", # Cron表达式（必填）
|
|
"executePolicy": 1, # 执行策略：1-GET, 2-POST, 3-PUT, 4-DELETE
|
|
"isSync": 1, # 是否同步：0-否, 1-是
|
|
"url": "/api/job/test/sms", # 执行URL（必填）
|
|
"note": "每天凌晨2点同步用户数据" # 备注（可选）
|
|
}
|
|
"""
|
|
# 解析请求参数
|
|
data, error = parse_request_body(request)
|
|
if error:
|
|
return R.failed(msg=error)
|
|
|
|
# 表单验证
|
|
form = forms.JobForm(data)
|
|
if not form.is_valid():
|
|
error_msg = regular.get_err(form)
|
|
logger.warning(f"添加任务参数验证失败: {error_msg}")
|
|
return R.failed(msg=error_msg)
|
|
|
|
try:
|
|
# 获取表单数据
|
|
cleaned_data = form.cleaned_data
|
|
|
|
# 处理URL：如果是相对路径且没有协议，确保以斜杠开头
|
|
url = cleaned_data.get('url')
|
|
if url and not url.startswith(('http://', 'https://', '/')):
|
|
url = '/' + url.lstrip('/')
|
|
|
|
# 创建任务记录
|
|
job = Job.objects.create(
|
|
job_name=cleaned_data.get('jobName'),
|
|
job_alias=cleaned_data.get('jobAlias'),
|
|
job_group=cleaned_data.get('jobGroup') or 'DEFAULT',
|
|
job_trigger=cleaned_data.get('jobTrigger'),
|
|
status=cleaned_data.get('status'),
|
|
cron_expression=cleaned_data.get('cronExpression'),
|
|
execute_policy=cleaned_data.get('executePolicy'),
|
|
is_sync=cleaned_data.get('isSync'),
|
|
url=url,
|
|
note=cleaned_data.get('note'),
|
|
create_user=get_username(request),
|
|
update_user=get_username(request),
|
|
is_delete=0
|
|
)
|
|
|
|
# 如果状态是运行中且Cron表达式不为空，添加到调度器
|
|
if job.status == 1 and job.cron_expression:
|
|
success = add_job_to_scheduler(job.id)
|
|
if success:
|
|
logger.info(f"任务已添加到调度器，ID: {job.id}")
|
|
else:
|
|
logger.warning(f"任务添加到调度器失败，ID: {job.id}")
|
|
|
|
logger.info(f"添加任务成功，ID: {job.id}, 名称: {job.job_name}")
|
|
return R.ok(msg="创建成功")
|
|
|
|
except Exception as e:
|
|
logger.error(f"添加定时任务异常: {str(e)}", exc_info=True)
|
|
return R.failed(msg="添加失败，请稍后重试")
|
|
|
|
|
|
# =============================================================
|
|
# 更新任务服务
|
|
# =============================================================
|
|
|
|
def update_job(request: HttpRequest) -> R:
|
|
"""
|
|
更新定时任务
|
|
|
|
更新现有任务信息，并根据状态变化自动同步调度器。
|
|
|
|
Args:
|
|
request: HttpRequest对象，包含任务ID和更新信息的JSON数据
|
|
|
|
Returns:
|
|
R: 操作结果
|
|
- 成功: R.ok(msg="更新成功")
|
|
- 失败: R.failed(msg="错误信息")
|
|
|
|
请求参数示例:
|
|
{
|
|
"id": 1, # 任务ID（必填）
|
|
"jobName": "更新后的任务名", # 任务名称
|
|
"jobAlias": "同步", # 任务别名
|
|
"jobGroup": "DEFAULT", # 任务分组
|
|
"jobTrigger": "sync_user", # 任务触发器
|
|
"status": 2, # 状态：暂停
|
|
"cronExpression": "0 3 * * *", # 新的Cron表达式
|
|
"executePolicy": 1, # 执行策略
|
|
"isSync": 1, # 是否同步
|
|
"url": "/api/job/test/sms", # 执行URL
|
|
"note": "更新后的备注" # 备注
|
|
}
|
|
"""
|
|
# 解析请求参数
|
|
data, error = parse_request_body(request)
|
|
if error:
|
|
return R.failed(msg=error)
|
|
|
|
# 验证任务ID
|
|
job_id = data.get('id')
|
|
if not job_id or not str(job_id).isdigit() or int(job_id) <= 0:
|
|
return R.failed("任务ID无效")
|
|
job_id = int(job_id)
|
|
|
|
# 表单验证
|
|
form = forms.JobForm(data)
|
|
if not form.is_valid():
|
|
error_msg = regular.get_err(form)
|
|
logger.warning(f"更新任务参数验证失败，ID: {job_id}, 错误: {error_msg}")
|
|
return R.failed(msg=error_msg)
|
|
|
|
try:
|
|
# 查询任务
|
|
job = Job.objects.filter(id=job_id, is_delete=0).first()
|
|
if not job:
|
|
logger.warning(f"任务不存在，ID: {job_id}")
|
|
return R.failed("任务不存在")
|
|
|
|
# 记录旧状态，用于后续调度器处理
|
|
old_status = job.status
|
|
old_cron = job.cron_expression
|
|
old_url = job.url
|
|
|
|
# 获取表单数据
|
|
cleaned_data = form.cleaned_data
|
|
|
|
# 处理URL
|
|
url = cleaned_data.get('url')
|
|
if url and not url.startswith(('http://', 'https://', '/')):
|
|
url = '/' + url.lstrip('/')
|
|
|
|
# 更新任务信息
|
|
job.job_name = cleaned_data.get('jobName')
|
|
job.job_alias = cleaned_data.get('jobAlias')
|
|
job.job_group = cleaned_data.get('jobGroup')
|
|
job.job_trigger = cleaned_data.get('jobTrigger')
|
|
job.status = cleaned_data.get('status')
|
|
job.cron_expression = cleaned_data.get('cronExpression')
|
|
job.execute_policy = cleaned_data.get('executePolicy')
|
|
job.is_sync = cleaned_data.get('isSync')
|
|
job.url = url
|
|
job.note = cleaned_data.get('note')
|
|
job.update_user = get_username(request)
|
|
job.save()
|
|
|
|
# 处理调度器中的任务
|
|
scheduler_result = _handle_scheduler_after_update(
|
|
job, old_status, old_cron, old_url
|
|
)
|
|
|
|
if scheduler_result:
|
|
logger.info(f"调度器同步成功，ID: {job_id}")
|
|
|
|
logger.info(f"更新任务成功，ID: {job_id}")
|
|
return R.ok(msg="更新成功")
|
|
|
|
except Exception as e:
|
|
logger.error(f"更新定时任务异常，ID: {job_id}, 错误: {str(e)}", exc_info=True)
|
|
return R.failed(msg="更新失败，请稍后重试")
|
|
|
|
|
|
def _handle_scheduler_after_update(job: Job, old_status: int, old_cron: str, old_url: str) -> bool:
|
|
"""
|
|
处理更新后的调度器逻辑
|
|
|
|
根据任务状态和Cron表达式的变化，决定如何同步调度器。
|
|
|
|
Args:
|
|
job: 更新后的任务对象
|
|
old_status: 旧状态
|
|
old_cron: 旧Cron表达式
|
|
old_url: 旧URL
|
|
|
|
Returns:
|
|
bool: 是否成功处理
|
|
"""
|
|
try:
|
|
# 情况1：从非运行中变为运行中，且Cron表达式不为空
|
|
if old_status != 1 and job.status == 1 and job.cron_expression:
|
|
logger.info(f"任务状态从{old_status}变为运行中，添加到调度器")
|
|
return add_job_to_scheduler(job.id)
|
|
|
|
# 情况2：从运行中变为非运行中
|
|
elif old_status == 1 and job.status != 1:
|
|
logger.info(f"任务从运行中变为状态{job.status}，从调度器移除")
|
|
return remove_job_from_scheduler(job.id)
|
|
|
|
# 情况3：保持运行中，但Cron表达式变化
|
|
elif job.status == 1 and old_cron != job.cron_expression:
|
|
logger.info(f"任务运行中，Cron表达式从{old_cron}变为{job.cron_expression}，重新添加到调度器")
|
|
if job.cron_expression:
|
|
return add_job_to_scheduler(job.id)
|
|
else:
|
|
return remove_job_from_scheduler(job.id)
|
|
|
|
# 情况4：保持运行中，但URL变化（不需要重新调度，因为URL只在执行时使用）
|
|
elif job.status == 1 and old_url != job.url:
|
|
logger.info(f"任务运行中，URL从{old_url}变为{job.url}，无需重新调度")
|
|
return True
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"处理调度器更新异常: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
|
|
# =============================================================
|
|
# 删除任务服务（支持批量）
|
|
# =============================================================
|
|
|
|
def delete_jobs(job_ids: str) -> R:
|
|
"""
|
|
删除定时任务（批量删除）
|
|
|
|
支持单个ID或多个ID（用逗号分隔）的批量删除。
|
|
删除前会先从调度器中移除任务。
|
|
|
|
Args:
|
|
job_ids: 任务ID字符串，多个ID用逗号分隔
|
|
例如: "1" 或 "1,2,3"
|
|
|
|
Returns:
|
|
R: 操作结果，包含实际删除的记录数
|
|
- 成功: R.ok(msg="本次共删除N条数据")
|
|
- 失败: R.failed(msg="错误信息")
|
|
"""
|
|
if not job_ids:
|
|
return R.failed("记录ID不存在")
|
|
|
|
try:
|
|
# 解析ID列表
|
|
id_list = []
|
|
for id_str in job_ids.split(','):
|
|
id_str = id_str.strip()
|
|
if id_str and id_str.isdigit():
|
|
id_list.append(int(id_str))
|
|
|
|
if not id_list:
|
|
return R.failed("无效的记录ID")
|
|
|
|
logger.info(f"开始删除任务，IDs: {id_list}")
|
|
|
|
# 先移除调度器中的任务
|
|
removed_count = 0
|
|
for job_id in id_list:
|
|
if remove_job_from_scheduler(job_id):
|
|
removed_count += 1
|
|
|
|
# 批量更新删除标识（软删除）
|
|
updated_count = Job.objects.filter(
|
|
id__in=id_list,
|
|
is_delete=0
|
|
).update(is_delete=1)
|
|
|
|
logger.info(f"删除任务完成，从调度器移除{removed_count}个，数据库删除{updated_count}条")
|
|
return R.ok(msg=f"本次共删除{updated_count}条数据")
|
|
|
|
except ValueError as e:
|
|
logger.error(f"记录ID格式错误: {job_ids}, 错误: {str(e)}")
|
|
return R.failed("记录ID格式错误")
|
|
except Exception as e:
|
|
logger.error(f"删除定时任务异常，IDs: {job_ids}, 错误: {str(e)}", exc_info=True)
|
|
return R.failed(msg="删除失败，请稍后重试")
|
|
|
|
|
|
# =============================================================
|
|
# 状态管理服务
|
|
# =============================================================
|
|
|
|
def status_job(request: HttpRequest) -> R:
|
|
"""
|
|
设置任务状态（启用/暂停）
|
|
|
|
更新任务的运行状态，并同步到调度器。
|
|
|
|
Args:
|
|
request: HttpRequest对象，包含任务ID和目标状态
|
|
|
|
Returns:
|
|
R: 操作结果
|
|
- 成功: R.ok(msg="已启用/已暂停/已恢复")
|
|
- 失败: R.failed(msg="错误信息")
|
|
|
|
请求参数示例:
|
|
{
|
|
"id": 1, # 任务ID（必填）
|
|
"status": 1 # 目标状态：0-未发布, 1-运行中, 2-暂停, 3-已完成
|
|
}
|
|
"""
|
|
# 解析请求参数
|
|
data, error = parse_request_body(request)
|
|
if error:
|
|
return R.failed(msg=error)
|
|
|
|
job_id = data.get('id')
|
|
target_status = data.get('status')
|
|
|
|
if not job_id:
|
|
return R.failed("任务ID不能为空")
|
|
|
|
if target_status is None:
|
|
return R.failed("目标状态不能为空")
|
|
|
|
try:
|
|
# 验证状态值
|
|
target_status = int(target_status)
|
|
if target_status not in [0, 1, 2, 3]:
|
|
return R.failed("无效的状态值")
|
|
|
|
# 查询任务
|
|
job = Job.objects.filter(id=job_id, is_delete=0).first()
|
|
if not job:
|
|
logger.warning(f"任务不存在，ID: {job_id}")
|
|
return R.failed("任务不存在")
|
|
|
|
# 记录旧状态
|
|
old_status = job.status
|
|
|
|
# 如果状态相同，无需处理
|
|
if old_status == target_status:
|
|
return R.ok(msg="状态未变化")
|
|
|
|
# 更新状态
|
|
job.status = target_status
|
|
job.save(update_fields=['status', 'update_time'])
|
|
|
|
# 根据新旧状态处理调度器
|
|
result = _handle_status_change(job, old_status, target_status)
|
|
|
|
status_text = dict(Job.STATUS_CHOICES).get(target_status, '未知')
|
|
logger.info(f"设置任务状态成功，ID: {job_id}, 新状态: {status_text}")
|
|
return R.ok(msg=f"已{status_text}")
|
|
|
|
except ValueError as e:
|
|
logger.error(f"状态值格式错误: {str(e)}")
|
|
return R.failed("状态值格式错误")
|
|
except Exception as e:
|
|
logger.error(f"设置任务状态异常，ID: {job_id}, 错误: {str(e)}", exc_info=True)
|
|
return R.failed(msg="设置失败，请稍后重试")
|
|
|
|
|
|
def _handle_status_change(job: Job, old_status: int, new_status: int) -> bool:
|
|
"""
|
|
处理状态变更时的调度器操作
|
|
|
|
Args:
|
|
job: 任务对象
|
|
old_status: 旧状态
|
|
new_status: 新状态
|
|
|
|
Returns:
|
|
bool: 是否成功处理
|
|
"""
|
|
try:
|
|
# 变为运行中
|
|
if new_status == 1:
|
|
if job.cron_expression:
|
|
logger.info(f"任务启用，添加到调度器，ID: {job.id}")
|
|
return add_job_to_scheduler(job.id)
|
|
else:
|
|
logger.warning(f"任务启用但Cron表达式为空，无法添加到调度器，ID: {job.id}")
|
|
return False
|
|
|
|
# 从运行中变为其他状态
|
|
elif old_status == 1:
|
|
if new_status == 2: # 暂停
|
|
logger.info(f"任务暂停，ID: {job.id}")
|
|
return pause_job_in_scheduler(job.id)
|
|
else: # 未发布、删除等
|
|
logger.info(f"任务停止，从调度器移除，ID: {job.id}")
|
|
return remove_job_from_scheduler(job.id)
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"处理状态变更异常: {str(e)}", exc_info=True)
|
|
return False
|
|
|
|
|
|
# =============================================================
|
|
# 暂停任务服务
|
|
# =============================================================
|
|
|
|
def pause_job(job_id: int) -> R:
|
|
"""
|
|
暂停任务
|
|
|
|
将任务状态设置为暂停，并从调度器中暂停执行。
|
|
|
|
Args:
|
|
job_id: 任务ID
|
|
|
|
Returns:
|
|
R: 操作结果
|
|
- 成功: R.ok(msg="暂停成功")
|
|
- 失败: R.failed(msg="错误信息")
|
|
"""
|
|
try:
|
|
# 参数校验
|
|
if not job_id or job_id <= 0:
|
|
return R.failed("任务ID无效")
|
|
|
|
# 查询任务
|
|
job = Job.objects.filter(id=job_id, is_delete=0).first()
|
|
if not job:
|
|
logger.warning(f"任务不存在，ID: {job_id}")
|
|
return R.failed("任务不存在")
|
|
|
|
# 如果已经是暂停状态
|
|
if job.status == 2:
|
|
return R.ok(msg="任务已处于暂停状态")
|
|
|
|
# 更新状态
|
|
job.status = 2
|
|
job.save(update_fields=['status', 'update_time'])
|
|
|
|
# 暂停调度器中的任务
|
|
if pause_job_in_scheduler(job.id):
|
|
logger.info(f"暂停任务成功，ID: {job_id}")
|
|
return R.ok(msg="暂停成功")
|
|
else:
|
|
# 调度器暂停失败，但数据库已更新
|
|
logger.warning(f"调度器暂停失败，但数据库已更新，ID: {job_id}")
|
|
return R.ok(msg="任务已暂停，但调度器同步失败")
|
|
|
|
except Exception as e:
|
|
logger.error(f"暂停任务异常，ID: {job_id}, 错误: {str(e)}", exc_info=True)
|
|
return R.failed(msg="暂停失败，请稍后重试")
|
|
|
|
|
|
# =============================================================
|
|
# 恢复任务服务
|
|
# =============================================================
|
|
|
|
def resume_job(job_id: int) -> R:
|
|
"""
|
|
恢复任务
|
|
|
|
将任务状态设置为运行中，并添加到调度器执行。
|
|
|
|
Args:
|
|
job_id: 任务ID
|
|
|
|
Returns:
|
|
R: 操作结果
|
|
- 成功: R.ok(msg="恢复成功")
|
|
- 失败: R.failed(msg="错误信息")
|
|
"""
|
|
try:
|
|
# 参数校验
|
|
if not job_id or job_id <= 0:
|
|
return R.failed("任务ID无效")
|
|
|
|
# 查询任务
|
|
job = Job.objects.filter(id=job_id, is_delete=0).first()
|
|
if not job:
|
|
logger.warning(f"任务不存在，ID: {job_id}")
|
|
return R.failed("任务不存在")
|
|
|
|
# 如果已经是运行中状态
|
|
if job.status == 1:
|
|
return R.ok(msg="任务已处于运行中状态")
|
|
|
|
# 检查Cron表达式
|
|
if not job.cron_expression:
|
|
return R.failed("任务Cron表达式为空，无法恢复")
|
|
|
|
# 更新状态
|
|
job.status = 1
|
|
job.save(update_fields=['status', 'update_time'])
|
|
|
|
# 添加到调度器
|
|
success = False
|
|
if job.cron_expression:
|
|
success = add_job_to_scheduler(job.id)
|
|
else:
|
|
success = resume_job_in_scheduler(job.id)
|
|
|
|
if success:
|
|
logger.info(f"恢复任务成功，ID: {job_id}")
|
|
return R.ok(msg="恢复成功")
|
|
else:
|
|
logger.warning(f"调度器恢复失败，但数据库已更新，ID: {job_id}")
|
|
return R.ok(msg="任务已恢复，但调度器同步失败")
|
|
|
|
except Exception as e:
|
|
logger.error(f"恢复任务异常，ID: {job_id}, 错误: {str(e)}", exc_info=True)
|
|
return R.failed(msg="恢复失败，请稍后重试")
|
|
|
|
|
|
# =============================================================
|
|
# 立即执行服务
|
|
# =============================================================
|
|
|
|
def run_job_once_service(job_id: int, params: Optional[Dict[str, Any]] = None) -> R:
|
|
"""
|
|
立即执行一次任务
|
|
|
|
手动触发任务执行，不依赖调度器的定时规则。
|
|
|
|
Args:
|
|
job_id: 任务ID
|
|
params: 执行参数（可选）
|
|
|
|
Returns:
|
|
R: 操作结果，包含执行结果数据
|
|
- 成功: R.ok(data=result, msg="执行成功")
|
|
- 失败: R.failed(msg="错误信息")
|
|
"""
|
|
try:
|
|
# 参数校验
|
|
if not job_id or job_id <= 0:
|
|
return R.failed("任务ID无效")
|
|
|
|
# 查询任务
|
|
job = Job.objects.filter(id=job_id, is_delete=0).first()
|
|
if not job:
|
|
logger.warning(f"任务不存在，ID: {job_id}")
|
|
return R.failed("任务不存在")
|
|
|
|
logger.info(f"开始立即执行任务，ID: {job_id}, 名称: {job.job_name}")
|
|
|
|
# 调用调度器的立即执行函数
|
|
result = scheduler_run_job_once(job.id, params)
|
|
|
|
if result.get('success'):
|
|
logger.info(f"任务立即执行成功，ID: {job_id}, 结果: {result}")
|
|
return R.ok(data=result, msg="执行成功")
|
|
else:
|
|
error_msg = result.get('error', '未知错误')
|
|
logger.error(f"任务立即执行失败，ID: {job_id}, 错误: {error_msg}")
|
|
return R.failed(msg=f"执行失败: {error_msg}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"立即执行任务异常，ID: {job_id}, 错误: {str(e)}", exc_info=True)
|
|
return R.failed(msg="执行失败，请稍后重试")
|
|
|
|
|
|
# =============================================================
|
|
# 调度器管理服务
|
|
# =============================================================
|
|
|
|
def reload_all_jobs() -> R:
|
|
"""
|
|
重新加载所有任务到调度器
|
|
|
|
清除当前调度器中的所有任务，然后从数据库重新加载所有运行中的任务。
|
|
|
|
Returns:
|
|
R: 操作结果，包含加载的任务数量
|
|
- 成功: R.ok(data=count, msg="已重新加载 N 个任务")
|
|
- 失败: R.failed(msg="错误信息")
|
|
"""
|
|
try:
|
|
logger.info("开始重新加载所有任务到调度器")
|
|
|
|
# 调用调度器的重新加载函数
|
|
loaded_count = reload_scheduler()
|
|
|
|
logger.info(f"重新加载任务完成，共加载 {loaded_count} 个任务")
|
|
return R.ok(data=loaded_count, msg=f"已重新加载 {loaded_count} 个任务")
|
|
|
|
except Exception as e:
|
|
logger.error(f"重新加载任务异常: {str(e)}", exc_info=True)
|
|
return R.failed(msg="重新加载失败，请稍后重试")
|
|
|
|
|
|
def get_scheduler_status() -> R:
|
|
"""
|
|
获取调度器当前状态
|
|
|
|
Returns:
|
|
R: 操作结果，包含调度器状态信息
|
|
- 成功: R.ok(data=status_info, msg="获取成功")
|
|
- 失败: R.failed(msg="错误信息")
|
|
"""
|
|
try:
|
|
from application.job.scheduler import scheduler, _print_scheduler_status
|
|
|
|
# 获取调度器中的任务列表
|
|
jobs = scheduler.get_jobs()
|
|
job_list = []
|
|
|
|
for job in jobs:
|
|
next_run = job.next_run_time
|
|
job_list.append({
|
|
'id': job.id,
|
|
'name': job.name,
|
|
'next_run_time': next_run.strftime('%Y-%m-%d %H:%M:%S') if next_run else None,
|
|
'pending': job.pending
|
|
})
|
|
|
|
status_info = {
|
|
'running': scheduler.running,
|
|
'job_count': len(jobs),
|
|
'jobs': job_list
|
|
}
|
|
|
|
return R.ok(data=status_info, msg="获取成功")
|
|
|
|
except Exception as e:
|
|
logger.error(f"获取调度器状态异常: {str(e)}", exc_info=True)
|
|
return R.failed(msg="获取失败")
|