207 lines
7.7 KiB
Python
207 lines
7.7 KiB
Python
"""按物料展开字段定义：解析共享分组引用、取字段默认值、派生前端属性面板 setters"""
|
||
|
||
from pydantic import JsonValue
|
||
|
||
from app.schemas.component_resource import ComponentCatalog, ComponentDefinition
|
||
from app.schemas.material_field import (
|
||
MISSING,
|
||
PANEL_CONTROLS,
|
||
FieldGroup,
|
||
MaterialField,
|
||
lookup,
|
||
)
|
||
|
||
|
||
def resolve_field_groups(
|
||
catalog: ComponentCatalog, component: ComponentDefinition
|
||
) -> list[FieldGroup]:
|
||
"""按物料声明的顺序展开字段分组，共享分组取目录中的定义；返回副本，不污染目录"""
|
||
shared = {group.key: group for group in catalog.field_groups}
|
||
return [
|
||
(shared[group] if isinstance(group, str) else group).model_copy(deep=True)
|
||
for group in component.field_groups
|
||
]
|
||
|
||
|
||
def field_default(component: ComponentDefinition, field: MaterialField) -> JsonValue:
|
||
"""物料默认配置中有该路径时取默认配置的值，否则取字段声明的 default"""
|
||
value = lookup(component.template(), field.segments)
|
||
return field.default if value is MISSING else value
|
||
|
||
|
||
def panel_setters(groups: list[FieldGroup]) -> list[dict]:
|
||
"""派生前端属性面板的 setters，格式与前端 SetterSchema 一致"""
|
||
setters = []
|
||
for group in groups:
|
||
for field in group.fields:
|
||
if not field.panel:
|
||
continue
|
||
props = {}
|
||
if field.options:
|
||
props["options"] = [
|
||
{"label": option.label, "value": option.value} for option in field.options
|
||
]
|
||
setter = {
|
||
"type": PANEL_CONTROLS[field.value_type],
|
||
"label": field.label,
|
||
"key": field.path,
|
||
"props": props,
|
||
}
|
||
if field.span is not None:
|
||
setter["span"] = field.span
|
||
setters.append(setter)
|
||
return setters
|
||
|
||
|
||
def frontend_catalog(catalog: ComponentCatalog) -> dict:
|
||
"""前端物料接口的数据：在保存的定义上附加由字段定义派生的 setters"""
|
||
data = catalog.model_dump(mode="json", by_alias=True)
|
||
for item, component in zip(data["components"], catalog.components, strict=True):
|
||
item["setters"] = panel_setters(resolve_field_groups(catalog, component))
|
||
return data
|
||
|
||
|
||
# ---- 字段查询：MCP 工具、动作提示词和动作校验共用同一套逻辑 ----
|
||
|
||
|
||
def find_material(catalog: ComponentCatalog, component_type: str) -> ComponentDefinition:
|
||
"""按节点 type 找物料；找不到或对应多个物料时报错，不猜测"""
|
||
matches = [item for item in catalog.components if item.type == component_type]
|
||
if len(matches) != 1:
|
||
known = "、".join(item.type for item in catalog.components)
|
||
reason = "没有对应的物料" if not matches else "对应多个物料"
|
||
raise ValueError(f"组件类型 {component_type} {reason}，可用类型：{known}")
|
||
return matches[0]
|
||
|
||
|
||
def describe_field(
|
||
component: ComponentDefinition, group: FieldGroup, field: MaterialField
|
||
) -> dict:
|
||
"""给模型看的字段说明：省略空值和属性面板专用的信息"""
|
||
data = field.model_dump(
|
||
mode="json",
|
||
exclude={"default", "panel", "span", "editable"},
|
||
exclude_none=True,
|
||
exclude_defaults=True,
|
||
)
|
||
data["default"] = field_default(component, field)
|
||
if not field.editable:
|
||
data["editable"] = False
|
||
data["group"] = group.key
|
||
return data
|
||
|
||
|
||
def list_field_groups(catalog: ComponentCatalog, component_type: str) -> dict:
|
||
"""物料的字段分组目录：每组的说明和字段名称，模型据此决定查看哪几组"""
|
||
component = find_material(catalog, component_type)
|
||
return {
|
||
"component_type": component.type,
|
||
"name": component.name,
|
||
"catalog_version": catalog.version,
|
||
"groups": [
|
||
{
|
||
"key": group.key,
|
||
"label": group.label,
|
||
"description": group.description,
|
||
"fields": [field.label for field in group.fields],
|
||
}
|
||
for group in resolve_field_groups(catalog, component)
|
||
],
|
||
}
|
||
|
||
|
||
def get_fields(
|
||
catalog: ComponentCatalog, component_type: str, group_keys: list[str]
|
||
) -> dict:
|
||
"""指定分组的完整字段定义"""
|
||
component = find_material(catalog, component_type)
|
||
groups = {group.key: group for group in resolve_field_groups(catalog, component)}
|
||
if unknown := [key for key in group_keys if key not in groups]:
|
||
raise ValueError(
|
||
f"{component.type} 没有字段分组 {unknown}，可用分组：{list(groups)}"
|
||
)
|
||
return {
|
||
"component_type": component.type,
|
||
"catalog_version": catalog.version,
|
||
"fields": [
|
||
describe_field(component, groups[key], field)
|
||
for key in dict.fromkeys(group_keys)
|
||
for field in groups[key].fields
|
||
],
|
||
}
|
||
|
||
|
||
def _match_score(field: MaterialField, query: str) -> int:
|
||
"""名称、别名完全一致最相关，其次是互相包含、路径末段一致，最后是说明中出现"""
|
||
names = [name.casefold() for name in (field.label, *field.aliases)]
|
||
if query in names:
|
||
return 4
|
||
if any(query in name or name in query for name in names):
|
||
return 3
|
||
if query == field.path.rsplit(".", 1)[-1].casefold():
|
||
return 2
|
||
if query in field.description.casefold() or query in field.path.casefold():
|
||
return 1
|
||
return 0
|
||
|
||
|
||
def search_fields(
|
||
catalog: ComponentCatalog, component_type: str, query: str, limit: int = 8
|
||
) -> dict:
|
||
"""按名称、别名、路径和说明搜索字段，按相关度返回"""
|
||
component = find_material(catalog, component_type)
|
||
keyword = query.strip().casefold()
|
||
if not keyword:
|
||
raise ValueError("搜索内容不能为空")
|
||
scored = [
|
||
(score, index, group, field)
|
||
for index, (group, field) in enumerate(
|
||
(group, field)
|
||
for group in resolve_field_groups(catalog, component)
|
||
for field in group.fields
|
||
)
|
||
if (score := _match_score(field, keyword))
|
||
]
|
||
scored.sort(key=lambda item: (-item[0], item[1]))
|
||
return {
|
||
"component_type": component.type,
|
||
"catalog_version": catalog.version,
|
||
"query": query,
|
||
"fields": [
|
||
describe_field(component, group, field)
|
||
for _, _, group, field in scored[:limit]
|
||
],
|
||
}
|
||
|
||
|
||
def check_update_field(catalog: ComponentCatalog | None, node_type: str, action) -> None:
|
||
"""update 只能修改字段定义中的可编辑字段，值须符合字段类型和取值约束
|
||
|
||
不合法时抛 ValueError，路径不存在时附上可能对应的字段供修正参考
|
||
"""
|
||
if catalog is None:
|
||
raise ValueError("物料目录未载入，无法确认组件有哪些字段")
|
||
component = find_material(catalog, node_type)
|
||
fields = {
|
||
field.path: field
|
||
for group in resolve_field_groups(catalog, component)
|
||
for field in group.fields
|
||
}
|
||
path = action.field_path
|
||
field = fields.get(path)
|
||
if field is None:
|
||
message = f"{node_type} 没有字段 {path}，只能修改字段定义中的路径"
|
||
keyword = next(
|
||
(str(item) for item in reversed(action.path) if isinstance(item, str)),
|
||
action.section,
|
||
)
|
||
if similar := search_fields(catalog, node_type, keyword, limit=3)["fields"]:
|
||
message += "；可能是：" + "、".join(
|
||
f"{item['label']}（{item['path']}）" for item in similar
|
||
)
|
||
raise ValueError(message)
|
||
if not field.editable:
|
||
raise ValueError(f"字段 {path}（{field.label}）不允许修改")
|
||
if error := field.value_error(action.value):
|
||
raise ValueError(f"字段 {path}（{field.label}）{error}")
|