Kris
a48f7ebf2a
本次提交新增了通道模块的完整基础实现： 1. 搭建了channel包的顶层导出结构，整合所有核心子模块接口 2. 实现了错误处理工具类与重试退避逻辑 3. 新增UI表单/页面/字段的schema定义与自动生成工具 4. 完成上下文管理模块，支持会话上下文、可见性过滤与运行时状态注册 5. 实现通道能力配置类，支持流式传输、功能开关等多维度能力定义
163 lines
4.7 KiB
Python
163 lines
4.7 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class UIFieldSchema:
|
|
"""UI 字段定义 — 描述前端表单中单个字段的元数据。"""
|
|
|
|
name: str
|
|
type: str
|
|
label: str = ""
|
|
description: str = ""
|
|
required: bool = False
|
|
default: Any = None
|
|
placeholder: str = ""
|
|
options: list[dict[str, Any]] = field(default_factory=list)
|
|
validation: dict[str, Any] = field(default_factory=dict)
|
|
sensitive: bool = False
|
|
hidden: bool = False
|
|
disabled: bool = False
|
|
order: int = 0
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"name": self.name,
|
|
"type": self.type,
|
|
"label": self.label,
|
|
"description": self.description,
|
|
"required": self.required,
|
|
"default": self.default,
|
|
"placeholder": self.placeholder,
|
|
"options": self.options,
|
|
"validation": self.validation,
|
|
"sensitive": self.sensitive,
|
|
"hidden": self.hidden,
|
|
"disabled": self.disabled,
|
|
"order": self.order,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class UIFormSchema:
|
|
"""UI 表单定义 — 描述前端表单的整体结构。"""
|
|
|
|
id: str
|
|
title: str = ""
|
|
description: str = ""
|
|
fields: list[UIFieldSchema] = field(default_factory=list)
|
|
submit_label: str = "Save"
|
|
cancel_label: str = "Cancel"
|
|
layout: str = "vertical"
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"title": self.title,
|
|
"description": self.description,
|
|
"fields": [f.to_dict() for f in sorted(self.fields, key=lambda x: x.order)],
|
|
"submit_label": self.submit_label,
|
|
"cancel_label": self.cancel_label,
|
|
"layout": self.layout,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class UISectionSchema:
|
|
"""UI 区块定义 — 描述前端页面中的一个区块。"""
|
|
|
|
id: str
|
|
title: str = ""
|
|
description: str = ""
|
|
forms: list[UIFormSchema] = field(default_factory=list)
|
|
order: int = 0
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"title": self.title,
|
|
"description": self.description,
|
|
"forms": [f.to_dict() for f in self.forms],
|
|
"order": self.order,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class UIPageSchema:
|
|
"""UI 页面定义 — 描述前端页面的完整结构。"""
|
|
|
|
id: str
|
|
title: str = ""
|
|
description: str = ""
|
|
sections: list[UISectionSchema] = field(default_factory=list)
|
|
route: str = ""
|
|
icon: str = ""
|
|
order: int = 0
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"title": self.title,
|
|
"description": self.description,
|
|
"sections": [s.to_dict() for s in sorted(self.sections, key=lambda x: x.order)],
|
|
"route": self.route,
|
|
"icon": self.icon,
|
|
"order": self.order,
|
|
}
|
|
|
|
|
|
def build_ui_schema_from_config_schema(config_schema: dict[str, Any]) -> UIFormSchema:
|
|
"""从配置 schema 自动生成 UI 表单 schema。"""
|
|
fields: list[UIFieldSchema] = []
|
|
properties = config_schema.get("properties", {})
|
|
required = set(config_schema.get("required", []))
|
|
|
|
for name, prop in properties.items():
|
|
if not isinstance(prop, dict):
|
|
continue
|
|
|
|
field_type = _map_json_schema_type_to_ui(prop.get("type", "string"))
|
|
field = UIFieldSchema(
|
|
name=name,
|
|
type=field_type,
|
|
label=prop.get("title", name.replace("_", " ").title()),
|
|
description=prop.get("description", ""),
|
|
required=name in required,
|
|
default=prop.get("default"),
|
|
placeholder=prop.get("examples", [""])[0] if prop.get("examples") else "",
|
|
sensitive=_is_sensitive_field(name),
|
|
)
|
|
|
|
if "enum" in prop:
|
|
field.options = [{"label": str(v), "value": v} for v in prop["enum"]]
|
|
|
|
fields.append(field)
|
|
|
|
return UIFormSchema(
|
|
id=config_schema.get("title", "form"),
|
|
title=config_schema.get("title", ""),
|
|
description=config_schema.get("description", ""),
|
|
fields=fields,
|
|
)
|
|
|
|
|
|
def _map_json_schema_type_to_ui(json_type: str | list[str]) -> str:
|
|
if isinstance(json_type, list):
|
|
json_type = json_type[0] if json_type else "string"
|
|
mapping = {
|
|
"string": "text",
|
|
"integer": "number",
|
|
"number": "number",
|
|
"boolean": "switch",
|
|
"array": "list",
|
|
"object": "group",
|
|
}
|
|
return mapping.get(json_type, "text")
|
|
|
|
|
|
def _is_sensitive_field(name: str) -> bool:
|
|
sensitive_keywords = {"password", "token", "secret", "key", "api_key", "private"}
|
|
return any(keyword in name.lower() for keyword in sensitive_keywords)
|