my-v0-server/app/processor/nodes/validator_node.py

"""校验动作范围，再按顺序修改大屏 JSON 副本，检查整批动作能否执行"""

from pydantic import BaseModel, ValidationError

from app.actions.plan import validate_plan_actions, validate_plan_result
from app.actions.registry import apply_action
from app.processor.base import BaseNode
from app.processor.state import QueryState
from app.schemas.action_validation import ActionValidationError, schema_errors
from app.schemas.actions import ActionBatch, AddAction, UpdateCanvasAction


class ValidatorNode(BaseNode):
    name = "validator_node"

    def process(self, state: QueryState) -> QueryState:
        updated_state: QueryState = state.copy()
        actions = state["actions"]
        try:
            # 重新校验模型内容，避免赋值后绕过字段校验
            batch = ActionBatch.model_validate(
                {
                    "actions": [
                        action.model_dump() if isinstance(action, BaseModel) else action
                        for action in actions
                    ]
                    if isinstance(actions, list)
                    else actions
                }
            )
        except ValidationError as error:
            errors = schema_errors(error)
            updated_state["validation_passed"] = False
            updated_state["validation_errors"] = errors
            updated_state["simulated_screen"] = None
            return updated_state

        # 需求范围检查集中在校验节点，错误统一进入修正循环
        requirement = state.get("requirement")
        if state.get("screen_plan") is not None:
            errors = validate_plan_actions(state, batch)
            if errors:
                updated_state["validation_passed"] = False
                updated_state["validation_errors"] = errors
                updated_state["simulated_screen"] = None
                return updated_state
        elif requirement is not None:
            # 此分支没有规划，只校验简单修改，新增动作需要走上面的规划分支
            if any(isinstance(action, AddAction) for action in batch.actions):
                updated_state["validation_passed"] = False
                updated_state["validation_errors"] = [
                    ActionValidationError(
                        action_index=index,
                        code="scope_error",
                        message="新增动作尚未接入需求范围校验，当前仅支持独立模拟执行",
                    )
                    for index, action in enumerate(batch.actions)
                    if isinstance(action, AddAction)
                ]
                updated_state["simulated_screen"] = None
                return updated_state
            allowed = {
                ("nodes", target, change.operation)
                for change in requirement.changes
                if change.target_scope == "nodes"
                for target in change.target_ids
            }
            allowed.update(
                ("canvas", "", "update_canvas")
                for change in requirement.changes
                if change.target_scope == "canvas"
                and change.operation in ("update", "resize", "update_canvas")
            )
            # 画布没有组件 ID，使用独立作用域避免与同名组件混淆
            targets = [
                ("canvas", "", action.type)
                if isinstance(action, UpdateCanvasAction)
                else ("nodes", action.target_id, action.type)
                for action in batch.actions
            ]
            actual = set(targets)
            errors = [
                ActionValidationError(
                    action_index=index,
                    code="scope_error",
                    message=f"动作超出需求范围: {' / '.join(target)}",
                )
                for index, target in enumerate(targets)
                if target not in allowed
            ]
            if missing := allowed - actual:
                errors.append(
                    ActionValidationError(
                        code="scope_error",
                        message=f"动作遗漏了需求中的目标或操作: {sorted(missing)}",
                    )
                )
            if errors:
                updated_state["validation_passed"] = False
                updated_state["validation_errors"] = errors
                updated_state["simulated_screen"] = None
                return updated_state

        # 在副本上试执行，不改变用户传入的大屏，也不操作前端画布
        screen = state["input_context"].screen.model_copy(deep=True)
        for index, action in enumerate(batch.actions):
            try:
                screen = apply_action(screen, action)
            except ValueError as error:
                # 首个失败即停止，后续动作可能依赖前序结果，不返回部分成功快照
                updated_state["validation_passed"] = False
                updated_state["validation_errors"] = [
                    ActionValidationError(
                        action_index=index, code="execution_error", message=str(error)
                    )
                ]
                updated_state["simulated_screen"] = None
                return updated_state

        if state.get("screen_plan") is not None:
            errors = validate_plan_result(state, screen)
            if errors:
                updated_state["validation_passed"] = False
                updated_state["validation_errors"] = errors
                updated_state["simulated_screen"] = None
                return updated_state

        updated_state["actions"] = batch.actions
        updated_state["validation_passed"] = True
        updated_state["validation_errors"] = []
        # 保存整批动作执行后的 JSON，供后续检查，不代表前端已经完成更新
        updated_state["simulated_screen"] = screen
        return updated_state


if __name__ == "__main__":
    import json

    from app.processor.nodes.input_node import InputNode
    from app.schemas.actions import MoveAction, ResizeAction, UpdateAction

    # 1. 创建 state，不需要调用模型
    state = InputNode().process(
        {
            "request": {
                "prompt": "移动、调整尺寸并修改标题",
                "screen": {
                    "canvas": {"width": 1920, "height": 1080},
                    "nodes": [
                        {
                            "id": "sales-chart",
                            "type": "chart",
                            "layout": {"x": 100, "y": 200, "width": 400, "height": 300},
                            "props": {"title": "销售额"},
                        }
                    ],
                },
            }
        }
    )
    state["actions"] = [
        MoveAction(target_id="sales-chart", delta_x=50),
        ResizeAction(target_id="sales-chart", width=500),
        UpdateAction(
            target_id="sales-chart", section="props", path=["title"], value="月度销售额"
        ),
    ]
    # 2. 执行节点并打印完整 state
    node = ValidatorNode()
    final_state = node.process(state)
    print(
        json.dumps(
            final_state,
            ensure_ascii=False,
            indent=2,
            default=lambda value: value.model_dump(mode="json"),
        )
    )

    # 3. 在第二个动作制造错误，验证整批失败且不返回部分结果
    failed_state = {
        **final_state,
        "actions": [
            state["actions"][0],
            MoveAction(target_id="missing", delta_x=50),
        ],
    }
    failed_result = node.process(failed_state)
    print("失败校验:", failed_result["validation_passed"])
    print("错误:", [error.model_dump() for error in failed_result["validation_errors"]])
    print("失败快照:", failed_result["simulated_screen"])