364 lines
11 KiB
Python
364 lines
11 KiB
Python
# ///
|
|
# plugin_config_schema.py
|
|
# 描述：插件配置Schema系统，支持变量定义和表单生成
|
|
# 作者：User
|
|
# 创建日期：2026-04-07
|
|
# ///
|
|
|
|
from typing import Any, Dict, List, Optional, Callable
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
|
|
|
|
class FieldType(Enum):
|
|
STRING = "string"
|
|
PASSWORD = "password"
|
|
NUMBER = "number"
|
|
BOOLEAN = "boolean"
|
|
SELECT = "select"
|
|
TEXTAREA = "textarea"
|
|
JSON = "json"
|
|
CRON = "cron"
|
|
DATASOURCE_LIST = "datasource_list"
|
|
|
|
|
|
@dataclass
|
|
class ConfigField:
|
|
name: str
|
|
label: str
|
|
field_type: FieldType = FieldType.STRING
|
|
required: bool = False
|
|
default: Any = None
|
|
description: str = ""
|
|
options: List[Dict[str, str]] = field(default_factory=list)
|
|
placeholder: str = ""
|
|
min_value: Any = None
|
|
max_value: Any = None
|
|
validation: Optional[Callable] = None
|
|
depends_on: Optional[str] = None
|
|
show_when: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
@dataclass
|
|
class ConfigSection:
|
|
name: str
|
|
label: str
|
|
fields: List[ConfigField] = field(default_factory=list)
|
|
description: str = ""
|
|
|
|
|
|
class PluginConfigSchema:
|
|
def __init__(self, plugin_name: str):
|
|
self.plugin_name = plugin_name
|
|
self.sections: List[ConfigSection] = []
|
|
self.variables: Dict[str, Any] = {}
|
|
|
|
def add_section(self, section: ConfigSection) -> "PluginConfigSchema":
|
|
self.sections.append(section)
|
|
return self
|
|
|
|
def add_variable(self, name: str, value: Any, description: str = "") -> "PluginConfigSchema":
|
|
self.variables[name] = {
|
|
"value": value,
|
|
"description": description
|
|
}
|
|
return self
|
|
|
|
def to_form_schema(self) -> Dict[str, Any]:
|
|
return {
|
|
"plugin_name": self.plugin_name,
|
|
"sections": [
|
|
{
|
|
"name": s.name,
|
|
"label": s.label,
|
|
"description": s.description,
|
|
"fields": [
|
|
{
|
|
"name": f.name,
|
|
"label": f.label,
|
|
"type": f.field_type.value,
|
|
"required": f.required,
|
|
"default": f.default,
|
|
"description": f.description,
|
|
"options": f.options,
|
|
"placeholder": f.placeholder,
|
|
"min_value": f.min_value,
|
|
"max_value": f.max_value,
|
|
"depends_on": f.depends_on,
|
|
"show_when": f.show_when
|
|
}
|
|
for f in s.fields
|
|
]
|
|
}
|
|
for s in self.sections
|
|
],
|
|
"variables": self.variables
|
|
}
|
|
|
|
def validate_config(self, config: Dict[str, Any]) -> tuple:
|
|
errors = []
|
|
for section in self.sections:
|
|
for field in section.fields:
|
|
if field.required and (field.name not in config or not config[field.name]):
|
|
errors.append(f"{section.label} - {field.label} 为必填项")
|
|
if field.validation and field.name in config:
|
|
try:
|
|
field.validation(config[field.name])
|
|
except Exception as e:
|
|
errors.append(f"{section.label} - {field.label} 验证失败: {str(e)}")
|
|
return len(errors) == 0, errors
|
|
|
|
def resolve_variables(self, config: Dict[str, Any]) -> Dict[str, Any]:
|
|
resolved = config.copy()
|
|
for var_name, var_info in self.variables.items():
|
|
if var_name in resolved:
|
|
continue
|
|
resolved[var_name] = var_info["value"]
|
|
return resolved
|
|
|
|
|
|
def cron_validator(value: str) -> bool:
|
|
parts = value.strip().split()
|
|
if len(parts) != 5:
|
|
raise ValueError("Cron表达式必须有5个部分")
|
|
return True
|
|
|
|
|
|
def get_basic_config_schema() -> PluginConfigSchema:
|
|
schema = PluginConfigSchema("basic")
|
|
|
|
schema.add_section(ConfigSection(
|
|
name="basic",
|
|
label="基本设置",
|
|
fields=[
|
|
ConfigField(
|
|
name="cron",
|
|
label="执行周期",
|
|
field_type=FieldType.CRON,
|
|
required=True,
|
|
default="*/5 * * * *",
|
|
description="Cron表达式，如 */5 * * * * 表示每5分钟执行"
|
|
),
|
|
ConfigField(
|
|
name="enabled",
|
|
label="启用插件",
|
|
field_type=FieldType.BOOLEAN,
|
|
default=False,
|
|
description="是否启用此插件"
|
|
)
|
|
]
|
|
))
|
|
|
|
schema.add_section(ConfigSection(
|
|
name="node",
|
|
label="节点设置",
|
|
fields=[
|
|
ConfigField(
|
|
name="node_id",
|
|
label="节点ID",
|
|
field_type=FieldType.STRING,
|
|
required=True,
|
|
default="wx1",
|
|
placeholder="如 wx1",
|
|
description="微信节点ID"
|
|
),
|
|
ConfigField(
|
|
name="receiver",
|
|
label="默认接收人",
|
|
field_type=FieldType.STRING,
|
|
required=True,
|
|
placeholder="联系人名称",
|
|
description="默认消息接收人"
|
|
)
|
|
]
|
|
))
|
|
|
|
return schema
|
|
|
|
|
|
def get_datasource_schema() -> PluginConfigSchema:
|
|
schema = PluginConfigSchema("datasource")
|
|
|
|
schema.add_section(ConfigSection(
|
|
name="datasource",
|
|
label="数据源配置",
|
|
fields=[
|
|
ConfigField(
|
|
name="fetch_timeout",
|
|
label="请求超时(秒)",
|
|
field_type=FieldType.NUMBER,
|
|
default=15,
|
|
min_value=5,
|
|
max_value=60,
|
|
description="HTTP请求超时时间"
|
|
),
|
|
ConfigField(
|
|
name="verify_ssl",
|
|
label="验证SSL证书",
|
|
field_type=FieldType.BOOLEAN,
|
|
default=True,
|
|
description="是否验证SSL证书"
|
|
)
|
|
]
|
|
))
|
|
|
|
schema.add_section(ConfigSection(
|
|
name="http_url",
|
|
label="HTTP数据源",
|
|
fields=[
|
|
ConfigField(
|
|
name="url",
|
|
label="请求URL",
|
|
field_type=FieldType.STRING,
|
|
required=True,
|
|
placeholder="https://example.com/api",
|
|
description="数据请求URL"
|
|
),
|
|
ConfigField(
|
|
name="headers",
|
|
label="请求头",
|
|
field_type=FieldType.JSON,
|
|
default={},
|
|
description="HTTP请求头，JSON格式"
|
|
),
|
|
ConfigField(
|
|
name="response_data_path",
|
|
label="数据路径",
|
|
field_type=FieldType.STRING,
|
|
description="从响应中提取数据的路径，如 data.items"
|
|
)
|
|
]
|
|
))
|
|
|
|
return schema
|
|
|
|
|
|
def get_external_db_schema() -> PluginConfigSchema:
|
|
schema = PluginConfigSchema("external_db")
|
|
|
|
schema.add_section(ConfigSection(
|
|
name="external_db",
|
|
label="外部数据库",
|
|
fields=[
|
|
ConfigField(
|
|
name="host",
|
|
label="数据库地址",
|
|
field_type=FieldType.STRING,
|
|
required=True,
|
|
placeholder="192.168.1.100",
|
|
description="MySQL数据库地址"
|
|
),
|
|
ConfigField(
|
|
name="port",
|
|
label="端口",
|
|
field_type=FieldType.NUMBER,
|
|
default=3306,
|
|
min_value=1,
|
|
max_value=65535,
|
|
description="数据库端口"
|
|
),
|
|
ConfigField(
|
|
name="user",
|
|
label="用户名",
|
|
field_type=FieldType.STRING,
|
|
required=True,
|
|
description="数据库用户名"
|
|
),
|
|
ConfigField(
|
|
name="password",
|
|
label="密码",
|
|
field_type=FieldType.PASSWORD,
|
|
description="数据库密码"
|
|
),
|
|
ConfigField(
|
|
name="database",
|
|
label="数据库名",
|
|
field_type=FieldType.STRING,
|
|
required=True,
|
|
description="数据库名称"
|
|
),
|
|
ConfigField(
|
|
name="charset",
|
|
label="字符集",
|
|
field_type=FieldType.STRING,
|
|
default="utf8mb4",
|
|
description="数据库字符集"
|
|
)
|
|
]
|
|
))
|
|
|
|
schema.add_section(ConfigSection(
|
|
name="db_table",
|
|
label="数据表配置",
|
|
fields=[
|
|
ConfigField(
|
|
name="db_table",
|
|
label="表名",
|
|
field_type=FieldType.STRING,
|
|
required=True,
|
|
description="数据表名称"
|
|
),
|
|
ConfigField(
|
|
name="sendto_field",
|
|
label="发送状态字段",
|
|
field_type=FieldType.STRING,
|
|
default="sendto",
|
|
description="标记发送状态的字段名"
|
|
)
|
|
]
|
|
))
|
|
|
|
return schema
|
|
|
|
|
|
def create_cookie_field(label: str, name: str = "cookie", required: bool = True) -> ConfigField:
|
|
return ConfigField(
|
|
name=name,
|
|
label=label,
|
|
field_type=FieldType.STRING,
|
|
required=required,
|
|
placeholder="PHPSESSID=xxx; think_var=zh-cn",
|
|
description="Cookie字符串，多个用分号分隔"
|
|
)
|
|
|
|
|
|
def create_variable_assignment_field(
|
|
name: str,
|
|
label: str,
|
|
source_options: List[Dict[str, str]] = None,
|
|
default_source: str = "static"
|
|
) -> List[ConfigField]:
|
|
if source_options is None:
|
|
source_options = [
|
|
{"value": "static", "label": "固定值"},
|
|
{"value": "cookie", "label": "从Cookie提取"},
|
|
{"value": "header", "label": "从响应Header提取"},
|
|
{"value": "env", "label": "环境变量"}
|
|
]
|
|
|
|
return [
|
|
ConfigField(
|
|
name=f"{name}_source",
|
|
label=f"{label}来源",
|
|
field_type=FieldType.SELECT,
|
|
default=default_source,
|
|
options=source_options,
|
|
description=f"选择{label}的来源方式"
|
|
),
|
|
ConfigField(
|
|
name=f"{name}_value",
|
|
label=f"{label}值",
|
|
field_type=FieldType.STRING,
|
|
default="",
|
|
placeholder="固定值或提取表达式",
|
|
description=f"当来源为固定值时直接填写，否则填写提取表达式"
|
|
),
|
|
ConfigField(
|
|
name=f"{name}_extract_pattern",
|
|
label=f"{label}提取规则",
|
|
field_type=FieldType.STRING,
|
|
placeholder="如: key=(.*?); 或正则表达式",
|
|
description="从原始值中提取的规则，支持正则或字符串截取"
|
|
)
|
|
]
|