291 lines
9.3 KiB
Python
291 lines
9.3 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
延时节点回调函数
|
|
当延时到期后，由调度器调用此函数推进工作流
|
|
"""
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_delay_job_id(instance_id: str, node_id: str) -> str:
|
|
"""生成延时任务的唯一 job_id"""
|
|
return f"wf_delay_{instance_id}_{node_id}"
|
|
|
|
|
|
async def workflow_delay_callback(instance_id: str, node_id: str):
|
|
"""
|
|
延时到期回调 - 推进工作流
|
|
|
|
由调度器在延时到期后调用，负责：
|
|
1. 加载流程实例，校验状态
|
|
2. 清除延时状态
|
|
3. 记录延时完成日志
|
|
4. 推进流程到下一节点
|
|
|
|
Args:
|
|
instance_id: 流程实例ID
|
|
node_id: 延时节点ID
|
|
"""
|
|
from app.database import AsyncSessionLocal
|
|
|
|
logger.info(f"延时回调触发: instance_id={instance_id}, node_id={node_id}")
|
|
|
|
async with AsyncSessionLocal() as db:
|
|
try:
|
|
await _execute_delay_resume(db, instance_id, node_id)
|
|
await db.commit()
|
|
logger.info(f"延时回调完成: instance_id={instance_id}, node_id={node_id}")
|
|
except Exception as e:
|
|
await db.rollback()
|
|
logger.error(f"延时回调失败: instance_id={instance_id}, node_id={node_id}, error={e}", exc_info=True)
|
|
raise
|
|
|
|
|
|
async def _execute_delay_resume(db: AsyncSession, instance_id: str, node_id: str):
|
|
"""执行延时恢复逻辑"""
|
|
from online_dev.workflow.model import WorkflowInstance, WorkflowDefinition, WorkflowLog
|
|
from online_dev.workflow.engine.base import ExecutionContext
|
|
from online_dev.workflow.engine.utils import FlowUtils
|
|
|
|
# 1. 加载流程实例
|
|
stmt = select(WorkflowInstance).where(
|
|
WorkflowInstance.id == instance_id,
|
|
WorkflowInstance.is_deleted == False,
|
|
)
|
|
result = await db.execute(stmt)
|
|
instance = result.scalar_one_or_none()
|
|
|
|
if not instance:
|
|
logger.warning(f"延时回调: 流程实例不存在 {instance_id}")
|
|
return
|
|
|
|
# 2. 校验实例状态
|
|
if instance.status != 'pending':
|
|
logger.info(f"延时回调: 流程实例状态非 pending ({instance.status})，跳过 {instance_id}")
|
|
return
|
|
|
|
# 3. 校验延时节点匹配
|
|
if instance.delay_node_id != node_id:
|
|
logger.warning(
|
|
f"延时回调: 节点不匹配，期望 {node_id}，实际 {instance.delay_node_id}，跳过"
|
|
)
|
|
return
|
|
|
|
# 4. 加载流程定义
|
|
stmt = select(WorkflowDefinition).where(
|
|
WorkflowDefinition.id == instance.workflow_id,
|
|
WorkflowDefinition.is_deleted == False,
|
|
)
|
|
result = await db.execute(stmt)
|
|
definition = result.scalar_one_or_none()
|
|
|
|
if not definition or not definition.flow_definition:
|
|
logger.error(f"延时回调: 流程定义不存在或无效 workflow_id={instance.workflow_id}")
|
|
return
|
|
|
|
flow_def = definition.flow_definition
|
|
|
|
# 5. 查找延时节点
|
|
delay_node = FlowUtils.find_node_by_id(flow_def.get('nodes'), node_id)
|
|
if not delay_node:
|
|
logger.error(f"延时回调: 延时节点不存在 node_id={node_id}")
|
|
return
|
|
|
|
# 6. 清除实例的延时状态
|
|
instance.delay_node_id = ""
|
|
instance.delay_until = None
|
|
db.add(instance)
|
|
await db.flush()
|
|
|
|
# 7. 记录延时完成日志
|
|
log = WorkflowLog(
|
|
instance_id=str(instance.id),
|
|
node_id=node_id,
|
|
node_name=delay_node.get('name', '延时等待'),
|
|
action='delay_complete',
|
|
operator_id='',
|
|
comment='延时等待结束，流程继续',
|
|
)
|
|
db.add(log)
|
|
await db.flush()
|
|
|
|
# 8. 构建执行上下文并推进流程
|
|
from online_dev.workflow.engine.utils import FormDataUtils
|
|
|
|
form_data = {}
|
|
if instance.form_code and instance.form_data_id:
|
|
try:
|
|
form_data = await FormDataUtils.load_form_data(db, instance.form_code, instance.form_data_id)
|
|
except Exception as e:
|
|
logger.warning(f"延时回调: 加载表单数据失败: {e}")
|
|
|
|
context = ExecutionContext(
|
|
instance=instance,
|
|
form_data=form_data,
|
|
current_user_id='',
|
|
flow_definition=flow_def,
|
|
db=db,
|
|
)
|
|
|
|
# 9. 推进到下一节点
|
|
from online_dev.workflow.engine.workflow_engine import WorkflowEngine
|
|
|
|
engine = WorkflowEngine()
|
|
await engine._advance_to_next(context, delay_node)
|
|
|
|
logger.info(f"延时回调: 流程已推进 instance_id={instance_id}")
|
|
|
|
|
|
async def recover_pending_delay_tasks():
|
|
"""
|
|
应用启动时恢复未完成的延时任务
|
|
|
|
查询所有处于延时等待状态的流程实例，重新注册定时任务：
|
|
- delay_until > now: 注册定时任务等待到期
|
|
- delay_until <= now: 直接执行回调推进流程
|
|
"""
|
|
from app.database import AsyncSessionLocal
|
|
from online_dev.workflow.model import WorkflowInstance
|
|
|
|
logger.info("开始恢复未完成的延时任务...")
|
|
|
|
try:
|
|
async with AsyncSessionLocal() as db:
|
|
# 查询所有有延时状态的 pending 实例
|
|
stmt = select(WorkflowInstance).where(
|
|
WorkflowInstance.status == 'pending',
|
|
WorkflowInstance.delay_node_id != '',
|
|
WorkflowInstance.delay_node_id.isnot(None),
|
|
WorkflowInstance.delay_until.isnot(None),
|
|
WorkflowInstance.is_deleted == False,
|
|
)
|
|
result = await db.execute(stmt)
|
|
instances = list(result.scalars().all())
|
|
|
|
if not instances:
|
|
logger.info("没有需要恢复的延时任务")
|
|
return
|
|
|
|
now = datetime.now()
|
|
recovered = 0
|
|
expired = 0
|
|
|
|
for instance in instances:
|
|
instance_id = str(instance.id)
|
|
node_id = instance.delay_node_id
|
|
delay_until = instance.delay_until
|
|
|
|
if delay_until > now:
|
|
# 未到期，重新注册定时任务
|
|
success = await _register_delay_job(instance_id, node_id, delay_until)
|
|
if success:
|
|
recovered += 1
|
|
logger.info(
|
|
f"恢复延时任务: instance={instance_id}, node={node_id}, "
|
|
f"到期时间={delay_until}"
|
|
)
|
|
else:
|
|
# 已过期，直接执行回调
|
|
expired += 1
|
|
logger.info(
|
|
f"延时任务已过期，立即执行: instance={instance_id}, node={node_id}"
|
|
)
|
|
try:
|
|
await workflow_delay_callback(instance_id, node_id)
|
|
except Exception as e:
|
|
logger.error(f"执行过期延时回调失败: {e}", exc_info=True)
|
|
|
|
logger.info(f"延时任务恢复完成: 重新注册 {recovered} 个, 立即执行 {expired} 个")
|
|
|
|
except Exception as e:
|
|
logger.error(f"恢复延时任务失败: {e}", exc_info=True)
|
|
|
|
|
|
async def _register_delay_job(instance_id: str, node_id: str, delay_until: datetime) -> bool:
|
|
"""
|
|
注册延时定时任务到调度器
|
|
|
|
Args:
|
|
instance_id: 流程实例ID
|
|
node_id: 延时节点ID
|
|
delay_until: 到期时间
|
|
|
|
Returns:
|
|
是否注册成功
|
|
"""
|
|
try:
|
|
from scheduler.service import scheduler_service
|
|
|
|
scheduler = scheduler_service.get_scheduler()
|
|
if not scheduler:
|
|
logger.warning("调度器未初始化，无法注册延时任务")
|
|
return False
|
|
|
|
job_id = get_delay_job_id(instance_id, node_id)
|
|
|
|
# 创建回调包装函数（闭包捕获参数）
|
|
_instance_id = instance_id
|
|
_node_id = node_id
|
|
|
|
async def delay_wrapper():
|
|
await workflow_delay_callback(_instance_id, _node_id)
|
|
|
|
# 注册任务
|
|
await scheduler.configure_task(job_id, func=delay_wrapper)
|
|
|
|
# 添加一次性调度
|
|
from apscheduler.triggers.date import DateTrigger
|
|
|
|
await scheduler.add_schedule(
|
|
func_or_task_id=job_id,
|
|
trigger=DateTrigger(run_time=delay_until),
|
|
id=job_id,
|
|
)
|
|
|
|
logger.info(f"延时任务已注册: job_id={job_id}, 到期时间={delay_until}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"注册延时任务失败: {e}", exc_info=True)
|
|
return False
|
|
|
|
|
|
async def remove_delay_job(instance_id: str, node_id: str) -> bool:
|
|
"""
|
|
从调度器移除延时任务
|
|
|
|
Args:
|
|
instance_id: 流程实例ID
|
|
node_id: 延时节点ID
|
|
|
|
Returns:
|
|
是否移除成功
|
|
"""
|
|
try:
|
|
from scheduler.service import scheduler_service
|
|
|
|
scheduler = scheduler_service.get_scheduler()
|
|
if not scheduler:
|
|
return False
|
|
|
|
job_id = get_delay_job_id(instance_id, node_id)
|
|
|
|
try:
|
|
await scheduler.remove_schedule(job_id)
|
|
logger.info(f"延时任务已移除: job_id={job_id}")
|
|
except Exception:
|
|
# 任务可能不存在（已执行或已清理）
|
|
pass
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"移除延时任务失败: {e}", exc_info=True)
|
|
return False
|