my-v0-server/app/actions/plan.py

"""将物料默认配置补入新增动作，并检查动作是否符合规划，不执行画布更新"""

from copy import deepcopy
from math import isclose

from app.processor.state import QueryState
from app.schemas.action_validation import ActionValidationError
from app.schemas.actions import (
    ActionBatch,
    AddAction,
    DeleteAction,
    MoveAction,
    ResizeAction,
    UpdateAction,
)
from app.schemas.screen import Screen, ScreenNode


def plan_block_reason(state: QueryState) -> str | None:
    """检查规划是否在当前支持范围内，不支持时返回原因"""
    plan = state["screen_plan"]
    resources = state.get("component_resources")
    if state["requirement"].needs_clarification:
        return "需求尚未澄清，不能生成规划动作"
    if plan.canvas_changes:
        return "规划中的画布修改尚未接入，当前仅支持无需规划的画布修改"
    if plan.resource_needs:
        return "规划仍有待获取资源，不能直接生成动作"
    additions = [item for item in plan.components if item.disposition == "add"]
    removals = {
        item.target_id for item in plan.components if item.disposition == "remove"
    }
    modifications = {
        item.target_id: item for item in plan.components if item.disposition == "modify"
    }
    if not additions and not modifications and not removals:
        return "规划未包含可执行的新增、调整或删除组件"
    if additions and (resources is None or resources.status != "resolved"):
        return "组件目录尚未就绪"
    supported = {item.type for item in resources.components} if resources else set()
    if any(item.component_type not in supported for item in additions):
        return "规划中的新增类型未在组件目录中找到"
    nodes = {node.id: node for node in state["input_context"].screen.nodes}
    for item in plan.components:
        if item.disposition == "add":
            continue
        if item.target_id not in nodes:
            return f"规划目标不存在: {item.target_id}"
        if (
            item.component_type is not None
            and item.component_type != nodes[item.target_id].type
        ):
            return "当前不支持更换已有组件类型"
    # 已有目标的修改诉求不能通过 keep 或遗漏目标绕过
    supported_operations = {"move", "resize", "update", "restyle", "reflow", "layout"}
    changes = state["requirement"].changes
    # 删除必须逐个目标获得明确授权，整体布局调整不隐含删除权限
    allowed_removals = {
        target
        for change in changes
        if change.target_scope == "nodes" and change.operation == "delete"
        for target in change.target_ids
    }
    if removals != allowed_removals:
        return "规划删除目标与需求不一致，不能遗漏或扩大删除范围"
    for change in changes:
        if change.target_scope == "screen" and change.operation == "add":
            if not additions:
                return "规划遗漏新增需求"
        elif change.target_scope == "nodes" and change.operation == "delete":
            continue
        elif (
            change.target_scope in {"nodes", "screen"}
            and change.operation in supported_operations
        ):
            if not modifications or any(
                target not in modifications for target in change.target_ids
            ):
                return "规划遗漏需要调整的组件"
        else:
            return "需求包含未支持的操作，当前不能只执行部分规划"
    if (
        modifications
        and changes
        and not any(
            change.target_scope == "screen" and change.operation in supported_operations
            for change in changes
        )
    ):
        allowed = {target for change in changes for target in change.target_ids}
        if set(modifications) - allowed:
            return "规划调整了需求范围外的组件"
    return None


def plan_materials(state: QueryState) -> list[dict]:
    """提取规划需要的完整物料配置，供动作生成和修正使用"""
    types = {
        item.component_type
        for item in state["screen_plan"].components
        if item.disposition == "add"
    }
    if not types:
        return []
    return [
        item.model_dump(mode="json", by_alias=True)
        for item in state["component_resources"].components
        if item.type in types
    ]


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 assemble_plan_actions(state: QueryState, batch: ActionBatch) -> ActionBatch:
    """用物料默认配置补全新增节点，模型显式提供的字段优先"""
    resources = state.get("component_resources")
    materials = {item.key: item for item in resources.components} if resources else {}
    result = batch.model_copy(deep=True)
    for action in result.actions:
        if not isinstance(action, AddAction) or action.component_key not in materials:
            continue
        material = materials[action.component_key]
        node = merge_config(
            material.schema_.model_dump(mode="json"),
            action.node.model_dump(mode="json", exclude_unset=True),
        )
        action.node = ScreenNode.model_validate(node)
        # 保守标记物料含示例数据，不能由模型声称已取得真实数据
        action.sample_data = material.sample_data
    return result


