payment-platform/backend/app/plugin/module_task/workflow/definition/service.py

import asyncio
from typing import Any

from app.api.v1.module_system.auth.schema import AuthSchema
from app.core.exceptions import CustomException

from ..engine.prefect_engine import run_prefect_workflow_sync, utc_now_iso, validate_workflow_graph
from ..node_type.crud import WorkflowNodeTypeCRUD
from .crud import WorkflowCRUD
from .schema import (
    WorkflowCreateSchema,
    WorkflowExecuteResultSchema,
    WorkflowExecuteSchema,
    WorkflowOutSchema,
    WorkflowPublishSchema,
    WorkflowQueryParam,
    WorkflowUpdateSchema,
)


class WorkflowService:
    """工作流：画布存储 + 发布校验 + Prefect 执行"""

    @staticmethod
    def _out(obj: Any) -> dict:
        return WorkflowOutSchema.model_validate(obj).model_dump(mode="json")

    @classmethod
    async def get_workflow_detail_service(cls, auth: AuthSchema, id: int) -> dict:
        """
        获取工作流详情。

        参数:
        - auth (AuthSchema): 认证信息。
        - id (int): 工作流 ID。

        返回:
        - dict: 序列化后的工作流详情。

        异常:
        - CustomException: 不存在时抛出。
        """
        obj = await WorkflowCRUD(auth).get_obj_by_id_crud(id=id)
        if not obj:
            raise CustomException(msg="工作流不存在")
        return cls._out(obj)

    @classmethod
    async def get_workflow_list_service(
        cls,
        auth: AuthSchema,
        search: WorkflowQueryParam | None = None,
        order_by: list[dict[str, str]] | None = None,
    ) -> list[dict]:
        """
        获取工作流列表（非分页）。

        参数:
        - auth (AuthSchema): 认证信息。
        - search (WorkflowQueryParam | None): 查询条件。
        - order_by (list[dict[str, str]] | None): 排序。

        返回:
        - list[dict]: 工作流字典列表。
        """
        if order_by is None:
            order_by = [{"updated_time": "desc"}]
        obj_list = await WorkflowCRUD(auth).get_obj_list_crud(
            search=search.__dict__ if search else None,
            order_by=order_by,
        )
        return [cls._out(o) for o in obj_list]

    @classmethod
    async def get_workflow_page_service(
        cls,
        auth: AuthSchema,
        page_no: int,
        page_size: int,
        search: WorkflowQueryParam | None = None,
        order_by: list[dict[str, str]] | None = None,
    ) -> dict:
        """
        分页查询工作流（数据库 OFFSET/LIMIT）。

        参数:
        - auth (AuthSchema): 认证信息。
        - page_no (int): 页码。
        - page_size (int): 每页条数。
        - search (WorkflowQueryParam | None): 查询条件。
        - order_by (list[dict[str, str]] | None): 排序。

        返回:
        - dict: 分页结果（items 已 JSON 友好序列化）。
        """
        offset = (page_no - 1) * page_size
        order = order_by or [{"updated_time": "desc"}]
        result = await WorkflowCRUD(auth).page(
            offset=offset,
            limit=page_size,
            order_by=order,
            search=search.__dict__ if search else {},
            out_schema=WorkflowOutSchema,
        )
        result["items"] = [
            WorkflowOutSchema.model_validate(item).model_dump(mode="json") for item in result["items"]
        ]
        return result

    @classmethod
    async def create_workflow_service(cls, auth: AuthSchema, data: WorkflowCreateSchema) -> dict:
        """
        创建工作流草稿。

        参数:
        - auth (AuthSchema): 认证信息。
        - data (WorkflowCreateSchema): 创建体。

        返回:
        - dict: 新建工作流字典。

        异常:
        - CustomException: 编码重复或创建失败。
        """
        exist = await WorkflowCRUD(auth).get(code=data.code)
        if exist:
            raise CustomException(msg="流程编码已存在")
        obj = await WorkflowCRUD(auth).create_obj_crud(data=data)
        if not obj:
            raise CustomException(msg="创建工作流失败")
        return cls._out(obj)

    @classmethod
    async def update_workflow_service(
        cls, auth: AuthSchema, id: int, data: WorkflowUpdateSchema
    ) -> dict:
        """
        更新工作流。

        参数:
        - auth (AuthSchema): 认证信息。
        - id (int): 工作流 ID。
        - data (WorkflowUpdateSchema): 更新体。

        返回:
        - dict: 更新后工作流字典。

        异常:
        - CustomException: 不存在、编码冲突或更新失败。
        """
        exist = await WorkflowCRUD(auth).get_obj_by_id_crud(id=id)
        if not exist:
            raise CustomException(msg="工作流不存在")
        if exist.code != data.code:
            other = await WorkflowCRUD(auth).get(code=data.code)
            if other:
                raise CustomException(msg="流程编码已存在")
        obj = await WorkflowCRUD(auth).update_obj_crud(id=id, data=data)
        if not obj:
            raise CustomException(msg="更新工作流失败")
        return cls._out(obj)

    @classmethod
    async def delete_workflow_service(cls, auth: AuthSchema, ids: list[int]) -> None:
        """
        批量删除工作流。

        参数:
        - auth (AuthSchema): 认证信息。
        - ids (list[int]): ID 列表。

        返回:
        - None

        异常:
        - CustomException: ID 为空时抛出。
        """
        if not ids:
            raise CustomException(msg="删除ID不能为空")
        await WorkflowCRUD(auth).delete_obj_crud(ids=ids)

    @classmethod
    async def publish_workflow_service(
        cls, auth: AuthSchema, id: int, body: WorkflowPublishSchema | None = None
    ) -> dict:
        """
        校验 DAG 后发布工作流。

        参数:
        - auth (AuthSchema): 认证信息。
        - id (int): 工作流 ID。
        - body (WorkflowPublishSchema | None): 可选附加参数。

        返回:
        - dict: 发布后工作流字典。

        异常:
        - CustomException: 不存在、图无效或发布失败。
        """
        obj = await WorkflowCRUD(auth).get_obj_by_id_crud(id=id)
        if not obj:
            raise CustomException(msg="工作流不存在")
        nodes = obj.nodes or []
        edges = obj.edges or []

        try:
            validate_workflow_graph(nodes, edges)
        except ValueError as e:
            raise CustomException(msg=str(e)) from e

        data = WorkflowUpdateSchema(
            name=obj.name,
            code=obj.code,
            description=obj.description,
            nodes=obj.nodes,
            edges=obj.edges,
            workflow_status="published",
        )
        updated = await WorkflowCRUD(auth).update_obj_crud(id=id, data=data)
        if not updated:
            raise CustomException(msg="发布失败")
        return cls._out(updated)

    @classmethod
    async def execute_workflow_service(
        cls, auth: AuthSchema, body: WorkflowExecuteSchema
    ) -> dict:
        """
        执行已发布工作流（Prefect 同步入口在线程池中运行）。

        参数:
        - auth (AuthSchema): 认证信息。
        - body (WorkflowExecuteSchema): 工作流 ID 与变量。

        返回:
        - dict: 执行结果摘要（成功或失败结构）。

        异常:
        - CustomException: 未发布、缺节点、节点类型未注册等。
        """
        obj = await WorkflowCRUD(auth).get_obj_by_id_crud(id=body.workflow_id)
        if not obj:
            raise CustomException(msg="工作流不存在")
        if obj.workflow_status != "published":
            raise CustomException(msg="仅已发布的工作流可执行")

        nodes = obj.nodes or []
        edges = obj.edges or []
        if not nodes:
            raise CustomException(msg="工作流没有节点")

        codes = {n.get("type") for n in nodes if n.get("type")}
        templates: dict[str, dict[str, Any]] = {}
        for code in codes:
            node_type = await WorkflowNodeTypeCRUD(auth).get(code=code)
            if not node_type:
                raise CustomException(msg=f"编排节点类型未注册（请在「工作流编排节点类型」中维护，非定时任务节点）: {code}")
            if not node_type.func or not str(node_type.func).strip():
                raise CustomException(msg=f"编排节点类型未配置 func 代码块: {code}")
            templates[code] = {
                "func": node_type.func,
                "args": node_type.args,
                "kwargs": node_type.kwargs,
            }

        variables = body.variables or {}
        start = utc_now_iso()
        try:
            raw = await asyncio.to_thread(
                run_prefect_workflow_sync,
                nodes,
                edges,
                templates,
                variables,
            )
        except ValueError as e:
            raise CustomException(msg=str(e)) from e
        except CustomException:
            raise
        except Exception as e:
            end = utc_now_iso()
            err = WorkflowExecuteResultSchema(
                workflow_id=obj.id,
                workflow_name=obj.name,
                status="failed",
                start_time=start,
                end_time=end,
                variables=variables,
                node_results=None,
                error=str(e),
            )
            return err.model_dump(mode="json")

        end = utc_now_iso()
        ok = WorkflowExecuteResultSchema(
            workflow_id=obj.id,
            workflow_name=obj.name,
            status="completed",
            start_time=start,
            end_time=end,
            variables=variables,
            node_results=raw.get("node_results"),
            error=None,
        )
        return ok.model_dump(mode="json")