diff --git a/CODEBUDDY.md b/CODEBUDDY.md index ff4376f..3b61cb8 100644 --- a/CODEBUDDY.md +++ b/CODEBUDDY.md @@ -40,9 +40,11 @@ python -m pytest # 期望全绿 配置见 `pytest.ini`（`testpaths = tests`）与根 `conftest.py`（把仓库根加入 `sys.path`， 并提供 `RENDER_WAIT=0` / `REPORT_DIR=tmp` 两个 fixture）。**注意**：`dwg_review.config` 的常量在导入期定值，测试改 `os.environ` 无效，须 `monkeypatch.setattr(config, ...)`。 +`tests/test_sources_pdf.py` 里的 `_build_pdf()` 手工构造最小 PDF，PDF 路径无需样本文件即可回归。 -**慢层 —— 端到端冒烟脚本**（仓库根 `verify_*.py`，**必须先启动已加载插件的 CAD，并确认 -8080 端口在监听**）： +**慢层 —— 端到端冒烟脚本**（仓库根 `verify_*.py`）： + +需要**已启动已加载插件的 CAD、8080 端口在监听**（`verify_pdf_source.py` 除外）： ```bash python verify_mcp_frame.py # 基座最小集：分帧 / id / isError 映射 @@ -57,6 +59,14 @@ python verify_collisions.py # find_collisions：全模型/逐图框粗 这些脚本通过 TCP 连接 `127.0.0.1:8080` 收发 JSON-RPC。 +**离线源验收（PDF / 图片，不需要 CAD）**： + +```bash +python verify_pdf_source.py "E:/path/图纸.pdf" # 离线：页序 / evidence 同形 / 瓦片字高实测 / 旋转映射自检 +python verify_pdf_source.py "E:/path/scan.png" --dump out/c1 +python verify_pdf_source.py "E:/path/图纸.pdf" --llm # 追加真实 python -m dwg_review 端到端 +``` + ## 架构 ### 两个构建目标 @@ -86,6 +96,33 @@ python verify_collisions.py # find_collisions：全模型/逐图框粗 第三方工具只需继承 `mcp_Tool`，实现纯虚 `Execute(args)`，并用 `EXPORT_MCP_TOOL(ToolClass, FactoryName)` 宏导出工厂 + 销毁函数。`mcp_Tool_Utility` 提供 `make_text_result` / `make_image_result` / `make_mixed_result` / `make_error` 等返回体构造器。`mcp_plugins.cpp` 中的 `mcp_Tool_ViewportScreenshot`（GDI+ 抓屏 → PNG → Base64 → 多模态返回）是完整参考实现。 +### Python 侧：图纸审查 agent（`dwg_review/`） + +CLI：`python -m dwg_review [source] [options]`。**来源决定运行模式**： + +| `source` | 模式 | 取证方式 | +| --- | --- | --- | +| 省略 / `.dwg` / `.dxf` | **CAD 在线** | 连 8080，逐图框调 MCP 工具 + 截屏 | +| `.pdf` / 常见图片格式 | **完全离线**（不连 CAD） | `sources.py` 内存渲染 + 墨迹瓦片 | + +四阶段流水线：`phases/planning`（识别图框 / 分页）→ `phases/evidence`（取证）→ +`phases/review`（LLM 审查 + 受控只读取证子循环）→ `phases/report`（汇总落盘）。 + +- **统一证据模型** `evidence_model.py`：`source_type` + `images:[{b64,mime,label,window}]`； + `window` 是该图**实际可见的世界范围**（供 `inspect_region` 把截图归一化坐标映射回 WCS）。 +- **多源分发 seam**：`phases/evidence.collect(ctx, index, unit)` 按 `source_type_of(unit)` + 分发；非 CAD 源走 `sources.collect`，产出与 CAD **同形**的 evidence（结构化键给 + `evidence_model.empty_envelope()`，PDF 另有 best-effort 文本层放进 `texts` 信封）。 +- **纯函数复用**：`tiling.py`（比例换算 / 瓦片切分）、`heatmap.py`（占用网格 / 聚类） + 与 CAD 无关，CAD 用**实体 bbox** 驱动、PDF/图片用**像素墨迹**驱动，链路完全一致。 +- **离线守卫**：`connect_cad=False` 时 `ctx.tools`/`ctx.client` 均为 `None`；所有访问点 + （`maximized_drawing_area` / `_review_tools` / `_run_readonly_tool` / `_reset_viewport` / + `run_deterministic_checks` / `annotator.annotate`&`clear`）都先判 `None`。 + `checklist.build_system_prompt(offline=True)` 换成纯视觉提示词（无工具、**禁止输出 handle**）。 +- **世界坐标系**（都 y 向上）：CAD = 图形单位；PDF = **页内点**（已应用 `/Rotate`）； + 图片 = **源像素**（`sources.snap_box` 对齐整数，保证「窗口 ↔ 像素」精确对应）。 +- 运行时依赖见 `requirements.txt`（`openai` / `Pillow` / `pypdfium2`）。 + ### 关键头文件 - `inc/cadsdk.h`：ARX 开发总伞头文件，一次性引入几乎所有 ObjectARX/ZWCAD 头文件，处理 ZWCAD 与 AutoCAD 的宏/库差异。业务 `.cpp` 通常只需 `#include "cadsdk.h"`。 @@ -102,5 +139,6 @@ python verify_collisions.py # find_collisions：全模型/逐图框粗 - **不要改动线程模型**：任何涉及 ARX API 的工具逻辑必须在主线程（`ExecuteTaskInMainThread` 派发链）中执行，不能在 `TcpServerLoop` 线程里直接调用。 - **C++ 标准随 SDK 版本浮动**：`cmakelists.txt` 按 `CAD_SDK_VERSION` 自动设定 C++98/03/14/20，跨版本改动需兼容旧标准（低版本 SDK 最高只到 C++14）。 -- 新增 CAD 版本支持时，需在 `cmakepresets.json` 同步添加 `configurePresets` 与 `buildPresets`，并在 `cmake/ThirdPartyDeps.cmake` 中确认对应库列表（如 `CAD_SDK_VERSION GREATER 23` 时追加 `acpal.lib`/`acgeoment.lib`）。 +- **新增 CAD 版本支持时**，需在 `cmakepresets.json` 同步添加 `configurePresets` 与 `buildPresets`，并在 `cmake/ThirdPartyDeps.cmake` 中确认对应库列表（如 `CAD_SDK_VERSION GREATER 23` 时追加 `acpal.lib`/`acgeoment.lib`）。 +- **Python 侧不得把 CAD 能力下沉到 MCP**：PDF / 图片的解析与渲染属于 agent 侧感知层（`dwg_review/sources.py`），**不塞进 MCP、不编 DLL**；只有「CAD 进程内的正交感知能力」才考虑新增 MCP 工具。 - `bin/`、`lib/`、`out/` 均为构建产物，已在 `.gitignore` 中忽略；源码在 `src/`、`inc/`、`cmake/`。 diff --git a/ai.memory/MEMORY.md b/ai.memory/MEMORY.md index f8b0f36..b8c9a73 100644 --- a/ai.memory/MEMORY.md +++ b/ai.memory/MEMORY.md @@ -12,5 +12,5 @@ - [用户协作偏好](feedback_preferences.md) — 调试用 Debug/R230、模型 deepseek-flash、DLL 集中在 cad_mcp_plugins 等 - [环境与外部资源](reference_environment.md) — 构建/运行要点、CAD SDK 路径、测试脚本 - [工具需求格式](tool_request_format.md) — 现有工具不够用时的结构化「工具需求」模板（面向 AI 自己实现） -- [Agent 路线与多源架构](project_agent_roadmap.md) — 总目标（多格式图纸）、视觉为最大公约数、第一步多源证据层+多图、后续路线 +- [Agent 路线与多源架构](project_agent_roadmap.md) — 总目标（多格式图纸）、视觉为最大公约数、多源证据层+多图、热力图/瓦片规划/inspect_region/find_collisions、**PDF+图片离线源（pypdfium2）**、后续路线 - [工具层与 agent 层的正交分工](feedback_tool_layering.md) — 反对无休止加工具；工具层正交稳定，业务规则进 agent 层 diff --git a/ai.memory/project_agent_roadmap.md b/ai.memory/project_agent_roadmap.md index d99cae9..06ca615 100644 --- a/ai.memory/project_agent_roadmap.md +++ b/ai.memory/project_agent_roadmap.md @@ -1,6 +1,6 @@ --- -name: Agent 路线与多源架构（2026-10-15） -description: dwg_review agent 的总目标、多源图纸分层架构（视觉为最大公约数）、工具层与 agent 层的正交分工、多源证据层、截图去 UI、grounding 实测、热力图 + 自适应瓦片规划 + VLM 分辨率标定、绘图区最大化（set_drawing_area）与空瓦片过滤、审查阶段分辨率对齐、agent 侧虚拟工具 inspect_region（放大取证闭环）、几何干涉检查 find_collisions（BVH + 容器过滤）、后续路线 +name: Agent 路线与多源架构（2026-10-16） +description: dwg_review agent 的总目标、多源图纸分层架构（视觉为最大公约数）、工具层与 agent 层的正交分工、多源证据层、截图去 UI、grounding 实测、热力图 + 自适应瓦片规划 + VLM 分辨率标定、绘图区最大化（set_drawing_area）与空瓦片过滤、审查阶段分辨率对齐、agent 侧虚拟工具 inspect_region（放大取证闭环）、几何干涉检查 find_collisions（BVH + 容器过滤）、PDF/图片离线源（pypdfium2 + 墨迹瓦片）、后续路线 type: project --- @@ -75,9 +75,8 @@ Agent 推理层（审查逻辑 / 清单 / 规则） 1. **VLM 主动放大（自适应取证）**：**2026-10-16 已落地**（见下节 `inspect_region`）。 剩余可选打磨：把「看清没」的判定进一步形式化（目前靠 agent 用标定数学**构造保证**可读 + 回合上限 + 截图预算兜底），以及 grounding→WCS 之外的其它指认方式。 -2. **PDF / 图片源**：在 `phases/evidence.collect()` 加分支，产出**同形** evidence - （`window=None`、结构化空信封、`images` 带 label）；规划阶段抽象"审查单元" - （CAD=图框 / PDF=页 / 图片=单）；批注对非 CAD 源降级（`annotator` 已天然跳过无 handle 的 issue）。 +2. **PDF / 图片源**：**2026-10-16 已落地**（见下节「C1：PDF / 图片离线源」）。 + 剩余：真实 PDF 图纸量大时按需增量取页；图片源的低分辨率原图无法放大（设计如此）。 3. **审图规范接入**：清单升级为规范库。 ## 热力图 → 自适应瓦片规划 → VLM 分辨率标定（2026-10-15，进行中） @@ -298,4 +297,75 @@ top 全是 `INSERT/40B × 一切` 且 ratio=1.0（图框完全包住小实体） LLM 在受控子循环里**主动调用** `find_collisions {'min_overlap': 0.2, 'narrow': True, 'limit': 60, 'window': <图框>}`，并结合 `inspect_region`；最终**未误报干涉**（正确真阴性）。 -**待办**：C1（PDF / 图片源适配，见「后续路线」第 2 条）仍未做。 +**待办**：无（C1 / C2 均已完成）。 + +## C1：PDF / 图片**离线源**（2026-10-16） + +**用户拍板的关键决定**： +- 支持 **PDF + 图片**两种源；PDF 先做，图片「顺手做了」。 +- **完全离线，不连 CAD**（`connect_cad=False`）。 +- **不做 OCR**：「都有 VLM 了，还 OCR 实在是画蛇添足」。 +- **文本抽取只是次要补充**：国标图纸的文字多为 **SHX 描边**（导出成线条），文本层几乎为空 + （实测该暖通 PDF：正文页 **0 个文本矩形**；有文本层的是目录页/表格页）。 +- **PDF 在内存渲染，复用既有瓦片规划**能力。 +- 后端选 **pypdfium2**（Apache-2.0/BSD、`py3-none-` wheel 与 Python 版本无关、 + ~4MB）而非 PyMuPDF（**AGPL-3.0** 有商业约束 + cpXX wheel）。文本抽取已降级为次要， + 瓦片规划改用**像素墨迹密度**而非矢量路径，故 pypdfium2 足够。 + +**新增 `dwg_review/sources.py`（agent 侧，不编 DLL）**： +- 世界坐标系统一（都 y 向上，与 `heatmap`/`tiling` 一致）：**PDF = 页内点**（已应用 `/Rotate`）、 + **图片 = 源像素**。故 `window`/`bbox` 与 CAD 侧同形，直接复用 `tiling`/`heatmap` 纯函数。 +- `detect_source(path)`（扩展名 → cad/pdf/image）、`plan_file(ctx)`（每页一个「图框」， + **保持原始页序**——不能再按 bbox 重排）、`collect(ctx, index, unit)`（同形 evidence）。 +- **墨迹网格 `ink_grid(image, box)`**（纯 PIL，无 numpy）：灰度 → 缩略 → 阈值二值化 → + `Image.BOX` 块均值降到网格 → 与 `heatmap.build_grid` **同形**的 grid + 占用格 bbox。 + 于是 `heatmap.cluster_cells` → `tiling.plan_regions` **原样复用**。 +- **每张瓦片的渲染倍率由瓦片世界尺寸反推**：`zoom = IMAGE_MAX_EDGE / max(w,h)`，夹到源上限 + （PDF `PDF_TILE_MAX_ZOOM`；图片 `IMAGE_MAX_ZOOM=1` **不放大**）。这与 `plan_regions` 的 + `s_text`/`s_line` 是**同一套数学**（`zoom = 1/s`）：规划出的文字簇瓦片恰好落在 + `zoom_text = TEXT_READABLE_PX / (TEXT_HEIGHT_MM · PT_PER_MM)`。 +- `PdfReader` / `ImageReader` 读取器（PDF 惰性 import pypdfium2）；`open_reader/close_reader` + 把读取器缓存在 `ctx.source_reader`（避免每页重复打开）。 + +**pypdfium2 的关键 API 事实（踩过的坑）**： +- `page.render(scale, rotation, crop)` 的 **`crop` 是「要裁掉的边距」(left,bottom,right,top)， + 不是绝对矩形**，单位是点、且在**旋转后**的页面上生效 → `crop_of(box, W, H) = (x0, y0, W-x1, H-y1)`。 + 写成绝对矩形会报 `Crop exceeds page dimensions`（或裁错区域）。 +- `page.get_size()` **已应用 `/Rotate`**，`page.get_mediabox()` 未应用。 +- **`tp.get_rect(i)` 前必须先调一次 `tp.count_rects()`**，否则 `PdfiumError: Failed to get rectangle`。 +- 文本 rect 是**未旋转**坐标 → `/Rotate 90` 映射 `(x,y) → (y, mw - x)`（**实测验证**： + 文本区墨迹命中率 0.335 vs 其它假设 ≤0.064，并目视确认「暖通专业图纸」正向）。 +- 退化 bbox（宽/高 ≤ 0）会让 crop 边距之和等于页尺寸 → 提前拒绝，别交给 pdfium。 + +**离线链路打通（守卫 `ctx.tools is None`）**： +`context.ReviewContext.create(connect_cad=False)` → `client/tools` 均为 None； +`phases/evidence.maximized_drawing_area`、`phases/review._review_tools`/`_run_readonly_tool`/ +`_reset_viewport`、`checks.run_deterministic_checks`、`annotator.annotate`/`clear` 全部先判 None。 +`checklist.build_system_prompt(offline=True)` 换成**纯视觉提示词**（无工具协议、明确要求 +**不要输出 handle**、说明文本层为空属正常）。 + +**CLI**：`python -m dwg_review `（位置参数）；`.pdf`/图片 → 离线模式， +`.dwg/.dxf` → 提示后按 CAD 在线模式（Python 打不开 DWG），未知格式/文件不存在 → exit 2。 +`--list-tools`/`--check`/`--clear`/`--restore` 在离线源下被拒绝。 + +**实测（`verify_pdf_source.py`，20 页暖通图纸）**： +- PDF 前 4 页 **20/20 PASS**；图片源（由该 PDF 第 3 页渲染成 2717×1945 PNG）**9/9 PASS**。 +- **分辨率自检（实测而非推导）**：每张瓦片的「纸面字高像素」≥ 线条下限 6px， + 典型瓦片达 **8.5~13.9px**（文字档 8px）；整体图 ~4.3px（整页缩略，**设计如此**，细节由瓦片负责）。 +- **旋转自检自动化**：文本区墨迹 0.311 vs 整页平均 0.012；反向对照（忽略 `/Rotate`）0.145 → 映射正确。 +- **真实 LLM 端到端**（`python -m dwg_review --frames 3`）：VLM **读懂了真实工程图**—— + 报出「设备标注框叠压家具/洁具/门扇弧线」「设备平面定位尺寸缺失」「标高以『梁底』代替数值不统一」 + 「1-1 剖面室外机未表达与墙柱间距」「材料表单重/共重两列全空」「审定未签署」等**具体且可核验**的问题， + 且**未编造 handle**（离线证据无 handle，全部 issue 无 handle 字段）。 + +**新增配置**：`PDF_MAX_PAGES`(0=全部) / `PDF_TEXT_MIN_CHARS`(2) / `PDF_OVERVIEW_MAX_ZOOM`(4.0) / +`PDF_TILE_MAX_ZOOM`(6.0) / `IMAGE_MAX_ZOOM`(1.0) / `INK_ANALYSIS_MAX_EDGE`(1024) / +`INK_THRESHOLD`(200) / `INK_MIN_CELL_FRAC`(0.01)；模块级 `PT_PER_MM = 72/25.4`。 + +**已知局限**： +- 图框/页边线是**细线**，在墨迹网格里把整页连成**一个大连通簇** → 瓦片常退化为 + 「整页 2×2 象限 + 少量密集小岛」。「空白区不出图」的收益因此打折（CAD 侧靠 + `drop_oversized` 丢掉图框块，离线侧暂无等价手段）。成本由 `TILE_MAX_COUNT`/`EVIDENCE_MAX_IMAGES` 兜底。 +- 图片源不能放大（`IMAGE_MAX_ZOOM=1`）：低分辨率扫描件无法凭空提升可读性。 +- 文本层对国标图纸基本无用（正文页 0 个文本矩形），**这是预期**，主力仍是 VLM 看图。 +- 依赖新增：`requirements.txt`（openai / Pillow / pypdfium2）。 diff --git a/ai.memory/reference_environment.md b/ai.memory/reference_environment.md index e6e8a72..b5d5ca7 100644 --- a/ai.memory/reference_environment.md +++ b/ai.memory/reference_environment.md @@ -49,11 +49,22 @@ type: reference | `verify_drawing_area.py` | 绘图区最大化验收：基线 → `set_drawing_area(max)` → `restore`，断言窗口矩形/最大化态/绘图区尺寸全回位（实测 3.68×） | | `verify_review_maximized.py` | 审查全链路验收：在 `collect`/`review_frame` 入口各采一次窗口矩形，断言**取证与审查两阶段都在最大化态**、结束精确还原；并统计子循环 zoom/截图次数 + 记录 LLM 实际调用的取证工具（含虚拟工具 `inspect_region`） | | `verify_collisions.py` | `find_collisions` 验收：工具注册 / 全模型与逐图框粗筛 / **容器过滤**（图框块剔除）/ `min_overlap` 阈值 / 分页 / `narrow` 细筛 / `--inject` 注入干涉正例（画 2 个深重叠矩形 → 应报出 ratio≈0.902 → 清理）。**只读**（`--inject` 段除外，会自动清理） | +| `verify_pdf_source.py` | **PDF / 图片离线源验收（不需要 CAD）**：离线上下文断言 / 页序 / evidence 同形 / 图片窗口在图框内 / **实测瓦片字高像素达标** / **旋转映射自检**（文本区墨迹 vs 忽略 `/Rotate` 反向对照）/ `--dump` 导出目视包 / `--llm` 追加真实 `python -m dwg_review` 端到端 | > 历史：根目录 `test.py` / `test_autoagent.py` / `countOfCircle.py` / `agent_loop.py` / > `dwg_review_agent.py` 五个遗留调试脚本已于 2026-10-14 删除（前三个含硬编码 API Key， > 后两个是重复实现）。`test_screenshot.png` 是 `test_autoagent.py` 的残留产物。 +## Python 运行时依赖（2026-10-16 起） + +- 仓库根新增 **`requirements.txt`**（运行时）与 `requirements-dev.txt`（仅 pytest）。 + - `openai>=1.0`（LLM 客户端，Key 走环境变量 `DEEPSEEK_API_KEY`） + - `Pillow>=10.0`（截图下采样、墨迹分析、离线标定） + - `pypdfium2>=4.30`（PDF 源适配：内存渲染 + 带 bbox 抽文本） + - 选它而非 PyMuPDF：**许可证干净**（Apache-2.0/BSD vs AGPL-3.0）、 + wheel 为 `py3-none-`（**与 Python 版本无关**，实测 Python 3.14.5 直接可用）、体积小（~4MB）。 +- 实测环境：Python **3.14.5**，`pypdfium2 5.14.0`（`py3-none-win_amd64`）。 + ## 两个构建目标 | 目标 | 产物 | 作用 | @@ -93,7 +104,9 @@ type: reference 本仓库已设置 `git config core.sshCommand "ssh -i ~/.ssh/id_ed25519_guojianming -o IdentitiesOnly=yes"`。 换机器时需补该配置或加入 `~/.ssh/config`。 -## 真实测试图纸（2026-10-15） +## 真实测试图纸 + +**CAD 源（2026-10-15）** - 图纸：`e:\desktop\agent_test.dwg`（用户制作的真实 Plant 3D 图纸：3 个图框 A0/A1/A2， 86 个实体：`AECDBEWFITTING` 40 / `AECDBEWPIPE` 29 / `AECDBEWEQUIPMENT` 14 / `INSERT` 3）。 @@ -101,3 +114,12 @@ type: reference —— 必须带 `/p SHS-ENVI2026` profile，否则**自定义实体无法显示**（该 profile 自动加载 用户写的显示插件）。 - 实测：`python -m dwg_review --frames 1` 端到端可跑通并出报告。 + +**PDF 源（2026-10-16）** + +- 图纸：`E:\desktop\05 暖通专业图纸.pdf`（**20 页**，A3/A4 混排，`/Rotate 90`）。 + 页 0 = 封面（仅一行标题字）；页 1 = 图纸目录（**有文本层**，43 个文本矩形）； + 页 2 起 = 真实施工图（**0 个文本矩形**，文字全是 SHX 描边）。 +- 实测：`python -m dwg_review "E:\desktop\05 暖通专业图纸.pdf" --frames 3`（**离线、不连 CAD**） + 端到端跑通，VLM 报出设备标注叠压、定位尺寸缺失、材料表空白等具体问题。 +- 图片源样本：由该 PDF 第 3 页渲染的 `out/verify_pdf_source/page2.png`（2717×1945，测试产物）。 diff --git a/dwg_review/__main__.py b/dwg_review/__main__.py index db0a277..5d3c55e 100644 --- a/dwg_review/__main__.py +++ b/dwg_review/__main__.py @@ -1,4 +1,9 @@ -"""CLI 入口：python -m dwg_review [options]""" +"""CLI 入口：python -m dwg_review [source] [options] + +两种运行模式（由 `source` 扩展名决定）： + - **CAD 在线**（.dwg/.dxf 或不带参数）：连 127.0.0.1:8080，审查 CAD 中当前打开的图纸； + - **离线**（.pdf / 图片）：完全离线，不连 CAD，内存渲染 + VLM 审查。 +""" import argparse import json @@ -8,10 +13,11 @@ from pathlib import Path from . import config from . import checks as checks_phase from . import findings as findings_mod +from . import sources from .annotator import annotate, clear from .checklist import build_system_prompt, checklist_summary from .context import ReviewContext -from .evidence_model import evidence_images +from .evidence_model import SOURCE_CAD, evidence_images from .mcp_client import McpError, McpToolError, result_json from .phases import evidence as evidence_phase from .phases import planning, report, review @@ -46,43 +52,69 @@ def cmd_check(ctx): def run_review(ctx, args): + offline = ctx.tools is None print("=== 图纸审查 Agent ===") print(f"清单：{checklist_summary(ctx.checklist)}") if ctx.llm is not None: print(f"模型：{ctx.llm.model}") + if offline: + print(f"来源：{ctx.source}（离线模式，不连 CAD） {ctx.source_path}") print() print("[1/4] 规划：识别图框与概览 ...") plan = planning.plan(ctx) frames = plan.get("frames", []) - print(f" 识别到 {len(frames)} 个图框") + unit = "页" if offline else "图框" + if offline: + info = plan.get("drawing_info") or {} + print( + f" 共 {len(frames)} {unit}（原文档 {info.get('page_count')} 页，" + f"坐标单位 {info.get('unit')}）" + ) + else: + print(f" 识别到 {len(frames)} 个图框") if not frames: - print("未识别到任何图框，退出。") + print(f"未识别到任何{unit}，退出。") return 1 - system_prompt = build_system_prompt(ctx.checklist, allow_tools=True) + system_prompt = build_system_prompt( + ctx.checklist, allow_tools=not offline, offline=offline + ) evidences = [] frame_findings = [] # 截屏前把绘图区最大化（窗口铺满 + 干净屏幕）。**取证与审查都要截图**——审查阶段的 # 受控子循环会按需放大复核，故一并包住，结束在 finally 里还原；旧版插件无此工具时静默降级。 + # 离线源没有 CAD，该上下文管理器内部会直接放行。 with evidence_phase.maximized_drawing_area(ctx): print("[2/4] 取证：逐图框采集（串行） ...") for idx, frame in enumerate(frames, 1): - print( - f" - 图框 {idx}/{len(frames)} kind={frame.get('kind')} " - f"sheet={frame.get('sheet_size')}" - ) + if offline: + print( + f" - {unit} {idx}/{len(frames)} 原页 {frame.get('page_index')} " + f"图幅 {frame.get('sheet_size') or '—'}" + ) + else: + print( + f" - {unit} {idx}/{len(frames)} kind={frame.get('kind')} " + f"sheet={frame.get('sheet_size')}" + ) ev = evidence_phase.collect(ctx, idx, frame) if ev is None: print(" [跳过] 缺少有效包围盒") continue evidences.append(ev) - print( - f" 图元 {(ev.get('entities') or {}).get('total')} 条，" - f"截图 {len(evidence_images(ev))} 张" - ) + if offline: + print( + f" 文本层 {(ev.get('texts') or {}).get('total')} 条，" + f"图片 {len(evidence_images(ev))} 张" + ) + else: + print( + f" 图元 {(ev.get('entities') or {}).get('total')} 条，" + f"截图 {len(evidence_images(ev))} 张" + ) if args.plan_only: path = _dump_evidence(evidences) @@ -91,7 +123,7 @@ def run_review(ctx, args): print("[3/4] 审查：确定性检查前置 + 逐图框 LLM 审查（含受控只读取证子循环） ...") for ev in evidences: - print(f" - 审查图框 {ev['index']} ...") + print(f" - 审查{unit} {ev['index']} ...") checks = ( checks_phase.run_deterministic_checks(ctx, ev) if config.DETERMINISTIC_CHECKS @@ -122,6 +154,9 @@ def run_review(ctx, args): if args.annotate: print() + if offline: + print("[批注回写] 当前为离线源（无 CAD 可回写），已跳过。") + return 0 print(f"[批注回写] {'(dry-run) ' if args.dry_run else ''}结构化问题 {len(issues)} 条") created, highlighted = annotate(ctx, issues, dry_run=args.dry_run) if not args.dry_run and (created or highlighted): @@ -136,12 +171,20 @@ def run_review(ctx, args): def build_parser(): parser = argparse.ArgumentParser( - prog="dwg_review", description="CAD 图纸审查 Agent（基于 cad_mcp_frame MCP 基座）" + prog="dwg_review", + description="图纸审查 Agent（CAD 走 cad_mcp_frame MCP 基座；PDF / 图片走离线渲染）", + ) + parser.add_argument( + "source", + nargs="?", + default=None, + help="图纸路径：.dwg/.dxf 走 CAD 在线模式（需先在 CAD 中打开该图纸）；" + ".pdf/图片 走完全离线模式。省略则按 CAD 在线模式运行。", ) parser.add_argument("--list-tools", action="store_true", help="列出基座注册的全部工具") parser.add_argument("--check", action="store_true", help="检查基座连通性并打印图纸概览") parser.add_argument("--plan-only", action="store_true", help="只做规划+取证，落盘 evidence.json") - parser.add_argument("--frames", type=int, default=None, help="最多审查的图框数（0=全部）") + parser.add_argument("--frames", type=int, default=None, help="最多审查的图框数/页数（0=全部）") parser.add_argument("--annotate", action="store_true", help="审查后把问题批注回写到图上") parser.add_argument("--dry-run", action="store_true", help="批注只打印不回写") parser.add_argument("--clear", default=None, help="删除给定 handle（逗号分隔）以撤销批注") @@ -150,17 +193,56 @@ def build_parser(): return parser +def _resolve_source_arg(source_arg): + """解析位置参数 → (source, source_path)；失败返回 (None, 错误信息)。""" + if not source_arg: + return None, None + try: + detected = sources.detect_source(source_arg) + except ValueError as exc: + return None, str(exc) + + if detected == SOURCE_CAD: + # Python 侧无法打开 DWG，只能提示用户在 CAD 里打开后运行 + print( + f"[!] 参数 {source_arg} 是 CAD 图纸：Python 无法直接打开，" + "将按 **CAD 在线模式** 审查 CAD 中当前打开的图纸。" + ) + print(" （请先在 CAD 中打开该图纸；不带参数运行效果相同。）") + return None, None + + path = Path(source_arg).expanduser().resolve() + if not path.is_file(): + return None, f"文件不存在：{path}" + return (detected, str(path)), None + + def main(argv=None): args = build_parser().parse_args(argv) if args.checklist: config.CHECKLIST_PATH = Path(args.checklist) - need_llm = not ( - args.list_tools or args.check or args.plan_only or args.clear or args.restore - ) + resolved, err = _resolve_source_arg(args.source) + if err: + print(f"[X] {err}") + return 2 + source, source_path = resolved if resolved else (None, None) + connect_cad = source is None + + cad_only = bool(args.list_tools or args.check or args.clear or args.restore) + if not connect_cad and cad_only: + print("[X] --list-tools / --check / --clear / --restore 需要连接 CAD，仅支持 CAD 在线模式。") + return 2 + + need_llm = not (args.list_tools or args.check or args.plan_only or args.clear or args.restore) try: - ctx = ReviewContext.create(need_llm=need_llm) + ctx = ReviewContext.create( + need_llm=need_llm, + connect_cad=connect_cad, + source=source, + source_path=source_path, + ) except McpError as exc: print(f"[X] 无法连接 CAD 基座：{exc}") return 1 @@ -188,6 +270,9 @@ def main(argv=None): except KeyboardInterrupt: print("\n[中断]") return 130 + finally: + if not connect_cad: + sources.close_reader(ctx) if __name__ == "__main__": diff --git a/dwg_review/annotator.py b/dwg_review/annotator.py index b13e644..d84255d 100644 --- a/dwg_review/annotator.py +++ b/dwg_review/annotator.py @@ -33,9 +33,15 @@ def _pick_point(detail): def annotate(ctx, issues, *, dry_run=False, log=print): - """按问题列表回写批注，返回 (created, highlighted) 两个 handle 列表。""" + """按问题列表回写批注，返回 (created, highlighted) 两个 handle 列表。 + + 离线源（PDF / 图片，`ctx.tools is None`）没有可回写的 CAD → 直接返回空结果。 + """ created = [] highlighted = [] + if ctx.tools is None: + log("[批注] 当前为离线源，跳过回写") + return created, highlighted seen = set() # 同一 handle 只处理一次，避免重复批注 for issue in issues: @@ -99,6 +105,8 @@ def annotate(ctx, issues, *, dry_run=False, log=print): def clear(ctx, handles=None, highlighted=None, *, log=print): """撤销批注：删除批注实体 + 还原被高亮的实体。""" + if ctx.tools is None: + return if handles: try: ctx.tools.call("delete_entities", {"handles": list(handles)}) diff --git a/dwg_review/checklist.py b/dwg_review/checklist.py index 0113ca0..cfa2a2e 100644 --- a/dwg_review/checklist.py +++ b/dwg_review/checklist.py @@ -37,16 +37,37 @@ def load_checklist(path=None): return DEFAULT_CHECKLIST -def build_system_prompt(checklist, *, allow_tools=True): - """根据审查清单动态生成 LLM 系统提示词。""" - lines = [ - "你是一名资深建筑与工业厂房图纸审查专家。", - "你将逐张收到某个图框（一张图纸）的：结构化图元数据、文字内容，以及该图框的视口截图。", - "", - f"请依据审查清单《{checklist.get('name', '审查清单')}》逐项审查，" - "指出**具体问题**（尽量引用实体 handle 或文字内容），并标注严重程度：", - "", - ] +def build_system_prompt(checklist, *, allow_tools=True, offline=False): + """根据审查清单动态生成 LLM 系统提示词。 + + `offline=True` 用于 PDF / 图片源：**没有 CAD、没有实体句柄、没有结构化数据**， + 输入只有渲染图（整体 + 局部瓦片）与少量文本层片段，结论必须完全基于图像。 + """ + if offline: + lines = [ + "你是一名资深建筑与工业厂房图纸审查专家。", + "你将逐张收到某**一页图纸的渲染图**（整体图 + 若干局部放大图；" + "局部图按内容分布自适应切分，可能重叠）。", + "", + "**重要**：该图纸是从 PDF / 图片导入的，**没有实体级结构化数据、也没有实体句柄**。" + "你看到的就是全部依据——请**只依据图像**判断。", + "国标图纸的文字多为 SHX 描边（导出后成为线条），因此附带的文本层往往为空或很少，" + "这是正常现象；**请直接从图上读文字与尺寸**，不要因为没有文本层就降低审查力度。", + "", + f"请依据审查清单《{checklist.get('name', '审查清单')}》逐项审查，" + "指出**具体问题**（说明问题出现在图上什么位置，例如「左上角平面图 A 轴附近」），" + "并标注严重程度：", + "", + ] + else: + lines = [ + "你是一名资深建筑与工业厂房图纸审查专家。", + "你将逐张收到某个图框（一张图纸）的：结构化图元数据、文字内容，以及该图框的视口截图。", + "", + f"请依据审查清单《{checklist.get('name', '审查清单')}》逐项审查，" + "指出**具体问题**（尽量引用实体 handle 或文字内容），并标注严重程度：", + "", + ] for cat in checklist.get("categories", []): lines.append(f"【{cat.get('title', '')}】（严重度：{cat.get('severity', 'medium')}）") for chk in cat.get("checks", []): @@ -77,6 +98,32 @@ def build_system_prompt(checklist, *, allow_tools=True): "", ] + if offline: + lines += [ + "要求：", + "- 只报告你**在图上确实看到**的问题，不要臆测；看不清就说看不清，不要编造。", + "- 每条问题请写明**位置**（用图上可见的轴线号 / 房间名 / 相对方位描述）。", + "- 局部放大图可能有重叠，同一问题只报一次。", + "- 按清单条目组织结论，未命中的条目可省略。", + "- 不要复述输入数据。", + "", + "输出格式（**必须严格遵守**）：", + "在你准备给出该页图纸的最终结论时，**只输出一个 JSON 对象**，" + "不要输出任何解释文字，也不要用 markdown 代码块包裹。结构如下：", + "{", + ' "summary": "<该页的一句话整体结论；若未发现问题，写“未发现明显问题”>",', + ' "issues": [', + " {", + ' "category": "<图层/文字/标注/几何/干涉/图面整体 之一>",', + ' "severity": "",', + ' "message": "<问题描述 + 位置，简洁具体>"', + " }", + " ]", + "}", + "**不要输出 `handle` 字段**（本来源没有实体句柄，编造会被系统丢弃）。", + ] + return "\n".join(lines) + lines += [ "要求：", "- 只报告你**有依据**的问题，不要臆测；数据与截图冲突时以数据为准。", diff --git a/dwg_review/checks.py b/dwg_review/checks.py index 15679f6..7e95c3a 100644 --- a/dwg_review/checks.py +++ b/dwg_review/checks.py @@ -187,7 +187,8 @@ def run_deterministic_checks(ctx, evidence, *, log=print): evidence, check_unclosed=config.CHECK_UNCLOSED_POLYLINES ) window = evidence.get("window") - if window: + # 离线源（PDF / 图片）没有 CAD 可查：tools 为 None，只跑纯 Python 派生检查。 + if window and ctx.tools is not None: issues += check_zero_length( ctx, window, tolerance=config.ZERO_LENGTH_TOLERANCE, diff --git a/dwg_review/config.py b/dwg_review/config.py index 2ea1550..fad8da3 100644 --- a/dwg_review/config.py +++ b/dwg_review/config.py @@ -171,6 +171,29 @@ HEATMAP_CLUSTER_DILATE = int(os.environ.get("HEATMAP_CLUSTER_DILATE", "0")) # 丢弃 bbox 面积超过图框该比例的实体（去图框边框/图签块） HEATMAP_MAX_AREA_FRAC = float(os.environ.get("HEATMAP_MAX_AREA_FRAC", "0.5")) +# --------------------------------------------------------------------------- +# 非 CAD 源（PDF / 图片）：完全离线，不连 CAD +# --------------------------------------------------------------------------- +# 毫米 → PDF 点（1pt = 1/72in）。PDF 世界坐标以**点**为单位，故纸面字高须换算成点。 +PT_PER_MM = 72.0 / 25.4 +# PDF 最多取几页（0 = 全部） +PDF_MAX_PAGES = int(os.environ.get("PDF_MAX_PAGES", "0")) +# 文本层片段的最小字符数（低于此视为噪声丢弃）。国标图纸文字多为 SHX 描边， +# 导出后是线条而非文本，文本层往往整页为空——**文本抽取只是次要补充**。 +PDF_TEXT_MIN_CHARS = int(os.environ.get("PDF_TEXT_MIN_CHARS", "2")) +# 整体图（整页/整图）渲染倍率上限（px/点）。整页长边同时受 IMAGE_MAX_EDGE 约束。 +PDF_OVERVIEW_MAX_ZOOM = float(os.environ.get("PDF_OVERVIEW_MAX_ZOOM", "4.0")) +# 瓦片渲染倍率上限（px/点）。规划器算出的倍率通常 ~1.4，此值只作安全兜底。 +PDF_TILE_MAX_ZOOM = float(os.environ.get("PDF_TILE_MAX_ZOOM", "6.0")) +# 图片源渲染倍率上限：1.0 = **不放大**（放大会插值出虚假细节）。 +IMAGE_MAX_ZOOM = float(os.environ.get("IMAGE_MAX_ZOOM", "1.0")) +# 墨迹分析用的缩略图长边上限（越大越精确但越慢） +INK_ANALYSIS_MAX_EDGE = int(os.environ.get("INK_ANALYSIS_MAX_EDGE", "1024")) +# 灰度阈值：像素值 < 该值视为墨迹（0=黑，255=白） +INK_THRESHOLD = int(os.environ.get("INK_THRESHOLD", "200")) +# 一个网格被判定为「有墨迹」所需的最小墨迹面积占比（滤掉零星噪点） +INK_MIN_CELL_FRAC = float(os.environ.get("INK_MIN_CELL_FRAC", "0.01")) + # --------------------------------------------------------------------------- # 绘图区最大化（截屏前把 CAD 窗口铺满屏幕 + 进入干净屏幕） # --------------------------------------------------------------------------- diff --git a/dwg_review/context.py b/dwg_review/context.py index 347bfff..aa2d925 100644 --- a/dwg_review/context.py +++ b/dwg_review/context.py @@ -2,7 +2,7 @@ from dataclasses import dataclass -from . import config +from . import config, evidence_model from .cad_tools import CadTools from .checklist import load_checklist from .llm_client import LlmClient @@ -11,10 +11,24 @@ from .mcp_client import McpClient @dataclass class ReviewContext: + """一次审查运行的共享上下文。 + + `connect_cad=False`（离线模式）时 `client` / `tools` 均为 `None`——供 PDF / 图片 + 等**纯离线源**使用：这些源没有 CAD 可连，全链路（取证 / 审查 / 批注）都必须 + 先判 `ctx.tools is None` 再决定是否走 CAD 工具。`source` 记录本次运行的来源类型， + 供规划阶段分发（CAD 走 `detect_drawing_frames`，文件源自带图框）。 + + `source_path` 为文件源（PDF / 图片）的磁盘路径；`source_reader` 是 `sources` + 打开的读取器（缓存，避免每页重复打开 PDF）。 + """ + client: McpClient tools: CadTools llm: LlmClient checklist: dict + source: str = evidence_model.SOURCE_CAD + source_path: str = None + source_reader: object = None max_frames: int = 0 frame_limit: int = 200 max_iters: int = 4 @@ -22,9 +36,11 @@ class ReviewContext: dry_run: bool = False @classmethod - def create(cls, *, need_llm=True, **overrides): - client = McpClient() - tools = CadTools(client) + def create(cls, *, need_llm=True, connect_cad=True, source=None, source_path=None, + **overrides): + client = McpClient() if connect_cad else None + # CadTools.__init__ 会立刻 list_tools() 去连 127.0.0.1:8080；离线模式必须跳过。 + tools = CadTools(client) if connect_cad else None llm = LlmClient() if need_llm else None checklist = load_checklist() params = dict( @@ -32,6 +48,8 @@ class ReviewContext: tools=tools, llm=llm, checklist=checklist, + source=source or evidence_model.SOURCE_CAD, + source_path=source_path, max_frames=config.MAX_FRAMES, frame_limit=config.FRAME_ENTITY_LIMIT, max_iters=config.REVIEW_MAX_ITERS, diff --git a/dwg_review/evidence_model.py b/dwg_review/evidence_model.py index 42e6c22..71b6f11 100644 --- a/dwg_review/evidence_model.py +++ b/dwg_review/evidence_model.py @@ -16,6 +16,16 @@ SOURCE_IMAGE = "image" STRUCTURED_SOURCES = ("entities", "texts", "dimensions", "blocks", "polylines") +def empty_envelope(total=0): + """构造空的**结构化信封** `{"total": N, "items": []}`。 + + CAD 侧各查询工具统一返回此形（`items` 明细 + `total` 计数）。非 CAD 源 + （PDF / 图片）没有结构化数据，用本工厂补齐同形键，使下游 handle 收集 / + 派生检查无需分支即可统一遍历（空 items 自然不产出任何检查项）。 + """ + return {"total": int(total), "items": []} + + def make_image(b64, mime="image/png", label=None, window=None): """构造 images 列表元素。 diff --git a/dwg_review/heatmap.py b/dwg_review/heatmap.py index 40d9b6f..29e6532 100644 --- a/dwg_review/heatmap.py +++ b/dwg_review/heatmap.py @@ -225,7 +225,7 @@ def cluster_cells(grid, min_count=1, dilate=0): stack.append((nr, nc)) if not cells: continue # 膨胀出来的孤立空格，无实体，丢弃 - boxes = [_cell_box(grid, cr, cc) for cr, cc in cells] + boxes = [cell_box(grid, cr, cc) for cr, cc in cells] clusters.append({ "bbox": union_box(boxes), "cells": sorted(cells), @@ -236,7 +236,7 @@ def cluster_cells(grid, min_count=1, dilate=0): return clusters -def _cell_box(grid, r, c): +def cell_box(grid, r, c): """格 (r,c) 的世界 bbox（row 0 在上）。""" win = grid["window"] min_x, max_y = float(win["min_x"]), float(win["max_y"]) @@ -315,7 +315,7 @@ def render_overlay(path, window, boxes, grid, clusters=None, *, size=1200, bg=(1 for c, n in enumerate(row): if n <= 0: continue - x1, y1, x2, y2 = _cell_box(grid, r, c) + x1, y1, x2, y2 = cell_box(grid, r, c) p1, p2 = to_px(x1, y2), to_px(x2, y1) t = min(1.0, n / max_count) color = (int(60 + 195 * t), int(200 - 140 * t), int(60 - 40 * t)) diff --git a/dwg_review/phases/evidence.py b/dwg_review/phases/evidence.py index 55d471b..a29277f 100644 --- a/dwg_review/phases/evidence.py +++ b/dwg_review/phases/evidence.py @@ -11,7 +11,7 @@ import contextlib import math import time -from .. import config, evidence_model, heatmap, tiling +from .. import config, evidence_model, heatmap, sources, tiling from ..llm_client import image_size_b64 from ..mcp_client import result_images, result_json @@ -34,8 +34,9 @@ def maximized_drawing_area(ctx): - 受 `config.DRAWING_AREA_MAXIMIZE` 控制； - 工具不存在（旧版插件）/ 调用失败 → **静默降级**：照常取证，只是绘图区小一些； - 退出时一定尝试还原；**还原失败不掩盖业务异常**（不 raise）。 + - 离线源（PDF / 图片）无 CAD 连接（`ctx.tools is None`）→ 直接放行。 """ - if not config.DRAWING_AREA_MAXIMIZE: + if not config.DRAWING_AREA_MAXIMIZE or ctx.tools is None: yield return @@ -144,21 +145,8 @@ def _text_boxes(entities, texts, dimensions): def _content_predicate(boxes): - """构造「该窗口内是否有实体」谓词；无实体返回 None。 - - 用于剔除空白瓦片：簇 / 象限的 bbox 只是**包围盒**，L 形或斜向分布会含大片空白。 - 这里直接按**实体 bbox** 判断——比先量化成占用格更精确（格子的跨界膨胀会把 - 相邻空白瓦片也判成有内容）。接触即算有内容（保守，宁可多拍不漏）。 - """ - boxes = [b for b in (boxes or []) if b] - if not boxes: - return None - - def _has_content(win): - box = (win["min_x"], win["min_y"], win["max_x"], win["max_y"]) - return any(tiling.boxes_intersect(box, b) for b in boxes) - - return _has_content + """构造「该窗口内是否有实体」谓词；无实体返回 None（委托 `tiling.content_predicate`）。""" + return tiling.content_predicate(boxes) def _apply_keep(regions, keep): @@ -309,11 +297,13 @@ def collect_frame(ctx, index, frame): def collect(ctx, index, unit): """按审查单元来源分发取证。 - 当前仅实现 CAD 源；PDF / 图片源预留（未来在对应分支接入离线解析 / 直接读图， - 产出与 CAD **同形**的 evidence：window=None、结构化键给空信封、images 带 label）。 + CAD 源走 `collect_frame`（CAD 工具 + 截屏）；PDF / 图片源走 `sources.collect` + （内存渲染 + 墨迹瓦片），产出与 CAD **同形**的 evidence：`window` 有值、 + 结构化键给空信封（PDF 另有 best-effort 文本层）、`images` 带 label。 """ source = evidence_model.source_type_of(unit) if source == evidence_model.SOURCE_CAD: return collect_frame(ctx, index, unit) - # TODO(pdf/image): 接入离线解析（DWG/PDF）与直接读图。 + if sources.is_file_source(source): + return sources.collect(ctx, index, unit) return None diff --git a/dwg_review/phases/planning.py b/dwg_review/phases/planning.py index 66387b4..3cb0616 100644 --- a/dwg_review/phases/planning.py +++ b/dwg_review/phases/planning.py @@ -1,8 +1,13 @@ """阶段 1：全局规划。 收集图纸概览 + 识别图框，按阅读顺序排序，产出审查计划。 + +按来源分发：CAD 源用 `detect_drawing_frames` 识别图框；PDF / 图片源直接由文件 +结构给出图框（PDF 每页一个、图片整图一个），**保持原始页序**（阅读顺序已由文档 +自身决定，不能再按 bbox 重排）。 """ +from .. import sources from ..mcp_client import result_json @@ -18,6 +23,9 @@ def _sort_frames(frames): def plan(ctx): """返回 {drawing_info, layers, layouts, frames}。""" + if sources.is_file_source(ctx.source): + return sources.plan_file(ctx) + drawing_info = result_json(ctx.tools.call("get_drawing_info", {}), {}) layers = result_json(ctx.tools.call("get_layers", {"count_entities": True}), {}) layouts = result_json(ctx.tools.call("get_layouts", {}), {}) diff --git a/dwg_review/phases/report.py b/dwg_review/phases/report.py index 6934446..603551b 100644 --- a/dwg_review/phases/report.py +++ b/dwg_review/phases/report.py @@ -19,7 +19,7 @@ _SUMMARY_SYSTEM = ( def format_finding(finding): """把一个 findings 元素渲染成给汇总 LLM 的文本块。""" - lines = [f"【图框 {finding.get('frame')}（图幅 {finding.get('sheet')}）】"] + lines = [f"【图框 {finding.get('frame')}（图幅 {finding.get('sheet') or '—'}）】"] summary = finding.get("summary") if summary: lines.append(str(summary)) diff --git a/dwg_review/phases/review.py b/dwg_review/phases/review.py index 59d151b..81d607f 100644 --- a/dwg_review/phases/review.py +++ b/dwg_review/phases/review.py @@ -56,7 +56,13 @@ def _run_readonly_tool(ctx, name, args, image_budget, *, evidence=None, state=No `inspect_region` 是 **agent 侧虚拟工具**（不走 MCP），在只读白名单校验**之前**拦截： 它内部只调白名单内的只读 MCP 工具。 + + 离线源（PDF / 图片，`ctx.tools is None`）没有任何 MCP 工具可用 → 一律拒绝； + `inspect_region` 依赖 CAD 截图，同样不可用。 """ + if ctx.tools is None: + return [{"type": "text", "text": f"当前为离线源，工具 {name} 不可用。"}], image_budget + if name == inspect.TOOL_NAME: if not config.INSPECT_ENABLED or evidence is None or state is None: return [{"type": "text", "text": "inspect_region 当前不可用。"}], image_budget @@ -96,7 +102,11 @@ def _review_tools(ctx): 启用 `inspect_region` 时，用它**取代**三个裸视口工具（`zoom_to_window` / `zoom_to_entity` / `get_viewport_screenshot`）——一个语义化入口比三个裸工具更好用， 且可读性由 agent 保证；关闭时维持原样（退回旧行为）。 + + 离线源（`ctx.tools is None`）不提供任何工具（也没有 CAD 可 zoom / 截图）。 """ + if ctx.tools is None: + return [] if config.INSPECT_ENABLED: names = READ_ONLY_TOOLS - _RAW_VIEW_TOOLS return ctx.tools.openai_tools(names=names) + [inspect.TOOL_SCHEMA] @@ -106,7 +116,7 @@ def _review_tools(ctx): def _reset_viewport(ctx, evidence): """把视口复位到该图框的整幅视图（审查中的放大不应残留给用户）。""" win = evidence.get("window") - if not win: + if not win or ctx.tools is None: return try: ctx.tools.call("zoom_to_window", dict(win, margin=0.02)) diff --git a/dwg_review/sources.py b/dwg_review/sources.py new file mode 100644 index 0000000..b0926bb --- /dev/null +++ b/dwg_review/sources.py @@ -0,0 +1,593 @@ +"""非 CAD 源适配：PDF / 图片图纸的**离线**取证（不连 CAD）。 + +设计要点 +-------- +1. **世界坐标系**（两种源统一，且都 y 向上，与 `heatmap` / `tiling` 约定一致）： + - PDF ：页内**点**坐标，原点左下，已应用 `/Rotate`（即 `page.get_size()` 的朝向）； + - 图片：**像素**坐标，原点左下。 + 故 `window` / `bbox` 与 CAD 侧同形，可直接喂给 `tiling` / `heatmap` 的纯函数。 + +2. **瓦片规划复用 CAD 侧链路**：把整页渲染（图片则读入）成一张缩略图 → 灰度二值化求 + 墨迹密度 → 量化成与 `heatmap.build_grid` 同形的占用网格 → `heatmap.cluster_cells` + 聚类 → `tiling.plan_regions` 出瓦片窗口。**无需实体数据**，只靠像素墨迹。 + +3. **每张瓦片的渲染倍率由瓦片世界尺寸反推**：`zoom = IMAGE_MAX_EDGE / max(w,h)`， + 再夹到源允许的上限（PDF `PDF_TILE_MAX_ZOOM`；图片 `IMAGE_MAX_ZOOM`=1，**不放大**）。 + 这与 `plan_regions` 的 `s_text` / `s_line` 是同一套数学：规划出的文字簇瓦片恰好 + 落在 `zoom_text = TEXT_READABLE_PX / (TEXT_HEIGHT_MM · PT_PER_MM)`。 + +4. **文本层是次要的**：国标图纸的文字多为 SHX 描边（导出后变成线条），文本层几乎为空 + （实测某暖通图纸：正文页 0 个文本矩形）。仅当某页确有文本层（目录页 / 表格）时 + best-effort 抽取，塞进 `texts` 信封；主力识别交给 VLM 看渲染图。 + +5. `pypdfium2` **惰性 import**：只在真的处理 PDF 时导入，CAD / 图片路径零额外依赖。 +""" + +import base64 +import io +import math +import os + +from . import config, evidence_model, heatmap, tiling + +_LABEL_OVERALL = "overall" + +_PDF_EXTS = frozenset({".pdf"}) +_IMAGE_EXTS = frozenset( + {".png", ".jpg", ".jpeg", ".bmp", ".tif", ".tiff", ".webp", ".gif"} +) +_CAD_EXTS = frozenset({".dwg", ".dxf", ".dwt"}) + + +# --------------------------------------------------------------------------- +# 来源判定 +# --------------------------------------------------------------------------- +def detect_source(path): + """按扩展名判定来源类型（`cad` / `pdf` / `image`）；未知格式抛 ValueError。""" + ext = os.path.splitext(str(path))[1].lower() + if ext in _PDF_EXTS: + return evidence_model.SOURCE_PDF + if ext in _IMAGE_EXTS: + return evidence_model.SOURCE_IMAGE + if ext in _CAD_EXTS: + return evidence_model.SOURCE_CAD + raise ValueError( + f"无法识别的图纸格式：{path}（支持 .dwg/.dxf/.pdf 与常见图片格式）" + ) + + +def is_file_source(source): + """是否为文件源（非 CAD）。""" + return source in (evidence_model.SOURCE_PDF, evidence_model.SOURCE_IMAGE) + + +# --------------------------------------------------------------------------- +# PDF 坐标换算（纯函数，可单测） +# --------------------------------------------------------------------------- +def rotate_point(x, y, media_w, media_h, rotation): + """PDF **未旋转**坐标 → **已旋转**（显示）坐标。 + + `media_w`/`media_h` 为 MediaBox 宽高（未旋转）。rotation 取 0/90/180/270。 + 注：仅 `/Rotate 90` 经实测验证（墨迹命中率 0.335 vs 其它假设 ≤0.064，并目视确认）。 + """ + r = int(rotation) % 360 + if r == 90: + return (y, media_w - x) + if r == 180: + return (media_w - x, media_h - y) + if r == 270: + return (media_h - y, x) + return (x, y) + + +def rect_to_world(rect, media_w, media_h, rotation): + """未旋转 rect(l,b,r,t) → 已旋转世界 bbox(minx,miny,maxx,maxy)。""" + x0, y0, x1, y1 = (float(v) for v in rect) + pts = [ + rotate_point(x0, y0, media_w, media_h, rotation), + rotate_point(x0, y1, media_w, media_h, rotation), + rotate_point(x1, y0, media_w, media_h, rotation), + rotate_point(x1, y1, media_w, media_h, rotation), + ] + xs = [p[0] for p in pts] + ys = [p[1] for p in pts] + return (min(xs), min(ys), max(xs), max(ys)) + + +def crop_of(box, page_w, page_h): + """世界 bbox（y 向上）→ `page.render(crop=...)` 的 (left, bottom, right, top) 裁边量。 + + `crop` 是**要裁掉的边距**（不是绝对矩形），单位为点，且在**旋转后**的页面上生效。 + """ + x0, y0, x1, y1 = (float(v) for v in box) + return ( + max(0.0, x0), + max(0.0, y0), + max(0.0, float(page_w) - x1), + max(0.0, float(page_h) - y1), + ) + + +def guess_sheet_size(w_pt, h_pt, tol=0.02): + """由页尺寸（点）反查标准幅面名（A0~A4，容差 2%）；不匹配返回 None。""" + w_mm, h_mm = float(w_pt) / config.PT_PER_MM, float(h_pt) / config.PT_PER_MM + longest, shortest = max(w_mm, h_mm), min(w_mm, h_mm) + for name, (sw, sh) in tiling.SHEET_SIZES_MM.items(): + if abs(longest - sw) <= sw * tol and abs(shortest - sh) <= sh * tol: + return name + return None + + +# --------------------------------------------------------------------------- +# 墨迹分析（纯 PIL，无 numpy） +# --------------------------------------------------------------------------- +def _empty_grid(box=None): + return { + "window": tiling.box_to_window(box) if box else None, + "cols": 0, "rows": 0, "cell_w": 0.0, "cell_h": 0.0, + "counts": [], "coverage": [], + } + + +def ink_grid(image, box, *, div=None, threshold=None, max_edge=None, min_frac=None): + """把一张图按**墨迹密度**量化成占用网格（与 `heatmap.build_grid` 同形）。 + + 返回 `(grid, boxes)`： + - `grid` ：可直接交给 `heatmap.cluster_cells`； + - `boxes`：占用格的**世界 bbox** 列表（供 `tiling.content_predicate` 剔除空白瓦片）。 + + 步骤：灰度 → 缩到长边 ≤ `max_edge` → 按 `threshold` 二值化（暗 = 墨迹）→ + 用 `Image.BOX` 块均值降到网格尺寸（每个像素即该格的墨迹面积占比）。 + + `counts` 取 `int(frac * 100)`（≥1），故**墨迹越密计数越大**，`cluster_cells` + 的降序排列即是「内容密度降序」，密集区优先出图。 + """ + from PIL import Image + + div = config.HEATMAP_CELL_DIV if div is None else int(div) + threshold = config.INK_THRESHOLD if threshold is None else int(threshold) + max_edge = config.INK_ANALYSIS_MAX_EDGE if max_edge is None else int(max_edge) + min_frac = config.INK_MIN_CELL_FRAC if min_frac is None else float(min_frac) + + box = tiling.norm_box(box) + if box is None or box[2] <= box[0] or box[3] <= box[1] or image is None: + return _empty_grid(box), [] + + win = tiling.box_to_window(box) + w, h = box[2] - box[0], box[3] - box[1] + cell = heatmap.div_to_cell(win, div) + if cell <= 0: + return _empty_grid(box), [] + cols = max(1, int(math.ceil(w / cell))) + rows = max(1, int(math.ceil(h / cell))) + + gray = image.convert("L") + gw, gh = gray.size + if max_edge > 0 and max(gw, gh) > max_edge: + k = max_edge / float(max(gw, gh)) + gray = gray.resize( + (max(1, int(round(gw * k))), max(1, int(round(gh * k)))), Image.BILINEAR + ) + # 二值化：暗于阈值 → 墨迹（255） + bw = gray.point(lambda p: 255 if p < threshold else 0, mode="L") + # 块均值降采样到网格尺寸：每个输出像素 = 该格的墨迹面积占比 × 255 + small = bw.resize((cols, rows), Image.BOX) + px = small.load() + + counts = [[0] * cols for _ in range(rows)] + coverage = [[0.0] * cols for _ in range(rows)] + for r in range(rows): + for c in range(cols): + frac = px[c, r] / 255.0 + coverage[r][c] = frac + if frac >= min_frac: + counts[r][c] = max(1, int(round(frac * 100))) + + grid = { + "window": win, "cols": cols, "rows": rows, + "cell_w": w / cols, "cell_h": h / rows, + "counts": counts, "coverage": coverage, + } + boxes = [heatmap.cell_box(grid, r, c) for r, c in heatmap.occupied_cells(grid)] + return grid, boxes + + +# --------------------------------------------------------------------------- +# 渲染倍率 +# --------------------------------------------------------------------------- +def _tile_zoom(box, max_zoom, max_edge=None): + """由瓦片世界尺寸反推渲染倍率：长边恰好落到 `max_edge` 像素，且不超过 `max_zoom`。 + + 与 `tiling.plan_regions` 的 `s_*` 是同一套数学（`zoom = 1/s`）：规划器把文字簇切到 + `s_text · IMAGE_MAX_EDGE` 世界宽度，这里就恰好还原成 `zoom_text`。 + """ + max_edge = config.IMAGE_MAX_EDGE if max_edge is None else int(max_edge) + box = tiling.norm_box(box) + if box is None: + return float(max_zoom) + longest = max(box[2] - box[0], box[3] - box[1]) + if longest <= 0 or max_edge <= 0: + return float(max_zoom) + return min(float(max_zoom), max_edge / longest) + + +def file_scales(source): + """文件源的 `(s_text, s_line)`——CAD 图形单位 / 像素。 + + PDF 的世界单位是**点**，且 PDF 内容本身就是**纸面 1:1**（出图比例已烘焙进图形）， + 故等价于 `frame_scale = 1`、纸面字高 = `TEXT_HEIGHT_MM · PT_PER_MM` 点。 + 图片的世界单位是**像素**：1 世界单位 = 1 像素，且不允许放大，故 s 恒为 1。 + """ + if source == evidence_model.SOURCE_IMAGE: + return 1.0, 1.0 + text_pt = config.TEXT_HEIGHT_MM * config.PT_PER_MM + return ( + tiling.world_per_px(1.0, text_pt, config.TEXT_READABLE_PX), + tiling.world_per_px(1.0, text_pt, config.LINE_READABLE_PX), + ) + + +def snap_box(box, bounds=None): + """把 bbox 对齐到**整数世界坐标**（图片源的世界单位就是源像素）。 + + 渲染出来的像素只能是整数，若不对齐，注册的 `window` 与实际像素会差最多 1px， + 使「窗口 ↔ 像素」映射不精确（未来按像素反算世界坐标会累积误差）。 + `bounds` 给定时顺带夹回边界；退化返回 None。 + """ + b = tiling.norm_box(box) + if b is None: + return None + x0, y0 = math.floor(b[0]), math.floor(b[1]) + x1, y1 = math.ceil(b[2]), math.ceil(b[3]) + if bounds is not None: + bb = tiling.norm_box(bounds) + if bb is not None: + x0, y0 = max(x0, math.floor(bb[0])), max(y0, math.floor(bb[1])) + x1, y1 = min(x1, math.ceil(bb[2])), min(y1, math.ceil(bb[3])) + if x1 <= x0 or y1 <= y0: + return None + return (float(x0), float(y0), float(x1), float(y1)) + + +def _png_b64(img): + buf = io.BytesIO() + if img.mode not in ("RGB", "L"): + img = img.convert("RGB") + img.save(buf, format="PNG", optimize=True) + return base64.b64encode(buf.getvalue()).decode("ascii") + + +# --------------------------------------------------------------------------- +# 读取器：PDF / 图片 +# --------------------------------------------------------------------------- +class PdfReader: + """PDF 源：按页取证（内存渲染 + best-effort 文本层）。惰性 import pypdfium2。""" + + def __init__(self, path): + import pypdfium2 as pdfium # 惰性：仅 PDF 路径需要 + + self.path = str(path) + self.doc = pdfium.PdfDocument(self.path) + self.page_count = len(self.doc) + self._metas = {} + + # -- 生命周期 ---------------------------------------------------------- + def close(self): + try: + self.doc.close() + except Exception: # noqa: BLE001 + pass + + # -- 页元数据 ---------------------------------------------------------- + def _meta(self, index): + """(旋转后宽, 旋转后高, MediaBox 宽, MediaBox 高, rotation)，均以点计。""" + m = self._metas.get(index) + if m is None: + page = self.doc[index] + media = page.get_mediabox() + mw = float(media[2]) - float(media[0]) + mh = float(media[3]) - float(media[1]) + size = page.get_size() # 已应用 /Rotate + rw, rh = float(size[0]), float(size[1]) + m = (rw, rh, mw, mh, int(page.get_rotation())) + self._metas[index] = m + return m + + # -- 规划 -------------------------------------------------------------- + def frames(self): + """每页一个「图框」（bbox = 页内点坐标，y 向上，已应用 /Rotate）。""" + out = [] + for i in range(self.page_count): + rw, rh, _mw, _mh, _rot = self._meta(i) + out.append( + { + "bbox": [0.0, 0.0, rw, rh], + "kind": "page", + "page_index": i, + "sheet_size": guess_sheet_size(rw, rh), + "sheet_scale": 1.0, + "layer": None, + "confidence": 1.0, + "source_type": evidence_model.SOURCE_PDF, + } + ) + return out + + # -- 渲染 / 文本 ------------------------------------------------------- + def render(self, box, zoom, page_index=0): + """把世界 bbox 渲染成 PIL 图（`crop` 为点、在旋转后页面上生效）。 + + 退化 bbox（宽或高 ≤ 0）会让 `crop` 的边距之和等于页尺寸，pdfium 直接报错， + 故在此提前拒绝。 + """ + box = tiling.norm_box(box) + if box is None or box[2] <= box[0] or box[3] <= box[1]: + return None + rw, rh, _mw, _mh, _rot = self._meta(page_index) + page = self.doc[page_index] + bitmap = page.render( + scale=float(zoom), rotation=0, crop=crop_of(box, rw, rh) + ) + img = bitmap.to_pil() + return img.convert("RGB") if img.mode not in ("RGB", "L") else img + + def text_items(self, index, box=None, min_chars=None, limit=0): + """best-effort 抽文本层：逐文本矩形取串 + 世界 bbox。 + + 国标图纸文字多为 SHX 描边（导出成线条），多数页**整页 0 个文本矩形**—— + 返回空列表属**预期**，不是 bug。真正的识别靠 VLM 看渲染图。 + """ + min_chars = config.PDF_TEXT_MIN_CHARS if min_chars is None else int(min_chars) + _rw, _rh, mw, mh, rot = self._meta(index) + page = self.doc[index] + try: + tp = page.get_textpage() + except Exception: # noqa: BLE001 - 无文本层 / 解析失败都按「无文本」处理 + return [] + try: + total = tp.count_rects() # get_rect 的前置调用（pdfium 要求） + items = [] + for k in range(total): + if limit and len(items) >= limit: + break + try: + rect = tp.get_rect(k) + text = (tp.get_text_bounded(*rect) or "").strip() + except Exception: # noqa: BLE001 - 单个矩形失败不影响整页 + continue + if len(text) < min_chars: + continue + world = rect_to_world(rect, mw, mh, rot) + if box is not None and not tiling.boxes_intersect(world, box): + continue + items.append({"content": text, "bbox": list(world)}) + return items + finally: + try: + tp.close() + except Exception: # noqa: BLE001 + pass + + +class ImageReader: + """图片源：整图即一页，按需裁剪（**只缩不放**，放大只会插值出假细节）。""" + + def __init__(self, path): + from PIL import Image + + self.path = str(path) + self.img = Image.open(self.path) + self.img.load() + self.width, self.height = self.img.size + self.page_count = 1 + + def close(self): + try: + self.img.close() + except Exception: # noqa: BLE001 + pass + + def frames(self): + return [ + { + "bbox": [0.0, 0.0, float(self.width), float(self.height)], + "kind": "image", + "page_index": 0, + "sheet_size": None, + "sheet_scale": 1.0, + "layer": None, + "confidence": 1.0, + "source_type": evidence_model.SOURCE_IMAGE, + } + ] + + def render(self, box, zoom, page_index=0): + from PIL import Image + + # 世界单位 = 源像素 → 对齐到整数，保证「窗口 ↔ 像素」精确对应 + box = snap_box(box, (0.0, 0.0, float(self.width), float(self.height))) + if box is None: + return None + # 世界 y 向上 → 图像 y 向下 + left = int(box[0]) + upper = int(self.height - box[3]) + right = int(box[2]) + lower = int(self.height - box[1]) + if right <= left or lower <= upper: + return None + out = self.img.crop((left, upper, right, lower)) + if zoom and 0 < zoom < 1.0: + out = out.resize( + ( + max(1, int(round(out.size[0] * zoom))), + max(1, int(round(out.size[1] * zoom))), + ), + Image.LANCZOS, + ) + return out.convert("RGB") if out.mode not in ("RGB", "L") else out + + def text_items(self, index=0, box=None, **kwargs): + """图片没有文本层。""" + return [] + + +# --------------------------------------------------------------------------- +# 读取器缓存（挂在 ctx 上，避免每页重复打开 PDF） +# --------------------------------------------------------------------------- +def open_reader(ctx): + """打开（并缓存到 `ctx.source_reader`）文件源读取器。""" + source = ctx.source + if not is_file_source(source): + raise ValueError(f"来源 {source} 不是文件源，无需读取器") + reader = getattr(ctx, "source_reader", None) + if reader is not None: + return reader + path = getattr(ctx, "source_path", None) + if not path: + raise ValueError("文件源未指定路径（ctx.source_path 为空）") + reader = PdfReader(path) if source == evidence_model.SOURCE_PDF else ImageReader(path) + ctx.source_reader = reader + return reader + + +def close_reader(ctx): + """释放文件源读取器（幂等）。""" + reader = getattr(ctx, "source_reader", None) + if reader is not None: + reader.close() + ctx.source_reader = None + + +# --------------------------------------------------------------------------- +# 规划（与 phases.planning 的 CAD 分支同形） +# --------------------------------------------------------------------------- +def plan_file(ctx): + """文件源的「规划」：产出与 CAD 同形的 plan（每页 / 整图一个图框，保持原始页序）。""" + reader = open_reader(ctx) + frames = reader.frames() + if ctx.source == evidence_model.SOURCE_PDF and config.PDF_MAX_PAGES > 0: + frames = frames[: config.PDF_MAX_PAGES] + if ctx.max_frames > 0: + frames = frames[: ctx.max_frames] + info = { + "source_type": ctx.source, + "path": ctx.source_path, + "page_count": getattr(reader, "page_count", len(frames)), + "unit": "pt" if ctx.source == evidence_model.SOURCE_PDF else "px", + } + return { + "drawing_info": info, + "layers": evidence_model.empty_envelope(), + "layouts": evidence_model.empty_envelope(), + "frames": frames, + } + + +# --------------------------------------------------------------------------- +# 取证 +# --------------------------------------------------------------------------- +def _same_box(a, b, tol=1e-6): + """两个 bbox 是否（近似）相同。""" + return a is not None and b is not None and all( + abs(float(x) - float(y)) <= tol for x, y in zip(a, b) + ) + + +def _capture_file(ctx, frame, box, reader, source, text_boxes): + """采集视觉证据：整体一张 + 自适应墨迹瓦片若干。""" + page_index = frame.get("page_index", 0) + is_pdf = source == evidence_model.SOURCE_PDF + max_zoom = config.PDF_TILE_MAX_ZOOM if is_pdf else config.IMAGE_MAX_ZOOM + overview_zoom = config.PDF_OVERVIEW_MAX_ZOOM if is_pdf else config.IMAGE_MAX_ZOOM + + images = [] + + def _add(img, label, win): + if img is None or len(images) >= config.EVIDENCE_MAX_IMAGES: + return False + images.append( + evidence_model.make_image( + _png_b64(img), "image/png", label, window=list(win) + ) + ) + return True + + # 整体图：整页 / 整图渲染（长边 ≈ IMAGE_MAX_EDGE，且不超过 overview 上限） + overall_zoom = _tile_zoom(box, overview_zoom) + _add(reader.render(box, overall_zoom, page_index), _LABEL_OVERALL, box) + + if not config.TILING_ENABLED: + return images + + # 墨迹分析：低倍缩略图 → 占用网格 → 簇 → 瓦片 + thumb = reader.render( + box, _tile_zoom(box, max_zoom, config.INK_ANALYSIS_MAX_EDGE), page_index + ) + grid, ink_boxes = ink_grid(thumb, box) + clusters = heatmap.cluster_cells(grid, dilate=config.HEATMAP_CLUSTER_DILATE) + if not clusters: + return images + + s_text, s_line = file_scales(source) + if s_text <= 0 or s_line <= 0: + return images + flags = tiling.cluster_text_flags(clusters, text_boxes) + text_idx = {i for i, is_text in enumerate(flags) if is_text} + + regions = tiling.plan_regions( + frame, clusters, + text_clusters=text_idx, s_text=s_text, s_line=s_line, + view_px=(config.IMAGE_MAX_EDGE, config.IMAGE_MAX_EDGE), + max_tiles=config.TILE_MAX_COUNT, overlap=config.TILE_OVERLAP, + keep=tiling.content_predicate(ink_boxes), + ) + n = 0 + for win in regions: + if len(images) >= config.EVIDENCE_MAX_IMAGES: + break + tile_box = tiling.window_to_box(win) + if tile_box is None: + continue + if not is_pdf: + # 图片源世界单位是源像素：先对齐整数，注册的 window 才与像素精确对应 + tile_box = snap_box(tile_box, box) + if tile_box is None: + continue + zoom = _tile_zoom(tile_box, max_zoom) + # 整页即一个簇时，瓦片窗口会与整体图重合且分辨率不更高 → 省掉这张重复图 + if _same_box(tile_box, box) and zoom <= overall_zoom + 1e-9: + continue + if _add(reader.render(tile_box, zoom, page_index), f"tile-{n + 1}", tile_box): + n += 1 + return images + + +def collect(ctx, index, unit): + """文件源取证入口；产出与 `phases.evidence.collect_frame` **同形**的 evidence。""" + reader = open_reader(ctx) + box = tiling.frame_bbox(unit) + if box is None: + return None + source = evidence_model.source_type_of(unit) + + evidence = { + "source_type": source, + "index": index, + "frame": unit, + "window": tiling.box_to_window(box), + "entities": evidence_model.empty_envelope(), + "texts": evidence_model.empty_envelope(), + "dimensions": evidence_model.empty_envelope(), + "blocks": evidence_model.empty_envelope(), + "polylines": evidence_model.empty_envelope(), + "images": [], + } + + # best-effort 文本层（仅 PDF 可能有；图片恒为空） + page_index = unit.get("page_index", 0) + items = reader.text_items(page_index, box, limit=ctx.frame_limit) + evidence["texts"] = {"total": len(items), "items": items} + text_boxes = [i["bbox"] for i in items if i.get("bbox")] + + evidence["images"] = _capture_file(ctx, unit, box, reader, source, text_boxes) + return evidence diff --git a/dwg_review/tiling.py b/dwg_review/tiling.py index b0ec5ee..beec5c8 100644 --- a/dwg_review/tiling.py +++ b/dwg_review/tiling.py @@ -223,6 +223,26 @@ def visible_box(box, view_px, margin=0.0): return (cx - vw / 2.0, cy - vh / 2.0, cx + vw / 2.0, cy + vh / 2.0) +def content_predicate(boxes): + """构造「该窗口内是否有内容」谓词；无 box 返回 None。 + + 用于剔除空白瓦片：簇 / 象限的 bbox 只是**包围盒**，L 形或斜向分布会含大片空白。 + 这里直接按**内容 bbox** 判断——比先量化成占用格更精确（格子的跨界膨胀会把 + 相邻空白瓦片也判成有内容）。接触即算有内容（保守，宁可多拍不漏）。 + + CAD 侧传实体 bbox；PDF / 图片侧传**占用墨迹格**的 bbox（见 `sources.ink_grid`）。 + """ + boxes = [b for b in (norm_box(x) for x in (boxes or [])) if b] + if not boxes: + return None + + def _has_content(win): + box = window_to_box(win) + return box is not None and any(boxes_intersect(box, b) for b in boxes) + + return _has_content + + def cluster_text_flags(clusters, text_boxes): """标记每个簇是否含文字类实体（bbox 相交即算，保守）。 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..b8a1e21 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,12 @@ +# dwg_review 运行时依赖 +# 安装：pip install -r requirements.txt +# +# 说明： +# - openai ：LLM 客户端（审查 / 汇总）。API Key 走环境变量 DEEPSEEK_API_KEY。 +# - Pillow ：截图下采样、PDF/图片墨迹分析、离线标定脚本。 +# - pypdfium2 ：PDF 源适配（内存渲染 + 带 bbox 抽文本）。 +# 选它而非 PyMuPDF 的原因：Apache-2.0/BSD 许可证干净、 +# wheel 为 py3-none-（与 Python 版本无关）、体积小（~4MB）。 +openai>=1.0 +Pillow>=10.0 +pypdfium2>=4.30 diff --git a/tests/fakes.py b/tests/fakes.py index f42a080..2e24d4e 100644 --- a/tests/fakes.py +++ b/tests/fakes.py @@ -281,12 +281,15 @@ class FakeMcpClient: # --------------------------------------------------------------------------- # 组合辅助 # --------------------------------------------------------------------------- -def make_ctx(client=None, llm=None, checklist=None, **over): - """直接拼装一个 ReviewContext（不触碰网络、不读密钥）。""" +def make_ctx(client=None, llm=None, checklist=None, connect_cad=True, **over): + """直接拼装一个 ReviewContext（不触碰网络、不读密钥）。 + + `connect_cad=False` → `client` / `tools` 置 None，模拟 PDF / 图片**离线源**上下文。 + """ client = client if client is not None else FakeMcpClient() params = dict( - client=client, - tools=CadTools(client), + client=client if connect_cad else None, + tools=CadTools(client) if connect_cad else None, llm=llm if llm is not None else FakeLlmClient(), checklist=checklist if checklist is not None else DEFAULT_CHECKLIST, max_frames=0, diff --git a/tests/test_annotator.py b/tests/test_annotator.py index c16c3c9..5fabd3e 100644 --- a/tests/test_annotator.py +++ b/tests/test_annotator.py @@ -165,3 +165,21 @@ def test_clear_swallows_tool_errors(): ctx = make_ctx(client=client) clear(ctx, handles=["N1"], log=logs.append) assert any("撤销失败" in line for line in logs) + + +# --------------------------------------------------------------------------- +# 离线源（tools=None） +# --------------------------------------------------------------------------- +def test_annotate_offline_returns_empty_without_touching_cad(): + logs = [] + ctx = make_ctx(connect_cad=False) + created, highlighted = annotate( + ctx, [{"handle": "H1", "severity": "high", "message": "x"}], log=logs.append + ) + assert created == [] and highlighted == [] + assert any("离线" in line for line in logs) + + +def test_clear_offline_noop(): + ctx = make_ctx(connect_cad=False) + clear(ctx, handles=["N1"], highlighted=["A"], log=lambda *a: None) # 不应抛异常 diff --git a/tests/test_checks.py b/tests/test_checks.py index 3d5ee0a..eb54a86 100644 --- a/tests/test_checks.py +++ b/tests/test_checks.py @@ -304,3 +304,19 @@ def test_run_collisions_when_enabled(monkeypatch): assert "C1" in by_handle assert by_handle["C1"]["category"] == "干涉" assert ctx.client.arguments_of("find_collisions")["min_overlap"] == 0.6 + + +# --------------------------------------------------------------------------- +# 离线源（tools=None） +# --------------------------------------------------------------------------- +def test_run_offline_only_derives_from_evidence(): + """离线源没有 CAD 工具：只跑纯 Python 派生检查，不调任何 MCP 工具。""" + ctx = make_ctx(connect_cad=False) + ev = make_evidence( + texts={"total": 1, "items": [{"handle": "T1", "content": " "}]}, + polylines={"total": 1, "items": [ + {"handle": "P1", "vertex_count": 4, "closed": False}]}, + ) + issues = run_deterministic_checks(ctx, ev, log=QUIET) + cats = {i["category"] for i in issues} + assert "文字" in cats and "几何" in cats diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..6dd555e --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,60 @@ +"""CLI 入口（`dwg_review.__main__`）参数解析与来源判定单测。 + +只测**纯逻辑**（不真连 CAD、不真调 LLM）：位置参数 → (source, source_path) 的判定， +以及 CAD-only 选项在离线源下被拒绝。 +""" + +from PIL import Image + +from dwg_review.__main__ import _resolve_source_arg, build_parser + + +def test_parser_accepts_positional_source(): + args = build_parser().parse_args(["a.pdf"]) + assert args.source == "a.pdf" + + +def test_parser_source_optional(): + assert build_parser().parse_args([]).source is None + + +def test_parser_keeps_existing_flags(): + args = build_parser().parse_args(["a.pdf", "--plan-only", "--frames", "3"]) + assert args.plan_only is True and args.frames == 3 + + +def test_resolve_no_source_is_cad_mode(): + assert _resolve_source_arg(None) == (None, None) + + +def test_resolve_pdf(tmp_path): + path = tmp_path / "d.pdf" + path.write_bytes(b"%PDF-1.4\n") + resolved, err = _resolve_source_arg(str(path)) + assert err is None + assert resolved[0] == "pdf" + assert resolved[1].endswith("d.pdf") + + +def test_resolve_image(tmp_path): + path = tmp_path / "d.png" + Image.new("RGB", (10, 10)).save(path) + resolved, err = _resolve_source_arg(str(path)) + assert err is None and resolved[0] == "image" + + +def test_resolve_dwg_falls_back_to_cad_mode(capsys): + """DWG 参数无法在 Python 侧打开 → 提示后按 CAD 在线模式运行（source=None）。""" + resolved, err = _resolve_source_arg("x.dwg") + assert resolved is None and err is None + assert "CAD 在线模式" in capsys.readouterr().out + + +def test_resolve_missing_file_errors(tmp_path): + resolved, err = _resolve_source_arg(str(tmp_path / "nope.pdf")) + assert resolved is None and "文件不存在" in err + + +def test_resolve_unknown_extension_errors(): + resolved, err = _resolve_source_arg("x.svg") + assert resolved is None and "无法识别" in err diff --git a/tests/test_context.py b/tests/test_context.py new file mode 100644 index 0000000..2fb7f7c --- /dev/null +++ b/tests/test_context.py @@ -0,0 +1,45 @@ +"""`dwg_review.context.ReviewContext` 单元测试（重点：离线模式不碰网络）。""" + +from dwg_review import evidence_model +from dwg_review.context import ReviewContext + + +def test_create_offline_has_no_client_or_tools(): + """connect_cad=False → client / tools 均为 None（不构造 CadTools，不连 8080）。""" + ctx = ReviewContext.create(need_llm=False, connect_cad=False) + assert ctx.client is None + assert ctx.tools is None + assert ctx.llm is None + # 离线上下文仍应加载好清单（纯文件读取） + assert isinstance(ctx.checklist, dict) + + +def test_create_offline_default_source_is_cad(): + ctx = ReviewContext.create(need_llm=False, connect_cad=False) + assert ctx.source == evidence_model.SOURCE_CAD + + +def test_create_offline_explicit_source(): + ctx = ReviewContext.create(need_llm=False, connect_cad=False, source=evidence_model.SOURCE_PDF) + assert ctx.source == evidence_model.SOURCE_PDF + + +def test_create_offline_overrides_apply(): + ctx = ReviewContext.create( + need_llm=False, connect_cad=False, max_frames=3, frame_limit=9, dry_run=True + ) + assert ctx.max_frames == 3 + assert ctx.frame_limit == 9 + assert ctx.dry_run is True + + +def test_create_offline_does_not_connect(monkeypatch): + """确保离线路径完全没有实例化 McpClient / CadTools（真连会抛 OSError）。""" + + def _boom(*a, **k): # pragma: no cover - 被调用即失败 + raise AssertionError("离线模式不应触碰 CAD 客户端") + + monkeypatch.setattr("dwg_review.context.McpClient", _boom) + monkeypatch.setattr("dwg_review.context.CadTools", _boom) + ctx = ReviewContext.create(need_llm=False, connect_cad=False) + assert ctx.client is None and ctx.tools is None diff --git a/tests/test_evidence.py b/tests/test_evidence.py index 9976ecc..39ce74b 100644 --- a/tests/test_evidence.py +++ b/tests/test_evidence.py @@ -169,7 +169,25 @@ def test_collect_returns_none_for_unsupported_source(): from dwg_review.phases.evidence import collect ctx = make_ctx(client=FakeMcpClient()) - assert collect(ctx, 1, {"source_type": "pdf", "bbox": [0, 0, 1, 1]}) is None + # 未知来源（非 cad/pdf/image）→ 无法取证 + assert collect(ctx, 1, {"source_type": "svg", "bbox": [0, 0, 1, 1]}) is None + + +def test_collect_dispatches_pdf_source_to_sources(monkeypatch): + """pdf / image 来源交给 `sources.collect`（此处打桩，避免真去解析文件）。""" + from dwg_review.phases import evidence as ev_mod + + calls = [] + + def _fake(ctx, index, unit): + calls.append((index, unit.get("source_type"))) + return {"source_type": unit.get("source_type"), "index": index} + + monkeypatch.setattr(ev_mod.sources, "collect", _fake) + ctx = make_ctx(client=FakeMcpClient()) + out = ev_mod.collect(ctx, 3, {"source_type": "pdf", "bbox": [0, 0, 1, 1]}) + assert out == {"source_type": "pdf", "index": 3} + assert calls == [(3, "pdf")] # --------------------------------------------------------------------------- @@ -402,6 +420,14 @@ def test_maximized_drawing_area_disabled_by_config(drawing_area_env): assert client.calls == [] +def test_maximized_drawing_area_offline_passthrough(drawing_area_env): + """离线源（tools=None）即使开启最大化也直接放行，不碰 CAD。""" + ctx = make_ctx(connect_cad=False) + assert ctx.tools is None + with maximized_drawing_area(ctx): + pass # 不应抛 AttributeError + + def test_maximized_drawing_area_restores_even_if_body_raises(drawing_area_env): client = _area_client() ctx = make_ctx(client=client) diff --git a/tests/test_evidence_model.py b/tests/test_evidence_model.py index a917edc..806594d 100644 --- a/tests/test_evidence_model.py +++ b/tests/test_evidence_model.py @@ -19,6 +19,21 @@ def test_make_image_empty_mime_falls_back(): assert em.make_image("x", "")["mime"] == "image/png" +def test_empty_envelope_default(): + assert em.empty_envelope() == {"total": 0, "items": []} + + +def test_empty_envelope_total(): + env = em.empty_envelope(7) + assert env == {"total": 7, "items": []} + # items 必须是**新列表**（调用方可能各自 append，不能共享同一对象） + assert em.empty_envelope()["items"] is not em.empty_envelope()["items"] + + +def test_empty_envelope_int_coercion(): + assert em.empty_envelope("3")["total"] == 3 + + def test_source_type_of_default_cad(): assert em.source_type_of({}) == em.SOURCE_CAD assert em.source_type_of(None) == em.SOURCE_CAD diff --git a/tests/test_offline.py b/tests/test_offline.py new file mode 100644 index 0000000..b3f5fce --- /dev/null +++ b/tests/test_offline.py @@ -0,0 +1,217 @@ +"""离线源（PDF / 图片）**全链路**单测：规划 → 取证 → 审查 → 汇总，全程不碰 CAD。 + +与 `test_sources*.py` 的分工：那边测 `sources` 模块本身；这里测**跨阶段贯通**—— +离线上下文能否一路走完 `planning` / `evidence` / `checks` / `review` / `findings` / +`report` / `annotator`，且任何环节都不会去访问 `ctx.tools`（离线时它恒为 None）。 +""" + +import json + +import pytest + +from dwg_review import config, evidence_model, sources +from dwg_review.annotator import annotate +from dwg_review.checks import run_deterministic_checks +from dwg_review.checklist import build_system_prompt +from dwg_review.findings import build_finding, flatten_issues +from dwg_review.phases import evidence as evidence_phase +from dwg_review.phases import planning, report, review +from tests.fakes import FakeLlmClient, llm_calls, llm_text, make_ctx +from tests.test_sources_pdf import _build_pdf + + +@pytest.fixture +def pdf2(tmp_path): + """两页 PDF：第 1 页有文本层，第 2 页整页有墨迹（会切瓦片）。""" + path = tmp_path / "two.pdf" + path.write_bytes(_build_pdf([{"text": "HELLO"}, {"fill": True, "text": ""}])) + return str(path) + + +@pytest.fixture +def offline_ctx(pdf2, monkeypatch): + monkeypatch.setattr(config, "TILING_ENABLED", True) + monkeypatch.setattr(config, "DETERMINISTIC_CHECKS", True) + monkeypatch.setattr(config, "VALIDATE_HANDLES", True) + ctx = make_ctx( + connect_cad=False, + source=evidence_model.SOURCE_PDF, + source_path=pdf2, + llm=FakeLlmClient(text=json.dumps({"summary": "未发现明显问题", "issues": []})), + ) + yield ctx + sources.close_reader(ctx) + + +# --------------------------------------------------------------------------- +# 提示词 +# --------------------------------------------------------------------------- +def test_offline_prompt_has_no_tool_protocol(): + prompt = build_system_prompt( + {"name": "T", "categories": [{"title": "几何", "checks": ["x"]}]}, + allow_tools=False, + offline=True, + ) + assert "inspect_region" not in prompt + assert "find_collisions" not in prompt + assert "handle" in prompt # 明确要求**不要**输出 handle + assert "不要输出 `handle`" in prompt + assert "没有实体级结构化数据" in prompt + + +def test_offline_prompt_still_lists_checklist(): + prompt = build_system_prompt( + {"name": "T", "categories": [{"title": "干涉/碰撞", "severity": "high", + "checks": ["构件侵入"]}]}, + allow_tools=False, + offline=True, + ) + assert "干涉/碰撞" in prompt and "构件侵入" in prompt + + +def test_cad_prompt_unchanged_by_default(): + prompt = build_system_prompt({"name": "T", "categories": []}, allow_tools=True) + assert "inspect_region" in prompt + assert "结构化图元数据" in prompt + + +# --------------------------------------------------------------------------- +# 全链路 +# --------------------------------------------------------------------------- +def test_offline_pipeline_end_to_end(offline_ctx, tmp_path, monkeypatch): + monkeypatch.setattr(config, "REPORT_DIR", str(tmp_path / "reports")) + ctx = offline_ctx + + # 1) 规划：每页一个图框，保持页序，完全不调 CAD + plan = planning.plan(ctx) + frames = plan["frames"] + assert [f["page_index"] for f in frames] == [0, 1] + + # 2) 取证：同形 evidence + evidences = [] + for i, frame in enumerate(frames, 1): + ev = evidence_phase.collect(ctx, i, frame) + assert ev is not None + evidences.append(ev) + assert evidences[0]["texts"]["total"] == 1 + assert evidences[0]["images"][0]["label"] == "overall" + + # 3) 确定性检查：离线只跑纯 Python 派生（空文字 / 未闭合多段线），不报错 + for ev in evidences: + assert run_deterministic_checks(ctx, ev, log=lambda *a: None) == [] + + # 4) 审查：离线提示词 + 无工具 + prompt = build_system_prompt(ctx.checklist, allow_tools=False, offline=True) + findings = [] + for ev in evidences: + conclusion = review.review_frame(ctx, ev, prompt, log=lambda *a: None) + findings.append(build_finding(ev, conclusion, [], validate_handles=True)) + assert all(f["summary"] == "未发现明显问题" for f in findings) + + # 5) 汇总 + 落盘（换一个纯文本假 LLM 用于汇总） + ctx.llm = FakeLlmClient(text="汇总：未发现明显问题。") + md = report.summarize(ctx, findings, log=lambda *a: None) + assert md == "汇总：未发现明显问题。" + md_path, json_path = report.save_report(md, findings, plan, flatten_issues(findings), + log=lambda *a: None) + assert md_path.exists() and json_path.exists() + saved = json.loads(json_path.read_text(encoding="utf-8")) + assert len(saved["frames"]) == 2 + + # 6) 批注：离线直接跳过，不抛异常 + created, highlighted = annotate(ctx, flatten_issues(findings), log=lambda *a: None) + assert created == [] and highlighted == [] + + +def test_offline_pipeline_never_touches_tools(offline_ctx): + """离线全链路的每一步都不应访问 `ctx.tools`（它恒为 None）。""" + ctx = offline_ctx + assert ctx.tools is None and ctx.client is None + + plan = planning.plan(ctx) + ev = evidence_phase.collect(ctx, 1, plan["frames"][0]) + conclusion = review.review_frame(ctx, ev, "sys", log=lambda *a: None) + finding = build_finding(ev, conclusion, [], validate_handles=True) + assert finding["frame"] == 1 + + +def test_offline_llm_handle_is_stripped(offline_ctx): + """离线证据里没有 handle → LLM 若硬报 handle 会被剥掉并标为未核实。""" + ctx = offline_ctx + plan = planning.plan(ctx) + ev = evidence_phase.collect(ctx, 1, plan["frames"][0]) + ctx.llm = FakeLlmClient( + text=json.dumps( + { + "summary": "有问题", + "issues": [{"category": "几何", "severity": "high", + "message": "m", "handle": "DEADBEEF"}], + } + ) + ) + conclusion = review.review_frame(ctx, ev, "sys", log=lambda *a: None) + finding = build_finding(ev, conclusion, [], validate_handles=True) + issue = finding["issues"][0] + assert issue["handle"] is None + assert issue["verified"] is False + assert finding["counts"]["unverified"] == 1 + + +def test_offline_review_provides_no_tools_to_llm(offline_ctx): + """离线审查给 LLM 的 tools 列表必须是空（不能让它请求 CAD 工具）。""" + ctx = offline_ctx + plan = planning.plan(ctx) + ev = evidence_phase.collect(ctx, 1, plan["frames"][0]) + review.review_frame(ctx, ev, "sys", log=lambda *a: None) + assert ctx.llm.calls[-1]["tools"] == [] + + +def test_offline_review_survives_tool_call_attempt(offline_ctx): + """LLM 仍尝试调工具时：被拒绝（离线提示），随后照常收口出结论。""" + ctx = offline_ctx + plan = planning.plan(ctx) + ev = evidence_phase.collect(ctx, 1, plan["frames"][0]) + ctx.llm = FakeLlmClient( + script=[ + llm_calls([("query_entities", {})]), + llm_text(json.dumps({"summary": "ok", "issues": []})), + ] + ) + conclusion = review.review_frame(ctx, ev, "sys", log=lambda *a: None) + assert conclusion["summary"] == "ok" + + +def test_offline_evidence_is_json_serializable(offline_ctx): + ctx = offline_ctx + plan = planning.plan(ctx) + ev = evidence_phase.collect(ctx, 1, plan["frames"][0]) + trimmed = {k: v for k, v in ev.items() if k not in ("images",)} + json.dumps(trimmed, ensure_ascii=False) # 不抛异常即可 + + +def test_offline_image_source_pipeline(tmp_path, monkeypatch): + """图片源同样能走完全链路。""" + from PIL import Image, ImageDraw + + monkeypatch.setattr(config, "TILING_ENABLED", True) + img = Image.new("RGB", (800, 600), (255, 255, 255)) + ImageDraw.Draw(img).rectangle([50, 50, 400, 300], outline=(0, 0, 0), width=3) + path = tmp_path / "plan.png" + img.save(path) + + ctx = make_ctx( + connect_cad=False, + source=evidence_model.SOURCE_IMAGE, + source_path=str(path), + llm=FakeLlmClient(text=json.dumps({"summary": "s", "issues": []})), + ) + try: + plan = planning.plan(ctx) + assert plan["drawing_info"]["unit"] == "px" + ev = evidence_phase.collect(ctx, 1, plan["frames"][0]) + assert ev["source_type"] == "image" + assert ev["images"][0]["label"] == "overall" + conclusion = review.review_frame(ctx, ev, "sys", log=lambda *a: None) + assert conclusion["summary"] == "s" + finally: + sources.close_reader(ctx) diff --git a/tests/test_planning.py b/tests/test_planning.py index c48685b..550e41c 100644 --- a/tests/test_planning.py +++ b/tests/test_planning.py @@ -91,3 +91,53 @@ def test_plan_tolerates_non_dict_frames_data(): def test_plan_tolerates_missing_items_key(): client = FakeMcpClient(responses={"detect_drawing_frames": ok_text({"other": 1})}) assert plan(make_ctx(client=client))["frames"] == [] + + +# --------------------------------------------------------------------------- +# 文件源分发（PDF / 图片） +# --------------------------------------------------------------------------- +def test_plan_dispatches_file_source_to_sources(tmp_path, monkeypatch): + """ctx.source 为文件源时走 `sources.plan_file`，**完全不碰 CAD 工具**。""" + from PIL import Image + + from dwg_review import sources + + path = tmp_path / "tiny.png" + Image.new("RGB", (60, 40), (255, 255, 255)).save(path) + + ctx = make_ctx(connect_cad=False, source="image", source_path=str(path)) + try: + result = plan(ctx) + finally: + sources.close_reader(ctx) + + assert result["drawing_info"]["unit"] == "px" + assert len(result["frames"]) == 1 + assert result["frames"][0]["source_type"] == "image" + # 离线：没有任何 MCP 调用（ctx.tools 为 None 本身就是证据） + assert ctx.tools is None + + +def test_plan_file_source_does_not_sort_frames(tmp_path, monkeypatch): + """文件源**保持原始页序**（阅读顺序由文档自身决定，不能按 bbox 重排）。""" + from dwg_review import sources + + ctx = make_ctx(connect_cad=False, source="pdf", source_path="unused.pdf", max_frames=3) + + class _Reader: + page_count = 3 + + def frames(self): + # 故意让 bbox 顺序与页序相反（后页在上方） + return [ + {"bbox": [0.0, 0.0, 10.0, 10.0], "page_index": 0, "source_type": "pdf"}, + {"bbox": [0.0, 50.0, 10.0, 60.0], "page_index": 1, "source_type": "pdf"}, + {"bbox": [0.0, 90.0, 10.0, 100.0], "page_index": 2, "source_type": "pdf"}, + ] + + def close(self): + pass + + ctx.source_reader = _Reader() + result = sources.plan_file(ctx) + assert [f["page_index"] for f in result["frames"]] == [0, 1, 2] diff --git a/tests/test_review.py b/tests/test_review.py index ca21a1e..e35f708 100644 --- a/tests/test_review.py +++ b/tests/test_review.py @@ -401,3 +401,45 @@ def test_readonly_inspect_region_unavailable_without_evidence(): ctx = make_ctx(client=FakeMcpClient()) content, _ = _run_readonly_tool(ctx, "inspect_region", {}, [6]) assert "不可用" in content[0]["text"] + + +# --------------------------------------------------------------------------- +# 离线源（tools=None） +# --------------------------------------------------------------------------- +def test_review_tools_offline_returns_empty(): + ctx = make_ctx(connect_cad=False) + assert ctx.tools is None + assert _review_tools(ctx) == [] + + +def test_readonly_tool_offline_rejected(): + ctx = make_ctx(connect_cad=False) + content, budget = _run_readonly_tool(ctx, "query_entities", {}, [3]) + assert "离线" in content[0]["text"] + assert budget == [3] + + +def test_readonly_tool_offline_rejects_inspect_region(): + """离线源连 inspect_region 虚拟工具也不可用（它依赖 CAD 截图）。""" + ctx = make_ctx(connect_cad=False) + content, _ = _run_readonly_tool(ctx, "inspect_region", {}, [3]) + assert "离线" in content[0]["text"] + + +def test_review_frame_offline_never_calls_cad(): + """离线源审查：LLM 即使请求工具也不会碰 CAD，直接收口给出结论。""" + llm = FakeLlmClient(script=[ + llm_calls([("query_entities", {})]), + llm_text(json.dumps({"summary": "s", "issues": []})), + ]) + ctx = make_ctx(llm=llm, connect_cad=False) + out = review_frame(ctx, make_evidence(), "sys", log=lambda *a: None) + assert out["summary"] == "s" + + +def test_reset_viewport_offline_noop(): + from dwg_review.phases.review import _reset_viewport + + ctx = make_ctx(connect_cad=False) + ev = make_evidence() + _reset_viewport(ctx, ev) # 不应抛异常 diff --git a/tests/test_sources.py b/tests/test_sources.py new file mode 100644 index 0000000..6b053b7 --- /dev/null +++ b/tests/test_sources.py @@ -0,0 +1,401 @@ +"""`dwg_review.sources` 单元测试（纯函数 + 图片源；PDF 真机在 verify_pdf_source.py）。 + +设计约束：本文件**不依赖 pypdfium2**（PDF 读取器惰性 import，只有真处理 PDF 才需要）。 +""" + +import base64 + +import pytest + +from dwg_review import config, heatmap, sources +from tests.fakes import make_ctx + + +# --------------------------------------------------------------------------- +# detect_source / is_file_source +# --------------------------------------------------------------------------- +@pytest.mark.parametrize( + "name,expect", + [ + ("a.dwg", "cad"), + ("a.DXF", "cad"), + ("b.pdf", "pdf"), + ("c.PDF", "pdf"), + ("d.png", "image"), + ("e.JPG", "image"), + ("f.tiff", "image"), + ("g.webp", "image"), + ], +) +def test_detect_source(name, expect): + assert sources.detect_source(name) == expect + + +def test_detect_source_unknown_raises(): + with pytest.raises(ValueError): + sources.detect_source("x.svg") + + +def test_is_file_source(): + assert sources.is_file_source("pdf") and sources.is_file_source("image") + assert not sources.is_file_source("cad") + + +# --------------------------------------------------------------------------- +# PDF 坐标换算 +# --------------------------------------------------------------------------- +def test_rotate_point_identity(): + assert sources.rotate_point(3, 4, 100, 200, 0) == (3, 4) + + +def test_rotate_point_90_matches_probe(): + """/Rotate 90：(x,y) → (y, mw - x)（实测验证过的映射）。""" + mw, mh = 842.0, 1191.0 + assert sources.rotate_point(0, 0, mw, mh, 90) == (0.0, 842.0) + assert sources.rotate_point(mw, mh, mw, mh, 90) == (mh, 0.0) + + +def test_rotate_point_180_and_270_stay_in_range(): + mw, mh = 842.0, 1191.0 + for rot, (rw, rh) in ((180, (mw, mh)), (270, (mh, mw))): + for x, y in ((0, 0), (mw, mh), (100, 200)): + px, py = sources.rotate_point(x, y, mw, mh, rot) + assert 0 <= px <= rw and 0 <= py <= rh + + +def test_rect_to_world_rot90(): + mw, mh = 842.0, 1191.0 + # 未旋转的 (10,20)-(30,40) → 旋转后 x 取 [20,40]，y 取 [842-30, 842-10] + world = sources.rect_to_world((10, 20, 30, 40), mw, mh, 90) + assert world == (20.0, 812.0, 40.0, 832.0) + + +def test_crop_of_is_margin_not_rect(): + """crop 是「裁掉的边距」（left,bottom,right,top），不是绝对矩形。""" + assert sources.crop_of((10, 20, 110, 220), 200, 300) == (10.0, 20.0, 90.0, 80.0) + + +def test_crop_of_clamps_negative(): + assert sources.crop_of((-5, -5, 300, 400), 200, 300) == (0.0, 0.0, 0.0, 0.0) + + +# --------------------------------------------------------------------------- +# 幅面反查 +# --------------------------------------------------------------------------- +def test_guess_sheet_size_a3_landscape(): + w = 420.0 * config.PT_PER_MM + h = 297.0 * config.PT_PER_MM + assert sources.guess_sheet_size(w, h) == "A3" + + +def test_guess_sheet_size_a4_portrait(): + w = 210.0 * config.PT_PER_MM + h = 297.0 * config.PT_PER_MM + assert sources.guess_sheet_size(w, h) == "A4" + + +def test_guess_sheet_size_unknown(): + assert sources.guess_sheet_size(500, 700) is None + + +# --------------------------------------------------------------------------- +# 渲染倍率 / 尺度 +# --------------------------------------------------------------------------- +def test_tile_zoom_uses_max_edge(): + assert sources._tile_zoom((0, 0, 1000, 500), 10.0, 2000) == 2.0 + + +def test_tile_zoom_capped_by_max_zoom(): + assert sources._tile_zoom((0, 0, 10, 10), 3.0, 2000) == 3.0 + + +def test_tile_zoom_degenerate_returns_max(): + assert sources._tile_zoom((0, 0, 0, 0), 4.0, 1000) == 4.0 + + +def test_file_scales_image_is_one(): + assert sources.file_scales("image") == (1.0, 1.0) + + +def test_file_scales_pdf_matches_readable_px(monkeypatch): + monkeypatch.setattr(config, "TEXT_HEIGHT_MM", 2.0) + monkeypatch.setattr(config, "TEXT_READABLE_PX", 8) + monkeypatch.setattr(config, "LINE_READABLE_PX", 6) + s_text, s_line = sources.file_scales("pdf") + text_pt = 2.0 * config.PT_PER_MM + assert s_text == pytest.approx(text_pt / 8) + assert s_line == pytest.approx(text_pt / 6) + assert s_text < s_line # 文字更细 → 单张覆盖更小 + + +# --------------------------------------------------------------------------- +# 墨迹网格（纯 PIL） +# --------------------------------------------------------------------------- +def _img(size, draw=None, mode="RGB", bg=(255, 255, 255)): + from PIL import Image, ImageDraw + + img = Image.new(mode, size, bg) + if draw: + draw(ImageDraw.Draw(img)) + return img + + +def test_ink_grid_marks_only_dark_regions(): + from PIL import ImageDraw + + # 左半白、右半黑 + img = _img((100, 100), lambda d: d.rectangle([50, 0, 99, 99], fill=(0, 0, 0))) + grid, boxes = sources.ink_grid(img, (0, 0, 100, 100), div=8, max_edge=0) + assert grid["cols"] == 8 and grid["rows"] == 8 + occupied = heatmap.occupied_cells(grid) + # 只应命中右半（col >= 4） + assert occupied and all(c >= 4 for _r, c in occupied) + assert boxes and len(boxes) == len(occupied) + + +def test_ink_grid_blank_image_has_no_boxes(): + grid, boxes = sources.ink_grid(_img((64, 64)), (0, 0, 64, 64), div=8, max_edge=0) + assert boxes == [] + + +def test_ink_grid_degenerate_box(): + grid, boxes = sources.ink_grid(_img((10, 10)), (0, 0, 0, 0)) + assert boxes == [] and grid["cols"] == 0 + + +def test_ink_grid_counts_reflect_density(): + from PIL import ImageDraw + + # 全黑 → 每格占比 ~1 → counts 应显著大于 1 + img = _img((80, 80), lambda d: d.rectangle([0, 0, 79, 79], fill=(0, 0, 0))) + grid, _ = sources.ink_grid(img, (0, 0, 80, 80), div=8, max_edge=0) + assert max(max(row) for row in grid["counts"]) >= 50 + + +def test_ink_grid_respects_threshold(): + from PIL import ImageDraw + + # 灰 210 比阈值 200 亮 → 不算墨迹 + img = _img((40, 40), lambda d: d.rectangle([0, 0, 39, 39], fill=(210, 210, 210))) + _grid, boxes = sources.ink_grid(img, (0, 0, 40, 40), div=8, max_edge=0) + assert boxes == [] + + +def test_ink_grid_box_is_in_world_coords(): + from PIL import ImageDraw + + img = _img((100, 100), lambda d: d.rectangle([0, 0, 49, 49], fill=(0, 0, 0))) + _grid, boxes = sources.ink_grid(img, (0, 0, 200, 200), div=4, max_edge=0) + for b in boxes: + assert 0 <= b[0] <= b[2] <= 200 + assert 0 <= b[1] <= b[3] <= 200 + + +# --------------------------------------------------------------------------- +# ImageReader +# --------------------------------------------------------------------------- +@pytest.fixture +def sample_png(tmp_path): + from PIL import ImageDraw + + img = _img((400, 200), lambda d: d.rectangle([10, 10, 110, 90], fill=(0, 0, 0))) + path = tmp_path / "sample.png" + img.save(path) + return str(path) + + +def test_image_reader_frames(sample_png): + reader = sources.ImageReader(sample_png) + try: + frames = reader.frames() + finally: + reader.close() + assert len(frames) == 1 + f = frames[0] + assert f["bbox"] == [0.0, 0.0, 400.0, 200.0] + assert f["source_type"] == "image" + assert f["page_index"] == 0 + + +def test_image_reader_render_crops_world_box(sample_png): + reader = sources.ImageReader(sample_png) + try: + # 世界 (0,0)-(100,100)（y 向上）→ 图像上半左 100x100 区域 + out = reader.render((0, 0, 100, 100), 1.0) + finally: + reader.close() + assert out.size == (100, 100) + + +def test_image_reader_render_never_upscales(sample_png): + reader = sources.ImageReader(sample_png) + try: + out = reader.render((0, 0, 50, 50), 4.0) # zoom>1 不放大 + finally: + reader.close() + assert out.size == (50, 50) + + +def test_image_reader_render_downscales(sample_png): + reader = sources.ImageReader(sample_png) + try: + out = reader.render((0, 0, 400, 200), 0.5) + finally: + reader.close() + assert out.size == (200, 100) + + +def test_image_reader_text_items_empty(sample_png): + reader = sources.ImageReader(sample_png) + try: + assert reader.text_items(0) == [] + finally: + reader.close() + + +# --------------------------------------------------------------------------- +# open_reader / plan_file / collect（图片端到端，离线） +# --------------------------------------------------------------------------- +def test_open_reader_rejects_cad(): + ctx = make_ctx(connect_cad=False) + with pytest.raises(ValueError): + sources.open_reader(ctx) + + +def test_open_reader_requires_path(): + ctx = make_ctx(connect_cad=False, source="image") + with pytest.raises(ValueError): + sources.open_reader(ctx) + + +def test_plan_file_image(sample_png): + ctx = make_ctx(connect_cad=False, source="image", source_path=sample_png) + try: + plan = sources.plan_file(ctx) + finally: + sources.close_reader(ctx) + assert plan["drawing_info"]["unit"] == "px" + assert plan["drawing_info"]["page_count"] == 1 + assert len(plan["frames"]) == 1 + assert plan["layers"] == {"total": 0, "items": []} + + +def test_plan_file_slices_frames_by_max_frames(sample_png): + """max_frames 对文件源同样生效（用假 reader 造多页，避免依赖真实 PDF）。""" + ctx = make_ctx(connect_cad=False, source="image", source_path=sample_png, max_frames=2) + ctx.source_reader = _FakeReader(page_count=5) + plan = sources.plan_file(ctx) + assert len(plan["frames"]) == 2 + assert plan["drawing_info"]["page_count"] == 5 + + +class _FakeReader: + """最小读取器替身：只实现 plan_file 需要的接口。""" + + def __init__(self, page_count=1): + self.page_count = page_count + + def frames(self): + return [ + {"bbox": [0.0, 0.0, 10.0, 10.0], "page_index": i, + "source_type": "image"} + for i in range(self.page_count) + ] + + def close(self): + pass + + +def test_collect_image_end_to_end_offline(sample_png, monkeypatch): + """图片源取证：同形 evidence + 至少一张整体图 + 世界坐标 window。""" + monkeypatch.setattr(config, "TILING_ENABLED", True) + monkeypatch.setattr(config, "IMAGE_MAX_EDGE", 1280) + ctx = make_ctx( + connect_cad=False, source="image", source_path=sample_png, frame_limit=200 + ) + try: + plan = sources.plan_file(ctx) + ev = sources.collect(ctx, 1, plan["frames"][0]) + finally: + sources.close_reader(ctx) + + assert ev["source_type"] == "image" + assert ev["window"] == {"min_x": 0.0, "min_y": 0.0, "max_x": 400.0, "max_y": 200.0} + assert ev["entities"] == {"total": 0, "items": []} + assert ev["texts"] == {"total": 0, "items": []} + assert ev["images"], "至少应有整体图" + assert ev["images"][0]["label"] == "overall" + # 图片确实可解码 + assert base64.b64decode(ev["images"][0]["b64"]).startswith(b"\x89PNG") + # 每张图都登记了世界可见范围 + for img in ev["images"]: + assert len(img["window"]) == 4 + + +def test_collect_image_without_tiling(sample_png, monkeypatch): + monkeypatch.setattr(config, "TILING_ENABLED", False) + ctx = make_ctx(connect_cad=False, source="image", source_path=sample_png) + try: + plan = sources.plan_file(ctx) + ev = sources.collect(ctx, 1, plan["frames"][0]) + finally: + sources.close_reader(ctx) + assert [i["label"] for i in ev["images"]] == ["overall"] + + +def test_collect_image_bad_bbox_returns_none(sample_png): + ctx = make_ctx(connect_cad=False, source="image", source_path=sample_png) + try: + assert sources.collect(ctx, 1, {"source_type": "image", "bbox": [0, 0, 1]}) is None + finally: + sources.close_reader(ctx) + + +def test_reader_is_cached_on_ctx(sample_png): + ctx = make_ctx(connect_cad=False, source="image", source_path=sample_png) + try: + r1 = sources.open_reader(ctx) + r2 = sources.open_reader(ctx) + assert r1 is r2 is ctx.source_reader + finally: + sources.close_reader(ctx) + assert ctx.source_reader is None + + +# --------------------------------------------------------------------------- +# snap_box（图片源：世界单位 = 源像素，必须整数对齐） +# --------------------------------------------------------------------------- +def test_snap_box_outward(): + assert sources.snap_box((1.2, 2.7, 10.4, 20.1)) == (1.0, 2.0, 11.0, 21.0) + + +def test_snap_box_with_bounds(): + assert sources.snap_box((-5.5, -1.2, 300.4, 50.0), (0.0, 0.0, 100.0, 100.0)) == ( + 0.0, 0.0, 100.0, 50.0) + + +def test_snap_box_degenerate_returns_none(): + assert sources.snap_box((0.0, 0.0, 0.0, 0.0)) is None + assert sources.snap_box(None) is None + + +def test_image_tile_window_matches_pixels_exactly(sample_png, monkeypatch): + """图片源瓦片：注册的 window 宽度必须与渲染像素数**精确相等**（倍率恰为 1）。""" + import io as _io + + from PIL import Image as _Image + + monkeypatch.setattr(config, "TILING_ENABLED", True) + ctx = make_ctx(connect_cad=False, source="image", source_path=sample_png) + try: + plan = sources.plan_file(ctx) + ev = sources.collect(ctx, 1, plan["frames"][0]) + finally: + sources.close_reader(ctx) + tiles = [i for i in ev["images"] if i["label"] != "overall"] + assert tiles + for im in tiles: + w = im["window"] + img = _Image.open(_io.BytesIO(base64.b64decode(im["b64"]))) + assert img.size[0] == int(w[2] - w[0]) + assert img.size[1] == int(w[3] - w[1]) diff --git a/tests/test_sources_pdf.py b/tests/test_sources_pdf.py new file mode 100644 index 0000000..8694dcc --- /dev/null +++ b/tests/test_sources_pdf.py @@ -0,0 +1,285 @@ +"""`dwg_review.sources` 的 **PDF** 部分单测。 + +用手工构造的最小 PDF（`_build_pdf`）驱动，不依赖任何 PDF 生成库、不依赖外部样本文件。 +真实工程图纸的端到端验收见仓库根 `verify_pdf_source.py`。 +""" + +import base64 + +import pytest + +from dwg_review import config, sources +from tests.fakes import make_ctx + + +def _build_pdf(pages): + """构造最小可解析 PDF。pages = [{"mediabox", "rotate", "text"}, ...]。""" + objs = [ + b"<< /Type /Catalog /Pages 2 0 R >>", + None, # obj 2：Pages（末尾回填，需先知道各页 obj 号） + b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>", + ] + page_ids = [] + for spec in pages: + mb = spec.get("mediabox", (0, 0, 200, 100)) + rot = spec.get("rotate", 0) + text = spec.get("text", "HELLO") + parts = [] + if spec.get("fill"): + # 整页填黑 → 墨迹铺满整页（用于「整页一个簇」的场景） + parts.append( + f"{mb[0]} {mb[1]} {mb[2] - mb[0]} {mb[3] - mb[1]} re f" + ) + if text: + parts.append(f"BT /F1 12 Tf 20 40 Td ({text}) Tj ET") + stream = "\n".join(parts).encode() + page_id = len(objs) + 1 + content_id = page_id + 1 + page_ids.append(page_id) + objs.append( + ( + f"<< /Type /Page /Parent 2 0 R " + f"/MediaBox [{mb[0]} {mb[1]} {mb[2]} {mb[3]}] /Rotate {rot} " + f"/Resources << /Font << /F1 3 0 R >> >> /Contents {content_id} 0 R >>" + ).encode() + ) + objs.append( + b"<< /Length " + str(len(stream)).encode() + b" >>\nstream\n" + + stream + b"\nendstream" + ) + kids = " ".join(f"{i} 0 R" for i in page_ids) + objs[1] = f"<< /Type /Pages /Kids [{kids}] /Count {len(page_ids)} >>".encode() + + out = bytearray(b"%PDF-1.4\n") + offsets = [] + for i, body in enumerate(objs, 1): + offsets.append(len(out)) + out += f"{i} 0 obj\n".encode() + body + b"\nendobj\n" + xref_pos = len(out) + n = len(objs) + 1 + out += f"xref\n0 {n}\n".encode() + b"0000000000 65535 f \n" + for off in offsets: + out += f"{off:010d} 00000 n \n".encode() + out += ( + f"trailer\n<< /Size {n} /Root 1 0 R >>\nstartxref\n{xref_pos}\n%%EOF\n" + ).encode() + return bytes(out) + + +@pytest.fixture +def pdf_path(tmp_path): + """把 `_build_pdf(pages)` 落盘，返回路径。""" + + def _make(pages): + path = tmp_path / "mini.pdf" + path.write_bytes(_build_pdf(pages)) + return str(path) + + return _make + + +# --------------------------------------------------------------------------- +# PdfReader +# --------------------------------------------------------------------------- +def test_pdf_reader_frames_basic(pdf_path): + reader = sources.PdfReader(pdf_path([{}])) + try: + frames = reader.frames() + finally: + reader.close() + assert len(frames) == 1 + assert frames[0]["bbox"] == [0.0, 0.0, 200.0, 100.0] + assert frames[0]["source_type"] == "pdf" + assert frames[0]["page_index"] == 0 + assert frames[0]["sheet_scale"] == 1.0 + + +def test_pdf_reader_frames_multipage_keeps_order(pdf_path): + reader = sources.PdfReader( + pdf_path([{"mediabox": (0, 0, 100, 50)}, {"mediabox": (0, 0, 300, 400)}]) + ) + try: + frames = reader.frames() + finally: + reader.close() + assert [f["page_index"] for f in frames] == [0, 1] + assert frames[1]["bbox"] == [0.0, 0.0, 300.0, 400.0] + + +def test_pdf_reader_rot90_swaps_page_size(pdf_path): + """带 /Rotate 90 的页：世界尺寸应是 (MediaBox 高, MediaBox 宽)。""" + reader = sources.PdfReader(pdf_path([{"mediabox": (0, 0, 200, 100), "rotate": 90}])) + try: + frames = reader.frames() + finally: + reader.close() + assert frames[0]["bbox"] == [0.0, 0.0, 100.0, 200.0] + + +def test_pdf_reader_render_region_size(pdf_path): + reader = sources.PdfReader(pdf_path([{}])) + try: + out = reader.render((0, 0, 100, 50), 2.0) + finally: + reader.close() + assert out.size == (200, 100) + + +def test_pdf_reader_render_bad_box(pdf_path): + reader = sources.PdfReader(pdf_path([{}])) + try: + assert reader.render((0, 0, 0, 0), 2.0) is None + finally: + reader.close() + + +def test_pdf_reader_text_items(pdf_path): + reader = sources.PdfReader(pdf_path([{"text": "HELLO"}])) + try: + items = reader.text_items(0) + finally: + reader.close() + assert len(items) == 1 + assert items[0]["content"] == "HELLO" + x0, y0, x1, y1 = items[0]["bbox"] + assert 0 <= x0 < x1 <= 200 and 0 <= y0 < y1 <= 100 + + +def test_pdf_reader_text_items_min_chars(pdf_path): + """短于 PDF_TEXT_MIN_CHARS 的片段被丢弃（滤噪）。""" + reader = sources.PdfReader(pdf_path([{"text": "AB"}])) + try: + assert reader.text_items(0, min_chars=8) == [] + assert len(reader.text_items(0, min_chars=2)) == 1 + finally: + reader.close() + + +def test_pdf_reader_text_items_respects_box_filter(pdf_path): + reader = sources.PdfReader(pdf_path([{"text": "HELLO"}])) + try: + # 文本在 y≈40~49，给一个完全错开的窗口 → 过滤掉 + assert reader.text_items(0, box=(0, 80, 200, 100)) == [] + finally: + reader.close() + + +def test_pdf_reader_text_items_rot90_maps_into_world(pdf_path): + reader = sources.PdfReader(pdf_path([{"mediabox": (0, 0, 200, 100), "rotate": 90}])) + try: + items = reader.text_items(0) + finally: + reader.close() + assert len(items) == 1 + x0, y0, x1, y1 = items[0]["bbox"] + # 世界尺寸为 (100, 200)（旋转后），bbox 必须落在其中 + assert 0 <= x0 < x1 <= 100.0 + assert 0 <= y0 < y1 <= 200.0 + + +def test_pdf_reader_empty_text_layer(pdf_path): + reader = sources.PdfReader(pdf_path([{"text": ""}])) + try: + assert reader.text_items(0) == [] + finally: + reader.close() + + +# --------------------------------------------------------------------------- +# PDF 端到端（离线，plan + collect） +# --------------------------------------------------------------------------- +def test_pdf_plan_and_collect_end_to_end(pdf_path, monkeypatch): + monkeypatch.setattr(config, "TILING_ENABLED", True) + path = pdf_path([{"text": "HELLO"}]) + ctx = make_ctx(connect_cad=False, source="pdf", source_path=path) + try: + plan = sources.plan_file(ctx) + assert plan["drawing_info"]["unit"] == "pt" + assert plan["drawing_info"]["page_count"] == 1 + assert len(plan["frames"]) == 1 + ev = sources.collect(ctx, 1, plan["frames"][0]) + finally: + sources.close_reader(ctx) + + assert ev["source_type"] == "pdf" + assert ev["window"] == {"min_x": 0.0, "min_y": 0.0, "max_x": 200.0, "max_y": 100.0} + assert ev["entities"] == {"total": 0, "items": []} + assert ev["texts"]["total"] == 1 + assert ev["texts"]["items"][0]["content"] == "HELLO" + assert ev["images"] and ev["images"][0]["label"] == "overall" + assert base64.b64decode(ev["images"][0]["b64"]).startswith(b"\x89PNG") + + +def test_pdf_pdf_max_pages_caps_frames(pdf_path, monkeypatch): + monkeypatch.setattr(config, "PDF_MAX_PAGES", 1) + path = pdf_path([{}, {}]) + ctx = make_ctx(connect_cad=False, source="pdf", source_path=path) + try: + plan = sources.plan_file(ctx) + finally: + sources.close_reader(ctx) + assert len(plan["frames"]) == 1 + assert plan["drawing_info"]["page_count"] == 2 + + +def test_pdf_blank_page_has_only_overall(pdf_path, monkeypatch): + """无墨迹的页：墨迹网格为空 → 只出整体图（不凭空造瓦片）。""" + monkeypatch.setattr(config, "TILING_ENABLED", True) + path = pdf_path([{"text": ""}]) + ctx = make_ctx(connect_cad=False, source="pdf", source_path=path) + try: + plan = sources.plan_file(ctx) + ev = sources.collect(ctx, 1, plan["frames"][0]) + finally: + sources.close_reader(ctx) + assert [i["label"] for i in ev["images"]] == ["overall"] + + +def test_pdf_collect_multi_page_uses_own_page(pdf_path): + """多页时每页取证必须渲染**自己的页**（页序不可错位）。""" + path = pdf_path( + [ + {"mediabox": (0, 0, 200, 100), "text": "FIRST"}, + {"mediabox": (0, 0, 400, 200), "text": "SECOND"}, + ] + ) + ctx = make_ctx(connect_cad=False, source="pdf", source_path=path) + try: + plan = sources.plan_file(ctx) + ev2 = sources.collect(ctx, 2, plan["frames"][1]) + finally: + sources.close_reader(ctx) + assert ev2["window"]["max_x"] == 400.0 + assert ev2["texts"]["items"][0]["content"] == "SECOND" + + +def test_pdf_dedupes_tile_equal_to_overall(pdf_path, monkeypatch): + """整页即一个簇且瓦片分辨率不更高时，不重复出一张与整体图相同的图。""" + monkeypatch.setattr(config, "TILING_ENABLED", True) + monkeypatch.setattr(config, "PDF_OVERVIEW_MAX_ZOOM", 4.0) + monkeypatch.setattr(config, "PDF_TILE_MAX_ZOOM", 4.0) + path = pdf_path([{"fill": True}]) # 墨迹铺满整页 → 整页即一个簇 + ctx = make_ctx(connect_cad=False, source="pdf", source_path=path) + try: + plan = sources.plan_file(ctx) + ev = sources.collect(ctx, 1, plan["frames"][0]) + finally: + sources.close_reader(ctx) + labels = [i["label"] for i in ev["images"]] + assert labels.count("overall") == 1 + assert "tile-1" not in labels # 与整体图完全重合 → 省掉 + + +def test_pdf_keeps_tile_when_higher_zoom_than_overall(pdf_path, monkeypatch): + """瓦片分辨率**高于**整体图时不能去重（否则丢掉细节）。""" + monkeypatch.setattr(config, "TILING_ENABLED", True) + monkeypatch.setattr(config, "PDF_OVERVIEW_MAX_ZOOM", 1.0) + monkeypatch.setattr(config, "PDF_TILE_MAX_ZOOM", 6.0) + path = pdf_path([{"fill": True}]) + ctx = make_ctx(connect_cad=False, source="pdf", source_path=path) + try: + plan = sources.plan_file(ctx) + ev = sources.collect(ctx, 1, plan["frames"][0]) + finally: + sources.close_reader(ctx) + assert [i["label"] for i in ev["images"]] == ["overall", "tile-1"] diff --git a/verify_pdf_source.py b/verify_pdf_source.py new file mode 100644 index 0000000..ab1bd9c --- /dev/null +++ b/verify_pdf_source.py @@ -0,0 +1,407 @@ +# -*- coding: utf-8 -*- +""" +verify_pdf_source.py +PDF / 图片**离线源**验收脚本。**不需要 CAD、不需要 8080 端口**（这是本方向的核心卖点）。 + +用法： + python verify_pdf_source.py "E:\\path\\图纸.pdf" + python verify_pdf_source.py "E:\\path\\图纸.pdf" --pages 3 # 只验前 3 页 + python verify_pdf_source.py "E:\\path\\scan.png" # 图片源 + python verify_pdf_source.py "E:\\path\\图纸.pdf" --dump out/c1 # 导出图片供目视 + python verify_pdf_source.py "E:\\path\\图纸.pdf" --llm # 追加真实 LLM 端到端 + +覆盖： + 0) 离线：全程 ctx.tools is None / ctx.client is None（绝不连 CAD） + 1) 规划：页数、页序保持、每页 bbox / 幅面反查 + 2) 取证：同形 evidence（空结构化信封）、文本层、整体图 + 墨迹瓦片 + 3) 分辨率自检：**实测**每张图的「纸面字高像素数」，不得低于线条可读下限 + 4) 旋转自检（/Rotate ≠ 0 且有文本层时）：文本矩形处的墨迹密度显著高于错误映射 + 5) --dump：导出 PNG 供人工目视 + 6) --llm ：跑真实 `python -m dwg_review `，检查报告落盘 +""" +import base64 +import io +import os +import subprocess +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from PIL import Image # noqa: E402 + +from dwg_review import config, evidence_model, sources, tiling # noqa: E402 +from dwg_review.context import ReviewContext # noqa: E402 +from dwg_review.phases import evidence as evidence_phase # noqa: E402 +from dwg_review.phases import planning # noqa: E402 + +RESULTS = [] + + +def record(name, ok, detail=""): + RESULTS.append((name, ok, detail)) + print(" [%s] %-34s %s" % ("PASS" if ok else "FAIL", name, detail)) + + +def info(msg): + print(" %s" % msg) + + +# ---------------------------------------------------------------- 小工具 +def dark_frac(img, threshold=None): + """一张图里「暗像素」（墨迹）的占比。纯 PIL，无 numpy。""" + threshold = config.INK_THRESHOLD if threshold is None else threshold + bw = img.convert("L").point(lambda p: 255 if p < threshold else 0, mode="L") + hist = bw.histogram() + total = sum(hist) or 1 + return hist[255] / float(total) + + +def clip_box(box, page_box): + """把 bbox 夹到页范围内；退化返回 None。""" + x1 = max(box[0], page_box[0]) + y1 = max(box[1], page_box[1]) + x2 = min(box[2], page_box[2]) + y2 = min(box[3], page_box[3]) + return None if x2 <= x1 or y2 <= y1 else (x1, y1, x2, y2) + + +def png_size(b64): + return Image.open(io.BytesIO(base64.b64decode(b64))).size + + +def render_scale_of(image_b64, window): + """实测该图的渲染倍率 = 图像像素宽 / 世界宽。 + + - PDF 源：世界单位是**点** → 倍率即 px/点，可直接换算纸面字高像素； + - 图片源：世界单位是**源像素** → 倍率是「相对原图的缩放」，应恒 ≤ 1（不放大）。 + """ + w_px, _h_px = png_size(image_b64) + w_world = float(window[2]) - float(window[0]) + return (w_px / w_world) if w_world > 0 else 0.0 + + +def text_px_of(image_b64, window): + """纸面字高（TEXT_HEIGHT_MM）在该图上占多少像素。**仅对 PDF 源有意义**。""" + return config.TEXT_HEIGHT_MM * config.PT_PER_MM * render_scale_of(image_b64, window) + + +# ---------------------------------------------------------------- 主流程 +def main(): + args = parse_args() + path = Path(args.path).expanduser().resolve() + if not path.is_file(): + print("[错误] 文件不存在：%s" % path) + return 2 + + print("=== PDF / 图片离线源验收 ===") + print("文件: %s" % path) + try: + source = sources.detect_source(str(path)) + except ValueError as exc: + print("[错误] %s" % exc) + return 2 + if not sources.is_file_source(source): + print("[错误] %s 是 CAD 源，请用 CAD 在线流程（不带参数运行 dwg_review）。" % source) + return 2 + print("来源: %s\n" % source) + + # ---------------------------------------------------------- 0) 离线 + print("0) 离线上下文（不连 CAD）") + ctx = ReviewContext.create(need_llm=False, connect_cad=False, + source=source, source_path=str(path)) + record("ctx.tools is None", ctx.tools is None, "离线源不得持有 CAD 工具表") + record("ctx.client is None", ctx.client is None, "离线源不得创建 MCP 客户端") + record("无需 CAD 端口", True, "本脚本全程未连接 %s:%d" % (config.HOST, config.PORT)) + + try: + return run_checks(ctx, path, args) + finally: + sources.close_reader(ctx) + + +def parse_args(): + import argparse + + p = argparse.ArgumentParser(description="PDF / 图片离线源验收") + p.add_argument("path", help="PDF 或图片文件路径") + p.add_argument("--pages", type=int, default=0, help="只验前 N 页（0=全部，默认前 5 页）") + p.add_argument("--dump", default=None, help="把证据图片导出到该目录供目视") + p.add_argument("--llm", action="store_true", help="追加真实 LLM 端到端（需 DEEPSEEK_API_KEY）") + a = p.parse_args() + if a.pages == 0: + a.pages = 5 + return a + + +def run_checks(ctx, path, args): + # ---------------------------------------------------------- 1) 规划 + print("\n1) 规划（每页一个图框，保持原始页序）") + plan = planning.plan(ctx) + frames = plan["frames"] + info("drawing_info: %s" % plan["drawing_info"]) + record("frames 非空", bool(frames), "共 %d 页" % len(frames)) + if not frames: + return 1 + idx = [f.get("page_index") for f in frames] + record("页序保持 0..N-1", idx == list(range(len(frames))), "page_index=%s" % idx) + bad = [f for f in frames if not f.get("bbox") or f["bbox"][2] <= 0] + record("每页 bbox 有效", not bad, "非法 %d 页" % len(bad)) + if frames: + info("页 0: bbox=%s 幅面=%s" % (frames[0]["bbox"], frames[0].get("sheet_size"))) + + # ---------------------------------------------------------- 2) 取证 + limit = min(args.pages, len(frames)) + source_is_pdf = ctx.source == evidence_model.SOURCE_PDF + print("\n2) 取证（前 %d 页）" % limit) + dumps = [] + stats = {"tiles": 0, "text_pages": 0, "text_items": 0, "min_text_px": None} + for i in range(limit): + frame = frames[i] + ev = evidence_phase.collect(ctx, i + 1, frame) + if ev is None: + record("页 %d 取证" % i, False, "collect 返回 None") + continue + + page_box = tiling.frame_bbox(frame) + win = ev["window"] + assert win is not None + # 非文本结构化键必须是空信封；`texts` 例外（PDF 的 best-effort 文本层就放这里） + struct_ok = all( + ev.get(k) == {"total": 0, "items": []} + for k in evidence_model.STRUCTURED_SOURCES + if k != "texts" + ) + texts_env = ev.get("texts") or {} + struct_ok = struct_ok and isinstance(texts_env, dict) and isinstance( + texts_env.get("items"), list + ) + imgs = evidence_model.evidence_images(ev) + labels = [im.get("label") for im in imgs] + ok_shape = ( + ev["source_type"] == evidence_model.source_type_of(frame) + and imgs and labels[0] == "overall" + ) + record( + "页 %d evidence 同形" % i, + struct_ok and ok_shape, + "结构化=空信封, 图=%s, 文本层=%d 条" + % (labels, (ev.get("texts") or {}).get("total", 0)), + ) + + # 每张图：PNG 可解码、世界窗口在图框内、实测渲染倍率 + bad_win = [] + tile_px, tile_zoom, overall_px = [], [], None + for im in imgs: + w = im.get("window") + if not w or len(w) != 4 or w[0] < -1e-6 or w[1] < -1e-6 or \ + w[2] > page_box[2] + 1e-6 or w[3] > page_box[3] + 1e-6: + bad_win.append(im.get("label")) + continue + try: + px = png_size(im["b64"]) + except Exception as exc: # noqa: BLE001 + bad_win.append("%s(解码失败:%s)" % (im.get("label"), exc)) + continue + scale = render_scale_of(im["b64"], w) + tpx = text_px_of(im["b64"], w) + # 整体图是「整页缩略」——字读不清属**设计如此**（细节由瓦片负责），不参与达标判定 + if im.get("label") == "overall": + overall_px = tpx + else: + tile_px.append(tpx) + tile_zoom.append(scale) + if max(px) > config.IMAGE_MAX_EDGE + 1: + bad_win.append("%s(超长边 %d)" % (im.get("label"), max(px))) + if args.dump: + dumps.append((i, im.get("label"), im["b64"], px, w, tpx)) + record("页 %d 图片世界窗口在图框内" % i, not bad_win, "越界/异常: %s" % (bad_win or "无")) + + # 分辨率自检 + floor = config.LINE_READABLE_PX - 0.5 + if tile_px: + page_min = min(tile_px) + stats["min_text_px"] = ( + page_min if stats["min_text_px"] is None + else min(stats["min_text_px"], page_min) + ) + n_text_ok = sum(1 for v in tile_px if v >= config.TEXT_READABLE_PX - 0.5) + detail = ("瓦片 %d 张，最差 %.1fpx（下限 %.1f；达文字档 %.1fpx 的 %d 张）；" + "整体图 %.1fpx" % ( + len(tile_px), page_min, floor, + config.TEXT_READABLE_PX, n_text_ok, + overall_px if overall_px is not None else -1)) + if source_is_pdf: + record("页 %d 瓦片分辨率达标" % i, page_min >= floor, detail) + else: + # 图片源不允许放大（IMAGE_MAX_ZOOM=1），倍率必须 ≤1 + record( + "页 %d 图片瓦片未放大" % i, + max(tile_zoom) <= config.IMAGE_MAX_ZOOM + 1e-6, + "倍率 %.2f~%.2f（上限 %.2f）；%s" + % (min(tile_zoom), max(tile_zoom), config.IMAGE_MAX_ZOOM, detail), + ) + + stats["tiles"] += max(0, len(imgs) - 1) + if (ev.get("texts") or {}).get("total"): + stats["text_pages"] += 1 + stats["text_items"] += ev["texts"]["total"] + + # 旋转自检 + check_rotation(ctx, frame, ev, i) + + info( + "瓦片合计 %d 张；有文本层的页 %d；文本片段 %d；%s" + % ( + stats["tiles"], stats["text_pages"], stats["text_items"], + ("最差字高像素 %s" % ("%.1f" % stats["min_text_px"]) + if stats["min_text_px"] is not None and source_is_pdf + else "（图片源：倍率 ≤1，不放大）" if not source_is_pdf else "n/a"), + ) + ) + if stats["min_text_px"] is not None and source_is_pdf: + record( + "全局瓦片分辨率达标", + stats["min_text_px"] >= config.LINE_READABLE_PX - 0.5, + "最差 %.1fpx（下限 %.1f）" % (stats["min_text_px"], config.LINE_READABLE_PX), + ) + + # ---------------------------------------------------------- 5) 导出 + if args.dump and dumps: + out = Path(args.dump) + out.mkdir(parents=True, exist_ok=True) + for i, label, b64, px, w, tpx in dumps: + name = "p%02d_%s.png" % (i, label) + (out / name).write_bytes(base64.b64decode(b64)) + info("导出 %s %sx%s 倍率=%.2f 世界=%s" + % (name, px[0], px[1], render_scale_of(b64, w), + [round(v, 1) for v in w])) + record("导出目视包", True, "%d 张 → %s" % (len(dumps), out)) + + # ---------------------------------------------------------- 6) LLM + if args.llm: + print("\n6) 真实 LLM 端到端（python -m dwg_review）") + if not config.has_api_key(): + record("LLM 端到端", True, "跳过：未配置 DEEPSEEK_API_KEY") + else: + report_dir = Path(args.dump or "out/verify_pdf_source") / "report" + env = dict(os.environ, REPORT_DIR=str(report_dir)) + cmd = [sys.executable, "-m", "dwg_review", str(path), "--frames", "1"] + print(" $ %s" % " ".join(cmd)) + proc = subprocess.run(cmd, env=env, capture_output=True, text=True, + encoding="utf-8", errors="replace") + tail = (proc.stdout or "")[-1500:] + print(tail) + if proc.stderr: + print("[stderr] %s" % proc.stderr[-800:]) + mds = sorted(report_dir.glob("review_*.md")) if report_dir.exists() else [] + jsons = sorted(report_dir.glob("review_*.json")) if report_dir.exists() else [] + record("退出码 0", proc.returncode == 0, "returncode=%s" % proc.returncode) + record("报告已落盘", bool(mds and jsons), + "%s" % (mds[-1].name if mds else "无")) + + # ---------------------------------------------------------- 汇总 + print("\n=== 汇总 ===") + failed = [r for r in RESULTS if not r[1]] + for name, _ok, detail in RESULTS: + print(" [%s] %-34s %s" % ("PASS" if _ok else "FAIL", name, detail)) + print("\n通过 %d / %d" % (len(RESULTS) - len(failed), len(RESULTS))) + if failed: + print("失败项：%s" % ", ".join(r[0] for r in failed)) + return 1 + return 0 + + +def _unrotate_point(X, Y, mw, mh, rotation): + """`sources.rotate_point` 的逆：世界坐标 → 未旋转坐标。""" + r = int(rotation) % 360 + if r == 90: + return (mw - Y, X) + if r == 180: + return (mw - X, mh - Y) + if r == 270: + return (Y, mh - X) + return (X, Y) + + +def _unrotate_box(world, mw, mh, rotation): + x0, y0, x1, y1 = world + pts = [ + _unrotate_point(x, y, mw, mh, rotation) + for x, y in ((x0, y0), (x0, y1), (x1, y0), (x1, y1)) + ] + xs = [p[0] for p in pts] + ys = [p[1] for p in pts] + return (min(xs), min(ys), max(xs), max(ys)) + + +def _render_raw(reader, page_index, box, zoom, dims): + """按**指定页尺寸**渲染（`dims` 决定 crop 基准）。 + + `PdfReader.render` 固定以「旋转后页尺寸」为 crop 基准；做「忽略旋转」对照时 + 需要用 MediaBox 尺寸作基准，故这里直接调 pdfium。 + """ + page = reader.doc[page_index] + bitmap = page.render(scale=zoom, rotation=0, + crop=sources.crop_of(box, dims[0], dims[1])) + return bitmap.to_pil() + + +def check_rotation(ctx, frame, ev, page_index): + """旋转自检：`/Rotate` 映射必须让「文本矩形」落在真有字的地方。 + + 两条判据（都来自当初的手工探针：rot90 命中率 0.335 vs 其它假设 ≤0.064）： + a) 文本矩形处的墨迹密度 **明显高于** 整页平均（映射错了就会落进空白）； + b) 同一文本矩形若**忽略** `/Rotate` 直接当世界坐标，密度应更低（反向对照）。 + """ + items = (ev.get("texts") or {}).get("items") or [] + reader = ctx.source_reader + if not items or not isinstance(reader, sources.PdfReader): + return + page_idx = frame.get("page_index", 0) + rw, rh, mw, mh, rot = reader._meta(page_idx) + if rot % 360 == 0: + return + + page_box = tiling.frame_bbox(frame) + rects = [it["bbox"] for it in items + if it.get("bbox") and it["bbox"][2] > it["bbox"][0]] + if not rects: + return + + zoom = 4.0 + page_img = reader.render(page_box, 2.0, page_idx) + page_density = dark_frac(page_img) if page_img is not None else 0.0 + + hits, wrongs = [], [] + for rect in rects[:20]: + correct = clip_box(rect, page_box) + if correct is None: + continue + img = reader.render(correct, zoom, page_idx) + if img is not None: + hits.append(dark_frac(img)) + wrong_box = _unrotate_box(rect, mw, mh, rot) + img_w = _render_raw(reader, page_idx, wrong_box, zoom, (mw, mh)) + if img_w is not None: + wrongs.append(dark_frac(img_w)) + + if not hits: + return + mean_hit = sum(hits) / len(hits) + record( + "页 %d 旋转映射命中文本区(/Rotate %d)" % (page_index, rot), + mean_hit > page_density * 1.5 and mean_hit > 0.05, + "文本区墨迹=%.3f vs 整页平均=%.3f（n=%d）" % (mean_hit, page_density, len(hits)), + ) + if wrongs: + mean_wrong = sum(wrongs) / len(wrongs) + record( + "页 %d 反向对照（忽略旋转更差）" % page_index, + mean_hit > mean_wrong, + "正确=%.3f vs 忽略旋转=%.3f" % (mean_hit, mean_wrong), + ) + + +if __name__ == "__main__": + sys.exit(main())