"""按物料展开字段定义:解析共享分组引用、取字段默认值、派生前端属性面板 setters""" import re from pydantic import JsonValue from app.schemas.actions.set_interaction import INTERACTION_KEYS from app.schemas.base import JsonObject from app.schemas.component_resource import ( ComponentCatalog, ComponentDefinition, ComponentSummary, ) from app.schemas.material_field import ( MISSING, PANEL_CONTROLS, FieldGroup, MaterialField, lookup, ) from app.schemas.screen import ScreenNode 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;跟随主题的颜色为 None""" 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 if field.themed: setter["themed"] = True setters.append(setter) return setters def frontend_catalog(catalog: ComponentCatalog) -> dict: """前端物料接口的数据:在保存的定义上附加由字段定义派生的 setters 和 defaults""" 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)) item["defaults"] = structural_defaults(catalog, component) # 定义里引用共享事件组,前端仍拿展开后的数组 item["eventOptions"] = catalog.event_options(component) return data def structural_defaults( catalog: ComponentCatalog, component: ComponentDefinition ) -> JsonObject: """默认配置去掉示例内容(字段定义里标了 example 的标题、数据……)后剩下的结构默认值 前端打开大屏时用它给旧节点补上物料后来新增的字段;不用完整的默认配置,否则用户删掉的标题 会被补回成示例标题,真实大屏上又出现演示内容 """ defaults = component.template() for field in component_fields(catalog, component): if field.example: *parents, last = field.segments try: container = lookup(defaults, parents) except ValueError: continue if isinstance(container, dict) and isinstance(last, str): container.pop(last, None) return defaults # ---- 字段查询: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(catalog: ComponentCatalog, component: ComponentDefinition) -> dict: """给模型看的物料配置:完整默认 schema、可用事件及使用说明,字段定义由字段查询按需提供""" data = component.model_dump( mode="json", by_alias=True, exclude={"field_groups", "event_groups"} ) data["eventOptions"] = catalog.event_options(component) return data 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(catalog, 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", "themed", "example", "identifies", "template", }, exclude_none=True, exclude_defaults=True, ) data["default"] = field_default(component, field) if field.themed: # 默认为 null:没单独设置时画面上按画布主题取色 data["follows_theme"] = True if field.example: # 默认值是演示内容,要按实际内容写,不能当成合理的默认 data["default_is_example"] = True 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 and (top := _node_top_field(action)): raise ValueError(top) 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}") # 组件顶层字段不能用 update 改时该用什么 _TOP_FIELD_HINTS = { "layout": "位置和尺寸用 move、resize 或 set_layout", "name": "组件名称用 update 的 section=name、path=[]", } def _node_top_field(action) -> str | None: """props / style 下的路径第一段是组件的顶层字段(visibleWhen、filters、layout……)时的说明,否则为 None 模型把 filters 写成 section=props、path=["filters"],只说"没有字段 props.filters"时它不知道该换哪个动作 """ if action.section == "name" or not action.path: return None key = action.path[0] # props.style.color 这类多套了一层:props、style 本身能用 update 改,交给后面按字段名给建议 if key not in ScreenNode.model_fields or key in ("props", "style"): return None if key in INTERACTION_KEYS: hint = "显示条件、数据筛选和事件用 set_interaction 设置" else: hint = _TOP_FIELD_HINTS.get(key, "不能用 update 修改") return f"{key} 是组件的顶层字段,不在 {action.section} 下;{hint}" def component_fields( catalog: ComponentCatalog, component: ComponentDefinition ) -> list[MaterialField]: return [ field for group in resolve_field_groups(catalog, component) for field in group.fields ] def check_add_fields( catalog: ComponentCatalog, node_type: str, written: JsonValue ) -> None: """新增组件时模型显式写的字段,和 update 用同一套字段定义检查;不合法时一次列出全部问题 没写的字段来自默认配置,载入目录时已经检查过;物料没有定义的路径不检查,新增时可以整段写 option """ problems = [] for field in component_fields(catalog, find_material(catalog, node_type)): try: value = lookup(written, field.segments) except ValueError: problems.append( f"字段 {field.path}({field.label})所在的上一层结构和物料不同," "按 get_component_materials 返回的默认配置写" ) continue if value is not MISSING and (error := field.value_error(value)): problems.append(f"字段 {field.path}({field.label}){error}") if problems: raise ValueError(";".join(problems)) def retained_examples( catalog: ComponentCatalog | None, node: dict ) -> list[tuple[MaterialField, JsonValue]]: """组件上还和默认配置一样的示例内容(标题、系列名、数据……),按字段定义的 example 判断 只比较现在的值,不管是新增时没写、写了一样的值,还是之后又改回来;物料不存在时为空 有数据行时被 props.bind 覆盖的字段不算:画面上显示的是数据行里的值,留着的示例值不会出现 """ if catalog is None: return [] try: component = find_material(catalog, node["type"]) except ValueError: return [] template = component.template() bound = bound_paths(component, node) retained = [] for field in component_fields(catalog, component): if not field.example or field.path in bound: continue try: current = lookup(node, field.segments) except ValueError: continue default = lookup(template, field.segments) if current is not MISSING and current == default: retained.append((field, default)) return retained def bound_paths(component: ComponentDefinition, node: dict) -> set[str]: """props.bind 覆盖的字段路径 不看数据行是否已经写了:同一批里常先加卡片、再填 rows,中途提示"沿用了示例数据"会让模型去手写静态值; 交付时 bind 配空数据行由 interaction_problems 拦下 """ props = node.get("props") or {} bind = props.get("bind") if component.rows_path is None or not isinstance(bind, dict) or not bind: return set() return {f"props.{key}" for key in bind} # 摘要里每段文字的长度上限:文本组件的内容可能很长,认出组件只需要开头 IDENTITY_MAX_CHARS = 40 def identifying_texts(catalog: ComponentCatalog | None, node: dict) -> list[str]: """组件在画面上的名字或内容:字段定义里标了 identifies 的字段的文字,过长时截断 路径途经数组而字段按对象写时(例如 ECharts 的多个 title),每一项都取 """ if catalog is None: return [] try: component = find_material(catalog, node["type"]) except ValueError: return [] texts = [] for field in component_fields(catalog, component): if field.identifies: texts.extend( text if len(text) <= IDENTITY_MAX_CHARS else text[:IDENTITY_MAX_CHARS] + "…" for value in _identity_values(node, field) if (text := value.strip()) ) return texts def _identity_values(node: dict, field: MaterialField) -> list[str]: """识别字段的文字:字符串字段取它本身,选项数组取每项的 label""" values = values_at(node, field.segments) if field.value_type == "array": values = [ item.get("label") for value in values if isinstance(value, list) for item in value if isinstance(item, dict) ] return [value for value in values if isinstance(value, str)] # 识别字段(标题、名称、文本内容)在画面上是一行短文字。换行、超过这个长度或带句号分号时是一段话: # 线上问答之后"把结果放到大屏上",模型把"关键发现 1. …"、"注:……;……。"写进了文本组件 LABEL_MAX_CHARS = 60 SENTENCE_MARKS = re.compile(r"[。;;]") def is_label(text: str) -> bool: text = text.strip() return ( "\n" not in text and len(text) <= LABEL_MAX_CHARS and not SENTENCE_MARKS.search(text) ) def paragraph_texts( catalog: ComponentCatalog | None, node: dict, paths: set[str] | None = None ) -> list[tuple[MaterialField, str]]: """组件的识别字段里写成一段话的文字;paths 给出时只看这些字段""" if catalog is None: return [] try: component = find_material(catalog, node["type"]) except ValueError: return [] return [ (field, value) for field in component_fields(catalog, component) if field.identifies and (paths is None or field.path in paths) for value in _identity_values(node, field) if not is_label(value) ] def values_at(value: JsonValue, segments: list[str | int]) -> list[JsonValue]: """路径上的所有值:路径途经数组而字段按对象写时(ECharts 的多个 title),每一项都取""" if not segments: return [value] head, *rest = segments if isinstance(value, dict) and isinstance(head, str) and head in value: return values_at(value[head], rest) if isinstance(value, list) and isinstance(head, int) and 0 <= head < len(value): return values_at(value[head], rest) if isinstance(value, list) and isinstance(head, str): return [found for item in value for found in values_at(item, segments)] return []