feat(phase4): 全部业务模块迁移到 modules/（结构迁移）
- 21 个模块目录（auth/system/dictionary/configuration/cms/member/region/link/notice/job/log/upload/index） - 每个模块 router.py（合并模块导出多个 router）/schemas.py/service.py/repository.py/models.py - routers/v1.py 与 routers/__init__.py 改为直接引用 modules，前缀与权限串原样保留 - core/middleware.py 懒加载指向 modules.system.menu / modules.log；core/redis_client.py 指向 modules.job.scheduler - apps/views|services|forms 全部改为 re-export 垫片 - 迁移后 grep 守门通过：modules/core 无 apps/extensions 残留引用 - 实测登录、27+ 模块接口、写操作 commit、操作日志落库均正常 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Claude
@@ -0,0 +1,57 @@
|
||||
"""
|
||||
任务执行器模块
|
||||
|
||||
根据触发器标识执行对应任务的功能。
|
||||
"""
|
||||
|
||||
import logging
|
||||
from typing import Any, Dict
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class JobExecutor:
|
||||
"""任务执行器：根据触发器标识调用对应的内部方法执行任务"""
|
||||
|
||||
def __init__(self):
|
||||
self._handlers = {}
|
||||
self._register_default_handlers()
|
||||
|
||||
def _register_default_handlers(self):
|
||||
"""注册默认的任务处理器"""
|
||||
self.register('send_sms', self._send_sms)
|
||||
self.register('send_email', self._send_email)
|
||||
self.register('sync_data', self._sync_data)
|
||||
|
||||
def register(self, trigger: str, handler):
|
||||
"""注册任务处理器"""
|
||||
self._handlers[trigger] = handler
|
||||
logger.info(f"注册任务处理器: {trigger}")
|
||||
|
||||
def execute(self, trigger: str, **kwargs) -> Any:
|
||||
"""执行任务"""
|
||||
if trigger not in self._handlers:
|
||||
raise KeyError(f"未注册的任务触发器: {trigger}（可用 job_executor.register() 注册处理器）")
|
||||
|
||||
handler = self._handlers[trigger]
|
||||
logger.info(f"执行任务: {trigger}, 参数: {kwargs}")
|
||||
return handler(**kwargs)
|
||||
|
||||
def _send_sms(self, **kwargs) -> Dict:
|
||||
"""发送短信任务"""
|
||||
logger.info(f"发送短信 [任务ID: {kwargs.get('job_id')}]")
|
||||
return {'success': True, 'message': '短信发送成功'}
|
||||
|
||||
def _send_email(self, **kwargs) -> Dict:
|
||||
"""发送邮件任务"""
|
||||
logger.info(f"发送邮件 [任务ID: {kwargs.get('job_id')}]")
|
||||
return {'success': True, 'message': '邮件发送成功'}
|
||||
|
||||
def _sync_data(self, **kwargs) -> Dict:
|
||||
"""数据同步任务"""
|
||||
logger.info(f"数据同步 [任务ID: {kwargs.get('job_id')}]")
|
||||
return {'success': True, 'message': '数据同步成功'}
|
||||
|
||||
|
||||
# 全局执行器实例
|
||||
job_executor = JobExecutor()
|
||||
@@ -0,0 +1,14 @@
|
||||
# +========== 迁移新增：本模块 Repository 层
|
||||
"""modules/job 数据访问层。"""
|
||||
from modules.job.models import Job, JobLog
|
||||
from shared.base_repository import BaseRepository
|
||||
|
||||
class JobRepository(BaseRepository[Job]):
|
||||
pass
|
||||
|
||||
job_repo = JobRepository(Job)
|
||||
|
||||
class JobLogRepository(BaseRepository[JobLog]):
|
||||
pass
|
||||
|
||||
joblog_repo = JobLogRepository(JobLog)
|
||||
@@ -0,0 +1,156 @@
|
||||
from fastapi import APIRouter, Request
|
||||
|
||||
from modules.job.schemas import JobForm, JobStatusForm
|
||||
from core.middleware import check_demo
|
||||
from core.middleware import permission_required
|
||||
from modules.job import service as job
|
||||
|
||||
from core import R
|
||||
|
||||
# 创建路由
|
||||
job_router = APIRouter()
|
||||
|
||||
|
||||
# 查询分页数据
|
||||
@job_router.get('/page', summary='查询分页数据')
|
||||
@permission_required("sys:job:page")
|
||||
def page(request: Request):
|
||||
result = job.get_job_page(request)
|
||||
return result
|
||||
|
||||
|
||||
# 查询任务详情
|
||||
@job_router.get('/detail/{id}', summary='查询任务详情')
|
||||
@permission_required("sys:job:detail")
|
||||
def detail(request: Request, id):
|
||||
data = job.get_job_detail(id)
|
||||
return R.ok(data=data)
|
||||
|
||||
|
||||
# 添加定时任务
|
||||
@job_router.post('/add', summary='添加定时任务')
|
||||
@permission_required("sys:job:add")
|
||||
@check_demo
|
||||
def add(request: Request, data: JobForm):
|
||||
result = job.add_job(request, data)
|
||||
return result
|
||||
|
||||
|
||||
# 更新定时任务
|
||||
@job_router.put('/update', summary='更新定时任务')
|
||||
@permission_required("sys:job:update")
|
||||
@check_demo
|
||||
def update(request: Request, data: JobForm):
|
||||
result = job.update_job(request, data)
|
||||
return result
|
||||
|
||||
|
||||
# 删除定时任务
|
||||
@job_router.delete('/delete/{id}', summary='删除定时任务')
|
||||
@permission_required("sys:job:delete")
|
||||
@check_demo
|
||||
def delete(request: Request, id):
|
||||
result = job.delete_job(id)
|
||||
return result
|
||||
|
||||
|
||||
# 设置任务状态
|
||||
@job_router.put('/status', summary='设置任务状态')
|
||||
@permission_required("sys:job:status")
|
||||
@check_demo
|
||||
def status(request: Request, data: JobStatusForm):
|
||||
result = job.update_job_status(request, data)
|
||||
return result
|
||||
|
||||
|
||||
# 暂停任务
|
||||
@job_router.put('/pause/{id}', summary='暂停任务')
|
||||
@permission_required("sys:job:pause")
|
||||
@check_demo
|
||||
def pause(request: Request, id):
|
||||
result = job.pause_job_by_id(id)
|
||||
return result
|
||||
|
||||
|
||||
# 恢复任务
|
||||
@job_router.put('/resume/{id}', summary='恢复任务')
|
||||
@permission_required("sys:job:resume")
|
||||
@check_demo
|
||||
def resume(request: Request, id):
|
||||
result = job.resume_job_by_id(id)
|
||||
return result
|
||||
|
||||
|
||||
# 立即执行一次
|
||||
@job_router.post('/runOnce/{id}', summary='立即执行一次')
|
||||
@permission_required("sys:job:run_once")
|
||||
@check_demo
|
||||
def run_once(request: Request, id):
|
||||
result = job.run_job_once_api(id)
|
||||
return result
|
||||
|
||||
|
||||
# 重新加载调度器
|
||||
@job_router.post('/reload', summary='重新加载调度器')
|
||||
@permission_required("sys:job:reload")
|
||||
@check_demo
|
||||
def reload(request: Request):
|
||||
result = job.reload_all_jobs()
|
||||
return result
|
||||
|
||||
|
||||
import asyncio
|
||||
|
||||
from fastapi import APIRouter, Request
|
||||
|
||||
from core.middleware import check_demo
|
||||
from core.middleware import permission_required
|
||||
from modules.job import service as job_log
|
||||
|
||||
from core import R
|
||||
from utils.request import parse_batch_ids
|
||||
|
||||
# 创建路由
|
||||
job_log_router = APIRouter()
|
||||
|
||||
|
||||
# 查询分页数据
|
||||
@job_log_router.get('/page', summary='查询分页数据')
|
||||
@permission_required("sys:job_log:page")
|
||||
def page(request: Request):
|
||||
result = job_log.get_job_log_page(request)
|
||||
return result
|
||||
|
||||
|
||||
# 查询日志详情
|
||||
@job_log_router.get('/detail/{id}', summary='查询日志详情')
|
||||
@permission_required("sys:job_log:detail")
|
||||
def detail(request: Request, id):
|
||||
data = job_log.get_job_log_detail(id)
|
||||
return R.ok(data=data)
|
||||
|
||||
|
||||
# 删除任务日志
|
||||
@job_log_router.delete('/delete/{id}', summary='删除任务日志')
|
||||
@permission_required("sys:job_log:delete")
|
||||
@check_demo
|
||||
def delete(request: Request, id):
|
||||
result = job_log.delete_job_log(id)
|
||||
return result
|
||||
|
||||
|
||||
# 批量删除任务日志
|
||||
@job_log_router.delete('/batchDelete', summary='批量删除任务日志')
|
||||
@permission_required("sys:job_log:batch_delete")
|
||||
@check_demo
|
||||
async def batch_delete(request: Request):
|
||||
"""
|
||||
批量删除任务日志
|
||||
|
||||
Request Body:
|
||||
[1, 2, 3] # 日志ID数组
|
||||
"""
|
||||
id_str, err = await parse_batch_ids(request)
|
||||
if err:
|
||||
return err
|
||||
return await asyncio.to_thread(job_log.delete_job_log, id_str)
|
||||
@@ -0,0 +1,368 @@
|
||||
"""
|
||||
定时任务调度核心模块
|
||||
|
||||
使用 APScheduler 管理和执行定时任务。
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
from datetime import datetime
|
||||
from typing import Dict, Optional
|
||||
|
||||
import requests
|
||||
from apscheduler.schedulers.background import BackgroundScheduler
|
||||
from apscheduler.triggers.cron import CronTrigger
|
||||
from apscheduler.jobstores.base import JobLookupError
|
||||
|
||||
from config import FASTAPI_HOST, FASTAPI_PORT
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# 调度器实例
|
||||
_scheduler: Optional[BackgroundScheduler] = None
|
||||
|
||||
|
||||
def get_scheduler() -> BackgroundScheduler:
|
||||
"""获取调度器单例"""
|
||||
global _scheduler
|
||||
if _scheduler is None:
|
||||
_scheduler = BackgroundScheduler(timezone='Asia/Shanghai')
|
||||
return _scheduler
|
||||
|
||||
|
||||
def build_full_url(url: str) -> str:
|
||||
"""构建完整URL，将相对路径拼接为带域名的URL"""
|
||||
if not url:
|
||||
return url
|
||||
if url.startswith(('http://', 'https://')):
|
||||
return url
|
||||
if url.startswith('/'):
|
||||
url = f"http://{FASTAPI_HOST}:{FASTAPI_PORT}{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] = None
|
||||
) -> Dict:
|
||||
"""执行任务（调度器调用入口），执行完成后写入任务日志"""
|
||||
start_time = datetime.now()
|
||||
result = {'success': False, 'error': '未知错误'}
|
||||
try:
|
||||
logger.info(f"执行定时任务 [ID: {job_id}, URL: {url}, Trigger: {job_trigger}]")
|
||||
|
||||
if url:
|
||||
result = _execute_by_url(job_id, url, is_sync, params)
|
||||
elif job_trigger:
|
||||
result = _execute_by_trigger(job_trigger, job_id, params)
|
||||
else:
|
||||
result = {'success': False, 'error': '任务既没有URL也没有触发器'}
|
||||
|
||||
return result
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"执行任务异常 [ID: {job_id}]: {str(e)}")
|
||||
result = {'success': False, 'error': str(e)}
|
||||
return result
|
||||
|
||||
finally:
|
||||
_record_execution_log(job_id, result, start_time)
|
||||
|
||||
|
||||
def _record_execution_log(job_id: int, result: Dict, start_time: datetime):
|
||||
"""记录一次任务执行日志"""
|
||||
try:
|
||||
from modules.job.service import record_job_log
|
||||
|
||||
end_time = datetime.now()
|
||||
success = bool(result.get('success'))
|
||||
status = 0 if success else 1
|
||||
if not success:
|
||||
job_message = '任务执行失败'
|
||||
exception_info = str(result.get('error'))
|
||||
elif result.get('status') == 'triggered':
|
||||
job_message = '异步任务已触发'
|
||||
exception_info = None
|
||||
else:
|
||||
job_message = '任务执行成功'
|
||||
exception_info = None
|
||||
|
||||
record_job_log(job_id, start_time, end_time, status,
|
||||
exception_info=exception_info, job_message=job_message)
|
||||
except Exception as e:
|
||||
logger.error(f"记录任务日志失败 [ID: {job_id}]: {str(e)}")
|
||||
|
||||
|
||||
def _execute_by_url(job_id: int, url: str, is_sync: int, params: Optional[Dict] = None) -> 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请求"""
|
||||
try:
|
||||
logger.info(f"发送同步请求 [ID: {job_id}, URL: {url}]")
|
||||
headers = {'Content-Type': 'application/json', 'User-Agent': 'Scheduler/1.0'}
|
||||
response = requests.post(url, json=data, timeout=30, headers=headers)
|
||||
response.raise_for_status()
|
||||
|
||||
try:
|
||||
result_data = response.json()
|
||||
except Exception:
|
||||
result_data = {'text': response.text[:200]}
|
||||
|
||||
return {'success': True, 'status_code': response.status_code, 'data': result_data}
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
return {'success': False, 'error': str(e)}
|
||||
|
||||
|
||||
def _send_async_request(url: str, data: Dict, job_id: int) -> Dict:
|
||||
"""发送异步HTTP请求（独立线程）"""
|
||||
|
||||
def async_request():
|
||||
try:
|
||||
headers = {'Content-Type': 'application/json'}
|
||||
resp = requests.post(url, json=data, timeout=30, headers=headers)
|
||||
logger.info(f"异步任务完成 [ID: {job_id}, 状态码: {resp.status_code}]")
|
||||
except Exception as e:
|
||||
logger.error(f"异步任务异常 [ID: {job_id}]: {str(e)}")
|
||||
|
||||
thread = threading.Thread(target=async_request)
|
||||
thread.start()
|
||||
return {'success': True, 'status': 'triggered'}
|
||||
|
||||
|
||||
def _execute_by_trigger(trigger: str, job_id: int, params: Optional[Dict] = None) -> Dict:
|
||||
"""通过触发器方式执行任务"""
|
||||
try:
|
||||
from modules.job.job_executor import job_executor
|
||||
# 剔除 job_id，避免与 execute(job_id=...) 关键字参数冲突
|
||||
trigger_params = dict(params or {})
|
||||
trigger_params.pop('job_id', None)
|
||||
result = job_executor.execute(trigger, job_id=job_id, **trigger_params)
|
||||
# 处理器返回 success=False 时透传失败，而非一律视为成功
|
||||
if isinstance(result, dict) and result.get('success') is False:
|
||||
error = result.get('message') or result.get('error') or '任务执行失败'
|
||||
return {'success': False, 'trigger': trigger, 'error': error}
|
||||
return {'success': True, 'trigger': trigger, 'data': result}
|
||||
except Exception as e:
|
||||
return {'success': False, 'error': str(e)}
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 调度器任务管理函数
|
||||
# ============================================================
|
||||
|
||||
def _parse_job_params(params: Optional[str]) -> Dict:
|
||||
"""解析任务参数（JSON字符串 -> dict），容错非法输入"""
|
||||
if not params:
|
||||
return {}
|
||||
try:
|
||||
parsed = json.loads(params)
|
||||
if isinstance(parsed, dict):
|
||||
return parsed
|
||||
except Exception:
|
||||
logger.warning(f"任务参数不是合法JSON，已忽略: {str(params)[:200]}")
|
||||
return {}
|
||||
|
||||
|
||||
def _build_misfire_config(execute_policy) -> Dict:
|
||||
"""根据执行策略映射 APScheduler 的 misfire/coalesce 参数"""
|
||||
if execute_policy == 3:
|
||||
# 放弃执行：错过的运行不补
|
||||
return {'misfire_grace_time': 0, 'coalesce': True}
|
||||
if execute_policy == 2:
|
||||
# 执行一次：短时间内错过的补跑一次
|
||||
return {'misfire_grace_time': 600, 'coalesce': True}
|
||||
# 立即执行（默认）：错过立即补跑一次
|
||||
return {'misfire_grace_time': 3600, 'coalesce': True}
|
||||
|
||||
|
||||
def add_job_to_scheduler(job_dict: dict) -> bool:
|
||||
"""添加任务到调度器"""
|
||||
try:
|
||||
job_id = job_dict.get('id')
|
||||
job_status = job_dict.get('status')
|
||||
job_cron = job_dict.get('cron_expression')
|
||||
job_url = job_dict.get('url')
|
||||
job_sync = job_dict.get('is_sync', 0)
|
||||
job_trigger = job_dict.get('job_trigger')
|
||||
job_name = job_dict.get('job_name', '')
|
||||
job_params = _parse_job_params(job_dict.get('params'))
|
||||
|
||||
if job_status != 1:
|
||||
return False
|
||||
|
||||
if not job_cron:
|
||||
logger.warning(f"跳过添加任务: Cron表达式为空 (id={job_id})")
|
||||
return False
|
||||
|
||||
# 验证Cron表达式
|
||||
try:
|
||||
trigger = CronTrigger.from_crontab(job_cron)
|
||||
next_run = trigger.get_next_fire_time(None, datetime.now())
|
||||
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
|
||||
|
||||
scheduler = get_scheduler()
|
||||
scheduler_job_id = f"job_{job_id}"
|
||||
|
||||
# 移除已存在的同名任务
|
||||
try:
|
||||
scheduler.remove_job(scheduler_job_id)
|
||||
except JobLookupError:
|
||||
pass
|
||||
|
||||
misfire_config = _build_misfire_config(job_dict.get('execute_policy', 1))
|
||||
|
||||
scheduler.add_job(
|
||||
func=execute_job,
|
||||
trigger=trigger,
|
||||
args=[job_id, job_url, job_sync, job_trigger, job_params],
|
||||
id=scheduler_job_id,
|
||||
name=job_name,
|
||||
replace_existing=True,
|
||||
**misfire_config
|
||||
)
|
||||
|
||||
logger.info(f"任务已添加到调度器 [ID: {job_id}, cron: {job_cron}]")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"添加任务到调度器异常: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
def remove_job_from_scheduler(job_id: int) -> bool:
|
||||
"""从调度器移除任务"""
|
||||
try:
|
||||
get_scheduler().remove_job(f"job_{job_id}")
|
||||
return True
|
||||
except JobLookupError:
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"移除任务异常: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
def pause_job_in_scheduler(job_id: int) -> bool:
|
||||
"""暂停调度器中的任务"""
|
||||
try:
|
||||
get_scheduler().pause_job(f"job_{job_id}")
|
||||
return True
|
||||
except JobLookupError:
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"暂停任务异常: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
def resume_job_in_scheduler(job_id: int) -> bool:
|
||||
"""恢复调度器中的任务"""
|
||||
try:
|
||||
get_scheduler().resume_job(f"job_{job_id}")
|
||||
return True
|
||||
except JobLookupError:
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"恢复任务异常: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
def run_job_once(job_id: int, params: Optional[Dict] = None) -> Dict:
|
||||
"""立即执行一次任务（不依赖调度器）"""
|
||||
try:
|
||||
from modules.job.models import Job
|
||||
from core.database import get_db_session
|
||||
|
||||
db = get_db_session()
|
||||
job = db.query(Job).filter(Job.id == job_id, Job.is_delete == 0).first()
|
||||
if not job:
|
||||
return {'success': False, 'error': '任务不存在'}
|
||||
|
||||
return execute_job(job.id, job.url, job.is_sync, job.job_trigger, params or {})
|
||||
|
||||
except Exception as e:
|
||||
return {'success': False, 'error': str(e)}
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 调度器初始化
|
||||
# ============================================================
|
||||
|
||||
def init_scheduler() -> int:
|
||||
"""初始化调度器，启动并从数据库加载所有启用的任务"""
|
||||
try:
|
||||
from modules.job.models import Job
|
||||
from core.database import SessionLocal
|
||||
|
||||
scheduler = get_scheduler()
|
||||
if not scheduler.running:
|
||||
scheduler.start()
|
||||
logger.info("调度器已启动")
|
||||
|
||||
# lifespan 启动阶段无请求上下文，需用独立会话而非请求级 get_db_session()
|
||||
session = SessionLocal()
|
||||
try:
|
||||
jobs = session.query(Job).filter(Job.status == 1, Job.is_delete == 0).all()
|
||||
# 转为字典避免 Session 分离问题
|
||||
job_dicts = [job.to_dict() for job in jobs]
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
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)} 个任务")
|
||||
return success_count
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"初始化调度器异常: {str(e)}")
|
||||
return 0
|
||||
|
||||
|
||||
def shutdown_scheduler():
|
||||
"""关闭调度器"""
|
||||
try:
|
||||
scheduler = get_scheduler()
|
||||
if scheduler.running:
|
||||
scheduler.shutdown(wait=False)
|
||||
logger.info("调度器已关闭")
|
||||
except Exception as e:
|
||||
logger.error(f"关闭调度器异常: {str(e)}")
|
||||
|
||||
|
||||
def reload_scheduler() -> int:
|
||||
"""重新加载调度器：清空所有任务，重新从数据库加载"""
|
||||
try:
|
||||
scheduler = get_scheduler()
|
||||
scheduler.remove_all_jobs()
|
||||
return init_scheduler()
|
||||
except Exception as e:
|
||||
logger.error(f"重新加载调度器异常: {str(e)}")
|
||||
return 0
|
||||
@@ -0,0 +1,32 @@
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel, Field, field_validator
|
||||
|
||||
|
||||
# 定时任务表单验证
|
||||
class JobForm(BaseModel):
|
||||
id: Optional[int] = Field(None, description="任务ID")
|
||||
job_name: str = Field(..., min_length=1, max_length=255, description="任务名称")
|
||||
job_alias: Optional[str] = Field(None, max_length=255, description="任务别名")
|
||||
job_group: Optional[str] = Field(None, max_length=255, description="任务分组")
|
||||
job_trigger: Optional[str] = Field(None, max_length=255, description="任务触发器")
|
||||
status: int = Field(..., ge=0, le=3, description="任务状态")
|
||||
cron_expression: Optional[str] = Field(None, max_length=255, description="Cron表达式")
|
||||
execute_policy: int = Field(..., ge=1, le=3, description="执行策略")
|
||||
is_sync: int = Field(..., ge=0, le=1, description="是否同步任务")
|
||||
url: Optional[str] = Field(None, max_length=255, description="任务URL")
|
||||
params: Optional[str] = Field(None, max_length=2000, description="任务参数")
|
||||
note: Optional[str] = Field(None, max_length=255, description="任务备注")
|
||||
|
||||
@field_validator('cron_expression')
|
||||
@classmethod
|
||||
def validate_cron(cls, v):
|
||||
if v:
|
||||
v = v.strip()
|
||||
return v
|
||||
|
||||
|
||||
# 任务状态表单
|
||||
class JobStatusForm(BaseModel):
|
||||
id: int = Field(..., gt=0, description="任务ID")
|
||||
status: int = Field(..., ge=0, le=3, description="任务状态")
|
||||
@@ -0,0 +1,382 @@
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from sqlalchemy import and_
|
||||
|
||||
from shared.enums import JOB_STATUS_LIST, JOB_EXECUTE_POLICY_LIST
|
||||
from shared.constants import PAGE_SIZE
|
||||
from modules.job.schemas import JobForm, JobStatusForm
|
||||
from modules.job.models import Job
|
||||
from modules.job.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 core.database import get_db_session
|
||||
from core import R
|
||||
from core.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} 个任务")
|
||||
|
||||
|
||||
import logging
|
||||
|
||||
from sqlalchemy import and_
|
||||
|
||||
from shared.enums import JOB_LOG_STATUS_LIST
|
||||
from shared.constants import PAGE_SIZE
|
||||
from modules.job.models import Job
|
||||
from modules.job.models import JobLog
|
||||
from core.database import get_db_session
|
||||
from core.database import SessionLocal
|
||||
from core import R
|
||||
from utils.request import like_contains, parse_id_list
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 日志写入（供调度器调用，可在请求上下文外执行）
|
||||
# ============================================================
|
||||
|
||||
def record_job_log(job_id, start_time, end_time, status, exception_info=None, job_message=None):
|
||||
"""
|
||||
记录任务执行日志
|
||||
|
||||
使用独立会话写入，调度器线程等无请求上下文场景也可调用。
|
||||
"""
|
||||
session = SessionLocal()
|
||||
try:
|
||||
job = session.query(Job).filter(Job.id == job_id, Job.is_delete == 0).first()
|
||||
consume_time = 0
|
||||
if end_time and start_time:
|
||||
consume_time = int((end_time - start_time).total_seconds() * 1000)
|
||||
log = JobLog(
|
||||
job_id=job_id,
|
||||
job_name=job.job_name if job else '',
|
||||
job_group=job.job_group if job else None,
|
||||
job_trigger=job.job_trigger if job else None,
|
||||
cron_expression=job.cron_expression if job else None,
|
||||
job_message=job_message,
|
||||
status=status,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
consume_time=consume_time,
|
||||
exception_info=exception_info,
|
||||
)
|
||||
session.add(log)
|
||||
session.commit()
|
||||
return log
|
||||
except Exception as e:
|
||||
session.rollback()
|
||||
logger.error(f"记录任务执行日志异常: {str(e)}")
|
||||
return None
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 日志查询/删除
|
||||
# ============================================================
|
||||
|
||||
def get_job_log_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(JobLog).filter(JobLog.del_flag == 0)
|
||||
# 任务名称
|
||||
job_name = request.query_params.get('jobName')
|
||||
if job_name:
|
||||
query = query.filter(JobLog.job_name.like(like_contains(job_name), escape='\\'))
|
||||
# 任务分组
|
||||
job_group = request.query_params.get('jobGroup')
|
||||
if job_group:
|
||||
query = query.filter(JobLog.job_group == job_group)
|
||||
# 执行状态
|
||||
status = request.query_params.get('status')
|
||||
if status:
|
||||
query = query.filter(JobLog.status == status)
|
||||
# 排序
|
||||
query = query.order_by(JobLog.id.desc())
|
||||
# 记录总数
|
||||
count = query.count()
|
||||
# 分页查询
|
||||
log_list = query.limit(limit).offset((page - 1) * limit).all()
|
||||
# 若当前页超出末页（数据被并发删除），自动回退到第一页
|
||||
if page > 1 and not log_list:
|
||||
page = 1
|
||||
count = query.count()
|
||||
log_list = query.limit(limit).offset(0).all()
|
||||
# 实例化结果
|
||||
result = []
|
||||
for item in log_list:
|
||||
data = item.to_dict()
|
||||
data['status_name'] = JOB_LOG_STATUS_LIST.get(item.status)
|
||||
result.append(data)
|
||||
return R.ok(data=result, count=count)
|
||||
|
||||
|
||||
# 根据ID查询日志详情
|
||||
def get_job_log_detail(job_log_id):
|
||||
db = get_db_session()
|
||||
log = db.query(JobLog).filter(and_(JobLog.id == job_log_id, JobLog.del_flag == 0)).first()
|
||||
if not log:
|
||||
return None
|
||||
data = log.to_dict()
|
||||
data['status_name'] = JOB_LOG_STATUS_LIST.get(log.status)
|
||||
return data
|
||||
|
||||
|
||||
# 删除任务日志（软删除）
|
||||
def delete_job_log(ids_str):
|
||||
id_list = parse_id_list(ids_str)
|
||||
if not id_list:
|
||||
return R.failed("记录ID不存在")
|
||||
db = get_db_session()
|
||||
result = db.query(JobLog).filter(JobLog.id.in_(id_list), JobLog.del_flag == 0).update(
|
||||
{"del_flag": 1}, synchronize_session=False
|
||||
)
|
||||
return R.ok(msg="本次共删除{0}条数据".format(result))
|
||||
Reference in New Issue
Block a user