cad-agent/dwg_review/checks.py

"""确定性检查：MCP 工具（find_zero_length / find_duplicate_entities）+ 纯 Python 派生。

产出的 issue 带 `source="deterministic"`，其 handle 来自工具/数据本身，
因此被视为**可信**，可直接进入批注链路，不依赖 LLM 回忆。

注意：`find_zero_length` / `find_duplicate_entities` 的 `window` 参数在 C++ 侧
（`parseCommonQuery`）**实际生效**，只是 JSON schema 未声明——Python 侧显式传参即可。
"""

from . import config
from .mcp_client import McpError, McpToolError, result_json


def make_issue(category, severity, message, *, handle=None, handles=None,
               source="deterministic"):
    """构造统一 issue（`verified` 由 findings 层补齐）。"""
    issue = {
        "category": category,
        "severity": severity,
        "message": message,
        "source": source,
    }
    if handle:
        issue["handle"] = handle
    if handles:
        issue["handles"] = list(handles)
    return issue


def _iter_items(envelope):
    """健壮迭代 items（`result_json` 解析失败会返回原始字符串）。"""
    if not isinstance(envelope, dict):
        return []
    items = envelope.get("items")
    if not isinstance(items, list):
        return []
    return [i for i in items if isinstance(i, dict)]


# ---------------------------------------------------------------------------
#  MCP 检查
# ---------------------------------------------------------------------------
def check_zero_length(ctx, window, *, tolerance, limit, log=print):
    """find_zero_length(window) → 逐条零长度实体转 issue。"""
    try:
        resp = ctx.tools.call(
            "find_zero_length",
            {"window": window, "tolerance": tolerance, "limit": limit},
        )
    except (McpError, McpToolError) as exc:
        log(f"      [确定性] find_zero_length 失败: {exc}")
        return []

    out = []
    for item in _iter_items(result_json(resp, {})):
        handle = item.get("handle")
        if not handle:
            continue
        out.append(
            make_issue(
                "几何", "medium",
                f"零长度/退化实体（类型 {item.get('type')}，长度 {item.get('length')}）",
                handle=handle,
            )
        )
    return out


def check_duplicate_entities(ctx, window, *, tolerance, group_limit, log=print):
    """find_duplicate_entities(window) → 每个重复组一条 issue。

    `limit` 的语义是「最大重复组数」（非条目数）。一组重复是一个问题，
    故只产出一条 issue，主 handle 取组内首个，全部 handle 放进 `handles`。
    """
    try:
        resp = ctx.tools.call(
            "find_duplicate_entities",
            {"window": window, "tolerance": tolerance, "limit": group_limit},
        )
    except (McpError, McpToolError) as exc:
        log(f"      [确定性] find_duplicate_entities 失败: {exc}")
        return []

    env = result_json(resp, {})
    groups = env.get("groups") if isinstance(env, dict) else None
    out = []
    for group in groups or []:
        if not isinstance(group, dict):
            continue
        handles = [h for h in (group.get("handles") or []) if h]
        if not handles:
            continue
        out.append(
            make_issue(
                "几何", "low",
                f"重复/重叠实体组（类型 {group.get('type')}，共 {len(handles)} 个几何相同）",
                handle=handles[0],
                handles=handles,
            )
        )
    return out


# ---------------------------------------------------------------------------
#  纯 Python 派生（零额外 MCP 往返）
# ---------------------------------------------------------------------------
def _empty_texts(evidence):
    out = []
    for item in _iter_items(evidence.get("texts")):
        content = item.get("content")
        if content is None or str(content).strip() == "":
            out.append(
                make_issue("文字", "medium", "存在空文字或仅含空格的文字",
                           handle=item.get("handle"))
            )
    return out


def _unclosed_polylines(evidence):
    out = []
    for item in _iter_items(evidence.get("polylines")):
        if item.get("vertex_count", 0) >= 3 and item.get("closed") is False:
            out.append(
                make_issue("几何", "low",
                           "多段线顶点≥3但未闭合，可能是应当闭合的轮廓",
                           handle=item.get("handle"))
            )
    return out


def derive_from_evidence(evidence, *, check_unclosed=True):
    """从已采集证据里做零成本派生检查。"""
    issues = _empty_texts(evidence)
    if check_unclosed:
        issues += _unclosed_polylines(evidence)
    return issues


# ---------------------------------------------------------------------------
#  入口
# ---------------------------------------------------------------------------
def run_deterministic_checks(ctx, evidence, *, log=print):
    """对单个图框跑全部确定性检查，返回 issue 列表。

    任何单项 MCP 失败都只记录并跳过，不抛出（不拖垮该图框）。
    """
    issues = derive_from_evidence(
        evidence, check_unclosed=config.CHECK_UNCLOSED_POLYLINES
    )
    window = evidence.get("window")
    if window:
        issues += check_zero_length(
            ctx, window,
            tolerance=config.ZERO_LENGTH_TOLERANCE,
            limit=config.FRAME_ENTITY_LIMIT,
            log=log,
        )
        issues += check_duplicate_entities(
            ctx, window,
            tolerance=config.DUP_TOLERANCE,
            group_limit=config.DUP_GROUP_LIMIT,
            log=log,
        )
    return issues