- git 初始化 + .gitignore 补全(根 .env / logs / instance / __pycache__ 等) - 修复 utils/.gitignore 误将整个 utils/ 源码层忽略的问题,补入全部工具源码(含 ip2region.xdb) - 含后端全部代码:application / extensions / utils / config / migrations - 含前端 ui/(后续整体迁走) - 不含根 .env(含密钥,本地保留) Co-Authored-By: Claude <noreply@anthropic.com>
166 lines
4.6 KiB
Python
166 lines
4.6 KiB
Python
"""
|
|
任务执行器模块
|
|
=============================================================
|
|
|
|
该模块实现了根据触发器标识执行对应任务的功能。
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, Any, Optional
|
|
|
|
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):
|
|
"""
|
|
注册任务处理器
|
|
|
|
Args:
|
|
trigger: 触发器标识
|
|
handler: 处理函数
|
|
"""
|
|
self._handlers[trigger] = handler
|
|
logger.info(f"注册任务处理器: {trigger}")
|
|
|
|
def execute(self, trigger: str, **kwargs) -> Any:
|
|
"""
|
|
执行任务
|
|
|
|
Args:
|
|
trigger: 触发器标识
|
|
**kwargs: 执行参数
|
|
|
|
Returns:
|
|
Any: 执行结果
|
|
"""
|
|
if trigger not in self._handlers:
|
|
logger.error(f"未找到触发器对应的方法: {trigger}")
|
|
raise KeyError(f"未找到触发器对应的方法: {trigger}")
|
|
|
|
handler = self._handlers[trigger]
|
|
logger.info(f"执行任务: {trigger}, 参数: {kwargs}")
|
|
return handler(**kwargs)
|
|
|
|
# ============================================================
|
|
# 默认任务处理器
|
|
# ============================================================
|
|
|
|
def _send_sms(self, **kwargs) -> Dict:
|
|
"""
|
|
发送短信任务
|
|
|
|
Args:
|
|
**kwargs: 包含 phone, template_code, template_params, sign_name 等参数
|
|
|
|
Returns:
|
|
Dict: 执行结果
|
|
"""
|
|
from application.services import sms_svc
|
|
|
|
phone = kwargs.get('phone')
|
|
template_code = kwargs.get('template_code')
|
|
template_params = kwargs.get('template_params')
|
|
sign_name = kwargs.get('sign_name')
|
|
job_id = kwargs.get('job_id')
|
|
|
|
logger.info(f"发送短信 [任务ID: {job_id}, 手机号: {phone}, 模板: {template_code}]")
|
|
|
|
success = sms_svc.send_sms(
|
|
phone=phone,
|
|
template_code=template_code,
|
|
template_params=template_params,
|
|
sign_name=sign_name,
|
|
)
|
|
|
|
return {
|
|
'success': success,
|
|
'message': '短信发送成功' if success else '短信发送失败',
|
|
'phone': phone,
|
|
'template_code': template_code
|
|
}
|
|
|
|
def _send_email(self, **kwargs) -> Dict:
|
|
"""
|
|
发送邮件任务
|
|
|
|
Args:
|
|
**kwargs: 包含 to, subject, body, html 等参数
|
|
|
|
Returns:
|
|
Dict: 执行结果
|
|
"""
|
|
from application.services import mail_svc
|
|
|
|
to = kwargs.get('to')
|
|
subject = kwargs.get('subject')
|
|
body = kwargs.get('body')
|
|
html = kwargs.get('html')
|
|
cc = kwargs.get('cc')
|
|
job_id = kwargs.get('job_id')
|
|
|
|
logger.info(f"发送邮件 [任务ID: {job_id}, 收件人: {to}, 主题: {subject}]")
|
|
|
|
success = mail_svc.send_mail(to=to, subject=subject, body=body, html=html, cc=cc)
|
|
|
|
return {
|
|
'success': success,
|
|
'message': '邮件发送成功' if success else '邮件发送失败',
|
|
'to': to,
|
|
'subject': subject
|
|
}
|
|
|
|
def _sync_data(self, **kwargs) -> Dict:
|
|
"""
|
|
数据同步任务
|
|
|
|
Args:
|
|
**kwargs: 包含 source, target, sync_type 等参数
|
|
|
|
Returns:
|
|
Dict: 执行结果
|
|
"""
|
|
source = kwargs.get('source')
|
|
target = kwargs.get('target')
|
|
sync_type = kwargs.get('sync_type')
|
|
job_id = kwargs.get('job_id')
|
|
|
|
logger.info(f"数据同步 [任务ID: {job_id}, 源: {source}, 目标: {target}, 类型: {sync_type}]")
|
|
|
|
# TODO: 调用实际的数据同步服务
|
|
|
|
return {
|
|
'success': True,
|
|
'message': '数据同步成功',
|
|
'source': source,
|
|
'target': target
|
|
}
|
|
|
|
|
|
# 创建全局执行器实例
|
|
job_executor = JobExecutor()
|