273 lines
8.3 KiB
Python
273 lines
8.3 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
工作流延时节点处理器
|
|
使用 APScheduler 实现延时等待功能
|
|
"""
|
|
import json
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def calculate_delay_datetime(duration: int, unit: str) -> datetime:
|
|
"""
|
|
计算延时后的执行时间
|
|
|
|
Args:
|
|
duration: 延时时长
|
|
unit: 延时单位 - minute/hour/day/workday
|
|
|
|
Returns:
|
|
datetime: 延时后的执行时间
|
|
"""
|
|
now = datetime.now()
|
|
|
|
if unit == 'minute':
|
|
return now + timedelta(minutes=duration)
|
|
elif unit == 'hour':
|
|
return now + timedelta(hours=duration)
|
|
elif unit == 'day':
|
|
return now + timedelta(days=duration)
|
|
elif unit == 'workday':
|
|
# 工作日计算（跳过周末）
|
|
result = now
|
|
days_added = 0
|
|
while days_added < duration:
|
|
result += timedelta(days=1)
|
|
# 周一到周五是工作日 (0-4)
|
|
if result.weekday() < 5:
|
|
days_added += 1
|
|
return result
|
|
else:
|
|
# 默认按小时
|
|
return now + timedelta(hours=duration)
|
|
|
|
|
|
async def create_delay_job(
|
|
db: AsyncSession,
|
|
instance_id: str,
|
|
node_id: str,
|
|
duration: int,
|
|
unit: str,
|
|
user_id: str,
|
|
) -> Optional[str]:
|
|
"""
|
|
创建延时任务
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
instance_id: 流程实例ID
|
|
node_id: 延时节点ID
|
|
duration: 延时时长
|
|
unit: 延时单位
|
|
user_id: 操作用户ID
|
|
|
|
Returns:
|
|
str: 任务编码，失败返回 None
|
|
"""
|
|
try:
|
|
from scheduler.model import SchedulerJob
|
|
from scheduler.service import SchedulerService
|
|
|
|
# 计算执行时间
|
|
run_date = calculate_delay_datetime(duration, unit)
|
|
|
|
# 生成唯一的任务编码
|
|
job_code = f"workflow_delay_{instance_id}_{node_id}"
|
|
|
|
# 检查是否已存在
|
|
stmt = select(SchedulerJob).where(SchedulerJob.code == job_code)
|
|
result = await db.execute(stmt)
|
|
existing = result.scalar_one_or_none()
|
|
|
|
if existing:
|
|
# 更新执行时间
|
|
existing.run_date = run_date
|
|
existing.status = 1 # 启用
|
|
db.add(existing)
|
|
await db.flush()
|
|
|
|
# 更新调度器中的任务
|
|
scheduler_service = SchedulerService()
|
|
if scheduler_service.is_running():
|
|
await scheduler_service.modify_job(existing)
|
|
|
|
logger.info(f"更新延时任务: {job_code}, 执行时间: {run_date}")
|
|
return job_code
|
|
|
|
# 创建新任务
|
|
job = SchedulerJob(
|
|
name=f"工作流延时-{instance_id[:8]}",
|
|
code=job_code,
|
|
description=f"流程实例 {instance_id} 的延时节点 {node_id}",
|
|
group='workflow_delay',
|
|
trigger_type='date',
|
|
run_date=run_date,
|
|
task_func='core.workflow.engine.delay_process.execute_delay_complete',
|
|
task_kwargs=json.dumps({
|
|
'instance_id': instance_id,
|
|
'node_id': node_id,
|
|
'user_id': user_id,
|
|
}),
|
|
status=1, # 启用
|
|
max_instances=1,
|
|
coalesce=True,
|
|
)
|
|
db.add(job)
|
|
await db.flush()
|
|
|
|
# 添加到调度器
|
|
scheduler_service = SchedulerService()
|
|
if scheduler_service.is_running():
|
|
await scheduler_service.add_job(job)
|
|
|
|
logger.info(f"创建延时任务: {job_code}, 执行时间: {run_date}")
|
|
return job_code
|
|
|
|
except Exception as e:
|
|
logger.error(f"创建延时任务失败: {e}")
|
|
return None
|
|
|
|
|
|
async def cancel_delay_job(db: AsyncSession, instance_id: str, node_id: str) -> bool:
|
|
"""
|
|
取消延时任务
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
instance_id: 流程实例ID
|
|
node_id: 延时节点ID
|
|
|
|
Returns:
|
|
bool: 是否成功
|
|
"""
|
|
try:
|
|
from scheduler.model import SchedulerJob
|
|
from scheduler.service import SchedulerService
|
|
|
|
job_code = f"workflow_delay_{instance_id}_{node_id}"
|
|
|
|
# 从调度器移除
|
|
scheduler_service = SchedulerService()
|
|
if scheduler_service.is_running():
|
|
await scheduler_service.remove_job(job_code)
|
|
|
|
# 禁用数据库记录
|
|
stmt = update(SchedulerJob).where(
|
|
SchedulerJob.code == job_code
|
|
).values(status=0)
|
|
await db.execute(stmt)
|
|
await db.flush()
|
|
|
|
logger.info(f"取消延时任务: {job_code}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"取消延时任务失败: {e}")
|
|
return False
|
|
|
|
|
|
async def execute_delay_complete(db: AsyncSession, instance_id: str, node_id: str, user_id: str):
|
|
"""
|
|
延时完成后执行的任务
|
|
|
|
Args:
|
|
db: 数据库会话
|
|
instance_id: 流程实例ID
|
|
node_id: 延时节点ID
|
|
user_id: 操作用户ID
|
|
"""
|
|
from online_dev.workflow.model import WorkflowInstance, WorkflowDefinition, WorkflowLog
|
|
from online_dev.workflow.engine.workflow_engine import WorkflowEngine
|
|
from online_dev.workflow.engine.base import ExecutionContext
|
|
from online_dev.workflow.engine.utils import FlowUtils
|
|
from core.user.model import User
|
|
|
|
logger.info(f"延时任务执行: instance={instance_id}, node={node_id}")
|
|
|
|
try:
|
|
# 获取流程实例
|
|
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.error(f"流程实例不存在: {instance_id}")
|
|
return f"流程实例不存在: {instance_id}"
|
|
|
|
# 检查流程状态
|
|
if instance.status != 'pending':
|
|
logger.warning(f"流程实例 {instance_id} 状态不是 pending，跳过延时完成")
|
|
return f"流程状态不是 pending: {instance.status}"
|
|
|
|
# 检查当前节点
|
|
if instance.current_node_id != node_id:
|
|
logger.warning(f"流程实例 {instance_id} 当前节点不是 {node_id}，跳过延时完成")
|
|
return f"当前节点不匹配: {instance.current_node_id}"
|
|
|
|
# 获取流程定义
|
|
stmt = select(WorkflowDefinition).where(WorkflowDefinition.id == instance.workflow_id)
|
|
result = await db.execute(stmt)
|
|
workflow = result.scalar_one_or_none()
|
|
|
|
if not workflow:
|
|
logger.error(f"流程定义不存在: {instance.workflow_id}")
|
|
return f"流程定义不存在"
|
|
|
|
# 获取用户
|
|
stmt = select(User).where(User.id == user_id)
|
|
result = await db.execute(stmt)
|
|
user = result.scalar_one_or_none()
|
|
|
|
if not user:
|
|
# 使用发起人
|
|
stmt = select(User).where(User.id == instance.initiator_id)
|
|
result = await db.execute(stmt)
|
|
user = result.scalar_one_or_none()
|
|
|
|
# 记录延时完成日志
|
|
log = WorkflowLog(
|
|
instance_id=str(instance.id),
|
|
node_id=node_id,
|
|
node_name=instance.current_node_name,
|
|
action='delay_complete',
|
|
operator_id=str(user.id) if user else '',
|
|
comment='延时等待完成',
|
|
)
|
|
db.add(log)
|
|
await db.flush()
|
|
|
|
# 创建执行上下文
|
|
engine = WorkflowEngine()
|
|
|
|
context = ExecutionContext(
|
|
instance=instance,
|
|
form_data={},
|
|
current_user=user,
|
|
flow_definition=workflow.flow_definition,
|
|
db=db,
|
|
)
|
|
|
|
# 查找延时节点并推进
|
|
delay_node = FlowUtils.find_node_by_id(context.flow_definition, node_id)
|
|
if delay_node:
|
|
await engine._advance_to_next(context, delay_node)
|
|
logger.info(f"延时节点 {node_id} 完成，流程继续推进")
|
|
return "延时完成，流程已推进"
|
|
else:
|
|
logger.error(f"找不到延时节点: {node_id}")
|
|
return f"找不到延时节点: {node_id}"
|
|
|
|
except Exception as e:
|
|
logger.exception(f"延时任务执行失败: {e}")
|
|
raise
|