"""CAD parser — DXF via ezdxf, DWG via ODA File Converter.""" from __future__ import annotations import os import subprocess import tempfile import ezdxf def parse_cad(file_path: str) -> dict: ext = os.path.splitext(file_path)[1].lower() if ext == ".dwg": dxf_path = _convert_dwg(file_path) if dxf_path is None: return {"markdown": "", "error": "DWG 需要 ODA File Converter,下载: https://www.opendesign.com/guestfiles/oda_file_converter"} file_path = dxf_path try: doc = ezdxf.readfile(file_path) except Exception as e: return {"markdown": "", "error": f"无法解析 DXF: {e}"} return _extract(doc) def _convert_dwg(dwg_path: str) -> str | None: candidates = [ "ODAFileConverter", "/usr/local/bin/ODAFileConverter", "/Applications/ODAFileConverter.app/Contents/MacOS/ODAFileConverter", r"C:\Program Files\ODA\ODAFileConverter\ODAFileConverter.exe", ] converter = None for c in candidates: if os.path.isfile(c) or _which(c): converter = c break if not converter: return None input_dir = os.path.dirname(os.path.abspath(dwg_path)) output_dir = tempfile.mkdtemp(prefix="engimind_cad_") filename = os.path.basename(dwg_path) try: subprocess.run([converter, input_dir, output_dir, "ACAD2018", "DXF", "0", "1", filename], check=True, timeout=60, capture_output=True) except Exception: return None base = os.path.splitext(filename)[0] dxf = os.path.join(output_dir, base + ".dxf") return dxf if os.path.isfile(dxf) else None def _which(name: str) -> bool: try: return subprocess.run(["which", name], capture_output=True, timeout=5).returncode == 0 except Exception: return False def _extract(doc: ezdxf.document.Drawing) -> dict: parts = ["## CAD 图纸解析结果\n"] # Layers layers = [{"name": l.dxf.name, "color": l.dxf.color} for l in doc.layers] if layers: parts.append("### 图层列表\n\n| 图层名 | 颜色编号 |\n| --- | --- |") for l in layers: parts.append(f"| {l['name']} | {l['color']} |") parts.append("") msp = doc.modelspace() entity_count = {} texts, dimensions, blocks = [], [], set() for e in msp: et = e.dxftype() entity_count[et] = entity_count.get(et, 0) + 1 if et == "TEXT": texts.append(e.dxf.text) elif et == "MTEXT": texts.append(e.text) elif et == "DIMENSION": try: dimensions.append(e.dxf.text or "测量值") except Exception: pass elif et == "INSERT": blocks.add(e.dxf.name) if entity_count: parts.append("### 实体统计\n\n| 实体类型 | 数量 |\n| --- | --- |") for et, cnt in sorted(entity_count.items()): parts.append(f"| {et} | {cnt} |") parts.append("") if texts: parts.append("### 文字标注\n") for t in texts[:200]: clean = t.strip().replace("\n", " ") if clean: parts.append(f"- {clean}") if len(texts) > 200: parts.append(f"\n> 共 {len(texts)} 条,仅显示前 200 条。") parts.append("") if dimensions: parts.append("### 尺寸标注\n") for d in dimensions[:100]: parts.append(f"- {d}") parts.append("") if blocks: parts.append("### 使用的图块\n") for b in sorted(blocks): parts.append(f"- {b}") parts.append("") return {"markdown": "\n".join(parts)}