diff --git a/dwg_review/checks.py b/dwg_review/checks.py new file mode 100644 index 0000000..65f8266 --- /dev/null +++ b/dwg_review/checks.py @@ -0,0 +1,164 @@ +"""确定性检查：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 diff --git a/dwg_review/config.py b/dwg_review/config.py index 6558eef..6a68d10 100644 --- a/dwg_review/config.py +++ b/dwg_review/config.py @@ -47,6 +47,14 @@ def _read_env(name: str) -> str: return "" +def _read_bool(name: str, default: bool = True) -> bool: + """读取布尔型环境变量；未设置或空串时返回 default。""" + value = os.environ.get(name) + if value is None or value.strip() == "": + return default + return value.strip().lower() in ("1", "true", "yes", "on") + + # --------------------------------------------------------------------------- # CAD MCP 基座（TCP JSON-RPC） # --------------------------------------------------------------------------- @@ -87,6 +95,22 @@ REPORT_DIR = _resolve_path(os.environ.get("REPORT_DIR", "reports")) # 批注回写使用的图层 ANNOTATE_LAYER = os.environ.get("ANNOTATE_LAYER", "AI_REVIEW") +# --------------------------------------------------------------------------- +# 可信度三件套：确定性检查前置 / 结论结构化 / 证据交叉校验 +# --------------------------------------------------------------------------- +# 是否在 LLM 审查前跑确定性检查（零长度、重复实体、空文字、未闭合多段线） +DETERMINISTIC_CHECKS = _read_bool("DETERMINISTIC_CHECKS", True) +# 是否校验 LLM 引用的 handle 确实出现在已采集证据中（不在则剥掉 handle） +VALIDATE_HANDLES = _read_bool("VALIDATE_HANDLES", True) +# 是否检查未闭合多段线（中心线/开放轮廓多时误报，可关闭） +CHECK_UNCLOSED_POLYLINES = _read_bool("CHECK_UNCLOSED_POLYLINES", True) +# find_zero_length 的长度阈值 +ZERO_LENGTH_TOLERANCE = float(os.environ.get("ZERO_LENGTH_TOLERANCE", "0.001")) +# find_duplicate_entities 的几何量化容差 +DUP_TOLERANCE = float(os.environ.get("DUP_TOLERANCE", "0.001")) +# find_duplicate_entities 的 limit（语义是「最大重复组数」，非条目数） +DUP_GROUP_LIMIT = int(os.environ.get("DUP_GROUP_LIMIT", "50")) + # --------------------------------------------------------------------------- # 截图（视觉 token 成本控制） # --------------------------------------------------------------------------- diff --git a/dwg_review/findings.py b/dwg_review/findings.py new file mode 100644 index 0000000..8779847 --- /dev/null +++ b/dwg_review/findings.py @@ -0,0 +1,207 @@ +"""结构化结论：解析 LLM 的 JSON 输出、交叉校验 handle、装配 findings。 + +设计要点： + - `review_frame` 的结论是**受约束的 JSON**（`{summary, issues[]}`），不再是自由文本， + 因此不需要第二次 LLM 调用来「抽取」结构。 + - 解析失败时降级为 `summary=原文, issues=[]`：信息不丢，只是不进批注链路。 + - LLM 引用的 handle 若不在已采集证据中 → **保留问题、剥掉 handle**（verified=False）， + 既不会丢发现，也绝不会批注到不存在的实体。 +""" + +import json +import re + +# evidence 中带 handle 的数据源（items[].handle） +_SOURCES = ("entities", "texts", "dimensions", "blocks", "polylines") +_SEVERITIES = ("high", "medium", "low") +_FENCE = re.compile(r"```(?:json)?\s*(.*?)```", re.S) +_MISS = object() # 解析失败的哨兵（区别于合法的 JSON null） + + +def _try_load(text): + try: + return json.loads(text) + except (json.JSONDecodeError, TypeError): + return _MISS + + +def _iter_items(envelope): + """健壮迭代数据信封的 items。 + + `result_json` 在解析失败时会返回**原始字符串**，因此必须做 dict/list 守卫。 + """ + 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)] + + +def _outermost_json(text): + """抓文本中第一个 `{`/`[` 起、括号配平的整段 JSON（正确处理字符串内的括号）。""" + start = None + for i, ch in enumerate(text): + if ch in "{[": + start, open_c = i, ch + break + if start is None: + return None + close_c = "}" if open_c == "{" else "]" + depth = 0 + in_str = False + esc = False + for i in range(start, len(text)): + ch = text[i] + if in_str: + if esc: + esc = False + elif ch == "\\": + esc = True + elif ch == '"': + in_str = False + continue + if ch == '"': + in_str = True + elif ch == open_c: + depth += 1 + elif ch == close_c: + depth -= 1 + if depth == 0: + cand = text[start : i + 1] + return cand if _try_load(cand) is not _MISS else None + return None + + +def extract_json_block(text): + """从可能含围栏/前后缀的文本中抽出可解析的 JSON 字符串；抽不到返回 None。""" + t = (text or "").strip() + if not t: + return None + if _try_load(t) is not _MISS: # 整体就是 JSON + return t + m = _FENCE.search(t) + if m: + inner = m.group(1).strip() + if _try_load(inner) is not _MISS: + return inner + t = inner + return _outermost_json(t) + + +def parse_conclusion(text, *, log=print): + """把 LLM 结论解析为 {"summary": str, "issues": [dict], "raw": str}。 + + 支持：`{summary, issues}` 对象、裸 issues 数组、围栏包裹、前后夹带解释文字。 + 解析不到结构化内容时降级为 summary=原文、issues=[]。 + """ + block = extract_json_block(text) + if block is not None: + data = json.loads(block) + if isinstance(data, dict): + raw_issues = data.get("issues") + issues = ( + [i for i in raw_issues if isinstance(i, dict)] + if isinstance(raw_issues, list) + else [] + ) + summary = data.get("summary") + return { + "summary": str(summary).strip() if summary is not None else "", + "issues": issues, + "raw": text, + } + if isinstance(data, list): + return { + "summary": "", + "issues": [i for i in data if isinstance(i, dict)], + "raw": text, + } + log("[结论解析] 非结构化输出，降级为纯文本 summary") + return {"summary": (text or "").strip(), "issues": [], "raw": text} + + +def collect_known_handles(evidence, extra_handles=None): + """汇总证据里出现过的全部 handle（统一大写），外加 extra_handles。""" + known = set() + for key in _SOURCES: + for item in _iter_items((evidence or {}).get(key)): + handle = item.get("handle") + if handle: + known.add(str(handle).upper()) + for handle in extra_handles or (): + if handle: + known.add(str(handle).upper()) + return known + + +def normalize_issue(issue, *, frame_index, source, known_handles, validate_handles): + """补 frame/source/verified 并做 handle 交叉校验，返回统一 issue 结构。""" + issue = issue if isinstance(issue, dict) else {} + handle = issue.get("handle") + handle = str(handle) if handle else None + verified = True + # 确定性问题的 handle 来自工具/数据本身，恒可信；只校验 LLM 的引用 + if source != "deterministic" and handle and validate_handles: + if handle.upper() not in known_handles: + verified, handle = False, None + + severity = issue.get("severity") + if severity not in _SEVERITIES: + severity = "low" + + out = { + "frame": frame_index, + "category": str(issue.get("category") or "图面整体"), + "severity": severity, + "message": str(issue.get("message") or "").strip(), + "handle": handle, + "source": source, + "verified": verified, + } + handles = issue.get("handles") + if isinstance(handles, list) and handles: + out["handles"] = [str(h) for h in handles if h] + return out + + +def build_finding(evidence, conclusion, checks_issues=None, *, + validate_handles=True, log=print): + """把确定性检查与 LLM 结论合并、校验后装配成一个 findings 元素。""" + evidence = evidence or {} + index = evidence.get("index") + checks = checks_issues or [] + known = collect_known_handles( + evidence, extra_handles=[i.get("handle") for i in checks] + ) + + det = [ + normalize_issue(i, frame_index=index, source="deterministic", + known_handles=known, validate_handles=validate_handles) + for i in checks + ] + llm = [ + normalize_issue(i, frame_index=index, source="llm", + known_handles=known, validate_handles=validate_handles) + for i in (conclusion or {}).get("issues", []) + ] + issues = det + llm + return { + "frame": index, + "sheet": (evidence.get("frame") or {}).get("sheet_size"), + "summary": (conclusion or {}).get("summary", ""), + "issues": issues, + "counts": { + "deterministic": len(det), + "llm": len(llm), + "unverified": sum(1 for i in issues if not i["verified"]), + }, + } + + +def flatten_issues(findings): + """拍平所有图框的 issues，供 annotator / save_report 使用。""" + out = [] + for finding in findings or []: + out.extend(finding.get("issues") or []) + return out diff --git a/tests/test_checks.py b/tests/test_checks.py new file mode 100644 index 0000000..6d9a757 --- /dev/null +++ b/tests/test_checks.py @@ -0,0 +1,212 @@ +"""`dwg_review.checks` 单元测试：确定性检查前置。""" + +from dwg_review import checks +from dwg_review.checks import ( + check_duplicate_entities, + check_zero_length, + derive_from_evidence, + make_issue, + run_deterministic_checks, +) +from dwg_review.mcp_client import McpError, McpToolError +from tests.fakes import FakeMcpClient, make_ctx, make_evidence, ok_text + +QUIET = lambda *a: None # noqa: E731 +WINDOW = {"min_x": 0.0, "min_y": 0.0, "max_x": 10.0, "max_y": 10.0} + + +# --------------------------------------------------------------------------- +# make_issue +# --------------------------------------------------------------------------- +def test_make_issue_minimal(): + issue = make_issue("几何", "medium", "零长度") + assert issue == {"category": "几何", "severity": "medium", + "message": "零长度", "source": "deterministic"} + + +def test_make_issue_with_handles(): + issue = make_issue("几何", "low", "重复", handle="A", handles=["A", "B"]) + assert issue["handle"] == "A" + assert issue["handles"] == ["A", "B"] + + +def test_make_issue_omits_empty_handle(): + assert "handle" not in make_issue("文字", "low", "空文字", handle=None) + + +# --------------------------------------------------------------------------- +# check_zero_length +# --------------------------------------------------------------------------- +def test_zero_length_turns_items_into_issues(): + client = FakeMcpClient(responses={"find_zero_length": ok_text({ + "total": 2, "returned": 2, "offset": 0, + "items": [ + {"handle": "A1", "type": "LINE", "layer": "0", "length": 0.0}, + {"handle": "A2", "type": "ARC", "layer": "0", "length": 0.0001}, + ], + })}) + ctx = make_ctx(client=client) + issues = check_zero_length(ctx, WINDOW, tolerance=0.001, limit=200, log=QUIET) + assert [i["handle"] for i in issues] == ["A1", "A2"] + assert all(i["category"] == "几何" and i["severity"] == "medium" for i in issues) + + +def test_zero_length_skips_items_without_handle(): + client = FakeMcpClient(responses={"find_zero_length": ok_text( + {"items": [{"type": "LINE"}, {"handle": "B1"}]} + )}) + ctx = make_ctx(client=client) + assert [i["handle"] for i in check_zero_length(ctx, WINDOW, tolerance=0.001, + limit=200, log=QUIET)] == ["B1"] + + +def test_zero_length_passes_window_and_params(): + client = FakeMcpClient(responses={"find_zero_length": ok_text({"items": []})}) + ctx = make_ctx(client=client) + check_zero_length(ctx, WINDOW, tolerance=0.5, limit=77, log=QUIET) + assert client.arguments_of("find_zero_length") == { + "window": WINDOW, "tolerance": 0.5, "limit": 77, + } + + +def test_zero_length_swallows_tool_error(): + logs = [] + client = FakeMcpClient(responses={"find_zero_length": McpToolError("炸了")}) + ctx = make_ctx(client=client) + assert check_zero_length(ctx, WINDOW, tolerance=0.001, limit=200, log=logs.append) == [] + assert any("find_zero_length 失败" in line for line in logs) + + +def test_zero_length_tolerates_non_dict_envelope(): + client = FakeMcpClient(responses={"find_zero_length": ok_text("oops")}) + ctx = make_ctx(client=client) + assert check_zero_length(ctx, WINDOW, tolerance=0.001, limit=200, log=QUIET) == [] + + +# --------------------------------------------------------------------------- +# check_duplicate_entities +# --------------------------------------------------------------------------- +def test_duplicate_group_becomes_single_issue_with_all_handles(): + client = FakeMcpClient(responses={"find_duplicate_entities": ok_text({ + "count": 1, + "groups": [{"reason": "identical_geometry", "type": "LINE", + "handles": ["B1", "B2", "B3"]}], + })}) + ctx = make_ctx(client=client) + issues = check_duplicate_entities(ctx, WINDOW, tolerance=0.001, group_limit=50, log=QUIET) + assert len(issues) == 1 # 一组 = 一个问题 + assert issues[0]["handle"] == "B1" + assert issues[0]["handles"] == ["B1", "B2", "B3"] + assert issues[0]["severity"] == "low" + + +def test_duplicate_skips_empty_groups(): + client = FakeMcpClient(responses={"find_duplicate_entities": ok_text({ + "groups": [{"handles": []}, {"handles": None}, {"handles": ["C1", "C2"]}], + })}) + ctx = make_ctx(client=client) + issues = check_duplicate_entities(ctx, WINDOW, tolerance=0.001, group_limit=50, log=QUIET) + assert [i["handle"] for i in issues] == ["C1"] + + +def test_duplicate_passes_group_limit(): + client = FakeMcpClient(responses={"find_duplicate_entities": ok_text({"groups": []})}) + ctx = make_ctx(client=client) + check_duplicate_entities(ctx, WINDOW, tolerance=0.2, group_limit=9, log=QUIET) + assert client.arguments_of("find_duplicate_entities") == { + "window": WINDOW, "tolerance": 0.2, "limit": 9, + } + + +def test_duplicate_swallows_tool_error(): + logs = [] + client = FakeMcpClient(responses={"find_duplicate_entities": McpError(-32603, "x")}) + ctx = make_ctx(client=client) + assert check_duplicate_entities(ctx, WINDOW, tolerance=0.001, group_limit=50, + log=logs.append) == [] + assert any("find_duplicate_entities 失败" in line for line in logs) + + +# --------------------------------------------------------------------------- +# derive_from_evidence +# --------------------------------------------------------------------------- +def test_derive_flags_empty_and_whitespace_texts(): + ev = make_evidence(texts={"items": [ + {"handle": "T1", "content": ""}, + {"handle": "T2", "content": " "}, + {"handle": "T3", "content": None}, + {"handle": "T4", "content": "正常文字"}, + ]}) + issues = derive_from_evidence(ev) + assert [i["handle"] for i in issues] == ["T1", "T2", "T3"] + assert all(i["category"] == "文字" for i in issues) + + +def test_derive_flags_unclosed_polyline(): + ev = make_evidence(polylines={"items": [ + {"handle": "P1", "vertex_count": 4, "closed": False}, + {"handle": "P2", "vertex_count": 4, "closed": True}, + {"handle": "P3", "vertex_count": 2, "closed": False}, + {"handle": "P4", "closed": False}, + ]}) + issues = derive_from_evidence(ev) + assert [i["handle"] for i in issues] == ["P1"] + + +def test_derive_can_disable_unclosed_check(): + ev = make_evidence(polylines={"items": [{"handle": "P1", "vertex_count": 4, "closed": False}]}) + assert derive_from_evidence(ev, check_unclosed=False) == [] + + +def test_derive_tolerates_missing_sources(): + assert derive_from_evidence({}) == [] + assert derive_from_evidence(make_evidence()) == [] + + +# --------------------------------------------------------------------------- +# run_deterministic_checks +# --------------------------------------------------------------------------- +def build_ctx(): + client = FakeMcpClient(responses={ + "find_zero_length": ok_text({"items": [{"handle": "Z1", "type": "LINE", "length": 0.0}]}), + "find_duplicate_entities": ok_text({"groups": [ + {"type": "LINE", "handles": ["D1", "D2"]}]}), + }) + return make_ctx(client=client) + + +def test_run_combines_derived_and_mcp(): + ctx = build_ctx() + ev = make_evidence( + texts={"items": [{"handle": "T1", "content": ""}]}, + polylines={"items": [{"handle": "P1", "vertex_count": 3, "closed": False}]}, + ) + issues = run_deterministic_checks(ctx, ev, log=QUIET) + by_handle = {i.get("handle"): i for i in issues} + assert set(by_handle) == {"T1", "P1", "Z1", "D1"} + assert by_handle["D1"]["handles"] == ["D1", "D2"] + + +def test_run_passes_window_to_mcp_tools(): + ctx = build_ctx() + ev = make_evidence(bbox=(0.0, 0.0, 10.0, 10.0)) + run_deterministic_checks(ctx, ev, log=QUIET) + assert ctx.client.arguments_of("find_zero_length")["window"] == ev["window"] + assert ctx.client.arguments_of("find_duplicate_entities")["window"] == ev["window"] + + +def test_run_without_window_skips_mcp_but_keeps_derived(): + ctx = build_ctx() + ev = make_evidence() + ev["window"] = None + ev["texts"] = {"items": [{"handle": "T1", "content": ""}]} + issues = run_deterministic_checks(ctx, ev, log=QUIET) + assert [i["handle"] for i in issues] == ["T1"] + assert ctx.client.calls == [] + + +def test_run_respects_unclosed_switch(monkeypatch): + monkeypatch.setattr(checks.config, "CHECK_UNCLOSED_POLYLINES", False) + ctx = build_ctx() + ev = make_evidence(polylines={"items": [{"handle": "P1", "vertex_count": 4, "closed": False}]}) + assert [i["handle"] for i in run_deterministic_checks(ctx, ev, log=QUIET)] == ["Z1", "D1"] diff --git a/tests/test_config.py b/tests/test_config.py index 2477279..5b63875 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -27,6 +27,26 @@ def test_read_env_missing_returns_empty(monkeypatch): assert config._read_env("CAD_TEST_DEFINITELY_MISSING_XYZ") == "" +def test_read_bool_defaults_when_unset(monkeypatch): + monkeypatch.delenv("CAD_TEST_BOOL_UNSET", raising=False) + assert config._read_bool("CAD_TEST_BOOL_UNSET", True) is True + assert config._read_bool("CAD_TEST_BOOL_UNSET", False) is False + + +def test_read_bool_empty_string_uses_default(monkeypatch): + monkeypatch.setenv("CAD_TEST_BOOL_EMPTY", " ") + assert config._read_bool("CAD_TEST_BOOL_EMPTY", True) is True + + +def test_read_bool_truthy_and_falsy(monkeypatch): + for raw in ("1", "true", "TRUE", "Yes", "on"): + monkeypatch.setenv("CAD_TEST_BOOL", raw) + assert config._read_bool("CAD_TEST_BOOL", False) is True + for raw in ("0", "false", "no", "off", "nope"): + monkeypatch.setenv("CAD_TEST_BOOL", raw) + assert config._read_bool("CAD_TEST_BOOL", True) is False + + def test_has_api_key_reflects_module_value(monkeypatch): monkeypatch.setattr(config, "API_KEY", "") assert config.has_api_key() is False @@ -50,6 +70,22 @@ def test_constants_have_expected_types(): assert isinstance(config.CHECKLIST_PATH, type(config.PROJECT_ROOT)) +def test_credibility_constants_have_expected_types(): + assert isinstance(config.DETERMINISTIC_CHECKS, bool) + assert isinstance(config.VALIDATE_HANDLES, bool) + assert isinstance(config.CHECK_UNCLOSED_POLYLINES, bool) + assert isinstance(config.ZERO_LENGTH_TOLERANCE, float) + assert isinstance(config.DUP_TOLERANCE, float) + assert isinstance(config.DUP_GROUP_LIMIT, int) + + +def test_credibility_defaults_are_on(): + # 默认开启，保证默认拿到可信度收益 + assert config.DETERMINISTIC_CHECKS is True + assert config.VALIDATE_HANDLES is True + assert config.CHECK_UNCLOSED_POLYLINES is True + + def test_checklist_path_resolves_relative_to_root(): assert config.CHECKLIST_PATH.name == "review_checklist.json" assert config.CHECKLIST_PATH.parent == config.PROJECT_ROOT diff --git a/tests/test_findings.py b/tests/test_findings.py new file mode 100644 index 0000000..6faee0d --- /dev/null +++ b/tests/test_findings.py @@ -0,0 +1,241 @@ +"""`dwg_review.findings` 单元测试：结构化解析 + handle 交叉校验 + 装配。""" + +from dwg_review.findings import ( + build_finding, + collect_known_handles, + extract_json_block, + flatten_issues, + normalize_issue, + parse_conclusion, +) +from tests.fakes import make_evidence + +QUIET = lambda *a: None # noqa: E731 + + +# --------------------------------------------------------------------------- +# extract_json_block +# --------------------------------------------------------------------------- +def test_extract_bare_object(): + assert extract_json_block('{"a": 1}') == '{"a": 1}' + + +def test_extract_bare_array(): + assert extract_json_block('[{"a": 1}]') == '[{"a": 1}]' + + +def test_extract_fenced_object(): + assert extract_json_block('```json\n{"a": 1}\n```') == '{"a": 1}' + + +def test_extract_single_line_fence(): + # 旧 extract_issues 的已知瑕疵（单行围栏剥离失败）在此不再复现 + assert extract_json_block('```json{"a": 1}```') == '{"a": 1}' + + +def test_extract_from_surrounding_prose(): + text = '好的，结论如下：\n{"summary": "ok", "issues": []}\n以上。' + assert extract_json_block(text) == '{"summary": "ok", "issues": []}' + + +def test_extract_handles_braces_inside_strings(): + text = 'prefix {"message": "包含 } 和 { 的文字", "n": 1} suffix' + assert extract_json_block(text) == '{"message": "包含 } 和 { 的文字", "n": 1}' + + +def test_extract_returns_none_for_garbage(): + assert extract_json_block("未发现明显问题") is None + assert extract_json_block("") is None + assert extract_json_block(None) is None + assert extract_json_block("{ not json") is None + + +# --------------------------------------------------------------------------- +# parse_conclusion +# --------------------------------------------------------------------------- +def test_parse_object_with_issues(): + text = '{"summary": "整体尚可", "issues": [{"category": "几何", "message": "x"}]}' + out = parse_conclusion(text, log=QUIET) + assert out["summary"] == "整体尚可" + assert out["issues"] == [{"category": "几何", "message": "x"}] + assert out["raw"] == text + + +def test_parse_object_without_issues_key(): + out = parse_conclusion('{"summary": "无问题"}', log=QUIET) + assert out["summary"] == "无问题" + assert out["issues"] == [] + + +def test_parse_bare_array_becomes_issues(): + out = parse_conclusion('[{"message": "a"}, {"message": "b"}]', log=QUIET) + assert out["summary"] == "" + assert [i["message"] for i in out["issues"]] == ["a", "b"] + + +def test_parse_filters_non_dict_issues(): + out = parse_conclusion('{"summary": "s", "issues": [{"message": "ok"}, "junk", 3]}', log=QUIET) + assert out["issues"] == [{"message": "ok"}] + + +def test_parse_degrades_prose_to_summary(): + logs = [] + out = parse_conclusion("未发现明显问题。", log=logs.append) + assert out["summary"] == "未发现明显问题。" + assert out["issues"] == [] + assert any("降级" in line for line in logs) + + +def test_parse_degrades_empty_text(): + out = parse_conclusion("", log=QUIET) + assert out["summary"] == "" and out["issues"] == [] + + +# --------------------------------------------------------------------------- +# collect_known_handles +# --------------------------------------------------------------------------- +def test_collect_known_handles_from_all_sources(): + ev = make_evidence( + entities={"total": 1, "items": [{"handle": "AA"}]}, + texts={"total": 1, "items": [{"handle": "BB"}]}, + dimensions={"total": 1, "items": [{"handle": "CC"}]}, + blocks={"total": 1, "items": [{"handle": "DD"}]}, + polylines={"total": 1, "items": [{"handle": "EE"}]}, + ) + assert collect_known_handles(ev) == {"AA", "BB", "CC", "DD", "EE"} + + +def test_collect_known_handles_uppercases(): + ev = make_evidence(entities={"items": [{"handle": "2f3a"}]}) + assert collect_known_handles(ev) == {"2F3A"} + + +def test_collect_known_handles_includes_extra(): + ev = make_evidence() + assert collect_known_handles(ev, extra_handles=["Ff", None, ""]) == {"FF"} + + +def test_collect_known_handles_tolerates_non_dict_envelope(): + # result_json 解析失败会返回原始字符串 + assert collect_known_handles({"entities": "oops", "texts": None}) == set() + + +def test_collect_known_handles_skips_items_without_handle(): + ev = make_evidence(entities={"items": [{"type": "LINE"}, {"handle": ""}]}) + assert collect_known_handles(ev) == set() + + +# --------------------------------------------------------------------------- +# normalize_issue +# --------------------------------------------------------------------------- +def test_normalize_llm_issue_with_known_handle_is_verified(): + out = normalize_issue( + {"category": "文字", "severity": "high", "message": "重叠", "handle": "2f3a"}, + frame_index=1, source="llm", known_handles={"2F3A"}, validate_handles=True, + ) + assert out["handle"] == "2f3a" # 保留原样（大小写归一仅用于比对） + assert out["verified"] is True + assert out["source"] == "llm" + assert out["frame"] == 1 + + +def test_normalize_llm_issue_with_unknown_handle_strips_it(): + out = normalize_issue( + {"message": "疑似问题", "handle": "DEAD"}, + frame_index=2, source="llm", known_handles={"2F3A"}, validate_handles=True, + ) + assert out["handle"] is None + assert out["verified"] is False + assert out["message"] == "疑似问题" # 问题保留 + + +def test_normalize_deterministic_never_stripped(): + out = normalize_issue( + {"message": "零长度", "handle": "ZZZZ"}, + frame_index=1, source="deterministic", known_handles=set(), validate_handles=True, + ) + assert out["handle"] == "ZZZZ" and out["verified"] is True + + +def test_normalize_validation_disabled_keeps_handle(): + out = normalize_issue( + {"message": "x", "handle": "DEAD"}, + frame_index=1, source="llm", known_handles=set(), validate_handles=False, + ) + assert out["handle"] == "DEAD" and out["verified"] is True + + +def test_normalize_llm_issue_without_handle_is_verified(): + out = normalize_issue({"message": "整体偏空"}, frame_index=1, source="llm", + known_handles=set(), validate_handles=True) + assert out["handle"] is None and out["verified"] is True + + +def test_normalize_falls_back_on_invalid_severity_and_category(): + out = normalize_issue({"severity": "critical", "message": "x"}, + frame_index=1, source="llm", known_handles=set(), + validate_handles=True) + assert out["severity"] == "low" + assert out["category"] == "图面整体" + + +def test_normalize_preserves_handles_list(): + out = normalize_issue({"message": "重复组", "handle": "A", "handles": ["A", "B"]}, + frame_index=1, source="deterministic", known_handles=set(), + validate_handles=True) + assert out["handles"] == ["A", "B"] + + +# --------------------------------------------------------------------------- +# build_finding +# --------------------------------------------------------------------------- +def test_build_finding_merges_deterministic_then_llm(): + ev = make_evidence(index=3, entities={"items": [{"handle": "AA"}]}) + checks = [{"category": "几何", "severity": "medium", "message": "零长度", "handle": "ZZ"}] + conclusion = { + "summary": "有 2 处问题", + "issues": [ + {"category": "文字", "severity": "high", "message": "重叠", "handle": "AA"}, + {"category": "几何", "severity": "low", "message": "臆测", "handle": "NOPE"}, + ], + } + finding = build_finding(ev, conclusion, checks, validate_handles=True, log=QUIET) + assert finding["frame"] == 3 + assert finding["sheet"] == "A1" + assert finding["summary"] == "有 2 处问题" + assert [i["source"] for i in finding["issues"]] == ["deterministic", "llm", "llm"] + assert finding["counts"] == {"deterministic": 1, "llm": 2, "unverified": 1} + # 被剥掉 handle 的那条仍在 + assert finding["issues"][2]["verified"] is False + + +def test_build_finding_deterministic_handle_counts_as_known(): + # 确定性检查命中的 handle 应进入已知集合，供同框 LLM 引用 + ev = make_evidence(entities={"items": []}) + checks = [{"message": "零长度", "handle": "ZZ"}] + conclusion = {"summary": "", "issues": [{"message": "同一条", "handle": "zz"}]} + finding = build_finding(ev, conclusion, checks, validate_handles=True, log=QUIET) + assert finding["issues"][1]["verified"] is True + + +def test_build_finding_without_checks_or_conclusion(): + finding = build_finding(make_evidence(), {}, None, log=QUIET) + assert finding["issues"] == [] + assert finding["counts"] == {"deterministic": 0, "llm": 0, "unverified": 0} + + +# --------------------------------------------------------------------------- +# flatten_issues +# --------------------------------------------------------------------------- +def test_flatten_issues_concatenates(): + findings = [ + {"issues": [{"message": "a"}, {"message": "b"}]}, + {"issues": [{"message": "c"}]}, + {}, + ] + assert [i["message"] for i in flatten_issues(findings)] == ["a", "b", "c"] + + +def test_flatten_issues_empty(): + assert flatten_issues([]) == [] + assert flatten_issues(None) == []