tianfeng_task_modules/config/task_config.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
任务配置模块
包含任务类型、状态、输入参数等配置信息
"""

from enum import Enum
from typing import Dict, Any, List, Optional

#################################################
# 任务类型相关配置
#################################################

class TaskType(str, Enum):
    """任务类型枚举"""
    NORMAL = "NORMAL"
    SCHEDULED = "SCHEDULED"

class TaskTypeConfig:
    """任务类型配置"""
    NORMAL = TaskType.NORMAL.value
    SCHEDULED = TaskType.SCHEDULED.value
    
    # 任务类型详细信息
    DETAILS = {
        NORMAL: {
            "name": "普通任务",
            "description": "立即执行的一次性任务"
        },
        SCHEDULED: {
            "name": "定时任务",
            "description": "按照预设时间执行的任务"
        }
    }
    
    @classmethod
    def get_name(cls, task_type_key):
        """
        获取任务类型的显示名称
        
        Args:
            task_type_key (str): 任务类型的键名
            
        Returns:
            str: 任务类型的显示名称
        """
        return cls.DETAILS.get(task_type_key, {}).get("name", task_type_key)
    
    @classmethod
    def get_description(cls, task_type_key):
        """
        获取任务类型的描述
        
        Args:
            task_type_key (str): 任务类型的键名
            
        Returns:
            str: 任务类型的描述
        """
        return cls.DETAILS.get(task_type_key, {}).get("description", "")
    
    @classmethod
    def get_all(cls):
        """
        获取所有任务类型
        
        Returns:
            list: 任务类型列表，每个元素包含key、name和description
        """
        task_types = []
        for key, config in cls.DETAILS.items():
            task_types.append({
                "key": key,
                "name": config["name"],
                "description": config["description"]
            })
        return task_types

#################################################
# 任务状态相关配置
#################################################

class TaskStatus(str, Enum):
    """任务状态枚举"""
    PENDING = "PENDING"
    RUNNING = "RUNNING"
    COMPLETED = "COMPLETED"
    CANCELLED = "CANCELLED"
    FAILED = "FAILED"
    PAUSED = "PAUSED"
    WAITING = "WAITING"

class TaskStatusConfig:
    """任务状态配置"""
    PENDING = TaskStatus.PENDING.value
    RUNNING = TaskStatus.RUNNING.value
    COMPLETED = TaskStatus.COMPLETED.value
    CANCELLED = TaskStatus.CANCELLED.value
    FAILED = TaskStatus.FAILED.value
    PAUSED = TaskStatus.PAUSED.value
    WAITING = TaskStatus.WAITING.value
    
    # 任务状态详细信息
    DETAILS = {
        PENDING: {
            "name": "待执行",
            "description": "任务已创建但尚未开始执行"
        },
        RUNNING: {
            "name": "执行中",
            "description": "任务正在执行"
        },
        COMPLETED: {
            "name": "已完成",
            "description": "任务已成功完成"
        },
        CANCELLED: {
            "name": "已取消",
            "description": "任务被用户取消"
        },
        FAILED: {
            "name": "执行失败",
            "description": "任务执行过程中出现错误"
        },
        PAUSED: {
            "name": "暂停中",
            "description": "任务被暂停"
        },
        WAITING: {
            "name": "等待中",
            "description": "任务正在等待某些条件满足"
        }
    }
    
    @classmethod
    def get_name(cls, task_status_key):
        """
        获取任务状态的显示名称
        
        Args:
            task_status_key (str): 任务状态的键名
            
        Returns:
            str: 任务状态的显示名称
        """
        return cls.DETAILS.get(task_status_key, {}).get("name", task_status_key)
    
    @classmethod
    def get_description(cls, task_status_key):
        """
        获取任务状态的描述
        
        Args:
            task_status_key (str): 任务状态的键名
            
        Returns:
            str: 任务状态的描述
        """
        return cls.DETAILS.get(task_status_key, {}).get("description", "")
    
    @classmethod
    def get_all(cls):
        """
        获取所有任务状态
        
        Returns:
            list: 任务状态列表，每个元素包含key、name和description
        """
        task_statuses = []
        for key, config in cls.DETAILS.items():
            task_statuses.append({
                "key": key,
                "name": config["name"],
                "description": config["description"]
            })
        return task_statuses

#################################################
# 任务输入参数相关配置
#################################################

class TaskInputParamType(str, Enum):
    """任务输入参数类型枚举"""
    STRING = "string"
    INTEGER = "integer"
    FLOAT = "float"
    BOOLEAN = "boolean"
    DATETIME = "datetime"
    JSON = "json"
    ARRAY = "array"
    OBJECT = "object"

# 任务输入参数类型UI显示名称
PARAM_TYPE_DISPLAY_NAMES = {
    TaskInputParamType.STRING.value: "字符串",
    TaskInputParamType.INTEGER.value: "整数",
    TaskInputParamType.FLOAT.value: "浮点数", 
    TaskInputParamType.BOOLEAN.value: "布尔",
    TaskInputParamType.DATETIME.value: "日期时间",
    TaskInputParamType.JSON.value: "JSON对象",
    TaskInputParamType.ARRAY.value: "JSON数组",
    TaskInputParamType.OBJECT.value: "对象"
}

class SystemParamKey(str, Enum):
    """系统参数键名枚举"""
    TASK_ID = "task_id"
    INSTANCE_ID = "instance_id"
    TASK_NAME = "task_name"
    CREATED_AT = "created_at"
    VARIABLES = "variables"
    PRIORITY = "priority"

class TaskInputParamConfig:
    """任务输入参数配置类"""
    
    # 系统默认参数配置
    SYSTEM_PARAMS = [
        {
            "key": SystemParamKey.TASK_ID,
            "name": "任务ID",
            "type": TaskInputParamType.STRING,
            "is_system": True,
            "is_readonly": True,
            "description": "任务的唯一标识符"
        },
        {
            "key": SystemParamKey.INSTANCE_ID,
            "name": "任务实例ID",
            "type": TaskInputParamType.STRING,
            "is_system": True,
            "is_readonly": True,
            "description": "任务实例的唯一标识符"
        },
        {
            "key": SystemParamKey.TASK_NAME,
            "name": "任务名称",
            "type": TaskInputParamType.STRING,
            "is_system": True,
            "is_readonly": True,
            "description": "任务的名称"
        },
        {
            "key": SystemParamKey.CREATED_AT,
            "name": "创建时间",
            "type": TaskInputParamType.DATETIME,
            "is_system": True,
            "is_readonly": True,
            "description": "任务实例的创建时间"
        },
        {
            "key": SystemParamKey.VARIABLES,
            "name": "任务变量",
            "type": TaskInputParamType.JSON,
            "is_system": True,
            "is_readonly": False,
            "description": "任务的变量集合"
        },
        {
            "key": SystemParamKey.PRIORITY,
            "name": "任务优先级",
            "type": TaskInputParamType.INTEGER,
            "is_system": True,
            "is_readonly": False,
            "description": "任务的优先级，数值越大优先级越高"
        }
    ]
    
    @classmethod
    def get_system_params(cls) -> List[Dict[str, Any]]:
        """获取系统默认参数配置"""
        return cls.SYSTEM_PARAMS
    
    @classmethod
    def get_system_param_by_key(cls, key: str) -> Optional[Dict[str, Any]]:
        """根据键名获取系统参数配置"""
        for param in cls.SYSTEM_PARAMS:
            if param["key"] == key:
                return param
        return None
    
    @classmethod
    def build_system_params(cls, task_id: str) -> List[Dict[str, Any]]:
        """
        构建系统参数列表（不带值）
        
        Args:
            task_id: 任务ID
            
        Returns:
            系统参数列表（不带值）
        """
        params = []
        
        # 复制系统参数配置但不添加具体值
        for param_config in cls.SYSTEM_PARAMS:
            param = param_config.copy()
            
            # 只为任务ID设置值，其他参数保持不变
            if param["key"] == SystemParamKey.TASK_ID:
                param["value"] = task_id
            
            params.append(param)
        
        return params
    
    @classmethod
    def build_system_params_with_values(cls, task_id: str, instance: Dict[str, Any]) -> List[Dict[str, Any]]:
        """
        构建带有值的系统参数列表
        
        Args:
            task_id: 任务ID
            instance: 任务实例数据
            
        Returns:
            带有值的系统参数列表
        """
        params = []
        
        # 复制系统参数配置并添加值
        for param_config in cls.SYSTEM_PARAMS:
            param = param_config.copy()
            
            # 根据参数键名设置值
            if param["key"] == SystemParamKey.TASK_ID:
                param["value"] = task_id
            elif param["key"] == SystemParamKey.INSTANCE_ID:
                param["value"] = instance.get("instance_id")
            elif param["key"] == SystemParamKey.TASK_NAME:
                param["value"] = instance.get("name")
            elif param["key"] == SystemParamKey.CREATED_AT:
                param["value"] = instance.get("created_at")
            elif param["key"] == SystemParamKey.VARIABLES:
                param["value"] = instance.get("variables")
            elif param["key"] == SystemParamKey.PRIORITY:
                param["value"] = instance.get("priority", 1)
            
            params.append(param)
        
        return params
        
    @classmethod
    def get_input_params_form_fields(cls) -> List[Dict[str, Any]]:
        """
        获取任务输入参数表单字段定义
        
        Returns:
            表单字段定义列表
        """
        # 构建参数类型选项
        param_type_options = []
        for param_type in TaskInputParamType:
            param_type_options.append({
                "value": param_type.value,
                "label": PARAM_TYPE_DISPLAY_NAMES.get(param_type.value, param_type.value)
            })
            
        # 返回表单字段定义
        return [
            {
                "field_name": "param_name",
                "label": "变量名",
                "field_type": "input",
                "required": True,
                "description": "参数的唯一标识符，用于在任务执行过程中引用该参数"
            },
            {
                "field_name": "label",
                "label": "标签",
                "field_type": "input",
                "required": True,
                "description": "参数的显示名称，用于在界面上展示"
            },
            {
                "field_name": "param_type",
                "label": "类型",
                "field_type": "select",
                "required": True,
                "default": TaskInputParamType.STRING.value,
                "options": param_type_options,
                "description": "参数的数据类型"
            },
            {
                "field_name": "required",
                "label": "是否必填",
                "field_type": "checkbox",
                "required": False,
                "default": False,
                "description": "标记参数是否为必填项"
            },
            {
                "field_name": "default_value",
                "label": "默认值",
                "field_type": "input",
                "required": False,
                "description": "参数的默认值，当用户未提供值时使用"
            },
            {
                "field_name": "description",
                "label": "说明",
                "field_type": "textarea",
                "required": False,
                "description": "参数的详细说明，用于帮助用户理解参数的用途"
            }
        ]

#################################################
# 默认值配置
#################################################

class DefaultConfig:
    """默认值配置"""
    TASK_DESCRIPTION = ""  # 默认任务描述（备注）
    TEMPLATE_DESCRIPTION = "用户自有模板"  # 默认模板描述

#################################################
# 向后兼容的常量和函数
#################################################

# 任务类型和状态配置常量
TASK_TYPE_CONFIG = TaskTypeConfig.DETAILS
TASK_STATUS_CONFIG = TaskStatusConfig.DETAILS

# 任务类型常量
TASK_TYPE_NORMAL = TaskTypeConfig.NORMAL
TASK_TYPE_SCHEDULED = TaskTypeConfig.SCHEDULED

# 默认值配置
DEFAULT_TASK_DESCRIPTION = DefaultConfig.TASK_DESCRIPTION
DEFAULT_TEMPLATE_DESCRIPTION = DefaultConfig.TEMPLATE_DESCRIPTION

# 获取任务类型名称
def get_task_type_name(task_type_key):
    """
    获取任务类型的显示名称
    
    Args:
        task_type_key (str): 任务类型的键名
        
    Returns:
        str: 任务类型的显示名称
    """
    return TaskTypeConfig.get_name(task_type_key)

# 获取任务类型描述
def get_task_type_description(task_type_key):
    """
    获取任务类型的描述
    
    Args:
        task_type_key (str): 任务类型的键名
        
    Returns:
        str: 任务类型的描述
    """
    return TaskTypeConfig.get_description(task_type_key)

# 获取任务状态名称
def get_task_status_name(task_status_key):
    """
    获取任务状态的显示名称
    
    Args:
        task_status_key (str): 任务状态的键名
        
    Returns:
        str: 任务状态的显示名称
    """
    return TaskStatusConfig.get_name(task_status_key)

# 获取任务状态描述
def get_task_status_description(task_status_key):
    """
    获取任务状态的描述
    
    Args:
        task_status_key (str): 任务状态的键名
        
    Returns:
        str: 任务状态的描述
    """
    return TaskStatusConfig.get_description(task_status_key)

# 获取所有任务类型
def get_all_task_types():
    """
    获取所有任务类型
    
    Returns:
        list: 任务类型列表，每个元素包含key、name和description
    """
    return TaskTypeConfig.get_all()

# 获取所有任务状态
def get_all_task_statuses():
    """
    获取所有任务状态
    
    Returns:
        list: 任务状态列表，每个元素包含key、name和description
    """
    return TaskStatusConfig.get_all()

# 获取任务类型枚举
def get_task_type_enum(task_type_key):
    """
    获取任务类型的枚举值
    
    Args:
        task_type_key (str): 任务类型的键名
        
    Returns:
        str: 任务类型的枚举值（小写）
    """
    return task_type_key.lower()

# 获取任务状态枚举
def get_task_status_enum(task_status_key):
    """
    获取任务状态的枚举值
    
    Args:
        task_status_key (str): 任务状态的键名
        
    Returns:
        str: 任务状态的枚举值（小写）
    """
    return task_status_key.lower()