263 lines
9.2 KiB
Python
263 lines
9.2 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
子流程节点处理器
|
|
处理子流程的启动和完成
|
|
"""
|
|
import logging
|
|
import uuid
|
|
from typing import Dict, TYPE_CHECKING
|
|
|
|
from sqlalchemy import select
|
|
|
|
from online_dev.workflow.engine.handlers.base_handler import BaseNodeHandler
|
|
|
|
if TYPE_CHECKING:
|
|
from online_dev.workflow.engine.base import ExecutionContext
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class SubflowHandler(BaseNodeHandler):
|
|
"""
|
|
子流程节点处理器
|
|
|
|
支持:
|
|
- 变量传递（全部/选择/无）
|
|
- 等待子流程完成
|
|
- 超时处理
|
|
- 结果回传
|
|
"""
|
|
|
|
async def execute(self, context: 'ExecutionContext', node: Dict) -> None:
|
|
"""
|
|
执行子流程节点：启动子流程
|
|
"""
|
|
from online_dev.workflow.model import WorkflowDefinition, WorkflowInstance
|
|
from online_dev.workflow.engine.base import ExecutionContext as EC
|
|
|
|
node_id = self.get_node_id(node)
|
|
node_name = self.get_node_name(node)
|
|
node_config = self.get_node_config(node)
|
|
|
|
subflow_id = node_config.get('subflowId', '')
|
|
subflow_name = node_config.get('subflowName', '')
|
|
var_pass_mode = node_config.get('varPassMode', 'all')
|
|
selected_vars = node_config.get('selectedVars', [])
|
|
wait_for_completion = node_config.get('waitForCompletion', True)
|
|
|
|
logger.info(f"子流程节点 - 启动子流程: {subflow_name} ({subflow_id})")
|
|
|
|
if not subflow_id:
|
|
logger.warning(f"子流程节点 {node_id} 未配置子流程，自动跳过")
|
|
await self.advance_to_next(context, node)
|
|
return
|
|
|
|
# 获取子流程定义
|
|
stmt = select(WorkflowDefinition).where(
|
|
WorkflowDefinition.id == subflow_id,
|
|
WorkflowDefinition.status == 'published',
|
|
WorkflowDefinition.is_deleted == False,
|
|
)
|
|
result = await context.db.execute(stmt)
|
|
subflow_definition = result.scalar_one_or_none()
|
|
|
|
if not subflow_definition:
|
|
logger.error(f"子流程定义不存在或未发布: {subflow_id}")
|
|
await self.advance_to_next(context, node)
|
|
return
|
|
|
|
# 更新实例当前节点
|
|
await self.update_instance_node(context, node)
|
|
|
|
# 准备传递给子流程的变量
|
|
if var_pass_mode == 'all':
|
|
subflow_data = context.form_data.copy()
|
|
elif var_pass_mode == 'selected':
|
|
subflow_data = {k: v for k, v in context.form_data.items() if k in selected_vars}
|
|
else:
|
|
subflow_data = {}
|
|
|
|
# 生成子流程实例编号
|
|
sub_instance_no = f"SUB-{context.instance.instance_no}-{uuid.uuid4().hex[:6].upper()}"
|
|
|
|
# 获取超时配置
|
|
timeout = self._calculate_timeout(node_config)
|
|
timeout_action = node_config.get('timeoutAction', 'skip')
|
|
|
|
# 创建子流程实例
|
|
sub_instance = WorkflowInstance(
|
|
workflow_id=str(subflow_definition.id),
|
|
instance_no=sub_instance_no,
|
|
title=f"[子流程] {subflow_name} - {context.instance.title}",
|
|
status='pending',
|
|
initiator_id=context.current_user_id or '',
|
|
form_code=subflow_definition.form_code,
|
|
form_data_id=context.instance.form_data_id,
|
|
is_subflow=True,
|
|
parent_instance_id=str(context.instance.id),
|
|
parent_node_id=node_id,
|
|
subflow_timeout=timeout,
|
|
subflow_timeout_action=timeout_action,
|
|
)
|
|
context.db.add(sub_instance)
|
|
await context.db.flush()
|
|
|
|
# 记录日志
|
|
await self.create_log(
|
|
context, node, 'subflow_start',
|
|
comment=f'启动子流程: {subflow_name}',
|
|
extra_data={
|
|
'subflow_id': subflow_id,
|
|
'subflow_name': subflow_name,
|
|
'sub_instance_id': str(sub_instance.id),
|
|
'sub_instance_no': sub_instance_no,
|
|
'var_pass_mode': var_pass_mode,
|
|
},
|
|
)
|
|
|
|
logger.info(f"子流程实例已创建: {sub_instance_no}")
|
|
|
|
# 启动子流程
|
|
sub_flow_def = subflow_definition.flow_definition
|
|
sub_context = EC(
|
|
instance=sub_instance,
|
|
form_data=subflow_data,
|
|
current_user_id=context.current_user_id,
|
|
flow_definition=sub_flow_def,
|
|
db=context.db,
|
|
)
|
|
|
|
# 找到子流程的开始节点并执行
|
|
start_node = sub_flow_def.get('nodes')
|
|
|
|
if start_node and start_node.get('type') == 'start':
|
|
await self.engine._advance_to_next(sub_context, start_node)
|
|
else:
|
|
logger.error(f"子流程没有开始节点: {subflow_id}")
|
|
|
|
# 如果不等待完成，主流程继续推进
|
|
if not wait_for_completion:
|
|
logger.info(f"子流程不等待完成，主流程继续推进")
|
|
await self.advance_to_next(context, node)
|
|
else:
|
|
logger.info(f"主流程等待子流程完成: {sub_instance_no}")
|
|
|
|
def _calculate_timeout(self, node_config: Dict) -> int:
|
|
"""计算超时秒数"""
|
|
timeout_enabled = node_config.get('timeoutEnabled', False)
|
|
timeout_value = node_config.get('timeout', 24)
|
|
timeout_unit = node_config.get('timeoutUnit', 'hour')
|
|
|
|
if not timeout_enabled or not timeout_value:
|
|
return None
|
|
|
|
if timeout_unit == 'minute':
|
|
return timeout_value * 60
|
|
elif timeout_unit == 'hour':
|
|
return timeout_value * 3600
|
|
elif timeout_unit == 'day':
|
|
return timeout_value * 86400
|
|
return timeout_value * 3600
|
|
|
|
async def resume_parent(self, db, sub_instance) -> None:
|
|
"""
|
|
子流程完成后恢复父流程
|
|
"""
|
|
from online_dev.workflow.model import WorkflowLog, WorkflowInstance, WorkflowDefinition
|
|
from online_dev.workflow.engine.base import ExecutionContext as EC
|
|
from online_dev.workflow.engine.utils import FlowUtils, FormDataUtils
|
|
|
|
if not sub_instance.parent_instance_id:
|
|
return
|
|
|
|
parent_node_id = sub_instance.parent_node_id
|
|
if not parent_node_id:
|
|
return
|
|
|
|
# 获取父流程实例
|
|
stmt = select(WorkflowInstance).where(WorkflowInstance.id == sub_instance.parent_instance_id)
|
|
result = await db.execute(stmt)
|
|
parent_instance = result.scalar_one_or_none()
|
|
|
|
if not parent_instance:
|
|
return
|
|
|
|
logger.info(f"子流程完成，恢复父流程: {parent_instance.instance_no}")
|
|
|
|
# 获取父流程定义
|
|
stmt = select(WorkflowDefinition).where(WorkflowDefinition.id == parent_instance.workflow_id)
|
|
result = await db.execute(stmt)
|
|
parent_workflow = result.scalar_one_or_none()
|
|
|
|
if not parent_workflow:
|
|
return
|
|
|
|
parent_flow_def = parent_workflow.flow_definition
|
|
|
|
# 查找父流程等待的节点
|
|
parent_node = FlowUtils.find_node_by_id(parent_flow_def, parent_node_id)
|
|
if not parent_node:
|
|
logger.error(f"父流程节点不存在: {parent_node_id}")
|
|
return
|
|
|
|
# 获取表单数据
|
|
form_data = await FormDataUtils.load_form_data(
|
|
db,
|
|
parent_instance.form_code,
|
|
parent_instance.form_data_id,
|
|
)
|
|
|
|
# 结果回传
|
|
node_config = parent_node.get('config', {})
|
|
result_pass_mode = node_config.get('resultPassMode', 'none')
|
|
result_vars = node_config.get('resultVars', [])
|
|
|
|
if result_pass_mode != 'none':
|
|
sub_form_data = await FormDataUtils.load_form_data(
|
|
db,
|
|
sub_instance.form_code,
|
|
sub_instance.form_data_id,
|
|
)
|
|
if sub_form_data:
|
|
if result_pass_mode == 'all':
|
|
form_data.update(sub_form_data)
|
|
elif result_pass_mode == 'selected':
|
|
for var in result_vars:
|
|
if var in sub_form_data:
|
|
form_data[var] = sub_form_data[var]
|
|
|
|
# 记录日志
|
|
log = WorkflowLog(
|
|
instance_id=str(parent_instance.id),
|
|
node_id=parent_node_id,
|
|
node_name=parent_node.get('name', '子流程'),
|
|
action='subflow_complete',
|
|
operator_id=sub_instance.initiator_id,
|
|
comment=f'子流程完成: {sub_instance.instance_no}, 状态: {sub_instance.status}',
|
|
extra_data={
|
|
'sub_instance_id': str(sub_instance.id),
|
|
'sub_instance_no': sub_instance.instance_no,
|
|
'sub_status': sub_instance.status,
|
|
},
|
|
)
|
|
db.add(log)
|
|
await db.flush()
|
|
|
|
# 构建上下文并推进父流程
|
|
parent_context = EC(
|
|
instance=parent_instance,
|
|
form_data=form_data,
|
|
current_user_id=sub_instance.initiator_id or '',
|
|
flow_definition=parent_flow_def,
|
|
db=db,
|
|
)
|
|
|
|
# 根据子流程结果决定父流程走向
|
|
if sub_instance.status == 'approved':
|
|
await self.engine._advance_to_next(parent_context, parent_node)
|
|
elif sub_instance.status == 'rejected':
|
|
await self.engine._end_instance(parent_context, 'rejected')
|
|
else:
|
|
await self.engine._advance_to_next(parent_context, parent_node)
|