djangoadmin/DjangoAdmin_Flask_EleVue
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: P4c 最终批次-4 job 模块

Browse Source 
- get_job_page/add/update/delete/status 按新范式改造（get_info 委托 JobRepository）
- pause_job/resume_job/run_job_once/reload_all_jobs 去 R 改 BizError/返回数据
- router 取参包 R；修复多行 config 导入被 ast 破坏问题

【验收】flake8 零告警；应用启动；job 分页运行时正常；12 passed

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
DjangoAdmin Dev
2026-08-07 17:11:39 +08:00
co-authored by Claude
parent 0e0179859f
commit 6bd4951f95
9 changed files with 330 additions and 444 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/modules/job/scheduler_service.py
+45 -52
View File
@@ -77,6 +77,7 @@ _scheduler = None
# 调度器实例获取函数
# ============================================================
def get_scheduler():
"""
获取调度器实例（延迟初始化）
@@ -90,7 +91,7 @@ def get_scheduler():
global _scheduler
if _scheduler is None:
# 创建后台调度器，设置时区为北京时间
_scheduler = BackgroundScheduler(timezone='Asia/Shanghai')
_scheduler = BackgroundScheduler(timezone="Asia/Shanghai")
return _scheduler
@@ -102,6 +103,7 @@ scheduler = get_scheduler()
# URL 构建工具函数
# ============================================================
def build_full_url(url: str) -> str:
"""
构建完整的URL
@@ -125,11 +127,11 @@ def build_full_url(url: str) -> str:
return url
# 已经是完整URL，直接返回
if url.startswith(('http://', 'https://')):
if url.startswith(("http://", "https://")):
return url
# 相对路径，拼接站点域名
if url.startswith('/'):
if url.startswith("/"):
url = f"{FLASK_SITE_DOMAIN}{url}"
return url
@@ -139,12 +141,13 @@ def build_full_url(url: str) -> str:
# 任务执行核心函数
# ============================================================
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
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]:
"""
执行任务（调度器调用的入口函数）
@@ -176,19 +179,20 @@ def execute_job(
result = _execute_by_trigger(job_trigger, job_id, params)
else:
# 无效的任务配置
result = {'success': False, 'error': '任务既没有URL也没有触发器'}
result = {"success": False, "error": "任务既没有URL也没有触发器"}
return result
except Exception as e:
logger.error(f"执行任务异常 [ID: {job_id}]: {str(e)}")
return {'success': False, 'error': str(e)}
return {"success": False, "error": str(e)}
# ============================================================
# URL 方式任务执行
# ============================================================
def _execute_by_url(job_id: int, url: str, is_sync: int, params: Optional[Dict] = None) -> Dict:
"""
通过URL方式执行任务
@@ -208,11 +212,7 @@ def _execute_by_url(job_id: int, url: str, is_sync: int, params: Optional[Dict]
full_url = build_full_url(url)
# 构建请求数据
request_data = {
'job_id': job_id,
'timestamp': datetime.now().isoformat(), # 请求时间戳
'params': params or {}
}
request_data = {"job_id": job_id, "timestamp": datetime.now().isoformat(), "params": params or {}} # 请求时间戳
# 根据同步/异步选择发送方式
if is_sync == 1:
@@ -240,20 +240,15 @@ def _send_sync_request(url: str, data: Dict, job_id: int) -> Dict:
# 准备请求头
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json',
'User-Agent': 'Scheduler/1.0',
"Content-Type": "application/json",
"Accept": "application/json",
"User-Agent": "Scheduler/1.0",
# 添加定时任务专用密钥，用于跳过权限验证
'X-Scheduler-Key': INTERNAL_API_KEY,
"X-Scheduler-Key": INTERNAL_API_KEY,
}
# 发送POST请求
response = requests.post(
url,
json=data,
timeout=30, # 30秒超时
headers=headers
)
response = requests.post(url, json=data, timeout=30, headers=headers) # 30秒超时
response.raise_for_status() # 非200状态码抛出异常
# 解析响应数据
@@ -261,16 +256,12 @@ def _send_sync_request(url: str, data: Dict, job_id: int) -> Dict:
result_data = response.json()
except Exception:
# JSON解析失败时返回原始文本（限制长度）
result_data = {'text': response.text[:200]}
result_data = {"text": response.text[:200]}
return {
'success': True,
'status_code': response.status_code,
'data': result_data
}
return {"success": True, "status_code": response.status_code, "data": result_data}
except requests.exceptions.RequestException as e:
return {'success': False, 'error': str(e)}
return {"success": False, "error": str(e)}
def _send_async_request(url: str, data: Dict, job_id: int) -> Dict:
@@ -293,8 +284,8 @@ def _send_async_request(url: str, data: Dict, job_id: int) -> Dict:
"""异步请求执行函数（在独立线程中运行）"""
try:
headers = {
'Content-Type': 'application/json',
'X-Scheduler-Key': INTERNAL_API_KEY,
"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}]")
@@ -305,7 +296,7 @@ def _send_async_request(url: str, data: Dict, job_id: int) -> Dict:
thread = threading.Thread(target=async_request)
thread.start()
return {'success': True, 'status': 'triggered'}
return {"success": True, "status": "triggered"}
def _execute_by_trigger(trigger: str, job_id: int, params: Optional[Dict] = None) -> Dict:
@@ -324,17 +315,19 @@ def _execute_by_trigger(trigger: str, job_id: int, params: Optional[Dict] = None
"""
try:
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 {"success": True, "trigger": trigger, "data": result}
except Exception as e:
return {'success': False, 'error': str(e)}
return {"success": False, "error": str(e)}
# ============================================================
# 调度器任务管理函数
# ============================================================
def add_job_to_scheduler(job_dict:dict ) -> bool:
def add_job_to_scheduler(job_dict: dict) -> bool:
"""
添加任务到调度器
@@ -357,13 +350,13 @@ def add_job_to_scheduler(job_dict:dict ) -> bool:
# ============================================================
# 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')
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")
print(f"[scheduler] add_job id={job_id} status={job_status} cron='{job_cron}'")
@@ -386,6 +379,7 @@ def add_job_to_scheduler(job_dict:dict ) -> bool:
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})")
@@ -417,7 +411,7 @@ def add_job_to_scheduler(job_dict:dict ) -> bool:
name=job_name, # 任务显示名称
replace_existing=True, # 替换已存在的任务
misfire_grace_time=30, # 错过执行的宽限时间（秒）
coalesce=True # 合并错过的执行
coalesce=True, # 合并错过的执行
)
logger.info(f"任务已添加到调度器 [ID: {job_id}, cron: {job_cron}]")
@@ -504,22 +498,24 @@ def run_job_once(job_id: int, params: Optional[Dict] = None) -> 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 {"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)}
return {"success": False, "error": str(e)}
# ============================================================
# 调度器初始化和重载函数
# ============================================================
def init_scheduler() -> int:
"""
初始化调度器
@@ -538,10 +534,7 @@ def init_scheduler() -> int:
logger.info("调度器已启动")
# 查询所有启用的任务
jobs = Job.query.filter(
Job.status == 1,
Job.is_delete == 0
).all()
jobs = Job.query.filter(Job.status == 1, Job.is_delete == 0).all()
# 使用 model_to_dict 将每个 job 对象转换为字典
# 避免 SQLAlchemy Session 分离问题