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

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

from pydantic import ValidationError

from app.actions.plan import validate_plan_actions, validate_plan_result
from app.actions.registry import apply_action
from app.actions.update import fit_existing_path
from app.processor.base import BaseNode
from app.processor.state import QueryState
from app.resources.component import get_catalog_cache
from app.resources.material_fields import check_update_field
from app.schemas.action_validation import ActionValidationError, schema_errors
from app.schemas.actions import (
    ActionBatch,
    AddAction,
    UpdateAction,
    UpdateCanvasAction,
)
from app.schemas.operations import is_canvas_operation


class ValidatorNode(BaseNode):
    name = "validator_node"

    def process(self, state: QueryState) -> QueryState:
        # 浅拷贝状态
        updated_state: QueryState = state.copy()
        # 校验失败仍可能进入修正循环，最终失败由 Fallback 标记
        updated_state["status"] = "running"
        # 获取本轮待校验的动作，先检查结构，再检查范围和模拟执行结果
        actions = state["actions"]

        try:
            # 转回字典重新校验，避免赋值后绕过字段校验
            batch = ActionBatch.model_validate(
                {"actions": [action.model_dump() for action in 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["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
        else:
            # 此分支没有规划，只校验简单修改，新增动作需要走上面的规划分支
            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

            # 将需求允许的节点操作表示为三元组：作用域、目标 ID、操作名称
            allowed = {
                ("nodes", target, change.operation)
                for change in requirement.changes
                if change.target_scope == "nodes"
                for target in change.target_ids
            }

            # 将需求中的画布更新和尺寸调整统一映射为 update_canvas 动作
            allowed.update(
                ("canvas", "", "update_canvas")
                for change in requirement.changes
                if is_canvas_operation(change.target_scope, change.operation)
            )

            # 画布没有组件 ID，使用独立作用域避免与同名组件混淆
            targets = [
                ("canvas", "", action.type)
                if isinstance(action, UpdateCanvasAction)
                else ("nodes", action.target_id, action.type)
                for action in batch.actions
            ]

            # 去重后的实际操作集合用于检查遗漏，原 targets 列表保留动作顺序和下标
            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)

        catalog = get_catalog_cache()
        # 保存补齐路径后的动作，前端按它执行
        actions = list(batch.actions)

        # 按动作顺序逐项执行，每一步都使用上一步更新后的大屏副本
        for index, action in enumerate(actions):
            # 属性更新先按物料字段定义检查，目标不存在时交给处理器报错
            node = (
                next(
                    (item for item in screen.nodes if item.id == action.target_id), None
                )
                if isinstance(action, UpdateAction)
                else None
            )
            if node is not None:
                try:
                    check_update_field(catalog, node.type, action)
                except ValueError as error:
                    updated_state["validation_passed"] = False
                    updated_state["validation_errors"] = [
                        ActionValidationError(
                            action_index=index, code="field_error", message=str(error)
                        )
                    ]
                    updated_state["simulated_screen"] = None
                    return updated_state
                # 字段定义允许、但组件中还没有的中间对象，折叠进值里再执行
                action = actions[index] = fit_existing_path(node, action)
            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"] = actions
        # 标记整批动作校验通过，并清空之前可能残留的错误
        updated_state["validation_passed"] = True
        updated_state["validation_errors"] = []
        # 保存整批动作执行后的 JSON，供后续检查，不代表前端已经完成更新
        updated_state["simulated_screen"] = screen
        # 仅表示后端动作已通过校验，前端执行结果需另行反馈
        updated_state["status"] = "completed"

        # 返回动作、校验结果和完整模拟快照，供后续输出使用
        return updated_state