489 lines
19 KiB
Python
489 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
"""Safe DXF inspection, preview rendering, and master-data candidate building.
|
|
|
|
This module deliberately stops at reviewable candidates. A drawing is evidence,
|
|
not authority for quantities, standard time, or production-resource assignments.
|
|
"""
|
|
|
|
from collections import Counter
|
|
from datetime import datetime, timezone
|
|
from hashlib import sha256
|
|
from html import escape
|
|
import math
|
|
from pathlib import Path
|
|
import re
|
|
from typing import Any, Iterable
|
|
|
|
MAX_DXF_BYTES = 25 * 1024 * 1024
|
|
MAX_ENTITIES = 20_000
|
|
MAX_TEXTS = 1_000
|
|
MAX_DIMENSIONS = 1_000
|
|
MAX_BLOCK_REFERENCES = 1_000
|
|
MAX_SVG_BYTES = 5 * 1024 * 1024
|
|
PREVIEW_ENTITY_LIMIT = 12_000
|
|
|
|
_DRAWING_NAME_RE = re.compile(
|
|
r"^(?P<drawing_no>[A-Za-z0-9]+(?:-[A-Za-z0-9]+)+)-(?P<revision>[A-Za-z])(?:\(\d+\))?$"
|
|
)
|
|
_ITEM_REF_RE = re.compile(r"^(?:[A-Z]?\d{2,}[.]\d{2,}[A-Z]?|\d{6,}(?:-\d+)?)$", re.I)
|
|
_SCALE_RE = re.compile(r"(?:SCALE\s*)?(\d+\s*:\s*\d+)", re.I)
|
|
|
|
|
|
def _utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
|
|
|
|
def _stable_id(kind: str, digest: str, suffix: str = "") -> str:
|
|
seed = f"{kind}:{digest}:{suffix}".encode("utf-8")
|
|
return f"{kind}_{sha256(seed).hexdigest()[:20]}"
|
|
|
|
|
|
def _safe_path(path: str | Path) -> Path:
|
|
p = Path(path).expanduser().resolve(strict=True)
|
|
if not p.is_file() or p.suffix.lower() != ".dxf":
|
|
raise ValueError("Only an existing .dxf file can be inspected")
|
|
size = p.stat().st_size
|
|
if size <= 0:
|
|
raise ValueError("DXF file is empty")
|
|
if size > MAX_DXF_BYTES:
|
|
raise ValueError(f"DXF file exceeds {MAX_DXF_BYTES} bytes")
|
|
return p
|
|
|
|
|
|
def _file_digest(path: Path) -> str:
|
|
h = sha256()
|
|
with path.open("rb") as stream:
|
|
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
|
|
h.update(chunk)
|
|
return h.hexdigest()
|
|
|
|
|
|
def _filename_metadata(path: Path) -> dict[str, str | None]:
|
|
match = _DRAWING_NAME_RE.match(path.stem)
|
|
if not match:
|
|
return {"drawingNumber": path.stem, "revision": None}
|
|
return {
|
|
"drawingNumber": match.group("drawing_no"),
|
|
"revision": match.group("revision").lower(),
|
|
}
|
|
|
|
|
|
def _clean_text(value: Any) -> str:
|
|
text = str(value or "")
|
|
text = text.replace("\\P", "\n").replace("%%d", "°")
|
|
text = re.sub(r"[{}]", "", text)
|
|
return "\n".join(part.strip() for part in text.splitlines() if part.strip())
|
|
|
|
|
|
def _point(value: Any) -> list[float] | None:
|
|
try:
|
|
return [round(float(value[0]), 6), round(float(value[1]), 6)]
|
|
except (TypeError, ValueError, IndexError):
|
|
return None
|
|
|
|
|
|
def _entity_evidence(entity: Any, text: str | None = None) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {
|
|
"entityHandle": str(getattr(entity.dxf, "handle", "") or ""),
|
|
"entityType": entity.dxftype(),
|
|
"layer": str(getattr(entity.dxf, "layer", "0") or "0"),
|
|
}
|
|
insert = _point(getattr(entity.dxf, "insert", None))
|
|
if insert:
|
|
payload["position"] = insert
|
|
if text is not None:
|
|
payload["rawText"] = text
|
|
return payload
|
|
|
|
|
|
def _bbox_payload(doc: Any, entities: Iterable[Any]) -> dict[str, Any] | None:
|
|
try:
|
|
from ezdxf import bbox
|
|
|
|
box = bbox.extents(entities, fast=True)
|
|
if not box.has_data:
|
|
return None
|
|
return {
|
|
"min": [round(box.extmin.x, 6), round(box.extmin.y, 6)],
|
|
"max": [round(box.extmax.x, 6), round(box.extmax.y, 6)],
|
|
"width": round(box.size.x, 6),
|
|
"height": round(box.size.y, 6),
|
|
}
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _field_candidate(
|
|
field: str,
|
|
value: Any,
|
|
confidence: float,
|
|
evidence: dict[str, Any],
|
|
*,
|
|
review_required: bool = False,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"field": field,
|
|
"value": value,
|
|
"confidence": round(max(0.0, min(float(confidence), 1.0)), 3),
|
|
"reviewRequired": bool(review_required),
|
|
"evidence": evidence,
|
|
}
|
|
|
|
|
|
def inspect_dxf(path: str | Path) -> dict[str, Any]:
|
|
"""Parse a DXF into a stable, JSON-serializable evidence contract."""
|
|
|
|
p = _safe_path(path)
|
|
digest = _file_digest(p)
|
|
try:
|
|
import ezdxf
|
|
from ezdxf.lldxf.const import acad_release
|
|
except ImportError as exc: # pragma: no cover - environment guard
|
|
raise RuntimeError("DXF parsing requires the 'ezdxf' package") from exc
|
|
|
|
try:
|
|
doc = ezdxf.readfile(p)
|
|
except Exception as exc:
|
|
raise ValueError(f"Invalid or unsupported DXF: {exc}") from exc
|
|
|
|
modelspace = doc.modelspace()
|
|
if len(modelspace) > MAX_ENTITIES:
|
|
raise ValueError(f"DXF contains more than {MAX_ENTITIES} modelspace entities")
|
|
|
|
entity_counts: Counter[str] = Counter()
|
|
layer_counts: Counter[str] = Counter()
|
|
texts: list[dict[str, Any]] = []
|
|
dimensions: list[dict[str, Any]] = []
|
|
block_refs: list[dict[str, Any]] = []
|
|
|
|
for entity in modelspace:
|
|
kind = entity.dxftype()
|
|
layer = str(getattr(entity.dxf, "layer", "0") or "0")
|
|
entity_counts[kind] += 1
|
|
layer_counts[layer] += 1
|
|
|
|
if kind in {"TEXT", "MTEXT", "ATTRIB"} and len(texts) < MAX_TEXTS:
|
|
raw = entity.text if kind == "MTEXT" else getattr(entity.dxf, "text", "")
|
|
clean = _clean_text(raw)
|
|
if clean:
|
|
texts.append({
|
|
"id": _stable_id("text", digest, str(getattr(entity.dxf, "handle", ""))),
|
|
"text": clean,
|
|
"evidence": _entity_evidence(entity, str(raw)),
|
|
})
|
|
elif kind == "DIMENSION" and len(dimensions) < MAX_DIMENSIONS:
|
|
try:
|
|
measurement = entity.get_measurement()
|
|
measurement = round(float(measurement), 6)
|
|
except Exception:
|
|
measurement = None
|
|
dimensions.append({
|
|
"id": _stable_id("dimension", digest, str(getattr(entity.dxf, "handle", ""))),
|
|
"measurement": measurement,
|
|
"dimensionType": int(getattr(entity.dxf, "dimtype", 0) or 0) & 15,
|
|
"textOverride": _clean_text(getattr(entity.dxf, "text", "")) or None,
|
|
"evidence": _entity_evidence(entity),
|
|
})
|
|
elif kind == "INSERT" and len(block_refs) < MAX_BLOCK_REFERENCES:
|
|
attrs = []
|
|
for attrib in getattr(entity, "attribs", []):
|
|
attrs.append({
|
|
"tag": str(getattr(attrib.dxf, "tag", "") or ""),
|
|
"value": _clean_text(getattr(attrib.dxf, "text", "")),
|
|
"handle": str(getattr(attrib.dxf, "handle", "") or ""),
|
|
})
|
|
block_refs.append({
|
|
"id": _stable_id("blockref", digest, str(getattr(entity.dxf, "handle", ""))),
|
|
"name": str(getattr(entity.dxf, "name", "") or ""),
|
|
"insert": _point(getattr(entity.dxf, "insert", None)),
|
|
"rotation": round(float(getattr(entity.dxf, "rotation", 0.0) or 0.0), 6),
|
|
"attributes": attrs,
|
|
"evidence": _entity_evidence(entity),
|
|
})
|
|
|
|
name_meta = _filename_metadata(p)
|
|
fields = [
|
|
_field_candidate(
|
|
"drawingNumber", name_meta["drawingNumber"], 0.99,
|
|
{"source": "filename", "filename": p.name},
|
|
)
|
|
]
|
|
if name_meta["revision"]:
|
|
fields.append(_field_candidate(
|
|
"revision", name_meta["revision"], 0.98,
|
|
{"source": "filename", "filename": p.name},
|
|
))
|
|
|
|
all_text = [row["text"] for row in texts]
|
|
scale_match = next((_SCALE_RE.search(text) for text in all_text if _SCALE_RE.search(text)), None)
|
|
if scale_match:
|
|
fields.append(_field_candidate(
|
|
"scale", scale_match.group(1).replace(" ", ""), 0.82,
|
|
{"source": "entityText", "matchedText": scale_match.group(0)}, review_required=True,
|
|
))
|
|
|
|
item_refs: list[dict[str, Any]] = []
|
|
seen_refs: set[str] = set()
|
|
for row in texts:
|
|
for line in row["text"].splitlines():
|
|
token = line.strip()
|
|
if _ITEM_REF_RE.fullmatch(token) and token not in seen_refs:
|
|
seen_refs.add(token)
|
|
item_refs.append({
|
|
"reference": token,
|
|
"confidence": 0.72,
|
|
"reviewRequired": True,
|
|
"evidence": row["evidence"],
|
|
})
|
|
|
|
layers = []
|
|
for layer in doc.layers:
|
|
name = str(layer.dxf.name)
|
|
layers.append({
|
|
"name": name,
|
|
"color": int(getattr(layer.dxf, "color", 7) or 7),
|
|
"linetype": str(getattr(layer.dxf, "linetype", "CONTINUOUS") or "CONTINUOUS"),
|
|
"entityCount": int(layer_counts.get(name, 0)),
|
|
"off": bool(layer.is_off()),
|
|
"frozen": bool(layer.is_frozen()),
|
|
})
|
|
|
|
header = {
|
|
"acadVersion": str(doc.dxfversion),
|
|
"release": str(acad_release.get(doc.dxfversion, "unknown")),
|
|
"encoding": str(doc.encoding),
|
|
"units": int(doc.header.get("$INSUNITS", 0) or 0),
|
|
}
|
|
asset_id = _stable_id("drawing", digest)
|
|
parsed = {
|
|
"contractVersion": "drawing-dxf.v1",
|
|
"asset": {
|
|
"id": asset_id,
|
|
"kind": "DXF",
|
|
"filename": p.name,
|
|
"sourcePath": str(p),
|
|
"sizeBytes": p.stat().st_size,
|
|
"sha256": digest,
|
|
"mediaType": "image/vnd.dxf",
|
|
"idempotencyKey": f"drawing:{digest}",
|
|
},
|
|
"drawing": {
|
|
"drawingNumber": name_meta["drawingNumber"],
|
|
"revision": name_meta["revision"],
|
|
"metadata": header,
|
|
"bbox": _bbox_payload(doc, modelspace),
|
|
"modelspaceEntityCount": len(modelspace),
|
|
"layerCount": len(layers),
|
|
"blockDefinitionCount": len(doc.blocks),
|
|
},
|
|
"layers": sorted(layers, key=lambda item: item["name"].casefold()),
|
|
"entityStatistics": dict(sorted(entity_counts.items())),
|
|
"texts": texts,
|
|
"dimensions": dimensions,
|
|
"blockReferences": block_refs,
|
|
"fieldCandidates": fields,
|
|
"itemReferences": item_refs,
|
|
"warnings": [
|
|
"DXF extraction produces reviewable evidence only; it does not prove BOM quantity, material specification, standard time, or resource assignment."
|
|
],
|
|
"parsedAt": _utc_now(),
|
|
}
|
|
from server.aps_domain.drawing_process import recognize_process_candidates
|
|
process = recognize_process_candidates(parsed)
|
|
for detail in process.get("details") or []:
|
|
parsed["fieldCandidates"].append(_field_candidate(
|
|
detail["field"], detail["value"], float(detail["confidence"]),
|
|
detail["evidence"], review_required=True,
|
|
))
|
|
return parsed
|
|
|
|
|
|
def _svg_num(value: Any) -> str:
|
|
number = float(value)
|
|
if not math.isfinite(number):
|
|
raise ValueError("non-finite coordinate")
|
|
return f"{number:.4f}".rstrip("0").rstrip(".") or "0"
|
|
|
|
|
|
def render_dxf_preview(path: str | Path) -> str:
|
|
"""Render a bounded standalone SVG; no scripts, external links, or foreignObject."""
|
|
|
|
p = _safe_path(path)
|
|
try:
|
|
import ezdxf
|
|
from ezdxf.lldxf.const import acad_release
|
|
except ImportError as exc: # pragma: no cover
|
|
raise RuntimeError("DXF preview requires the 'ezdxf' package") from exc
|
|
try:
|
|
doc = ezdxf.readfile(p)
|
|
except Exception as exc:
|
|
raise ValueError(f"Invalid or unsupported DXF: {exc}") from exc
|
|
|
|
modelspace = doc.modelspace()
|
|
box = _bbox_payload(doc, modelspace)
|
|
if not box:
|
|
raise ValueError("DXF has no renderable extents")
|
|
min_x, min_y = box["min"]
|
|
width = max(float(box["width"]), 1.0)
|
|
height = max(float(box["height"]), 1.0)
|
|
stroke = max(width, height) / 2500.0
|
|
parts = [
|
|
'<svg xmlns="http://www.w3.org/2000/svg" role="img" aria-label="DXF drawing preview"',
|
|
f' viewBox="{_svg_num(min_x)} {_svg_num(-min_y-height)} {_svg_num(width)} {_svg_num(height)}">',
|
|
'<rect width="100%" height="100%" fill="#fff"/>',
|
|
f'<g transform="scale(1,-1)" fill="none" stroke="#1f2937" stroke-width="{_svg_num(stroke)}" vector-effect="non-scaling-stroke">',
|
|
]
|
|
rendered = 0
|
|
for entity in modelspace:
|
|
if rendered >= PREVIEW_ENTITY_LIMIT:
|
|
break
|
|
try:
|
|
kind = entity.dxftype()
|
|
handle = escape(str(getattr(entity.dxf, "handle", "") or ""), quote=True)
|
|
layer = escape(str(getattr(entity.dxf, "layer", "0") or "0"), quote=True)
|
|
attrs = f' data-handle="{handle}" data-layer="{layer}"'
|
|
if kind == "LINE":
|
|
a, b = entity.dxf.start, entity.dxf.end
|
|
parts.append(f'<line{attrs} x1="{_svg_num(a.x)}" y1="{_svg_num(a.y)}" x2="{_svg_num(b.x)}" y2="{_svg_num(b.y)}"/>')
|
|
elif kind == "CIRCLE":
|
|
c = entity.dxf.center
|
|
parts.append(f'<circle{attrs} cx="{_svg_num(c.x)}" cy="{_svg_num(c.y)}" r="{_svg_num(abs(entity.dxf.radius))}"/>')
|
|
elif kind == "ARC":
|
|
c, r = entity.dxf.center, abs(float(entity.dxf.radius))
|
|
a1, a2 = math.radians(float(entity.dxf.start_angle)), math.radians(float(entity.dxf.end_angle))
|
|
x1, y1 = c.x + r * math.cos(a1), c.y + r * math.sin(a1)
|
|
x2, y2 = c.x + r * math.cos(a2), c.y + r * math.sin(a2)
|
|
sweep = (float(entity.dxf.end_angle) - float(entity.dxf.start_angle)) % 360.0
|
|
large = 1 if sweep > 180.0 else 0
|
|
parts.append(f'<path{attrs} d="M {_svg_num(x1)} {_svg_num(y1)} A {_svg_num(r)} {_svg_num(r)} 0 {large} 1 {_svg_num(x2)} {_svg_num(y2)}"/>')
|
|
elif kind == "LWPOLYLINE":
|
|
points = list(entity.get_points("xy"))[:5000]
|
|
if len(points) >= 2:
|
|
pts = " ".join(f"{_svg_num(x)},{_svg_num(y)}" for x, y in points)
|
|
tag = "polygon" if entity.closed else "polyline"
|
|
parts.append(f'<{tag}{attrs} points="{pts}"/>')
|
|
elif kind == "POINT":
|
|
c = entity.dxf.location
|
|
parts.append(f'<circle{attrs} cx="{_svg_num(c.x)}" cy="{_svg_num(c.y)}" r="{_svg_num(stroke*2)}" fill="#1f2937"/>')
|
|
else:
|
|
continue
|
|
rendered += 1
|
|
except (AttributeError, TypeError, ValueError, OverflowError):
|
|
continue
|
|
if sum(len(part) for part in parts) > MAX_SVG_BYTES - 256:
|
|
break
|
|
parts.append("</g>")
|
|
parts.append(f'<metadata>renderedEntities={rendered}; sourceSha256={_file_digest(p)}</metadata>')
|
|
parts.append("</svg>")
|
|
svg = "".join(parts)
|
|
if len(svg.encode("utf-8")) > MAX_SVG_BYTES:
|
|
raise ValueError("Generated SVG exceeds safe output limit")
|
|
return svg
|
|
|
|
|
|
def build_drawing_master_candidates(parsed: dict[str, Any]) -> dict[str, Any]:
|
|
"""Build non-authoritative material, BOM-reference, and routing candidates."""
|
|
|
|
asset = parsed.get("asset") or {}
|
|
drawing = parsed.get("drawing") or {}
|
|
digest = str(asset.get("sha256") or "")
|
|
if not digest or len(digest) != 64:
|
|
raise ValueError("parsed drawing is missing a valid sha256")
|
|
drawing_no = str(drawing.get("drawingNumber") or "").strip()
|
|
if not drawing_no:
|
|
raise ValueError("parsed drawing is missing drawingNumber")
|
|
revision = drawing.get("revision")
|
|
evidence_base = {
|
|
"drawingAssetId": asset.get("id"),
|
|
"drawingNumber": drawing_no,
|
|
"revision": revision,
|
|
"sourceSha256": digest,
|
|
}
|
|
|
|
material = {
|
|
"candidateId": _stable_id("materialCandidate", digest, drawing_no),
|
|
"candidateType": "MATERIAL",
|
|
"code": drawing_no,
|
|
"name": f"图纸件 {drawing_no}",
|
|
"drawingNumber": drawing_no,
|
|
"revision": revision,
|
|
"confidence": 0.76,
|
|
"reviewRequired": True,
|
|
"unknownFields": ["materialType", "baseUnit", "specification"],
|
|
"evidence": evidence_base,
|
|
}
|
|
|
|
bom_candidates = []
|
|
for row in parsed.get("itemReferences") or []:
|
|
reference = str(row.get("reference") or "").strip()
|
|
if not reference:
|
|
continue
|
|
bom_candidates.append({
|
|
"candidateId": _stable_id("bomCandidate", digest, reference),
|
|
"candidateType": "BOM_REFERENCE",
|
|
"parentMaterialCode": drawing_no,
|
|
"componentReference": reference,
|
|
"componentMaterialCode": None,
|
|
"quantity": None,
|
|
"unit": None,
|
|
"confidence": round(min(float(row.get("confidence") or 0.0), 0.79), 3),
|
|
"reviewRequired": True,
|
|
"resolutionStatus": "UNRESOLVED",
|
|
"unknownFields": ["componentMaterialCode", "quantity", "unit"],
|
|
"evidence": {**evidence_base, "entity": row.get("evidence")},
|
|
})
|
|
|
|
from server.aps_domain.drawing_process import recognize_process_candidates
|
|
process = recognize_process_candidates(parsed)
|
|
operation_hints: list[tuple[str, str, float]] = [
|
|
(row["operationCode"], row["operationName"], float(row["confidence"]))
|
|
for row in (process.get("operations") or [])
|
|
]
|
|
if parsed.get("dimensions"):
|
|
operation_hints.append(("DIMENSION_INSPECTION", "尺寸检验", 0.70))
|
|
if not operation_hints:
|
|
operation_hints.append(("ENGINEERING_REVIEW", "工程评审", 0.45))
|
|
|
|
routing_candidates = []
|
|
for sequence, (code, name, confidence) in enumerate(operation_hints, start=10):
|
|
routing_candidates.append({
|
|
"candidateId": _stable_id("routingCandidate", digest, code),
|
|
"candidateType": "ROUTING_OPERATION",
|
|
"productCode": drawing_no,
|
|
"sequenceHint": sequence,
|
|
"operationCode": code,
|
|
"operationName": name,
|
|
"standardTime": None,
|
|
"resourceCode": None,
|
|
"confidence": confidence,
|
|
"reviewRequired": True,
|
|
"unknownFields": ["standardTime", "resourceCode"],
|
|
"evidence": evidence_base,
|
|
})
|
|
|
|
return {
|
|
"contractVersion": "drawing-master-candidates.v1",
|
|
"changeSetId": _stable_id("drawingChangeSet", digest),
|
|
"idempotencyKey": f"drawing-master:{digest}",
|
|
"source": evidence_base,
|
|
"status": "PENDING_REVIEW",
|
|
"reviewRequired": True,
|
|
"materials": [material],
|
|
"bomReferences": bom_candidates,
|
|
"routingOperations": routing_candidates,
|
|
"processDetails": process.get("details") or [],
|
|
"processBlocks": process.get("blocks") or [],
|
|
"processAssemblies": process.get("assemblies") or [],
|
|
"processLocations": process.get("locations") or [],
|
|
"processSequenceSteps": process.get("sequenceSteps") or [],
|
|
"prohibitedAutoFill": ["BOM.quantity", "BOM.unit", "routing.standardTime", "routing.resourceCode"],
|
|
"warnings": [
|
|
"Candidates must pass engineering review/P2 approval before master-data write.",
|
|
"No BOM quantity, standard time, or production resource was inferred from geometry alone.",
|
|
],
|
|
}
|
|
|
|
|
|
__all__ = ["inspect_dxf", "render_dxf_preview", "build_drawing_master_candidates"]
|