Files
TK-BOM-Project/backend/app/services/pipe_classifier.py
Hyungi Ahn 9e5250a8f9 자재 분류 시스템 개선 및 통합 분류기 구현
- 통합 분류기 구현으로 키워드 우선순위 체계 적용
- HEX.PLUG → FITTING 분류 수정 (기존 VALVE 오분류 해결)
- 플랜지/밸브가 볼트로 오분류되는 문제 해결 (A193, A194 재질 키워드 우선순위 적용)
- 피팅 재질(A234, A403, A420) 기반 분류 추가
- 니플 길이 정보 보존 로직 개선
- 파이프 끝단 가공 정보를 구매 단계에서 제외
- PostgreSQL 사용으로 RULES.md 업데이트
- 상호 배타적 키워드 시스템 구현 (Level 1 키워드 우선)
2025-07-23 14:38:49 +09:00

374 lines
12 KiB
Python

"""
PIPE 분류 전용 모듈
재질 분류 + 파이프 특화 분류
"""
import re
from typing import Dict, List, Optional
from .material_classifier import classify_material, get_manufacturing_method_from_material
# ========== 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": ["BOE", "BOTH END", "BOTH BEVELED", "양쪽개선"],
"cutting_note": "양쪽 개선",
"machining_required": True,
"confidence": 0.95
},
"ONE_END_BEVELED": {
"codes": ["BE", "BEV", "PBE", "PIPE BEVELED END"],
"cutting_note": "한쪽 개선",
"machining_required": True,
"confidence": 0.95
},
"NO_BEVEL": {
"codes": ["PE", "PLAIN END", "PPE", "평단", "무개선"],
"cutting_note": "무 개선",
"machining_required": False,
"confidence": 0.95
}
}
# ========== 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 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)
# 5. 길이(절단 치수) 처리
length_info = extract_pipe_length_info(length, description)
# 6. 최종 결과 조합
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')
},
# 전체 신뢰도
"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) -> Dict:
"""파이프 스케줄 분류"""
desc_upper = description.upper()
# 1. 스케줄 패턴 확인
for pattern in PIPE_SCHEDULE["patterns"]:
match = re.search(pattern, desc_upper)
if match:
schedule_num = match.group(1)
return {
"schedule": f"SCH {schedule_num}",
"schedule_number": schedule_num,
"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",
"confidence": 0.9,
"matched_pattern": pattern
}
# 3. 기본값
return {
"schedule": "UNKNOWN",
"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