673 lines
28 KiB
Python
673 lines
28 KiB
Python
# ============================================================
|
||
# 工程图纸 DXF 自动识别（moduleId: domain-dxf-drawing, 可重生 ✅）
|
||
# 图纸识别 → 主数据候选的安全管线（只读解析 + 候选生成 + SVG 预览）：
|
||
# - parse_dxf()：DXF → ParsedDrawing（版本/范围/图层/实体统计/标题栏/
|
||
# 位置号/技术说明），全部字段携带 source 与 confidence（0-1）。
|
||
# - 标题栏双通道：①块属性（ATTDEF/ATTRIB，如 MTU SF_STD_A 的
|
||
# ZEICHNUNGSNUMMER/BENENNUNG/WERKSTOFF）置信 0.9；②右下角文字区
|
||
# 中文标签（图号/名称/材料/比例…）兜底置信 0.6；③文件名兜底置信 0.5。
|
||
# - drawing_to_master_candidates()：物料/BOM/工艺路线候选——只生成图纸
|
||
# 确实承载的信息；材质/板厚、数量、工序、工时、外协等图纸无法可靠
|
||
# 获得的字段一律标记 missing + requiresConfirm，绝不编造（fail closed）。
|
||
# - dxf_to_svg()：模型空间 → SVG（ezdxf drawing 插件，白底），供前端预览。
|
||
# - discover_dxf_files()：项目工程目录 DXF 发现（只读扫描）。
|
||
# 写入边界：本模块绝不写世界状态；写入一律走 /api/dxf/stage →
|
||
# master.*.upsert 的既有 P2 确认卡（/api/actions/confirm 唯一执行路径）。
|
||
# ============================================================
|
||
from __future__ import annotations # 前向类型引用
|
||
|
||
import hashlib
|
||
import logging
|
||
import os
|
||
import re
|
||
import tempfile
|
||
from typing import Any
|
||
|
||
World = dict[str, Any] # 世界状态类型别名
|
||
|
||
_log = logging.getLogger(__name__) # 模块日志（几何降级等诊断）
|
||
|
||
# 空值占位符：MTU 等标题栏用 "."/"-" 表示未填写
|
||
_EMPTY_TOKENS = {"", ".", "-", "--", "---", "/", "—"}
|
||
|
||
# 位置号（Positionsnummer）：如 14.001 / 20.027 / 15.001M（装配图明细索引）
|
||
_POSITION_RE = re.compile(r"^\d{1,3}\.\d{1,3}[A-Za-z]?$")
|
||
|
||
# 文件名中的 图号-张次-版本：5060102101-001-e(1).dxf → number/sheet/rev
|
||
_FILENAME_RE = re.compile(
|
||
r"^(?P<number>\d{6,})-(?P<sheet>\d{1,4})(?:-(?P<rev>[A-Za-z]))?(?:\(\d+\))?$")
|
||
|
||
# 中文标题栏标签 → 标准字段（右下角文字区兜底通道）
|
||
_CN_LABEL_MAP: tuple[tuple[re.Pattern[str], str], ...] = tuple(
|
||
(re.compile(pat), field) for pat, field in (
|
||
(r"^(?:图号|图纸编号|图样编号)\s*[:：]?\s*(.+)$", "drawingNumber"),
|
||
(r"^(?:名称|零件名称|产品名称)\s*[:：]?\s*(.+)$", "title"),
|
||
(r"^(?:材料|材质)\s*[:：]?\s*(.+)$", "materialSpec"),
|
||
(r"^(?:板厚|厚度)\s*[:：]?\s*(.+)$", "thickness"),
|
||
(r"^(?:数量|单件数量)\s*[:：]?\s*(.+)$", "quantity"),
|
||
(r"^比例\s*[:：]?\s*(.+)$", "scale"),
|
||
(r"^(?:重量|质量)\s*[:：]?\s*(.+)$", "mass"),
|
||
(r"^(?:设计|制图|绘制)\s*[:：]?\s*(.+)$", "drawnBy"),
|
||
(r"^校对\s*[:：]?\s*(.+)$", "checkedBy"),
|
||
(r"^审核\s*[:：]?\s*(.+)$", "approvedBy"),
|
||
(r"^日期\s*[:：]?\s*(.+)$", "drawnDate"),
|
||
(r"^(?:单位|公司名称)\s*[:：]?\s*(.+)$", "company"),
|
||
)
|
||
)
|
||
|
||
# 块属性标签 → 标准字段（MTU/西门子系德文标题栏 + 常见英文标签）
|
||
# 值为标签前缀（GEN-TITLE-SCA{5.42} 这类带后缀的用前缀匹配）
|
||
_ATTR_TAG_MAP: tuple[tuple[str, str], ...] = (
|
||
("ZEICHNUNGSNUMMER", "drawingNumber"),
|
||
("DRAWING", "drawingNumber"),
|
||
("BLATTNUMMER", "sheet"),
|
||
("SHEET", "sheet"),
|
||
("FORMAT", "format"),
|
||
("GEN-TITLE-SCA", "scale"),
|
||
("MASSSTAB", "scale"),
|
||
("SCALE", "scale"),
|
||
("BENENNUNG", "titlePart"), # BENENNUNG1..6 逐段拼接
|
||
("TITLE", "titlePart"),
|
||
("ZEICHNUNGSART", "drawingType"),
|
||
("WERKSTOFF", "materialSpec"),
|
||
("MATERIAL", "materialSpec"),
|
||
("HALBZEUG", "semiFinished"),
|
||
("MASSE", "mass"),
|
||
("WEIGHT", "mass"),
|
||
("ERSTELLER", "drawnBy"),
|
||
("DRAWN", "drawnBy"),
|
||
("ERSTELLDATUM", "drawnDate"),
|
||
("PRUEFBEARBEITER", "checkedBy"),
|
||
("CHECKED", "checkedBy"),
|
||
("PRUEFDATUM", "checkedDate"),
|
||
("NORMBEARBEITER", "approvedBy"),
|
||
("NORMDATUM", "approvedDate"),
|
||
("VERWENDBARKEIT", "applicableTo"),
|
||
("APPLICABLE", "applicableTo"),
|
||
("AUFTRAGSNUMMER", "orderNo"),
|
||
("REFERENZNUMMER", "referenceNo"),
|
||
("B_FREI", "releaseLetter"), # 版本发放字母（freigabe）
|
||
("AE_FREI", "releaseNote"),
|
||
)
|
||
# 至少命中 N 个已知标签才认定为标题栏块
|
||
_TITLE_BLOCK_MIN_HITS = 3
|
||
|
||
# 技术说明判定：MTEXT 长文本 / 含换行；或明确技术要求前缀
|
||
_NOTE_MIN_LEN = 40
|
||
_NOTE_PREFIX_RE = re.compile(
|
||
r"技术要求|技术条件|notes?\s*[:：]|hinweis|specification", re.IGNORECASE)
|
||
|
||
_PARSE_CACHE: dict[str, dict[str, Any]] = {} # drawingId → 解析缓存
|
||
_CACHE_MAX = 8
|
||
|
||
|
||
def _clean(value: Any) -> str:
|
||
"""标题栏取值归一化：去空白；占位符（. / - 等）视为空。"""
|
||
text = str(value or "").strip()
|
||
return "" if text in _EMPTY_TOKENS else text
|
||
|
||
|
||
def _mtext_plain(text: str) -> str:
|
||
"""MTEXT 去格式码：{...} 分组、\\P 换行、\\A; 对齐码等。"""
|
||
out = re.sub(r"[{}]", "", str(text or ""))
|
||
out = out.replace("\\P", "\n")
|
||
out = re.sub(r"\\[ACcFfHhQqTtWw][^;\\]*;?", "", out)
|
||
out = re.sub(r"\\[pX]", "", out)
|
||
out = re.sub(r"%%d", "°", out)
|
||
out = re.sub(r"%%[cu]", "", out)
|
||
return out.strip()
|
||
|
||
|
||
def _filename_meta(filename: str) -> dict[str, str]:
|
||
"""文件名线索：图号/张次/版本（5060102101-001-e(1) → 三段）。"""
|
||
stem = os.path.splitext(os.path.basename(filename))[0]
|
||
m = _FILENAME_RE.match(stem)
|
||
if not m:
|
||
return {"stem": stem}
|
||
return {
|
||
"stem": stem,
|
||
"drawingNumberNormalized": m.group("number"),
|
||
"sheet": m.group("sheet"),
|
||
"revision": (m.group("rev") or "").lower(),
|
||
}
|
||
|
||
|
||
def _entity_insert_xy(entity: Any) -> tuple[float, float]:
|
||
try:
|
||
ins = entity.dxf.insert
|
||
return float(ins.x), float(ins.y)
|
||
except AttributeError:
|
||
return 0.0, 0.0
|
||
|
||
|
||
def _collect_layout_texts(layout: Any, *, cap: int) -> list[dict[str, Any]]:
|
||
"""收集布局内 TEXT/MTEXT 文字（坐标/图层/字高/来源实体类型）。"""
|
||
texts: list[dict[str, Any]] = []
|
||
for e in layout.query("TEXT MTEXT"):
|
||
if len(texts) >= cap:
|
||
break
|
||
if e.dxftype() == "TEXT":
|
||
value = str(e.dxf.text or "").strip()
|
||
height = float(getattr(e.dxf, "height", 0.0) or 0.0)
|
||
else:
|
||
value = _mtext_plain(e.text)
|
||
height = float(getattr(e.dxf, "char_height", 0.0) or 0.0)
|
||
if not value:
|
||
continue
|
||
x, y = _entity_insert_xy(e)
|
||
texts.append({
|
||
"text": value, "x": round(x, 3), "y": round(y, 3),
|
||
"height": round(height, 2), "layer": str(e.dxf.layer),
|
||
"entityType": e.dxftype(),
|
||
})
|
||
return texts
|
||
|
||
|
||
def _title_block_from_attribs(doc: Any) -> dict[str, Any]:
|
||
"""通道①：布局内带 ATTRIB 的 INSERT → 标题栏字段（置信 0.9）。
|
||
|
||
遍历所有布局（模型/图纸空间），按已知标签命中数挑选最像标题栏的块引用；
|
||
字段全部标注来源 ATTRIB 标签，未填写的占位字段保持空（不编造）。
|
||
"""
|
||
best: dict[str, Any] | None = None
|
||
for layout in doc.layouts:
|
||
for ins in layout.query("INSERT"):
|
||
attribs = [(str(a.dxf.tag).upper(), a.dxf.text) for a in ins.attribs]
|
||
if not attribs:
|
||
continue
|
||
hits = sum(1 for tag, _ in attribs
|
||
if any(tag.startswith(prefix) for prefix, _ in _ATTR_TAG_MAP))
|
||
if hits < _TITLE_BLOCK_MIN_HITS:
|
||
continue
|
||
if best is None or hits > best["tagHits"]:
|
||
best = {
|
||
"tagHits": hits, "layout": layout.name,
|
||
"blockName": str(ins.dxf.name), "attribs": attribs,
|
||
}
|
||
if best is None:
|
||
return {"found": False, "source": "block-attribs", "fields": {}}
|
||
|
||
fields: dict[str, dict[str, Any]] = {}
|
||
title_parts: list[str] = []
|
||
for tag, raw in best["attribs"]:
|
||
value = _clean(raw)
|
||
mapped = next((field for prefix, field in _ATTR_TAG_MAP
|
||
if tag.startswith(prefix)), None)
|
||
if mapped is None:
|
||
continue
|
||
if mapped == "titlePart":
|
||
if value:
|
||
title_parts.append(value)
|
||
continue
|
||
if not value:
|
||
fields.setdefault(mapped, {
|
||
"value": "", "confidence": 0.0,
|
||
"source": f"ATTRIB {tag}（未填写）", "missing": True})
|
||
continue
|
||
fields[mapped] = {"value": value, "confidence": 0.9,
|
||
"source": f"ATTRIB {tag}", "missing": False}
|
||
if title_parts:
|
||
fields["title"] = {"value": " / ".join(title_parts), "confidence": 0.9,
|
||
"source": "ATTRIB BENENNUNG1..n", "missing": False}
|
||
return {
|
||
"found": True, "source": "block-attribs",
|
||
"blockName": best["blockName"], "layout": best["layout"],
|
||
"fields": fields,
|
||
}
|
||
|
||
|
||
def _title_block_from_cn_texts(texts: list[dict[str, Any]],
|
||
extents: dict[str, float]) -> dict[str, Any]:
|
||
"""通道②：右下角文字区中文标签兜底（置信 0.6，必须人工确认）。"""
|
||
if not texts:
|
||
return {"found": False, "source": "cn-labels", "fields": {}}
|
||
width = max(1e-6, extents["maxX"] - extents["minX"])
|
||
height = max(1e-6, extents["maxY"] - extents["minY"])
|
||
fields: dict[str, dict[str, Any]] = {}
|
||
for t in texts:
|
||
rel_x = (t["x"] - extents["minX"]) / width
|
||
rel_y = (t["y"] - extents["minY"]) / height
|
||
if rel_x < 0.5 or rel_y > 0.5: # 标题栏通常在右下象限
|
||
continue
|
||
line = t["text"].split("\n", 1)[0].strip()
|
||
for pattern, field in _CN_LABEL_MAP:
|
||
m = pattern.match(line)
|
||
if m and field not in fields:
|
||
fields[field] = {
|
||
"value": _clean(m.group(1)), "confidence": 0.6,
|
||
"source": f"TEXT ({t['x']:.0f},{t['y']:.0f}) layer={t['layer']}",
|
||
"missing": False}
|
||
break
|
||
return {"found": bool(fields), "source": "cn-labels", "fields": fields}
|
||
|
||
|
||
def _merge_title_block(primary: dict[str, Any],
|
||
fallback: dict[str, Any]) -> dict[str, Any]:
|
||
"""通道合并：属性通道优先，中文标签通道补缺（不覆盖已有字段）。"""
|
||
if not primary.get("found"):
|
||
return fallback
|
||
if not fallback.get("found"):
|
||
return primary
|
||
merged = dict(primary)
|
||
fields = dict(primary.get("fields") or {})
|
||
for key, val in (fallback.get("fields") or {}).items():
|
||
cur = fields.get(key)
|
||
if cur is None or cur.get("missing"):
|
||
fields[key] = val
|
||
merged["fields"] = fields
|
||
merged["fallbackSource"] = fallback.get("source")
|
||
return merged
|
||
|
||
|
||
def _position_numbers(texts: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""装配位置号聚类：值 + 出现次数 + 图层（BOM 行候选的索引，不含名称/数量）。"""
|
||
found: dict[str, dict[str, Any]] = {}
|
||
for t in texts:
|
||
value = t["text"].strip().replace("oder", "").strip()
|
||
if not _POSITION_RE.match(value):
|
||
continue
|
||
row = found.setdefault(value, {"position": value, "count": 0, "layers": []})
|
||
row["count"] += 1
|
||
if t["layer"] not in row["layers"]:
|
||
row["layers"].append(t["layer"])
|
||
return sorted(found.values(), key=lambda r: r["position"])
|
||
|
||
|
||
def _technical_notes(texts: list[dict[str, Any]], *, cap: int = 30) -> list[dict[str, Any]]:
|
||
"""技术说明：多行/长文本/明确技术要求前缀（原文保留，仅参考，不推断工艺）。"""
|
||
notes: list[dict[str, Any]] = []
|
||
for t in texts:
|
||
text = t["text"]
|
||
if len(text) < _NOTE_MIN_LEN and not _NOTE_PREFIX_RE.search(text):
|
||
continue
|
||
if _POSITION_RE.match(text.strip()):
|
||
continue
|
||
notes.append({"text": text[:500], "layer": t["layer"],
|
||
"x": t["x"], "y": t["y"]})
|
||
if len(notes) >= cap:
|
||
break
|
||
return notes
|
||
|
||
|
||
def _extents_of(msp: Any) -> dict[str, float]:
|
||
"""模型空间几何范围（ezdxf bbox；空图回退 0）。"""
|
||
try:
|
||
import ezdxf.bbox
|
||
box = ezdxf.bbox.extents(msp)
|
||
if box.has_data:
|
||
return {"minX": round(box.extmin.x, 3), "minY": round(box.extmin.y, 3),
|
||
"maxX": round(box.extmax.x, 3), "maxY": round(box.extmax.y, 3)}
|
||
except Exception as exc: # noqa: BLE001 — ezdxf 几何异常类型不固定，降级不阻断
|
||
_log.debug("DXF 几何范围计算降级（不影响文字识别）：%s", exc)
|
||
return {"minX": 0.0, "minY": 0.0, "maxX": 0.0, "maxY": 0.0}
|
||
|
||
|
||
def parse_dxf(path: str, *, max_texts: int = 400,
|
||
display_name: str | None = None) -> dict[str, Any]:
|
||
"""DXF → ParsedDrawing（只读解析，绝不写世界状态）。
|
||
|
||
display_name：上传场景的真实文件名（临时文件路径不具备文件名线索）。
|
||
失败语义（fail closed）：文件不存在/不可读/非 DXF → 显式 ValueError；
|
||
几何范围异常仅降级为空范围，不阻断标题栏与文字识别。
|
||
"""
|
||
if not os.path.isfile(path):
|
||
raise ValueError(f"DXF 文件不存在：{path}")
|
||
try:
|
||
import ezdxf
|
||
except ImportError as exc: # 依赖缺失显式报错，不静默降级
|
||
raise ValueError("缺少 ezdxf 依赖，无法解析 DXF（pip install ezdxf）") from exc
|
||
try:
|
||
doc = ezdxf.readfile(path)
|
||
except Exception as exc:
|
||
raise ValueError(f"DXF 解析失败：{os.path.basename(path)}（{exc}）") from exc
|
||
|
||
with open(path, "rb") as fh:
|
||
raw = fh.read()
|
||
sha = hashlib.sha256(raw).hexdigest()
|
||
filename = display_name or os.path.basename(path)
|
||
file_meta = _filename_meta(filename)
|
||
|
||
msp = doc.modelspace()
|
||
entity_stats: dict[str, int] = {}
|
||
layer_counts: dict[str, int] = {}
|
||
for e in msp:
|
||
entity_stats[e.dxftype()] = entity_stats.get(e.dxftype(), 0) + 1
|
||
layer = str(getattr(e.dxf, "layer", "0"))
|
||
layer_counts[layer] = layer_counts.get(layer, 0) + 1
|
||
layers = [{"name": layer.dxf.name, "color": int(layer.dxf.color),
|
||
"entityCount": layer_counts.get(layer.dxf.name, 0)}
|
||
for layer in doc.layers]
|
||
|
||
texts = _collect_layout_texts(msp, cap=max_texts)
|
||
extents = _extents_of(msp)
|
||
title_block = _merge_title_block(
|
||
_title_block_from_attribs(doc),
|
||
_title_block_from_cn_texts(texts, extents))
|
||
|
||
fields = dict(title_block.get("fields") or {})
|
||
# 文件名交叉校验：图号数字一致则提置信，不一致仅提示（不覆盖属性值）
|
||
warnings: list[str] = []
|
||
fn_number = file_meta.get("drawingNumberNormalized", "")
|
||
dn = fields.get("drawingNumber")
|
||
if dn and not dn.get("missing"):
|
||
normalized = re.sub(r"\s+", "", dn["value"])
|
||
dn["normalized"] = normalized
|
||
if fn_number and fn_number == normalized:
|
||
dn["confidence"] = 0.95
|
||
dn["source"] += " + 文件名一致"
|
||
elif fn_number:
|
||
warnings.append(
|
||
f"标题栏图号 {normalized} 与文件名图号 {fn_number} 不一致，需人工核对")
|
||
elif fn_number:
|
||
fields["drawingNumber"] = {
|
||
"value": fn_number, "normalized": fn_number, "confidence": 0.7,
|
||
"source": "文件名（标题栏未找到图号）", "missing": False}
|
||
title_block["found"] = True
|
||
if not fields.get("title") or fields["title"].get("missing"):
|
||
fields.setdefault("title", {
|
||
"value": file_meta.get("stem", filename), "confidence": 0.5,
|
||
"source": "文件名（标题栏未找到名称）", "missing": False})
|
||
# 版本：文件名后缀优先与标题栏发放字母互证
|
||
rev = file_meta.get("revision", "")
|
||
release = fields.get("releaseLetter", {}).get("value", "")
|
||
if rev and release and rev == release.lower():
|
||
fields["revision"] = {"value": rev, "confidence": 0.9,
|
||
"source": "文件名后缀 + ATTRIB B_FREI 一致", "missing": False}
|
||
elif rev:
|
||
fields["revision"] = {"value": rev, "confidence": 0.7,
|
||
"source": "文件名后缀", "missing": False}
|
||
elif release:
|
||
fields["revision"] = {"value": release, "confidence": 0.7,
|
||
"source": "ATTRIB B_FREI", "missing": False}
|
||
|
||
title_block["fields"] = fields
|
||
drawing_id = sha[:16]
|
||
parsed = {
|
||
"drawingId": drawing_id,
|
||
"filename": filename,
|
||
"path": os.path.abspath(path),
|
||
"sizeBytes": len(raw),
|
||
"sha256": sha,
|
||
"dxfVersion": str(doc.dxfversion),
|
||
"encoding": str(doc.encoding),
|
||
"extents": extents,
|
||
"layers": layers,
|
||
"entityStats": entity_stats,
|
||
"textCount": len(texts),
|
||
"texts": texts,
|
||
"titleBlock": title_block,
|
||
"positionNumbers": _position_numbers(texts),
|
||
"technicalNotes": _technical_notes(texts),
|
||
"fileMeta": file_meta,
|
||
"warnings": warnings,
|
||
}
|
||
_cache_put(drawing_id, {"parsed": parsed, "path": os.path.abspath(path)})
|
||
return parsed
|
||
|
||
|
||
def _cache_put(drawing_id: str, entry: dict[str, Any]) -> None:
|
||
"""解析缓存：上限 _CACHE_MAX 份（先进先出，避免大图纸堆积内存）。
|
||
|
||
上传场景产生的临时文件（tempPath）随缓存淘汰一并清理。
|
||
"""
|
||
if drawing_id in _PARSE_CACHE:
|
||
_cache_evict(drawing_id)
|
||
while len(_PARSE_CACHE) >= _CACHE_MAX:
|
||
_cache_evict(next(iter(_PARSE_CACHE)))
|
||
_PARSE_CACHE[drawing_id] = entry
|
||
|
||
|
||
def _cache_evict(drawing_id: str) -> None:
|
||
entry = _PARSE_CACHE.pop(drawing_id, None)
|
||
tmp = (entry or {}).get("tempPath")
|
||
if tmp:
|
||
try:
|
||
os.unlink(tmp)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def get_cached_drawing(drawing_id: str) -> dict[str, Any] | None:
|
||
"""取解析缓存（SVG/确认卡复用，避免重复解析大文件）。"""
|
||
return _PARSE_CACHE.get(drawing_id)
|
||
|
||
|
||
def _field_value(fields: dict[str, Any], name: str, default: str = "") -> str:
|
||
f = fields.get(name) or {}
|
||
if f.get("missing"):
|
||
return default
|
||
return str(f.get("value") or default)
|
||
|
||
|
||
def _field_conf(fields: dict[str, Any], name: str, default: float = 0.0) -> float:
|
||
f = fields.get(name) or {}
|
||
return float(f.get("confidence", default))
|
||
|
||
|
||
def _infer_material_type(fields: dict[str, Any]) -> tuple[str, float, str]:
|
||
"""由图纸种类推断物料类型（推断值置信封顶 0.6，必须人工确认）。"""
|
||
kind = _field_value(fields, "drawingType").upper()
|
||
title = _field_value(fields, "title").lower()
|
||
if any(k in kind for k in ("EB", "INSTALLATION", "ZK", "ASSEMBLY", "GESAMT")) \
|
||
or any(k in title for k in ("installation", "assembly", "安装", "总装")):
|
||
return "FINISHED_PRODUCT", 0.6, "图纸种类为安装/装配图（推断，需确认）"
|
||
if any(k in kind for k in ("ET", "TEIL", "PART")) \
|
||
or any(k in title for k in ("零件", "part")):
|
||
return "SEMI_FINISHED", 0.6, "图纸种类为零件图（推断，需确认）"
|
||
return "SEMI_FINISHED", 0.4, "图纸种类未知（默认半成品，需确认）"
|
||
|
||
|
||
def drawing_to_master_candidates(parsed: dict[str, Any]) -> dict[str, Any]:
|
||
"""ParsedDrawing → 物料/BOM/工艺路线候选（只生成图纸承载的信息）。
|
||
|
||
安全边界（fail closed，不编造）：
|
||
- 物料：编码/名称来自标题栏；类型/单位为推断值（置信 ≤0.6）；
|
||
- BOM：仅有装配位置号，无明细表时名称/数量/子件编码全部 missing；
|
||
- 工艺路线：图纸无工艺信息 → steps 空 + complete False；
|
||
- 材质/板厚/采购属性/标准工时/外协：图纸未承载 → 显式 missing。
|
||
"""
|
||
fields = (parsed.get("titleBlock") or {}).get("fields") or {}
|
||
file_meta = parsed.get("fileMeta") or {}
|
||
stem = file_meta.get("stem") or parsed.get("filename", "")
|
||
|
||
normalized = _field_value(fields, "drawingNumber")
|
||
normalized = re.sub(r"\s+", "", normalized) if normalized else ""
|
||
sheet = _field_value(fields, "sheet")
|
||
# 张次格式对齐文件名习惯：001 vs 1 数值一致时沿用文件名写法
|
||
fn_sheet = str(file_meta.get("sheet") or "")
|
||
if sheet and fn_sheet and sheet.isdigit() and fn_sheet.isdigit() \
|
||
and int(sheet) == int(fn_sheet):
|
||
sheet = fn_sheet
|
||
if normalized and sheet:
|
||
code = f"{normalized}-{sheet}"
|
||
elif normalized:
|
||
code = normalized
|
||
else:
|
||
code = stem
|
||
name = _field_value(fields, "title") or stem
|
||
mtype, _type_conf, type_source = _infer_material_type(fields)
|
||
|
||
spec_parts = []
|
||
if _field_value(fields, "applicableTo"):
|
||
spec_parts.append(f"适用 {_field_value(fields, 'applicableTo')}")
|
||
if _field_value(fields, "scale"):
|
||
spec_parts.append(f"比例 {_field_value(fields, 'scale')}")
|
||
if _field_value(fields, "revision"):
|
||
spec_parts.append(f"版本 {_field_value(fields, 'revision')}")
|
||
if _field_value(fields, "drawingType"):
|
||
spec_parts.append(_field_value(fields, "drawingType"))
|
||
|
||
material_missing: list[str] = []
|
||
if not code:
|
||
material_missing.append("code")
|
||
if not name or name == stem and _field_conf(fields, "title") < 0.6:
|
||
material_missing.append("name（标题栏未识别，需确认）")
|
||
material_conf = round(min(
|
||
_field_conf(fields, "drawingNumber", 0.5),
|
||
_field_conf(fields, "title", 0.5),
|
||
0.95), 4)
|
||
material = {
|
||
"code": code, "name": name, "type": mtype, "unit": "件",
|
||
"spec": " · ".join(spec_parts),
|
||
"confidence": material_conf,
|
||
"requiresConfirm": True,
|
||
"missingFields": material_missing,
|
||
"fieldSources": {
|
||
"code": (fields.get("drawingNumber") or {}).get("source", "文件名"),
|
||
"name": (fields.get("title") or {}).get("source", "文件名"),
|
||
"type": type_source,
|
||
"unit": "默认值「件」（图纸未承载计量单位，需确认）",
|
||
},
|
||
"notFromDrawing": [
|
||
"材质/板厚（WERKSTOFF 未填写或图纸未标注）",
|
||
"库存/安全库存/采购前置期/采购属性",
|
||
],
|
||
"materialSpec": _field_value(fields, "materialSpec"),
|
||
"thickness": _field_value(fields, "thickness"),
|
||
}
|
||
|
||
bom_items = [{
|
||
"position": p["position"],
|
||
"materialCode": "", "name": "", "quantity": None,
|
||
"confidence": 0.2,
|
||
"missing": ["materialCode", "name", "quantity"],
|
||
"source": f"图纸位置号（图层 {','.join(p['layers'])} · 出现 {p['count']} 次）",
|
||
} for p in parsed.get("positionNumbers") or []]
|
||
bom = {
|
||
"productCode": code,
|
||
"version": "V1.0",
|
||
"items": bom_items,
|
||
"complete": False,
|
||
"requiresConfirm": True,
|
||
"note": ("图纸仅含装配位置号、无明细表（Stückliste）："
|
||
"子件编码/名称/数量均需人工补全后才能建 BOM"
|
||
if bom_items else "图纸未识别到装配位置号，BOM 需人工编制"),
|
||
}
|
||
|
||
routing = {
|
||
"productCode": code,
|
||
"version": "V1.0",
|
||
"steps": [],
|
||
"complete": False,
|
||
"requiresConfirm": True,
|
||
"note": "图纸不承载工艺信息：工序/顺序/标准工时/外协判断均需人工确认后编制",
|
||
}
|
||
|
||
return {
|
||
"drawingId": parsed.get("drawingId"),
|
||
"filename": parsed.get("filename"),
|
||
"material": material,
|
||
"bom": bom,
|
||
"routing": routing,
|
||
"boundary": ("候选仅含图纸可识别信息；标注 missing/requiresConfirm 的字段"
|
||
"必须经 P2 人工确认，确认前不写正式主数据"),
|
||
}
|
||
|
||
|
||
def dxf_to_svg(path: str, *, background: str = "white",
|
||
include_layers: list[str] | None = None) -> str:
|
||
"""模型空间 → SVG 预览（白底；几何渲染异常显式报错，不返回半成品）。
|
||
|
||
include_layers：只渲染指定图层（前端图层开关）；None 渲染全部。
|
||
图层关闭仅作用于本次渲染的临时 doc（每次都重新读文件，不污染源文件）。
|
||
"""
|
||
if not os.path.isfile(path):
|
||
raise ValueError(f"DXF 文件不存在：{path}")
|
||
import ezdxf
|
||
from ezdxf.addons.drawing import Frontend, RenderContext
|
||
from ezdxf.addons.drawing.config import BackgroundPolicy, Configuration
|
||
from ezdxf.addons.drawing.layout import Page, Settings
|
||
from ezdxf.addons.drawing.svg import SVGBackend
|
||
doc = ezdxf.readfile(path)
|
||
if include_layers is not None:
|
||
wanted = set(include_layers)
|
||
for layer in doc.layers:
|
||
if layer.dxf.name not in wanted:
|
||
layer.off()
|
||
msp = doc.modelspace()
|
||
policy = (BackgroundPolicy.WHITE if background == "white"
|
||
else BackgroundPolicy.BLACK)
|
||
ctx = RenderContext(doc)
|
||
backend = SVGBackend()
|
||
config = Configuration(background_policy=policy)
|
||
Frontend(ctx, backend, config=config).draw_layout(msp, finalize=True)
|
||
page = Page.from_dxf_layout(msp)
|
||
return backend.get_string(page, settings=Settings(fit_page=True))
|
||
|
||
|
||
_SVG_VARIANT_MAX = 16
|
||
|
||
|
||
def dxf_svg_cached(drawing_id: str, include_layers: list[str] | None = None) -> str:
|
||
"""按 drawingId 生成/复用 SVG（缓存内取路径，避免重复解析大文件）。
|
||
|
||
图层过滤变体按排序后图层键缓存（上限 _SVG_VARIANT_MAX，防组合爆炸）。
|
||
"""
|
||
entry = get_cached_drawing(drawing_id)
|
||
if entry is None:
|
||
raise ValueError(f"图纸未解析或缓存已淘汰：{drawing_id}（请先调用解析接口）")
|
||
if include_layers is None:
|
||
if "svg" not in entry:
|
||
entry["svg"] = dxf_to_svg(entry["path"])
|
||
return entry["svg"]
|
||
key = ",".join(sorted(set(include_layers)))
|
||
variants: dict[str, str] = entry.setdefault("svgVariants", {})
|
||
if key not in variants:
|
||
while len(variants) >= _SVG_VARIANT_MAX:
|
||
variants.pop(next(iter(variants)))
|
||
variants[key] = dxf_to_svg(entry["path"], include_layers=include_layers)
|
||
return variants[key]
|
||
|
||
|
||
_DXF_EXT_RE = re.compile(r"\.dxf$", re.IGNORECASE)
|
||
|
||
|
||
def discover_dxf_files(root: str, *, recursive: bool = True,
|
||
limit: int = 200) -> dict[str, Any]:
|
||
"""项目工程目录 DXF 发现（只读扫描；目录不存在显式报错）。"""
|
||
if not os.path.isdir(root):
|
||
raise ValueError(f"工程目录不存在：{root}")
|
||
files: list[dict[str, Any]] = []
|
||
if recursive:
|
||
walker = os.walk(root)
|
||
entries = (os.path.join(dir_path, name)
|
||
for dir_path, _, names in walker for name in names)
|
||
else:
|
||
entries = (os.path.join(root, name) for name in os.listdir(root))
|
||
for path in entries:
|
||
if not _DXF_EXT_RE.search(path):
|
||
continue
|
||
try:
|
||
stat = os.stat(path)
|
||
except OSError:
|
||
continue
|
||
files.append({
|
||
"path": os.path.abspath(path),
|
||
"filename": os.path.basename(path),
|
||
"sizeBytes": stat.st_size,
|
||
"modifiedAt": int(stat.st_mtime),
|
||
"fileMeta": _filename_meta(path),
|
||
})
|
||
if len(files) >= limit:
|
||
break
|
||
files.sort(key=lambda f: f["filename"])
|
||
return {"root": os.path.abspath(root), "count": len(files),
|
||
"truncated": len(files) >= limit, "files": files}
|
||
|
||
|
||
def parse_dxf_bytes(filename: str, data: bytes) -> dict[str, Any]:
|
||
"""字节流解析入口（上传场景）：写临时文件后走 parse_dxf。
|
||
|
||
临时文件随缓存保留（SVG 复用需要真实路径），缓存淘汰时一并清理。
|
||
"""
|
||
suffix = os.path.splitext(filename)[1] or ".dxf"
|
||
fd, tmp = tempfile.mkstemp(prefix="aps-dxf-", suffix=suffix)
|
||
with os.fdopen(fd, "wb") as fh:
|
||
fh.write(data)
|
||
parsed = parse_dxf(tmp, display_name=filename)
|
||
# 路径保留临时文件（SVG 复用），缓存淘汰时一并清理
|
||
entry = get_cached_drawing(parsed["drawingId"])
|
||
if entry is not None:
|
||
entry["tempPath"] = tmp # 缓存淘汰时清理由 _cache_evict 负责
|
||
return parsed
|