# /// # 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="从原始值中提取的规则，支持正则或字符串截取" ) ]