ClaudeandClaude
dbcce79536
重构前保存现有 apps/ 结构基线，供后续各阶段回滚。 Co-Authored-By: Claude <noreply@anthropic.com>
260 lines
7.5 KiB
Python
260 lines
7.5 KiB
Python
import logging
|
||
from datetime import datetime, timezone
|
||
|
||
from sqlalchemy import and_
|
||
|
||
from apps.constants.constants import JOB_STATUS_LIST, JOB_EXECUTE_POLICY_LIST
|
||
from apps.constants.message import PAGE_SIZE
|
||
from apps.forms.job import JobForm, JobStatusForm
|
||
from apps.models.job import Job
|
||
from apps.services.scheduler import (
|
||
add_job_to_scheduler,
|
||
remove_job_from_scheduler,
|
||
pause_job_in_scheduler,
|
||
get_scheduler,
|
||
run_job_once as scheduler_run_job_once,
|
||
reload_scheduler,
|
||
)
|
||
from extensions import get_db_session
|
||
from utils import R
|
||
from utils.security import get_user_id
|
||
from utils.request import like_contains, parse_id_list
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# 查询定时任务分页数据
|
||
|
||
def get_job_page(request):
|
||
db = get_db_session()
|
||
# 页码
|
||
page = int(request.query_params.get("pageNo", 1))
|
||
# 每页数
|
||
limit = int(request.query_params.get("pageSize", PAGE_SIZE))
|
||
# 实例化查询对象
|
||
query = db.query(Job).filter(Job.is_delete == 0)
|
||
# 任务名称
|
||
job_name = request.query_params.get('jobName')
|
||
if job_name:
|
||
query = query.filter(Job.job_name.like(like_contains(job_name), escape='\\'))
|
||
# 任务状态
|
||
status = request.query_params.get('status')
|
||
if status:
|
||
query = query.filter(Job.status == status)
|
||
# 任务分组
|
||
job_group = request.query_params.get('jobGroup')
|
||
if job_group:
|
||
query = query.filter(Job.job_group == job_group)
|
||
# 排序
|
||
query = query.order_by(Job.id.desc())
|
||
# 记录总数
|
||
count = query.count()
|
||
# 分页查询
|
||
job_list = query.limit(limit).offset((page - 1) * limit).all()
|
||
# 若当前页超出末页（数据被并发删除），自动回退到第一页
|
||
if page > 1 and not job_list:
|
||
page = 1
|
||
count = query.count()
|
||
job_list = query.limit(limit).offset(0).all()
|
||
# 实例化结果
|
||
scheduler = get_scheduler()
|
||
result = []
|
||
if len(job_list) > 0:
|
||
for item in job_list:
|
||
data = item.to_dict()
|
||
data['status_name'] = JOB_STATUS_LIST.get(item.status)
|
||
data['execute_policy_name'] = JOB_EXECUTE_POLICY_LIST.get(item.execute_policy)
|
||
data['next_run_time'] = _get_next_run_time(scheduler, item.id)
|
||
result.append(data)
|
||
return R.ok(data=result, count=count)
|
||
|
||
|
||
def _get_next_run_time(scheduler, job_id):
|
||
"""从调度器获取任务下次执行时间"""
|
||
try:
|
||
scheduler_job = scheduler.get_job(f"job_{job_id}")
|
||
if scheduler_job and scheduler_job.next_run_time:
|
||
return scheduler_job.next_run_time.strftime('%Y-%m-%d %H:%M:%S')
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
# 根据ID查询任务详情
|
||
|
||
def get_job_detail(job_id):
|
||
db = get_db_session()
|
||
job = db.query(Job).filter(and_(Job.id == job_id, Job.is_delete == 0)).first()
|
||
if not job:
|
||
return None
|
||
data = job.to_dict()
|
||
data['status_name'] = JOB_STATUS_LIST.get(job.status)
|
||
data['execute_policy_name'] = JOB_EXECUTE_POLICY_LIST.get(job.execute_policy)
|
||
return data
|
||
|
||
|
||
# 添加定时任务
|
||
|
||
def add_job(request, data: JobForm):
|
||
db = get_db_session()
|
||
# 表单数据赋值给对象
|
||
job = Job(**data.model_dump(exclude={'id'}))
|
||
job.create_user = get_user_id(request)
|
||
# 插入数据
|
||
job.save()
|
||
|
||
# 状态为运行中且有Cron表达式，添加到调度器
|
||
if job.status == 1 and job.cron_expression:
|
||
add_job_to_scheduler(job.to_dict())
|
||
|
||
return R.ok(msg="添加成功")
|
||
|
||
|
||
# 更新定时任务
|
||
|
||
def update_job(request, data: JobForm):
|
||
db = get_db_session()
|
||
# 记录ID判空
|
||
if not data.id or data.id <= 0:
|
||
return R.failed("记录ID不能为空")
|
||
|
||
# 根据ID查询记录
|
||
job = db.query(Job).filter(and_(Job.id == data.id, Job.is_delete == 0)).first()
|
||
if not job:
|
||
return R.failed("任务不存在")
|
||
|
||
old_status = job.status
|
||
old_cron = job.cron_expression
|
||
|
||
# 构造更新数据（排除 id）
|
||
updData = data.model_dump(exclude={'id'})
|
||
updData['update_user'] = get_user_id(request)
|
||
updData['update_time'] = datetime.now(tz=timezone.utc)
|
||
result = db.query(Job).filter_by(id=data.id).update(updData)
|
||
if not result:
|
||
return R.failed("更新失败")
|
||
|
||
# 处理调度器更新
|
||
_handle_scheduler_update(data.model_dump(), old_status, old_cron)
|
||
|
||
return R.ok(msg="更新成功")
|
||
|
||
|
||
def _handle_scheduler_update(job_data, old_status, old_cron):
|
||
"""处理调度器更新（job_data 为完整任务数据，含 execute_policy/params）"""
|
||
job_id = job_data.get('id')
|
||
new_status = job_data.get('status')
|
||
new_cron = job_data.get('cron_expression')
|
||
if old_status != 1 and new_status == 1 and new_cron:
|
||
add_job_to_scheduler(job_data)
|
||
elif old_status == 1 and new_status != 1:
|
||
remove_job_from_scheduler(job_id)
|
||
elif new_status == 1 and old_cron != new_cron:
|
||
if new_cron:
|
||
add_job_to_scheduler(job_data)
|
||
else:
|
||
remove_job_from_scheduler(job_id)
|
||
|
||
|
||
# 删除定时任务
|
||
|
||
def delete_job(job_id):
|
||
# 从调度器移除
|
||
id_list = parse_id_list(job_id)
|
||
for jid in id_list:
|
||
remove_job_from_scheduler(jid)
|
||
return Job.batch_delete_r(job_id)
|
||
|
||
|
||
# 设置任务状态
|
||
|
||
def update_job_status(request, data: JobStatusForm):
|
||
db = get_db_session()
|
||
# 根据ID查询记录
|
||
job = db.query(Job).filter(and_(Job.id == data.id, Job.is_delete == 0)).first()
|
||
if not job:
|
||
return R.failed("任务不存在")
|
||
|
||
old_status = job.status
|
||
|
||
# 更新记录
|
||
result = db.query(Job).filter_by(id=data.id).update({
|
||
"status": data.status,
|
||
"update_user": get_user_id(request),
|
||
"update_time": datetime.now(tz=timezone.utc)
|
||
})
|
||
if not result:
|
||
return R.failed("设置失败")
|
||
|
||
# 处理调度器
|
||
if data.status == 1 and job.cron_expression:
|
||
job_dict = job.to_dict()
|
||
job_dict['status'] = data.status
|
||
add_job_to_scheduler(job_dict)
|
||
elif data.status == 2:
|
||
pause_job_in_scheduler(data.id)
|
||
elif old_status == 1 and data.status != 1:
|
||
remove_job_from_scheduler(data.id)
|
||
|
||
return R.ok(msg="设置成功")
|
||
|
||
|
||
# 暂停任务
|
||
|
||
def pause_job_by_id(job_id):
|
||
db = get_db_session()
|
||
job = db.query(Job).filter(and_(Job.id == job_id, Job.is_delete == 0)).first()
|
||
if not job:
|
||
return R.failed("任务不存在")
|
||
|
||
if job.status == 2:
|
||
return R.ok(msg="任务已处于暂停状态")
|
||
|
||
job.status = 2
|
||
job.update_user = job.create_user
|
||
job.update_time = datetime.now(tz=timezone.utc)
|
||
db.commit()
|
||
pause_job_in_scheduler(job_id)
|
||
|
||
return R.ok(msg="暂停成功")
|
||
|
||
|
||
# 恢复任务
|
||
|
||
def resume_job_by_id(job_id):
|
||
db = get_db_session()
|
||
job = db.query(Job).filter(and_(Job.id == job_id, Job.is_delete == 0)).first()
|
||
if not job:
|
||
return R.failed("任务不存在")
|
||
|
||
if job.status == 1:
|
||
return R.ok(msg="任务已处于运行中状态")
|
||
|
||
if not job.cron_expression:
|
||
return R.failed("Cron表达式为空，无法恢复")
|
||
|
||
job.status = 1
|
||
job.update_user = job.create_user
|
||
job.update_time = datetime.now(tz=timezone.utc)
|
||
db.commit()
|
||
add_job_to_scheduler(job.to_dict())
|
||
|
||
return R.ok(msg="恢复成功")
|
||
|
||
|
||
# 立即执行一次
|
||
|
||
def run_job_once_api(job_id):
|
||
result = scheduler_run_job_once(job_id)
|
||
if result.get('success'):
|
||
return R.ok(data=result, msg="执行成功")
|
||
else:
|
||
return R.failed(msg=result.get('error', '执行失败'))
|
||
|
||
|
||
# 重新加载调度器
|
||
|
||
def reload_all_jobs():
|
||
count = reload_scheduler()
|
||
return R.ok(data=count, msg=f"已重新加载 {count} 个任务")
|