97 lines
4.2 KiB
Python
97 lines
4.2 KiB
Python
"""整批模拟执行：新增组件补全物料默认配置，再按顺序在大屏副本上执行，遇到第一个错误即停止
|
||
|
||
纯函数，不依赖 agent；agent 的 apply_actions 工具和测试共用
|
||
"""
|
||
|
||
from copy import deepcopy
|
||
|
||
from app.actions.layout import EXPANDERS
|
||
from app.actions.registry import apply_action
|
||
from app.actions.update import fit_existing_path
|
||
from app.resources.material_fields import check_update_field, find_material
|
||
from app.schemas.actions import Action, AddAction, AgentAction, UpdateAction
|
||
from app.schemas.capabilities import DATA_BINDING_ERROR, DATA_BINDING_SUPPORTED
|
||
from app.schemas.component_resource import ComponentCatalog
|
||
from app.schemas.screen import Screen, ScreenNode
|
||
|
||
|
||
class BatchError(ValueError):
|
||
"""整批中第 index 个动作（从 0 开始）失败，消息里带上序号和目标，模型据此修正"""
|
||
|
||
def __init__(self, index: int, action: AgentAction, reason: str):
|
||
self.index = index
|
||
self.reason = reason
|
||
if isinstance(action, AddAction):
|
||
target = action.node.id
|
||
elif ids := getattr(action, "target_ids", None):
|
||
target = "、".join(ids)
|
||
else:
|
||
target = getattr(action, "target_id", "画布")
|
||
super().__init__(f"第 {index + 1} 个动作（{action.type} {target}）：{reason}")
|
||
|
||
|
||
def merge_config(defaults: dict, overrides: dict) -> dict:
|
||
"""对象逐层覆盖，数组和标量整体替换，不修改传入的数据"""
|
||
result = deepcopy(defaults)
|
||
for key, value in overrides.items():
|
||
result[key] = (
|
||
merge_config(result[key], value)
|
||
if isinstance(result.get(key), dict) and isinstance(value, dict)
|
||
else deepcopy(value)
|
||
)
|
||
return result
|
||
|
||
|
||
def complete_add(action: AddAction, catalog: ComponentCatalog) -> AddAction:
|
||
"""按 node.type 找物料补全新增组件，模型显式写出的字段优先
|
||
|
||
sample_data 模型写了就以模型为准（只有它知道是否换成了自己的数据），没写时按物料设置
|
||
"""
|
||
material = find_material(catalog, action.node.type)
|
||
if not DATA_BINDING_SUPPORTED and action.node.model_extra.get("dataId"):
|
||
raise ValueError(DATA_BINDING_ERROR)
|
||
overrides = action.node.model_dump(mode="json", exclude_unset=True)
|
||
node = ScreenNode.model_validate(merge_config(material.template(), overrides))
|
||
sample_data = (
|
||
action.sample_data
|
||
if "sample_data" in action.model_fields_set
|
||
else material.sample_data
|
||
)
|
||
return action.model_copy(update={"node": node, "sample_data": sample_data})
|
||
|
||
|
||
def run_batch(
|
||
screen: Screen, catalog: ComponentCatalog, actions: list[AgentAction]
|
||
) -> tuple[list[Action], Screen]:
|
||
"""从原始大屏开始按顺序模拟执行，返回前端要执行的动作和执行后的大屏；第一个错误抛 BatchError
|
||
|
||
布局意图（arrange、align、distribute、set_layout）在这里按当时的大屏展开成 move 和 resize
|
||
"""
|
||
current = screen.model_copy(deep=True)
|
||
completed: list[Action] = []
|
||
for index, action in enumerate(actions):
|
||
try:
|
||
if expander := EXPANDERS.get(action.type):
|
||
expanded = expander(current, action)
|
||
for primitive in expanded:
|
||
current = apply_action(current, primitive)
|
||
completed.extend(expanded)
|
||
continue
|
||
if isinstance(action, AddAction):
|
||
action = complete_add(action, catalog)
|
||
elif isinstance(action, UpdateAction):
|
||
# 目标不存在时交给执行器报错
|
||
node = next(
|
||
(item for item in current.nodes if item.id == action.target_id),
|
||
None,
|
||
)
|
||
if node is not None:
|
||
check_update_field(catalog, node.type, action)
|
||
# 字段定义允许、但组件中还没有的中间对象，折叠进值里再执行
|
||
action = fit_existing_path(node, action)
|
||
current = apply_action(current, action)
|
||
except ValueError as error: # 含 pydantic ValidationError
|
||
raise BatchError(index, action, str(error)) from error
|
||
completed.append(action)
|
||
return completed, current
|