# ============================================================ # 图纸/工艺文档关键信息识别（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[\u4e00-\u9fff]{1,16}?)(?:分段)?[（(]?(?P\d{3,}(?:[、，,、/\s]*\d{3,})*)[）)]?(?:\s*分段)?" r"|(?P\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"]