330 lines
14 KiB
Python
330 lines
14 KiB
Python
# ============================================================
|
||
# 图纸/工艺文档关键信息识别（moduleId: domain-drawing-process, 可重生 ✅）
|
||
# 只产出 PENDING_REVIEW 候选：工艺/工序、分段/组立、场地、装配顺序、
|
||
# 表面粗糙度、公差、螺纹、深度、材料等。标准工时/资源/数量一律不推断。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from hashlib import sha256
|
||
from typing import Any
|
||
|
||
PROCESS_VERSION = "drawing-process.v1"
|
||
|
||
_PROCESS_TERMS: tuple[tuple[re.Pattern[str], str, str, str], ...] = (
|
||
(re.compile(r"\bDREHEN\b|\bTURNING\b|\bTURN\b", re.IGNORECASE), "TURNING", "车削", "MACHINING"),
|
||
(re.compile(r"\bFR[AÄ]SEN\b|\bMILLING\b|\bMILL\b", re.IGNORECASE), "MILLING", "铣削", "MACHINING"),
|
||
(re.compile(r"\bBOHREN\b|\bDRILLING\b|\bDRILL\b", re.IGNORECASE), "DRILLING", "钻孔", "MACHINING"),
|
||
(re.compile(r"\bSCHLEIFEN\b|\bGRINDING\b|\bGRIND\b", re.IGNORECASE), "GRINDING", "磨削", "MACHINING"),
|
||
(
|
||
re.compile(r"\bGEWINDE\b|\bTHREAD(?:ING)?\b|[MG]\d+(?:[xX*]\d+(?:\.\d+)?)?(?!\s*[:：]\s*\d)", re.IGNORECASE),
|
||
"THREADING", "螺纹加工", "MACHINING",
|
||
),
|
||
(re.compile(r"\bSCHWEISSEN\b|\bWELD(?:ING)?\b|焊接", re.IGNORECASE), "WELDING", "焊接", "JOINING"),
|
||
(re.compile(r"拼板|\bPANEL\s*FABRICATION\b", re.IGNORECASE), "PANEL_FABRICATION", "拼板", "FABRICATION"),
|
||
(re.compile(r"装焊|\bFIT\s*-?\s*UP\s*WELD", re.IGNORECASE), "FIT_UP_WELDING", "装焊", "FABRICATION"),
|
||
(re.compile(r"划线|\bMARK(?:ING)?\b", re.IGNORECASE), "MARKING", "划线", "FABRICATION"),
|
||
(re.compile(r"下料|\bCUTTING\b|\bCUT\b", re.IGNORECASE), "CUTTING", "下料", "FABRICATION"),
|
||
(re.compile(r"卷制|\bROLL(?:ING)?\b", re.IGNORECASE), "ROLLING", "卷制", "FABRICATION"),
|
||
(re.compile(r"翻身|\bTURN\s*-?\s*OVER\b", re.IGNORECASE), "TURNOVER", "翻身", "FABRICATION"),
|
||
(re.compile(r"反造|\bINVERTED\s*BUILD\b", re.IGNORECASE), "INVERTED_BUILD", "反造", "FABRICATION"),
|
||
(re.compile(r"舾装|\bOUTFIT(?:TING)?\b", re.IGNORECASE), "OUTFITTING", "舾装", "OUTFITTING"),
|
||
(re.compile(r"管系|\bPIPING\b|\bPIPE\b", re.IGNORECASE), "PIPING", "管系", "OUTFITTING"),
|
||
(re.compile(r"胎架|平胎位|\bJIG\s*SETUP\b", re.IGNORECASE), "JIG_SETUP", "胎架定位", "FABRICATION"),
|
||
(re.compile(r"大组|\bMAJOR\s*ASSEMBLY\b", re.IGNORECASE), "MAJOR_ASSEMBLY", "大组", "FABRICATION"),
|
||
(re.compile(r"单面焊|\bSINGLE\s*-?\s*SIDE\s*WELD", re.IGNORECASE), "SINGLE_SIDE_WELDING", "单面焊", "WELDING"),
|
||
(
|
||
re.compile(
|
||
r"\bMONTAGE\b|\bASSEMBLY\b|\bEINBAU\b|\bINSTALLATION\b|\bMOUNT(?:ING)?\b|"
|
||
r"\bANSCHLUSS\b|\bCONNECTION\b|装配|组装",
|
||
re.IGNORECASE,
|
||
),
|
||
"ASSEMBLY", "装配/连接", "ASSEMBLY",
|
||
),
|
||
(re.compile(r"\bENTGRATEN\b|\bDEBURRING\b|\bDEBURR\b", re.IGNORECASE), "DEBURRING", "去毛刺", "FINISH"),
|
||
(re.compile(r"\bOBERFL[AÄ]CHE\b|\bSURFACE\b", re.IGNORECASE), "SURFACE_TREATMENT", "表面处理", "SURFACE"),
|
||
(
|
||
re.compile(r"\bKORROSION\b|\bCORROSION\b|\bSCHUTZ\b|\bPROTECTION\b", re.IGNORECASE),
|
||
"CORROSION_PROTECTION", "防腐保护", "SURFACE",
|
||
),
|
||
)
|
||
|
||
_DETAIL_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
|
||
("surfaceRoughness", re.compile(r"\bR[aA]\s*\d+(?:\.\d+)?\b", re.IGNORECASE)),
|
||
("tolerance", re.compile(r"±\s*\d+(?:\.\d+)?|\bISO\s*2768\b|\bTOLERANZ\b|\bTOLERANCE\b", re.IGNORECASE)),
|
||
("threadSpecification", re.compile(r"\b[MG]\d+(?:[xX*]\d+(?:\.\d+)?)?(?!\s*[:：]\s*\d)", re.IGNORECASE)),
|
||
("depthMm", re.compile(r"\b(\d+)\s*(?:DEEP|TIEF|T)\b", re.IGNORECASE)),
|
||
(
|
||
"materialSpecification",
|
||
re.compile(
|
||
r"\b(?:STAHL|STEEL|ALU(?:MINIUM)?|KUPFER|COPPER|EDELSTAHL|STAINLESS|MESSING|BRASS)\b",
|
||
re.IGNORECASE,
|
||
),
|
||
),
|
||
)
|
||
|
||
_BLOCK_GROUP_RE = re.compile(
|
||
r"(?P<name1>[\u4e00-\u9fff]{1,16}?)(?:分段)?[（(]?(?P<code1>\d{3,}(?:[、，,、/\s]*\d{3,})*)[）)]?(?:\s*分段)?"
|
||
r"|(?P<code2>\d{3,}(?:[、，,、/\s]*\d{3,})*)[\u4e00-\u9fff]{0,16}分段"
|
||
r"|FR\s*\d+[\u4e00-\u9fff]*组立",
|
||
re.IGNORECASE,
|
||
)
|
||
_ASSEMBLY_CUTS = (
|
||
"完成焊接后翻身装配到",
|
||
"翻身装配到",
|
||
"装配到",
|
||
"为基装配",
|
||
"先以",
|
||
"再装配",
|
||
"依次",
|
||
"然后",
|
||
"完成后",
|
||
"形成",
|
||
"分为",
|
||
"包含",
|
||
"主要",
|
||
"再",
|
||
"先",
|
||
"以",
|
||
"等",
|
||
)
|
||
_LOCATION_RE = re.compile(r"上建场地|联合厂房|平胎位|大组胎架|分片制作场地|片制作场地|分片场地")
|
||
_SEQUENCE_MARKER_RE = re.compile(r"先|再|然后|依次|完成后|完成|以[\u4e00-\u9fff]{1,12}为胎|根据|最后|随后|之后")
|
||
|
||
|
||
def _stable_id(kind: str, digest: str, suffix: str) -> str:
|
||
seed = f"{kind}:{digest}:{suffix}".encode()
|
||
return f"{kind}_{sha256(seed).hexdigest()[:20]}"
|
||
|
||
|
||
def _line_key(line: str) -> str:
|
||
return re.sub(r"\s+", " ", str(line or "").strip()).upper()
|
||
|
||
|
||
def _evidence(evidence: dict[str, Any], source: str, text: str) -> dict[str, Any]:
|
||
payload = {"source": source, "text": text}
|
||
for key, value in evidence.items():
|
||
if key == "source":
|
||
continue
|
||
payload[key] = value
|
||
return payload
|
||
|
||
|
||
def _scan_sources(
|
||
digest: str,
|
||
sources: list[tuple[str, dict[str, Any]]],
|
||
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||
operations: dict[str, dict[str, Any]] = {}
|
||
details: dict[str, dict[str, Any]] = {}
|
||
for raw, evidence in sources:
|
||
line = str(raw or "").strip()
|
||
if not line:
|
||
continue
|
||
for pattern, code, name, category in _PROCESS_TERMS:
|
||
if not pattern.search(line):
|
||
continue
|
||
key = f"{code}:{_line_key(line)}"
|
||
if key in operations:
|
||
continue
|
||
operations[key] = {
|
||
"candidateId": _stable_id("processOperation", digest, key),
|
||
"candidateType": "PROCESS_OPERATION",
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"processCode": code,
|
||
"processName": name,
|
||
"category": category,
|
||
"operationCode": code,
|
||
"operationName": name,
|
||
"standardTime": None,
|
||
"resourceCode": None,
|
||
"confidence": 0.7 if evidence.get("source") == "drawingText" else 0.62,
|
||
"unknownFields": ["standardTime", "resourceCode", "operationDescription"],
|
||
"evidence": _evidence(evidence, str(evidence.get("source") or "drawingText"), line),
|
||
}
|
||
for field, pattern in _DETAIL_PATTERNS:
|
||
match = pattern.search(line)
|
||
if field == "tolerance":
|
||
numeric = re.search(r"±\s*\d+(?:\.\d+)?", line)
|
||
match = numeric or match
|
||
if not match:
|
||
continue
|
||
value = match.group(0)
|
||
key = f"{field}:{_line_key(line)}"
|
||
if key in details:
|
||
continue
|
||
details[key] = {
|
||
"candidateId": _stable_id("processDetail", digest, key),
|
||
"candidateType": "PROCESS_DETAIL",
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"field": field,
|
||
"value": value,
|
||
"confidence": 0.62,
|
||
"unknownFields": ["mappedMasterField"],
|
||
"evidence": _evidence(evidence, str(evidence.get("source") or "drawingText"), line),
|
||
}
|
||
ordered_ops = list(operations.values())
|
||
for index, row in enumerate(ordered_ops):
|
||
row["sequenceHint"] = (index + 1) * 10
|
||
return ordered_ops, list(details.values())
|
||
|
||
|
||
def _scan_structured(
|
||
blob: str,
|
||
digest: str,
|
||
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]:
|
||
blocks: dict[str, dict[str, Any]] = {}
|
||
for match in _BLOCK_GROUP_RE.finditer(blob):
|
||
raw = match.group(0)
|
||
code = (match.group("code1") or match.group("code2") or "").strip()
|
||
if not code and raw.upper().startswith("FR"):
|
||
code = re.match(r"FR\s*\d+", raw, re.IGNORECASE).group(0)
|
||
if not code:
|
||
continue
|
||
name = (match.group("name1") or "").strip()
|
||
if not name and match.group("code2"):
|
||
name = re.sub(r"^[0-9、，,、/\s]+", "", raw).replace("分段", "").strip()
|
||
if not name and raw.upper().startswith("FR"):
|
||
name = re.sub(r"^FR\s*\d+", "", raw).replace("组立", "").strip()
|
||
key = _line_key(code)
|
||
if key in blocks:
|
||
continue
|
||
blocks[key] = {
|
||
"candidateId": _stable_id("processBlock", digest, key),
|
||
"candidateType": "BLOCK_GROUP",
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"blockCode": re.sub(r"\s+", "", code),
|
||
"blockName": name or None,
|
||
"confidence": 0.66,
|
||
"unknownFields": ["blockDrawing", "weight", "quantity"],
|
||
"evidence": {"source": "processText", "text": match.group(0)},
|
||
}
|
||
|
||
assemblies: dict[str, dict[str, Any]] = {}
|
||
for match in re.finditer(r"组立", blob):
|
||
start = max(0, match.start() - 20)
|
||
raw = blob[start:match.end()]
|
||
segments = re.split(r"[、，。；：/\s]+", raw)
|
||
name = segments[-1].strip() if segments else raw
|
||
name = re.sub(r"^\d+(?:\.\d+)?", "", name)
|
||
for separator in _ASSEMBLY_CUTS:
|
||
index = name.find(separator)
|
||
if index >= 0:
|
||
name = name[index + len(separator):]
|
||
break
|
||
name = re.split(r"[，,和):：]+", name)[-1].strip()
|
||
name = re.sub(r"(?:等|以及|、|。|；|：)$", "", name).strip("（()）")[:16]
|
||
if name == "组立" or not name.endswith("组立") or not re.search(r"[\u4e00-\u9fff]{2}", name):
|
||
continue
|
||
key = _line_key(name)
|
||
if key in assemblies:
|
||
continue
|
||
assemblies[key] = {
|
||
"candidateId": _stable_id("processAssembly", digest, key),
|
||
"candidateType": "ASSEMBLY_GROUP",
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"assemblyName": name,
|
||
"confidence": 0.64,
|
||
"unknownFields": ["memberParts", "sequence", "drawing"],
|
||
"evidence": {"source": "processText", "text": name},
|
||
}
|
||
|
||
locations: dict[str, dict[str, Any]] = {}
|
||
for match in _LOCATION_RE.finditer(blob):
|
||
name = match.group(0)
|
||
key = _line_key(name)
|
||
if key in locations:
|
||
continue
|
||
locations[key] = {
|
||
"candidateId": _stable_id("processLocation", digest, key),
|
||
"candidateType": "PROCESS_LOCATION",
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"location": name,
|
||
"confidence": 0.7,
|
||
"evidence": {"source": "processText", "text": name},
|
||
}
|
||
|
||
sequence_steps: list[dict[str, Any]] = []
|
||
for index, sentence in enumerate(re.split(r"[。；\n]+", blob), start=1):
|
||
sentence = sentence.strip()
|
||
if not sentence:
|
||
continue
|
||
markers = sorted({m for m in _SEQUENCE_MARKER_RE.findall(sentence) if m})
|
||
if not markers:
|
||
continue
|
||
sequence_steps.append({
|
||
"candidateId": _stable_id("processSequence", digest, f"{index}:{_line_key(sentence)}"),
|
||
"candidateType": "PROCESS_SEQUENCE_STEP",
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
"stepIndex": index,
|
||
"sentence": sentence,
|
||
"markers": markers,
|
||
"confidence": 0.6,
|
||
"unknownFields": ["parentBlock", "workstation", "duration"],
|
||
"evidence": {"source": "processText", "text": sentence},
|
||
})
|
||
return list(blocks.values()), list(assemblies.values()), list(locations.values()), sequence_steps
|
||
|
||
|
||
def recognize_process_text(text: str, *, source: str = "processDocument") -> dict[str, Any]:
|
||
"""Parse free-form process/construction text into structured reviewable candidates."""
|
||
digest = sha256(str(text or "").encode("utf-8")).hexdigest()
|
||
sentences = [part.strip() for part in re.split(r"[。；\n]+", str(text or "")) if part.strip()]
|
||
sources = [(sentence, {"source": source}) for sentence in sentences]
|
||
operations, details = _scan_sources(digest, sources)
|
||
blocks, assemblies, locations, sequence_steps = _scan_structured(str(text or ""), digest)
|
||
return {
|
||
"contractVersion": PROCESS_VERSION,
|
||
"operations": operations,
|
||
"details": details,
|
||
"blocks": blocks,
|
||
"assemblies": assemblies,
|
||
"locations": locations,
|
||
"sequenceSteps": sequence_steps,
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
}
|
||
|
||
|
||
def recognize_process_candidates(
|
||
parsed: dict[str, Any],
|
||
part_list_rows: list[dict[str, Any]] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Scan drawing text and part-list names for reviewable process/operation/detail candidates."""
|
||
asset = parsed.get("asset") or {}
|
||
digest = str(asset.get("sha256") or "")
|
||
sources: list[tuple[str, dict[str, Any]]] = []
|
||
text_parts: list[str] = []
|
||
for row in parsed.get("texts") or []:
|
||
evidence = dict(row.get("evidence") or {})
|
||
raw = str(row.get("text") or "")
|
||
sources.append((raw, {"source": "drawingText", **evidence}))
|
||
text_parts.append(raw)
|
||
for row in part_list_rows or []:
|
||
name = str(row.get("componentName") or "").strip()
|
||
if name:
|
||
evidence = dict(row.get("evidence") or {})
|
||
sources.append((name, {"source": "partListName", **evidence}))
|
||
text_parts.append(name)
|
||
operations, details = _scan_sources(digest, sources)
|
||
blocks, assemblies, locations, sequence_steps = _scan_structured("\n".join(text_parts), digest)
|
||
return {
|
||
"contractVersion": PROCESS_VERSION,
|
||
"operations": operations,
|
||
"details": details,
|
||
"blocks": blocks,
|
||
"assemblies": assemblies,
|
||
"locations": locations,
|
||
"sequenceSteps": sequence_steps,
|
||
"status": "PENDING_REVIEW",
|
||
"reviewRequired": True,
|
||
}
|
||
|
||
|
||
__all__ = ["PROCESS_VERSION", "recognize_process_candidates", "recognize_process_text"]
|