500 lines
18 KiB
Python
500 lines
18 KiB
Python
"""按物料展开字段定义：解析共享分组引用、取字段默认值、派生前端属性面板 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 []
|