payment-platform/backend/app/plugin/module_task/cronjob/node/service.py

from apscheduler.jobstores.base import JobLookupError

from app.api.v1.module_system.auth.schema import AuthSchema
from app.core.ap_scheduler import SchedulerUtil
from app.core.exceptions import CustomException
from app.utils.cron_util import CronUtil

from .crud import NodeCRUD
from .schema import (
    NodeCreateSchema,
    NodeExecuteSchema,
    NodeOutSchema,
    NodeQueryParam,
    NodeUpdateSchema,
)


class NodeService:
    """
    节点管理模块服务层

    设计原则:
    1. 节点CRUD只操作数据库，不直接操作调度器
    2. 调度器节点通过"执行"操作来创建和管理
    3. 支持预设函数和自定义代码块两种执行方式
    """

    @classmethod
    async def get_node_options_service(cls, auth: AuthSchema) -> list[dict]:
        """
        获取定时任务节点（task_node）列表，供调度/调试使用。

        工作流画布请使用 ``app.plugin.module_task.workflow.node_type`` 模块下接口（/task/workflow/node-type/options）。

        参数:
        - auth (AuthSchema): 认证信息模型

        返回:
        - list[dict]: 节点类型选项列表
        """
        obj_list = await NodeCRUD(auth).get_obj_list_crud()
        return [
            {
                "id": obj.id,
                "name": obj.name,
                "code": obj.code,
                "func": obj.func,
                "args": obj.args,
                "kwargs": obj.kwargs,
            }
            for obj in obj_list
        ]

    @classmethod
    async def get_node_detail_service(cls, auth: AuthSchema, id: int) -> dict:
        """
        获取节点详情

        参数:
        - auth (AuthSchema): 认证信息模型
        - id (int): 节点ID

        返回:
        - Dict: 节点详情字典
        """
        obj = await NodeCRUD(auth).get_obj_by_id_crud(id=id)
        return NodeOutSchema.model_validate(obj).model_dump()

    @classmethod
    async def get_node_list_service(
        cls,
        auth: AuthSchema,
        search: NodeQueryParam | None = None,
        order_by: list[dict[str, str]] | None = None,
    ) -> list[dict]:
        """
        获取节点列表

        参数:
        - auth (AuthSchema): 认证信息模型
        - search (NodeQueryParam | None): 查询参数模型
        - order_by (list[dict[str, str]] | None): 排序参数列表

        返回:
        - List[Dict]: 节点详情字典列表
        """
        obj_list = await NodeCRUD(auth).get_obj_list_crud(search=search.__dict__, order_by=order_by)
        return [NodeOutSchema.model_validate(obj).model_dump() for obj in obj_list]

    @classmethod
    async def get_node_page_service(
        cls,
        auth: AuthSchema,
        page_no: int,
        page_size: int,
        search: NodeQueryParam | None = None,
        order_by: list[dict[str, str]] | None = None,
    ) -> dict:
        """
        分页查询定时任务节点（数据库 OFFSET/LIMIT）。

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

        返回:
        - dict: 分页结果。
        """
        offset = (page_no - 1) * page_size
        return await NodeCRUD(auth).page(
            offset=offset,
            limit=page_size,
            order_by=order_by or [{"id": "asc"}],
            search=search.__dict__ if search else {},
            out_schema=NodeOutSchema,
        )

    @classmethod
    async def create_node_service(cls, auth: AuthSchema, data: NodeCreateSchema) -> dict:
        """
        创建节点 - 只保存到数据库，不创建调度器任务

        参数:
        - auth (AuthSchema): 认证信息模型
        - data (NodeCreateSchema): 节点创建模型

        返回:
        - Dict: 节点详情字典
        """
        exist_obj = await NodeCRUD(auth).get(name=data.name)
        if exist_obj:
            raise CustomException(msg="创建失败，该节点已存在")

        obj = await NodeCRUD(auth).create_obj_crud(data=data)
        if not obj:
            raise CustomException(msg="创建失败")
        return NodeOutSchema.model_validate(obj).model_dump()

    @classmethod
    async def update_node_service(cls, auth: AuthSchema, id: int, data: NodeUpdateSchema) -> dict:
        """
        更新节点 - 只更新数据库，不修改调度器任务

        参数:
        - auth (AuthSchema): 认证信息模型
        - id (int): 节点ID
        - data (NodeUpdateSchema): 节点更新模型

        返回:
        - dict: 节点详情字典
        """
        exist_obj = await NodeCRUD(auth).get_obj_by_id_crud(id=id)
        if not exist_obj:
            raise CustomException(msg="更新失败，该节点不存在")

        obj = await NodeCRUD(auth).update_obj_crud(id=id, data=data)
        if not obj:
            raise CustomException(msg="更新失败")
        return NodeOutSchema.model_validate(obj).model_dump()

    @classmethod
    async def delete_node_service(cls, auth: AuthSchema, ids: list[int]) -> None:
        """
        删除节点 - 只删除数据库记录，同时移除调度器中的任务

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

        返回:
        - None
        """
        if len(ids) < 1:
            raise CustomException(msg="删除失败，删除对象不能为空")
        for id in ids:
            exist_obj = await NodeCRUD(auth).get_obj_by_id_crud(id=id)
            if not exist_obj:
                raise CustomException(msg="删除失败，该节点不存在")
            try:
                SchedulerUtil.remove_job(job_id=id)
            except JobLookupError:
                # 作业不存在，忽略异常，继续删除数据库记录
                pass
        await NodeCRUD(auth).delete_obj_crud(ids=ids)

    @classmethod
    async def clear_node_service(cls, auth: AuthSchema) -> None:
        """
        清空所有节点

        参数:
        - auth (AuthSchema): 认证信息模型

        返回:
        - None
        """
        SchedulerUtil.clear_jobs()
        await NodeCRUD(auth).clear_obj_crud()

    @classmethod
    async def execute_node_service(cls, auth: AuthSchema, id: int, execute_data: NodeExecuteSchema) -> dict:
        """
        调试节点 - 根据任务配置创建调度器任务并执行

        参数:
        - auth (AuthSchema): 认证信息模型
        - id (int): 节点ID
        - execute_data (NodeExecuteSchema): 执行参数模型

        返回:
        - dict: 调试结果
        """
        obj = await NodeCRUD(auth).get_obj_by_id_crud(id=id)
        if not obj:
            raise CustomException(msg="调试失败，该节点不存在")

        trigger = execute_data.trigger
        trigger_args = execute_data.trigger_args
        start_date = execute_data.start_date
        end_date = execute_data.end_date

        if trigger == "now":
            SchedulerUtil.add_and_run_job_now(job_info=obj)
        elif trigger == "cron":
            if not trigger_args:
                raise CustomException(msg="Cron执行需要提供Cron表达式")
            if not CronUtil.validate_cron_expression(trigger_args):
                raise CustomException(msg=f"Cron表达式不正确: {trigger_args}")
            SchedulerUtil.add_cron_job(
                job_info=obj,
                trigger_args=trigger_args,
                start_date=start_date,
                end_date=end_date,
            )
        elif trigger == "interval":
            if not trigger_args:
                raise CustomException(msg="间隔执行需要提供间隔参数")
            SchedulerUtil.add_interval_job(
                job_info=obj,
                trigger_args=trigger_args,
                start_date=start_date,
                end_date=end_date,
            )
        elif trigger == "date":
            if not trigger_args:
                raise CustomException(msg="指定时间执行需要提供执行时间")
            SchedulerUtil.add_date_job(job_info=obj, run_date=trigger_args)
        else:
            raise CustomException(msg=f"不支持的触发方式: {trigger}")

        return {"job_id": id, "status": "executed", "trigger": trigger}