"""按物料展开字段定义：解析共享分组引用、取字段默认值、派生前端属性面板 setters""" from pydantic import JsonValue from app.schemas.component_resource import ( ComponentCatalog, ComponentDefinition, ComponentSummary, ) 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 material_payload(component: ComponentDefinition) -> dict: """给模型看的物料配置：完整默认 schema 及使用说明，字段定义由字段查询按需提供""" return component.model_dump(mode="json", by_alias=True, exclude={"field_groups"}) def list_components(catalog: ComponentCatalog) -> dict: """组件目录摘要：选类型用，默认配置和字段定义另行查询""" return { "catalog_version": catalog.version, "components": [ ComponentSummary( key=item.key, type=item.type, name=item.name, description=item.description, default_width=item.schema_.layout.width, default_height=item.schema_.layout.height, ).model_dump(mode="json") for item in catalog.components ], } def get_materials(catalog: ComponentCatalog, component_types: list[str]) -> dict: """指定类型的物料默认配置""" known = {item.type for item in catalog.components} if unknown := [item for item in component_types if item not in known]: raise ValueError( f"没有对应的物料类型 {unknown}，可用类型：{'、'.join(sorted(known))}" ) types = set(component_types) return { "catalog_version": catalog.version, "materials": [ material_payload(item) for item in catalog.components if item.type in types ], } 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}")