Files
TK-BOM-Project/backend/app/services/pipe_classifier.py
hyungi 22baea38e1
Some checks failed
SonarQube Analysis / SonarQube Scan (push) Has been cancelled
feat: BOM 페이지 개선 및 구매관리 기능 향상
- BOM 페이지에서 모든 카테고리 이모지 제거 (깔끔한 UI)
- 구매신청된 자재 비활성화 기능 개선
- 구매관리 페이지에서 선택된 프로젝트만 표시하도록 수정
- 파이프 카테고리 Excel 내보내기 개선:
  * 끝단처리, 압력등급 컬럼 제거
  * 사용자요구(분류기 추출) 및 추가요청사항 컬럼 추가
  * 품목명에서 끝단처리 정보 제거
  * 납기일 P열 고정 및 관리항목 자동 채움
- 파이프 분류기에서 사용자 요구사항 추출 기능 추가
- 재질별 스케줄 표시 개선 (SUS: SCH 40S, Carbon: SCH 40)
2025-10-16 13:27:14 +09:00

574 lines
20 KiB
Python

"""
PIPE 분류 전용 모듈
재질 분류 + 파이프 특화 분류
"""
import re
from typing import Dict, List, Optional
from .material_classifier import classify_material, get_manufacturing_method_from_material
# ========== PIPE USER 요구사항 키워드 ==========
PIPE_USER_REQUIREMENTS = {
"IMPACT_TEST": {
"keywords": ["IMPACT", "충격시험", "CHARPY", "CVN", "IMPACT TEST", "충격", "NOTCH"],
"description": "충격시험 요구",
"confidence": 0.95
},
"ASME_CODE": {
"keywords": ["ASME", "ASME CODE", "CODE", "B31.1", "B31.3", "B31.4", "B31.8", "VIII"],
"description": "ASME 코드 준수",
"confidence": 0.95
},
"STRESS_RELIEF": {
"keywords": ["STRESS RELIEF", "SR", "응력제거", "열처리", "HEAT TREATMENT"],
"description": "응력제거 열처리",
"confidence": 0.90
},
"RADIOGRAPHIC_TEST": {
"keywords": ["RT", "RADIOGRAPHIC", "방사선시험", "X-RAY", "엑스레이"],
"description": "방사선 시험",
"confidence": 0.90
},
"ULTRASONIC_TEST": {
"keywords": ["UT", "ULTRASONIC", "초음파시험", "초음파"],
"description": "초음파 시험",
"confidence": 0.90
},
"MAGNETIC_PARTICLE": {
"keywords": ["MT", "MAGNETIC PARTICLE", "자분탐상", "자분"],
"description": "자분탐상 시험",
"confidence": 0.90
},
"LIQUID_PENETRANT": {
"keywords": ["PT", "LIQUID PENETRANT", "침투탐상", "침투"],
"description": "침투탐상 시험",
"confidence": 0.90
},
"HYDROSTATIC_TEST": {
"keywords": ["HYDROSTATIC", "수압시험", "PRESSURE TEST", "압력시험"],
"description": "수압 시험",
"confidence": 0.90
},
"LOW_TEMPERATURE": {
"keywords": ["LOW TEMP", "저온", "LTCS", "LOW TEMPERATURE", "CRYOGENIC"],
"description": "저온용",
"confidence": 0.85
},
"HIGH_TEMPERATURE": {
"keywords": ["HIGH TEMP", "고온", "HTCS", "HIGH TEMPERATURE"],
"description": "고온용",
"confidence": 0.85
}
}
# ========== PIPE 제조 방법별 분류 ==========
PIPE_MANUFACTURING = {
"SEAMLESS": {
"keywords": ["SEAMLESS", "SMLS", "심리스", "무계목"],
"confidence": 0.95,
"characteristics": "이음새 없는 고품질 파이프"
},
"WELDED": {
"keywords": ["WELDED", "WLD", "ERW", "SAW", "용접", "전기저항용접"],
"confidence": 0.95,
"characteristics": "용접으로 제조된 파이프"
},
"CAST": {
"keywords": ["CAST", "주조"],
"confidence": 0.85,
"characteristics": "주조로 제조된 파이프"
}
}
# ========== PIPE 끝 가공별 분류 ==========
PIPE_END_PREP = {
"BOTH_ENDS_BEVELED": {
"codes": ["BBE", "BOE", "BOTH END", "BOTH BEVELED", "양쪽개선"],
"cutting_note": "양쪽 개선",
"machining_required": True,
"confidence": 0.95
},
"ONE_END_BEVELED": {
"codes": ["BE", "BEV", "PBE", "PIPE BEVELED END", "POE"],
"cutting_note": "한쪽 개선",
"machining_required": True,
"confidence": 0.95
},
"NO_BEVEL": {
"codes": ["PE", "PLAIN END", "PPE", "평단", "무개선"],
"cutting_note": "무 개선",
"machining_required": False,
"confidence": 0.95
},
"THREADED": {
"codes": ["TOE", "THE", "THREADED", "나사", "스레드"],
"cutting_note": "나사 가공",
"machining_required": True,
"confidence": 0.90
}
}
# ========== 구매용 파이프 분류 (끝단 가공 제외) ==========
def get_purchase_pipe_description(description: str) -> str:
"""구매용 파이프 설명 - 끝단 가공 정보 제거"""
# 모든 끝단 가공 코드들을 수집
end_prep_codes = []
for prep_data in PIPE_END_PREP.values():
end_prep_codes.extend(prep_data["codes"])
# 설명에서 끝단 가공 코드 제거
clean_description = description.upper()
# 끝단 가공 코드들을 길이 순으로 정렬 (긴 것부터 처리)
end_prep_codes.sort(key=len, reverse=True)
for code in end_prep_codes:
# 단어 경계를 고려하여 제거 (부분 매칭 방지)
pattern = r'\b' + re.escape(code) + r'\b'
clean_description = re.sub(pattern, '', clean_description, flags=re.IGNORECASE)
# 끝단 가공 관련 패턴들 추가 제거
# BOE-POE, POE-TOE 같은 조합 패턴들
end_prep_patterns = [
r'\b[A-Z]{2,3}E-[A-Z]{2,3}E\b', # BOE-POE, POE-TOE 등
r'\b[A-Z]{2,3}E-[A-Z]{2,3}\b', # BOE-TO, POE-TO 등
r'\b[A-Z]{2,3}-[A-Z]{2,3}E\b', # BO-POE, PO-TOE 등
r'\b[A-Z]{2,3}-[A-Z]{2,3}\b', # BO-PO, PO-TO 등
]
for pattern in end_prep_patterns:
clean_description = re.sub(pattern, '', clean_description, flags=re.IGNORECASE)
# 남은 하이픈과 공백 정리
clean_description = re.sub(r'\s*-\s*', ' ', clean_description) # 하이픈 제거
clean_description = re.sub(r'\s+', ' ', clean_description).strip() # 연속 공백 정리
return clean_description
def extract_end_preparation_info(description: str) -> Dict:
"""파이프 설명에서 끝단 가공 정보 추출"""
desc_upper = description.upper()
# 끝단 가공 코드 찾기
for prep_type, prep_data in PIPE_END_PREP.items():
for code in prep_data["codes"]:
if code in desc_upper:
return {
"end_preparation_type": prep_type,
"end_preparation_code": code,
"machining_required": prep_data["machining_required"],
"cutting_note": prep_data["cutting_note"],
"confidence": prep_data["confidence"],
"matched_pattern": code,
"original_description": description,
"clean_description": get_purchase_pipe_description(description)
}
# 기본값: PBE (양쪽 무개선)
return {
"end_preparation_type": "NO_BEVEL", # PBE로 매핑될 예정
"end_preparation_code": "PBE",
"machining_required": False,
"cutting_note": "양쪽 무개선 (기본값)",
"confidence": 0.5,
"matched_pattern": "DEFAULT",
"original_description": description,
"clean_description": get_purchase_pipe_description(description)
}
# ========== PIPE 스케줄별 분류 ==========
PIPE_SCHEDULE = {
"patterns": [
r"SCH\s*(\d+)",
r"SCHEDULE\s*(\d+)",
r"스케줄\s*(\d+)"
],
"common_schedules": ["10", "20", "40", "80", "120", "160"],
"wall_thickness_patterns": [
r"(\d+(?:\.\d+)?)\s*mm\s*THK",
r"(\d+(?:\.\d+)?)\s*THK"
]
}
def extract_pipe_user_requirements(description: str) -> List[str]:
"""
파이프 설명에서 User 요구사항 추출
Args:
description: 파이프 설명
Returns:
발견된 요구사항 리스트
"""
desc_upper = description.upper()
found_requirements = []
for req_type, req_data in PIPE_USER_REQUIREMENTS.items():
for keyword in req_data["keywords"]:
if keyword in desc_upper:
found_requirements.append(req_data["description"])
break # 같은 타입에서 중복 방지
return found_requirements
def classify_pipe_for_purchase(dat_file: str, description: str, main_nom: str,
length: Optional[float] = None) -> Dict:
"""구매용 파이프 분류 - 끝단 가공 정보 제외"""
# 끝단 가공 정보 제거한 설명으로 분류
clean_description = get_purchase_pipe_description(description)
# 기본 파이프 분류 수행
result = classify_pipe(dat_file, clean_description, main_nom, length)
# 구매용임을 표시
result["purchase_classification"] = True
result["original_description"] = description
result["clean_description"] = clean_description
return result
def classify_pipe(dat_file: str, description: str, main_nom: str,
length: Optional[float] = None) -> Dict:
"""
완전한 PIPE 분류
Args:
dat_file: DAT_FILE 필드
description: DESCRIPTION 필드
main_nom: MAIN_NOM 필드 (사이즈)
length: LENGTH 필드 (절단 치수)
Returns:
완전한 파이프 분류 결과
"""
desc_upper = description.upper()
# 1. 명칭 우선 확인 (다른 자재 타입 키워드가 있으면 파이프가 아님)
other_material_keywords = [
'FLG', 'FLANGE', '플랜지', # 플랜지
'ELBOW', 'ELL', 'TEE', 'REDUCER', 'CAP', 'COUPLING', '엘보', '', '리듀서', # 피팅
'VALVE', 'BALL', 'GATE', 'GLOBE', 'CHECK', '밸브', # 밸브
'BOLT', 'STUD', 'NUT', 'SCREW', '볼트', '너트', # 볼트
'GASKET', 'GASK', '가스켓', # 가스켓
'GAUGE', 'SENSOR', 'TRANSMITTER', 'INSTRUMENT', '계기' # 계기
]
for keyword in other_material_keywords:
if keyword in desc_upper:
return {
"category": "UNKNOWN",
"overall_confidence": 0.0,
"reason": f"다른 자재 키워드 발견: {keyword}"
}
# 2. 파이프 키워드 확인
pipe_keywords = ['PIPE', 'TUBE', '파이프', '배관', 'SMLS', 'SEAMLESS']
has_pipe_keyword = any(keyword in desc_upper for keyword in pipe_keywords)
# 파이프 재질 확인 (A106, A333, A312, A53)
pipe_materials = ['A106', 'A333', 'A312', 'A53']
has_pipe_material = any(material in desc_upper for material in pipe_materials)
# 파이프 키워드도 없고 파이프 재질도 없으면 UNKNOWN
if not has_pipe_keyword and not has_pipe_material:
return {
"category": "UNKNOWN",
"overall_confidence": 0.0,
"reason": "파이프 키워드 및 재질 없음"
}
# 3. 재질 분류 (공통 모듈 사용)
material_result = classify_material(description)
# 2. 제조 방법 분류
manufacturing_result = classify_pipe_manufacturing(description, material_result)
# 3. 끝 가공 분류
end_prep_result = classify_pipe_end_preparation(description)
# 4. 스케줄 분류 (재질 정보 전달)
schedule_result = classify_pipe_schedule(description, material_result)
# 5. 길이(절단 치수) 처리
length_info = extract_pipe_length_info(length, description)
# 6. User 요구사항 추출
user_requirements = extract_pipe_user_requirements(description)
# 7. 최종 결과 조합
return {
"category": "PIPE",
# 재질 정보 (공통 모듈)
"material": {
"standard": material_result.get('standard', 'UNKNOWN'),
"grade": material_result.get('grade', 'UNKNOWN'),
"material_type": material_result.get('material_type', 'UNKNOWN'),
"confidence": material_result.get('confidence', 0.0)
},
# 파이프 특화 정보
"manufacturing": {
"method": manufacturing_result.get('method', 'UNKNOWN'),
"confidence": manufacturing_result.get('confidence', 0.0),
"evidence": manufacturing_result.get('evidence', [])
},
"end_preparation": {
"type": end_prep_result.get('type', 'UNKNOWN'),
"cutting_note": end_prep_result.get('cutting_note', ''),
"machining_required": end_prep_result.get('machining_required', False),
"confidence": end_prep_result.get('confidence', 0.0)
},
"schedule": {
"schedule": schedule_result.get('schedule', 'UNKNOWN'),
"wall_thickness": schedule_result.get('wall_thickness', ''),
"confidence": schedule_result.get('confidence', 0.0)
},
"length_info": length_info,
"size_info": {
"nominal_size": main_nom,
"length_mm": length_info.get('length_mm')
},
# User 요구사항
"user_requirements": user_requirements,
# 전체 신뢰도
"overall_confidence": calculate_pipe_confidence({
"material": material_result.get('confidence', 0),
"manufacturing": manufacturing_result.get('confidence', 0),
"end_prep": end_prep_result.get('confidence', 0),
"schedule": schedule_result.get('confidence', 0)
})
}
def classify_pipe_manufacturing(description: str, material_result: Dict) -> Dict:
"""파이프 제조 방법 분류"""
desc_upper = description.upper()
# 1. DESCRIPTION 키워드 우선 확인
for method, method_data in PIPE_MANUFACTURING.items():
for keyword in method_data["keywords"]:
if keyword in desc_upper:
return {
"method": method,
"confidence": method_data["confidence"],
"evidence": [f"KEYWORD: {keyword}"],
"characteristics": method_data["characteristics"]
}
# 2. 재질 규격으로 추정
material_manufacturing = get_manufacturing_method_from_material(material_result)
if material_manufacturing in ["SEAMLESS", "WELDED"]:
return {
"method": material_manufacturing,
"confidence": 0.8,
"evidence": [f"MATERIAL_STANDARD: {material_result.get('standard')}"],
"characteristics": PIPE_MANUFACTURING.get(material_manufacturing, {}).get("characteristics", "")
}
# 3. 기본값
return {
"method": "UNKNOWN",
"confidence": 0.0,
"evidence": ["NO_MANUFACTURING_INFO"]
}
def classify_pipe_end_preparation(description: str) -> Dict:
"""파이프 끝 가공 분류"""
desc_upper = description.upper()
# 우선순위: 양쪽 > 한쪽 > 무개선
for prep_type, prep_data in PIPE_END_PREP.items():
for code in prep_data["codes"]:
if code in desc_upper:
return {
"type": prep_type,
"cutting_note": prep_data["cutting_note"],
"machining_required": prep_data["machining_required"],
"confidence": prep_data["confidence"],
"matched_code": code
}
# 기본값: 무개선
return {
"type": "NO_BEVEL",
"cutting_note": "무 개선 (기본값)",
"machining_required": False,
"confidence": 0.5,
"matched_code": "DEFAULT"
}
def classify_pipe_schedule(description: str, material_result: Dict = None) -> Dict:
"""파이프 스케줄 분류 - 재질별 표현 개선"""
desc_upper = description.upper()
# 재질 정보 확인
material_type = "CARBON" # 기본값
if material_result:
material_grade = material_result.get('grade', '').upper()
material_standard = material_result.get('standard', '').upper()
# 스테인리스 스틸 판단
if any(sus_indicator in material_grade or sus_indicator in material_standard
for sus_indicator in ['SUS', 'SS', 'A312', 'A358', 'A376', '304', '316', '321', '347']):
material_type = "STAINLESS"
# 1. 스케줄 패턴 확인
for pattern in PIPE_SCHEDULE["patterns"]:
match = re.search(pattern, desc_upper)
if match:
schedule_num = match.group(1)
# 재질별 스케줄 표현
if material_type == "STAINLESS":
# 스테인리스 스틸: SCH 40S, SCH 80S
if schedule_num in ["10", "20", "40", "80", "120", "160"]:
schedule_display = f"SCH {schedule_num}S"
else:
schedule_display = f"SCH {schedule_num}"
else:
# 카본 스틸: SCH 40, SCH 80
schedule_display = f"SCH {schedule_num}"
return {
"schedule": schedule_display,
"schedule_number": schedule_num,
"material_type": material_type,
"confidence": 0.95,
"matched_pattern": pattern
}
# 2. 두께 패턴 확인
for pattern in PIPE_SCHEDULE["wall_thickness_patterns"]:
match = re.search(pattern, desc_upper)
if match:
thickness = match.group(1)
return {
"schedule": f"{thickness}mm THK",
"wall_thickness": f"{thickness}mm",
"material_type": material_type,
"confidence": 0.9,
"matched_pattern": pattern
}
# 3. 기본값
return {
"schedule": "UNKNOWN",
"material_type": material_type,
"confidence": 0.0
}
def extract_pipe_length_info(length: Optional[float], description: str) -> Dict:
"""파이프 길이(절단 치수) 정보 추출"""
length_info = {
"length_mm": None,
"source": None,
"confidence": 0.0,
"note": ""
}
# 1. LENGTH 필드에서 추출 (우선)
if length and length > 0:
length_info.update({
"length_mm": round(length, 1),
"source": "LENGTH_FIELD",
"confidence": 0.95,
"note": f"도면 명기 길이: {length}mm"
})
# 2. DESCRIPTION에서 백업 추출
else:
desc_length = extract_length_from_description(description)
if desc_length:
length_info.update({
"length_mm": desc_length,
"source": "DESCRIPTION_PARSED",
"confidence": 0.8,
"note": f"설명란에서 추출: {desc_length}mm"
})
else:
length_info.update({
"source": "NO_LENGTH_INFO",
"confidence": 0.0,
"note": "길이 정보 없음 - 도면 확인 필요"
})
return length_info
def extract_length_from_description(description: str) -> Optional[float]:
"""DESCRIPTION에서 길이 정보 추출"""
length_patterns = [
r'(\d+(?:\.\d+)?)\s*mm',
r'(\d+(?:\.\d+)?)\s*M',
r'(\d+(?:\.\d+)?)\s*LG',
r'L\s*=\s*(\d+(?:\.\d+)?)',
r'길이\s*(\d+(?:\.\d+)?)'
]
for pattern in length_patterns:
match = re.search(pattern, description.upper())
if match:
return float(match.group(1))
return None
def calculate_pipe_confidence(confidence_scores: Dict) -> float:
"""파이프 분류 전체 신뢰도 계산"""
scores = [score for score in confidence_scores.values() if score > 0]
if not scores:
return 0.0
# 가중 평균 (재질이 가장 중요)
weights = {
"material": 0.4,
"manufacturing": 0.25,
"end_prep": 0.2,
"schedule": 0.15
}
weighted_sum = sum(
confidence_scores.get(key, 0) * weight
for key, weight in weights.items()
)
return round(weighted_sum, 2)
def generate_pipe_cutting_plan(pipe_data: Dict) -> Dict:
"""파이프 절단 계획 생성"""
cutting_plan = {
"material_spec": f"{pipe_data['material']['grade']} {pipe_data['schedule']['schedule']}",
"length_mm": pipe_data['length_info']['length_mm'],
"end_preparation": pipe_data['end_preparation']['cutting_note'],
"machining_required": pipe_data['end_preparation']['machining_required']
}
# 절단 지시서 생성
if cutting_plan["length_mm"]:
cutting_plan["cutting_instruction"] = f"""
재질: {cutting_plan['material_spec']}
절단길이: {cutting_plan['length_mm']}mm
끝가공: {cutting_plan['end_preparation']}
가공여부: {'베벨가공 필요' if cutting_plan['machining_required'] else '직각절단만'}
""".strip()
else:
cutting_plan["cutting_instruction"] = "도면 확인 후 길이 정보 입력 필요"
return cutting_plan