Codex
99 lines
3.0 KiB
Python
99 lines
3.0 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
工作流执行任务 - 供定时任务调度器调用
|
||
"""
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
async def execute_workflow_task(
|
||
job_code: str = None,
|
||
workflow_code: str = None,
|
||
inputs: dict = None,
|
||
**kwargs
|
||
):
|
||
"""
|
||
执行工作流任务
|
||
|
||
在定时任务中调用已发布的数据处理/自动化工作流。
|
||
|
||
Args:
|
||
job_code: 任务编码（由调度器自动传入）
|
||
workflow_code: 要执行的工作流编码（必填）
|
||
inputs: 工作流输入变量
|
||
**kwargs: 其他参数（包含 task_logger）
|
||
|
||
Returns:
|
||
dict: 执行结果
|
||
"""
|
||
if not workflow_code:
|
||
raise ValueError("workflow_code 参数不能为空，请在 task_kwargs 中配置")
|
||
|
||
inputs = inputs or {}
|
||
inputs['job_code'] = job_code
|
||
|
||
from app.database import AsyncSessionLocal
|
||
from ai_platform.services.workflow_service import AIWorkflowService
|
||
from ai_platform.models.workflow import AIWorkflow
|
||
from sqlalchemy import select
|
||
from scheduler.task_utils import TaskLoggerWrapper
|
||
|
||
log = TaskLoggerWrapper(job_code, kwargs.get('task_logger'))
|
||
await log.info(f"开始执行工作流: {workflow_code}")
|
||
|
||
async with AsyncSessionLocal() as db:
|
||
result = await db.execute(
|
||
select(AIWorkflow).where(
|
||
AIWorkflow.code == workflow_code,
|
||
AIWorkflow.is_deleted == False
|
||
)
|
||
)
|
||
workflow = result.scalar_one_or_none()
|
||
|
||
if not workflow:
|
||
raise ValueError(f"工作流不存在: {workflow_code}")
|
||
|
||
if workflow.workflow_type not in ('data_process', 'automation'):
|
||
raise ValueError(
|
||
f"工作流类型不支持定时执行: {workflow.workflow_type}，"
|
||
f"仅支持 data_process 和 automation 类型"
|
||
)
|
||
|
||
if workflow.status != 'published':
|
||
raise ValueError(f"工作流未发布: {workflow_code}")
|
||
|
||
await log.info(
|
||
f"工作流类型: {workflow.workflow_type}, "
|
||
f"工作流名称: {workflow.name}, "
|
||
f"输入变量: {list(inputs.keys())}"
|
||
)
|
||
|
||
service = AIWorkflowService(db)
|
||
run = await service.run_workflow(
|
||
workflow_id=str(workflow.id),
|
||
inputs=inputs,
|
||
trigger_type='api',
|
||
)
|
||
|
||
await log.info(
|
||
f"工作流执行完成: status={run.status}, "
|
||
f"elapsed_time={run.elapsed_time}ms, "
|
||
f"total_steps={run.total_steps}"
|
||
)
|
||
|
||
if run.status == 'failed':
|
||
raise RuntimeError(f"工作流执行失败: {run.error_message}")
|
||
|
||
return {
|
||
'run_id': str(run.id),
|
||
'workflow_code': workflow_code,
|
||
'workflow_name': workflow.name,
|
||
'status': run.status,
|
||
'elapsed_time': run.elapsed_time,
|
||
'total_steps': run.total_steps,
|
||
'total_tokens': run.total_tokens,
|
||
'outputs': run.outputs,
|
||
}
|