def validate_plan_actions(
    state: QueryState, batch: ActionBatch
) -> list[ActionValidationError]:
    """检查动作是否覆盖规划，禁止修改 keep 或未提及的组件"""
    if reason := plan_block_reason(state):
        return [ActionValidationError(code="scope_error", message=reason)]
    expected = {
        item.plan_id: item
        for item in state["screen_plan"].components
        if item.disposition == "add"
    }
    modifications = {
        item.target_id: item
        for item in state["screen_plan"].components
        if item.disposition == "modify"
    }
    resources = state.get("component_resources")
    materials = {item.key: item for item in resources.components} if resources else {}
    seen = set()
    modified = set()
    removals = {
        item.target_id
        for item in state["screen_plan"].components
        if item.disposition == "remove"
    }
    removed = set()
    errors = []
    for index, action in enumerate(batch.actions):
        message = None
        if isinstance(action, DeleteAction):
            if action.target_id not in removals:
                message = f"不能删除非 remove 规划目标: {action.target_id}"
            elif action.target_id in removed:
                message = f"不能重复删除组件: {action.target_id}"
            else:
                removed.add(action.target_id)
        elif not isinstance(action, AddAction):
            if not isinstance(action, (MoveAction, ResizeAction, UpdateAction)):
                message = "规划调整仅支持 move、resize 和 update"
            elif action.target_id not in modifications:
                message = f"不能修改保留或未提及的组件: {action.target_id}"
            elif (
                isinstance(action, UpdateAction)
                and not modifications[action.target_id].design
            ):
                message = f"规划未要求修改属性: {action.target_id}"
            else:
                modified.add(action.target_id)
        elif action.plan_id not in expected:
            message = f"动作未关联有效的新增规划: {action.plan_id}"
        elif action.plan_id in seen:
            message = f"同一规划组件不能重复新增: {action.plan_id}"
        elif action.node.id in {
            node.id for node in state["input_context"].screen.nodes
        }:
            message = f"新增组件不能复用原有 ID: {action.node.id}"
        else:
            seen.add(action.plan_id)
            item = expected[action.plan_id]
            material = materials.get(action.component_key)
            if (
                material is None
                or material.type != item.component_type
                or action.node.type != item.component_type
            ):
                message = "新增节点类型、物料和规划类型不一致"
            elif item.layout is not None and action.node.layout != item.layout:
                message = f"新增布局与规划不一致: {item.plan_id}"
            elif action.sample_data != material.sample_data:
                message = "示例数据标记与物料不一致"
            elif action.node.model_extra.get("dataId"):
                message = "当前尚未接入数据绑定，不能生成 dataId"
        if message:
            errors.append(
                ActionValidationError(
                    action_index=index, code="scope_error", message=message
                )
            )
    if missing := set(expected) - seen:
        errors.append(
            ActionValidationError(
                code="scope_error", message=f"遗漏新增规划组件: {sorted(missing)}"
            )
        )
    if missing := set(modifications) - modified:
        errors.append(
            ActionValidationError(
                code="scope_error", message=f"遗漏调整规划组件: {sorted(missing)}"
            )
        )
    if missing := removals - removed:
        errors.append(
            ActionValidationError(
                code="scope_error", message=f"遗漏删除规划组件: {sorted(missing)}"
            )
        )
    return errors


def validate_plan_result(
    state: QueryState, screen: Screen
) -> list[ActionValidationError]:
    """整批执行后检查已有组件的最终布局，避免动作合法但没有达到规划位置"""
    nodes = {node.id: node for node in screen.nodes}
    return [
        ActionValidationError(
            code="scope_error", message=f"执行后的布局与规划不一致: {item.target_id}"
        )
        for item in state["screen_plan"].components
        if item.disposition == "modify"
        and item.layout is not None
        and not all(
            isclose(
                getattr(nodes[item.target_id].layout, key),
                value,
                rel_tol=0,
                abs_tol=1e-9,
            )
            if key in {"x", "y", "width", "height"}
            else nodes[item.target_id].layout.model_dump().get(key) == value
            for key, value in item.layout.model_dump().items()
        )
    ]