From afea8428b21d4089ae5db482d53a18ca060b52d7 Mon Sep 17 00:00:00 2001 From: Hyungi Ahn Date: Thu, 8 Jan 2026 11:14:25 +0900 Subject: [PATCH] =?UTF-8?q?=EC=97=91=EC=85=80=20=ED=8C=8C=EC=8B=B1=20?= =?UTF-8?q?=EC=9D=B4=EC=9B=90=ED=99=94(=ED=91=9C=EC=A4=80/=EC=9D=B8?= =?UTF-8?q?=EB=B2=A4=ED=84=B0)=20=EB=B0=8F=20=EC=9E=90=EC=9E=AC=20?= =?UTF-8?q?=EB=B6=84=EB=A5=98=EA=B8=B0(Plate,=20H-Beam,=20Swagelok)=20?= =?UTF-8?q?=EA=B0=9C=EC=84=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/routers/files.py | 1245 +---------------- backend/app/services/excel_parser.py | 300 ++++ backend/app/services/fitting_classifier.py | 85 +- backend/app/services/integrated_classifier.py | 43 +- backend/app/services/material_service.py | 592 ++++++++ backend/app/services/plate_classifier.py | 50 + backend/app/services/structural_classifier.py | 34 + 7 files changed, 1059 insertions(+), 1290 deletions(-) create mode 100644 backend/app/services/excel_parser.py create mode 100644 backend/app/services/material_service.py create mode 100644 backend/app/services/plate_classifier.py create mode 100644 backend/app/services/structural_classifier.py diff --git a/backend/app/routers/files.py b/backend/app/routers/files.py index 1f04a55..a703435 100644 --- a/backend/app/routers/files.py +++ b/backend/app/routers/files.py @@ -16,19 +16,13 @@ from ..database import get_db from ..auth.middleware import get_current_user from ..services.activity_logger import ActivityLogger, log_activity_from_request from ..utils.logger import get_logger -from app.services.material_classifier import classify_material +from app.services.excel_parser import BOMParser +from app.services.material_service import MaterialService # [신규] 자재 서비스 임포트 # 로거 설정 logger = get_logger(__name__) -from app.services.integrated_classifier import classify_material_integrated, should_exclude_material -from app.services.bolt_classifier import classify_bolt -from app.services.flange_classifier import classify_flange -from app.services.fitting_classifier import classify_fitting -from app.services.gasket_classifier import classify_gasket -from app.services.instrument_classifier import classify_instrument -from app.services.pipe_classifier import classify_pipe -from app.services.valve_classifier import classify_valve -from app.services.revision_comparator import get_revision_comparison +# [REMOVED] classify_material_integrated 등 직접 임포트 제거 (서비스에서 사용) +from app.services.revision_comparator import get_revision_comparison, perform_simple_revision_comparison router = APIRouter() @@ -190,142 +184,8 @@ async def add_missing_columns(db: Session = Depends(get_db)): db.rollback() return {"success": False, "error": f"컬럼 추가 실패: {str(e)}"} -def validate_file_extension(filename: str) -> bool: - return Path(filename).suffix.lower() in ALLOWED_EXTENSIONS - -def generate_unique_filename(original_filename: str) -> str: - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - unique_id = str(uuid.uuid4())[:8] - stem = Path(original_filename).stem - suffix = Path(original_filename).suffix - return f"{stem}_{timestamp}_{unique_id}{suffix}" - -def parse_dataframe(df): - df = df.dropna(how='all') - # 원본 컬럼명 출력 - # 로그 제거 - df.columns = df.columns.str.strip().str.lower() - # 로그 제거 - - column_mapping = { - 'description': ['description', 'item', 'material', '품명', '자재명'], - 'quantity': ['qty', 'quantity', 'ea', '수량'], - 'main_size': ['main_nom', 'nominal_diameter', 'nd', '주배관'], - 'red_size': ['red_nom', 'reduced_diameter', '축소배관'], - 'length': ['length', 'len', '길이'], - 'weight': ['weight', 'wt', '중량'], - 'dwg_name': ['dwg_name', 'drawing', '도면명'], - 'line_num': ['line_num', 'line_number', '라인번호'] - } - - mapped_columns = {} - for standard_col, possible_names in column_mapping.items(): - for possible_name in possible_names: - if possible_name in df.columns: - mapped_columns[standard_col] = possible_name - break - - print(f"📋 엑셀 컬럼 매핑 결과: {mapped_columns}") - print(f"📋 원본 컬럼명들: {list(df.columns)}") - - materials = [] - for index, row in df.iterrows(): - description = str(row.get(mapped_columns.get('description', ''), '')) - - # WELD GAP 항목은 업로드 단계에서 제외 (불필요한 계산용 항목) - description_upper = description.upper() - if ('WELD GAP' in description_upper or 'WELDING GAP' in description_upper or - '웰드갭' in description_upper or '용접갭' in description_upper): - print(f"⚠️ WELD GAP 항목 제외: {description}") - continue - - quantity_raw = row.get(mapped_columns.get('quantity', ''), 0) - - try: - quantity = float(quantity_raw) if pd.notna(quantity_raw) else 0 - except: - quantity = 0 - - material_grade = "" - if "ASTM" in description.upper(): - # ASTM 표준과 등급만 추출, end_preparation(BOE, POE, BBE 등)은 제외 - # A\d{3,4} 패턴으로 3-4자리 숫자 보장, 등급도 포함 - astm_match = re.search(r'ASTM\s+(A\d{3,4}[A-Z]*(?:\s+(?:GR\s+)?[A-Z0-9]+)?)', description.upper()) - if astm_match: - material_grade = astm_match.group(0).strip() - - main_size = str(row.get(mapped_columns.get('main_size', ''), '')) - red_size = str(row.get(mapped_columns.get('red_size', ''), '')) - - # main_nom과 red_nom 별도 저장 (원본 값 유지) - main_nom = main_size if main_size != 'nan' and main_size != '' else None - red_nom = red_size if red_size != 'nan' and red_size != '' else None - - # 기존 size_spec도 유지 (호환성을 위해) - if main_size != 'nan' and red_size != 'nan' and red_size != '': - size_spec = f"{main_size} x {red_size}" - elif main_size != 'nan' and main_size != '': - size_spec = main_size - else: - size_spec = "" - - # LENGTH 정보 추출 - length_raw = row.get(mapped_columns.get('length', ''), '') - length_value = None - if pd.notna(length_raw) and str(length_raw).strip() != '': - try: - length_value = float(str(length_raw).strip()) - except (ValueError, TypeError): - length_value = None - - # DWG_NAME 정보 추출 - dwg_name_raw = row.get(mapped_columns.get('dwg_name', ''), '') - dwg_name = None - if pd.notna(dwg_name_raw) and str(dwg_name_raw).strip() not in ['', 'nan', 'None']: - dwg_name = str(dwg_name_raw).strip() - if index < 3: # 처음 3개만 로그 - print(f"📐 도면번호 파싱: {dwg_name}") - - # LINE_NUM 정보 추출 - line_num_raw = row.get(mapped_columns.get('line_num', ''), '') - line_num = None - if pd.notna(line_num_raw) and str(line_num_raw).strip() not in ['', 'nan', 'None']: - line_num = str(line_num_raw).strip() - if index < 3: # 처음 3개만 로그 - print(f"📍 라인번호 파싱: {line_num}") - - if description and description not in ['nan', 'None', '']: - materials.append({ - 'original_description': description, - 'quantity': quantity, - 'unit': "EA", - 'size_spec': size_spec, - 'main_nom': main_nom, - 'red_nom': red_nom, - 'material_grade': material_grade, - 'length': length_value, - 'dwg_name': dwg_name, - 'line_num': line_num, - 'line_number': index + 1, - 'row_number': index + 1 - }) - - return materials - -def parse_file_data(file_path): - file_extension = Path(file_path).suffix.lower() - - try: - if file_extension == ".csv": - df = pd.read_csv(file_path, encoding='utf-8') - elif file_extension in [".xlsx", ".xls"]: - df = pd.read_excel(file_path, sheet_name=0) - else: - raise HTTPException(status_code=400, detail="지원하지 않는 파일 형식") - - return parse_dataframe(df) - except Exception as e: - raise HTTPException(status_code=400, detail=f"파일 파싱 실패: {str(e)}") +# [REMOVED] 구형 파싱 함수들 (validate_file_extension, generate_unique_filename, parse_dataframe, parse_file_data) +# 이 기능들은 이제 app/services/excel_parser.py의 BOMParser 클래스로 이관되었습니다. @router.post("/upload") async def upload_file( @@ -362,32 +222,33 @@ async def upload_file( except Exception as e: print(f"⚠️ 트랜잭션 초기화 중 오류: {e}") # 오류 발생 시에도 계속 진행 - # 로그 제거 - if not validate_file_extension(file.filename): + + # [변경] BOMParser 사용하여 확장자 검증 + if not BOMParser.validate_extension(file.filename): raise HTTPException( status_code=400, - detail=f"지원하지 않는 파일 형식입니다. 허용된 확장자: {', '.join(ALLOWED_EXTENSIONS)}" + detail=f"지원하지 않는 파일 형식입니다. 허용된 확장자: xlsx, xls, csv" ) if file.size and file.size > 10 * 1024 * 1024: raise HTTPException(status_code=400, detail="파일 크기는 10MB를 초과할 수 없습니다") - unique_filename = generate_unique_filename(file.filename) + # [변경] BOMParser 사용하여 유니크 파일명 생성 + unique_filename = BOMParser.generate_unique_filename(file.filename) file_path = UPLOAD_DIR / unique_filename try: - # 로그 제거 with open(file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) - # 로그 제거 except Exception as e: raise HTTPException(status_code=500, detail=f"파일 저장 실패: {str(e)}") try: - # 로그 제거 - materials_data = parse_file_data(str(file_path)) + # [변경] BOMParser 사용하여 파일 파싱 (자동 양식 감지 포함) + print(f"🚀 파일 파싱 시작: {file_path}") + materials_data = BOMParser.parse_file(str(file_path)) parsed_count = len(materials_data) - # 로그 제거 + print(f"✅ 파싱 완료: {parsed_count}개 자재 추출됨") # 신규 자재 카운트 초기화 new_materials_count = 0 @@ -546,1075 +407,23 @@ async def upload_file( import traceback traceback.print_exc() - print(f"🔧 자재 분류 시작: {len(materials_to_classify)}개 자재") + print(f"🔧 자재 분류 및 저장 시작: {len(materials_to_classify)}개 자재") - # 자재 데이터 저장 (분류 포함) - 배치 처리로 성능 개선 - materials_to_insert = [] - pipe_details_to_insert = [] - fitting_details_to_insert = [] - bolt_details_to_insert = [] - gasket_details_to_insert = [] - flange_details_to_insert = [] - - materials_inserted = 0 - - # 변경없는 자재 먼저 처리 (기존 분류 결과 재사용) - if revision_comparison and revision_comparison.get("has_previous_confirmation", False): - unchanged_materials = revision_comparison.get("unchanged_materials", []) - for material_data in unchanged_materials: - previous_item = material_data.get("previous_item", {}) - - # 기존 분류 결과 재사용 - materials_to_insert.append({ - "file_id": file_id, - "original_description": material_data["original_description"], - "classified_category": previous_item.get("category", "UNCLASSIFIED"), - "confidence": 1.0, # 확정된 자료이므로 신뢰도 100% - "quantity": material_data["quantity"], - "unit": material_data.get("unit", "EA"), - "size_spec": material_data.get("size_spec", ""), - "material_grade": previous_item.get("material", ""), - "specification": previous_item.get("specification", ""), - "reused_from_confirmation": True - }) - materials_inserted += 1 - - # 리비전 업로드의 경우 변경사항 추적 - changed_materials_keys = set() - new_materials_keys = set() - - if parent_file_id is not None: - print(f"🔄 리비전 업로드: 전체 {len(materials_to_classify)}개 자재 저장 (변경사항 추적 포함)") - - # 이전 리비전의 자재 조회 (도면번호 기준) - try: - # 🎯 트랜잭션 오류 방지: 쿼리 실행 전 롤백 - db.rollback() - print("🔄 리비전 자재 조회 전 트랜잭션 롤백") - - prev_materials_query = text(""" - SELECT original_description, size_spec, material_grade, main_nom, - drawing_name, line_no, quantity - FROM materials - WHERE file_id = :parent_file_id - """) - prev_materials_result = db.execute(prev_materials_query, {"parent_file_id": parent_file_id}).fetchall() - print(f"✅ 이전 리비전 자재 조회 성공: {len(prev_materials_result)}개") - - except Exception as e: - print(f"❌ 이전 리비전 자재 조회 실패: {e}") - # 오류 발생 시 빈 결과로 처리 - prev_materials_result = [] - - # 이전 자재를 딕셔너리로 변환 (도면번호 + 설명 + 크기 + 재질로 키 생성) - prev_materials_dict = {} - for prev_mat in prev_materials_result: - if prev_mat.drawing_name: - key = f"{prev_mat.drawing_name}|{prev_mat.original_description}|{prev_mat.size_spec or ''}|{prev_mat.material_grade or ''}" - elif prev_mat.line_no: - key = f"{prev_mat.line_no}|{prev_mat.original_description}|{prev_mat.size_spec or ''}|{prev_mat.material_grade or ''}" - else: - key = f"{prev_mat.original_description}|{prev_mat.size_spec or ''}|{prev_mat.material_grade or ''}" - - prev_materials_dict[key] = { - "quantity": float(prev_mat.quantity) if prev_mat.quantity else 0, - "description": prev_mat.original_description - } - - # 새 자재와 비교하여 변경사항 감지 - for material_data in materials_to_classify: - desc = material_data["original_description"] - size_spec = material_data.get("size_spec", "") - material_grade = material_data.get("material_grade", "") - dwg_name = material_data.get("dwg_name") - line_num = material_data.get("line_num") - - if dwg_name: - new_key = f"{dwg_name}|{desc}|{size_spec}|{material_grade}" - elif line_num: - new_key = f"{line_num}|{desc}|{size_spec}|{material_grade}" - else: - new_key = f"{desc}|{size_spec}|{material_grade}" - - if new_key in prev_materials_dict: - # 기존에 있던 자재 - 수량 변경 확인 - prev_qty = prev_materials_dict[new_key]["quantity"] - new_qty = float(material_data.get("quantity", 0)) - - if abs(prev_qty - new_qty) > 0.001: - # 수량이 변경됨 - changed_materials_keys.add(new_key) - print(f"🔄 변경 감지: {desc[:40]}... (수량: {prev_qty} → {new_qty})") - else: - # 새로 추가된 자재 - new_materials_keys.add(new_key) - print(f"➕ 신규 감지: {desc[:40]}...") - - print(f"📊 변경사항 요약: 변경 {len(changed_materials_keys)}개, 신규 {len(new_materials_keys)}개") - else: - print(f"🆕 신규 업로드: 전체 {len(materials_to_classify)}개 분류 처리") - - # 분류가 필요한 자재 처리 - print(f"분류할 자재 총 개수: {len(materials_to_classify)}") - - for material_data in materials_to_classify: - # 자재 타입 분류기 적용 (PIPE, FITTING, VALVE 등) - description = material_data["original_description"] - size_spec = material_data["size_spec"] - - # 각 분류기로 시도 (올바른 매개변수 사용) - print(f"분류 시도: {description}") - - # LENGTH 정보 추출 - length_value = None - if "length" in material_data: - try: - length_value = float(material_data["length"]) - except (ValueError, TypeError): - length_value = None - - # main_nom과 red_nom 추출 - main_nom = material_data.get("main_nom") - red_nom = material_data.get("red_nom") - - # 1. 통합 분류기로 자재 타입 결정 - integrated_result = classify_material_integrated(description, main_nom or "", red_nom or "", length_value) - print(f"[분류] {description}") - print(f"통합 분류 결과: {integrated_result.get('category', 'UNCLASSIFIED')} (신뢰도: {integrated_result.get('confidence', 0)}, Level: {integrated_result.get('classification_level', 'NONE')})") - - # 2. 제외 대상 확인 - if should_exclude_material(description): - classification_result = { - "category": "EXCLUDE", - "overall_confidence": 0.95, - "reason": "제외 대상 자재" - } - else: - # 3. 타입별 상세 분류기 실행 - material_type = integrated_result.get('category', 'UNCLASSIFIED') - - if material_type == "PIPE": - from ..services.pipe_classifier import classify_pipe_for_purchase - classification_result = classify_pipe_for_purchase("", description, main_nom or "", length_value) - elif material_type == "FITTING": - classification_result = classify_fitting("", description, main_nom or "", red_nom) - elif material_type == "FLANGE": - classification_result = classify_flange("", description, main_nom or "", red_nom) - elif material_type == "VALVE": - classification_result = classify_valve("", description, main_nom or "") - elif material_type == "BOLT": - classification_result = classify_bolt("", description, main_nom or "") - print(f"🔧 BOLT 분류 결과: {classification_result}") - print(f"🔧 원본 설명: {description}") - print(f"🔧 main_nom: {main_nom}") - - # 길이 정보 확인 - dimensions_info = classification_result.get("dimensions", {}) - print(f"🔧 길이 정보: {dimensions_info}") - - # 재질 정보 확인 - material_info = classification_result.get("material", {}) - print(f"🔧 재질 정보: {material_info}") - elif material_type == "GASKET": - classification_result = classify_gasket("", description, main_nom or "") - elif material_type == "INSTRUMENT": - classification_result = classify_instrument("", description, main_nom or "") - elif material_type == "SUPPORT": - from ..services.support_classifier import classify_support - classification_result = classify_support("", description, main_nom or "") - elif material_type == "SPECIAL": - # SPECIAL 카테고리는 별도 분류기 없이 통합 분류 결과 사용 - classification_result = { - "category": "SPECIAL", - "overall_confidence": integrated_result.get('confidence', 1.0), - "reason": integrated_result.get('reason', 'SPECIAL 키워드 발견'), - "details": { - "description": description, - "main_nom": main_nom or "", - "drawing_required": True # 도면 필요 - } - } - else: - # UNCLASSIFIED 처리 - classification_result = { - "category": "UNCLASSIFIED", - "overall_confidence": integrated_result.get('confidence', 0.0), - "reason": f"분류 불가: {integrated_result.get('evidence', [])}" - } - - # 통합 분류기의 신뢰도가 더 낮으면 조정 - if integrated_result.get('confidence', 0) < 0.5: - classification_result['overall_confidence'] = min( - classification_result.get('overall_confidence', 1.0), - integrated_result.get('confidence', 0.0) + 0.2 - ) - - print(f"최종 분류 결과: {classification_result.get('category', 'UNCLASSIFIED')}") - - # 전체 재질명 추출 - from ..services.material_grade_extractor import extract_full_material_grade - full_material_grade = extract_full_material_grade(description) - if not full_material_grade and material_data.get("material_grade"): - full_material_grade = material_data["material_grade"] - - # 🔥 구매확정 상태 확인 (리비전인 경우) - is_purchase_confirmed = False - purchase_confirmed_at = None - purchase_confirmed_by = None - - if parent_file_id and purchased_materials_map: - # 자재 식별 키 생성 (리비전 비교와 동일한 방식) - material_key = f"{description.strip().upper()}|{size_spec or ''}" - - if material_key in purchased_materials_map: - # 이전 리비전에서 구매확정된 자재 - purchased_info = purchased_materials_map[material_key] - is_purchase_confirmed = True - purchase_confirmed_at = purchased_info.get("purchase_confirmed_at") - purchase_confirmed_by = purchased_info.get("purchase_confirmed_by") - print(f"🔥 구매확정 상태 상속: {description[:50]}...") - - # 기본 자재 정보 저장 - material_insert_query = text(""" - INSERT INTO materials ( - file_id, original_description, quantity, unit, size_spec, - main_nom, red_nom, material_grade, full_material_grade, line_number, row_number, - classified_category, classification_confidence, is_verified, - drawing_name, line_no, created_at, - purchase_confirmed, purchase_confirmed_at, purchase_confirmed_by - ) - VALUES ( - :file_id, :original_description, :quantity, :unit, :size_spec, - :main_nom, :red_nom, :material_grade, :full_material_grade, :line_number, :row_number, - :classified_category, :classification_confidence, :is_verified, - :drawing_name, :line_no, :created_at, - :purchase_confirmed, :purchase_confirmed_at, :purchase_confirmed_by - ) - RETURNING id - """) - - # 첫 번째 자재에 대해서만 디버그 출력 - if materials_inserted == 0: - print(f"첫 번째 자재 저장:") - print(f" size_spec: '{material_data['size_spec']}'") - print(f" original_description: {material_data['original_description']}") - print(f" category: {classification_result.get('category', 'UNCLASSIFIED')}") - print(f" drawing_name: {material_data.get('dwg_name')}") - print(f" line_no: {material_data.get('line_num')}") - print(f" purchase_confirmed: {is_purchase_confirmed}") - - material_result = db.execute(material_insert_query, { - "file_id": file_id, - "original_description": material_data["original_description"], - "quantity": material_data["quantity"], - "unit": material_data["unit"], - "size_spec": material_data["size_spec"], - "main_nom": material_data.get("main_nom"), - "red_nom": material_data.get("red_nom"), - "material_grade": material_data["material_grade"], - "full_material_grade": full_material_grade, - "line_number": material_data["line_number"], - "row_number": material_data["row_number"], - "classified_category": classification_result.get("category", "UNCLASSIFIED"), - "classification_confidence": classification_result.get("overall_confidence", 0.0), - "is_verified": False, - "drawing_name": material_data.get("dwg_name"), - "line_no": material_data.get("line_num"), - "created_at": datetime.now(), - "purchase_confirmed": is_purchase_confirmed, - "purchase_confirmed_at": purchase_confirmed_at, - "purchase_confirmed_by": purchase_confirmed_by - }) - - material_id = material_result.fetchone()[0] - materials_inserted += 1 - - # 리비전 업로드인 경우 변경/신규 자재 표시 및 구매신청 정보 상속 - if parent_file_id is not None: - # 현재 자재의 키 생성 - dwg_name = material_data.get("dwg_name") - line_num = material_data.get("line_num") - - if dwg_name: - current_key = f"{dwg_name}|{description}|{size_spec}|{material_data.get('material_grade', '')}" - elif line_num: - current_key = f"{line_num}|{description}|{size_spec}|{material_data.get('material_grade', '')}" - else: - current_key = f"{description}|{size_spec}|{material_data.get('material_grade', '')}" - - # 변경 또는 신규 자재에 상태 표시 - if current_key in changed_materials_keys: - # 변경된 자재 - update_status_query = text(""" - UPDATE materials SET revision_status = 'changed' - WHERE id = :material_id - """) - db.execute(update_status_query, {"material_id": material_id}) - elif current_key in new_materials_keys: - # 신규 자재 (추가됨) - update_status_query = text(""" - UPDATE materials SET revision_status = 'active' - WHERE id = :material_id - """) - db.execute(update_status_query, {"material_id": material_id}) - - # 구매신청 정보 상속은 전체 자재 저장 후 일괄 처리하도록 변경 - # (개별 처리는 비효율적이고 수량 계산이 복잡함) - - # PIPE 분류 결과인 경우 상세 정보 저장 - if classification_result.get("category") == "PIPE": - print("PIPE 상세 정보 저장 시작") - - # 끝단 가공 정보 추출 및 저장 - from ..services.pipe_classifier import extract_end_preparation_info - end_prep_info = extract_end_preparation_info(description) - - # 끝단 가공 정보 테이블에 저장 - end_prep_insert_query = text(""" - INSERT INTO pipe_end_preparations ( - material_id, file_id, end_preparation_type, end_preparation_code, - machining_required, cutting_note, original_description, clean_description, - confidence, matched_pattern - ) VALUES ( - :material_id, :file_id, :end_preparation_type, :end_preparation_code, - :machining_required, :cutting_note, :original_description, :clean_description, - :confidence, :matched_pattern - ) - """) - - db.execute(end_prep_insert_query, { - "material_id": material_id, - "file_id": file_id, - "end_preparation_type": end_prep_info["end_preparation_type"], - "end_preparation_code": end_prep_info["end_preparation_code"], - "machining_required": end_prep_info["machining_required"], - "cutting_note": end_prep_info["cutting_note"], - "original_description": end_prep_info["original_description"], - "clean_description": end_prep_info["clean_description"], - "confidence": end_prep_info["confidence"], - "matched_pattern": end_prep_info["matched_pattern"] - }) - - # 길이 정보 추출 - 분류 결과의 length_info 우선 사용 - length_info = classification_result.get("length_info", {}) - length_mm = length_info.get("length_mm") or material_data.get("length", 0.0) if material_data.get("length") else None - - # material_id도 함께 저장하도록 수정 - pipe_detail_insert_query = text(""" - INSERT INTO pipe_details ( - material_id, file_id, outer_diameter, schedule, - material_spec, manufacturing_method, end_preparation, length_mm - ) VALUES ( - :material_id, :file_id, :outer_diameter, :schedule, - :material_spec, :manufacturing_method, :end_preparation, :length_mm - ) - """) - - # 재질 정보 - material_info = classification_result.get("material", {}) - manufacturing_info = classification_result.get("manufacturing", {}) - end_prep_info = classification_result.get("end_preparation", {}) - schedule_info = classification_result.get("schedule", {}) - size_info = classification_result.get("size_info", {}) - - # main_nom을 outer_diameter로 활용 - outer_diameter = material_data.get("main_nom") or material_data.get("size_spec", "") - - # end_preparation 정보 추출 (분류 결과에서) - end_prep = "" - if isinstance(end_prep_info, dict): - end_prep = end_prep_info.get("type", "") - else: - end_prep = str(end_prep_info) if end_prep_info else "" - - # 재질 정보 - 분류 결과에서 상세 정보 추출 - material_grade_from_classifier = "" - if isinstance(material_info, dict): - material_grade_from_classifier = material_info.get("grade", "") - - # 분류기에서 더 상세한 재질 정보가 나왔으면 업데이트 - if material_grade_from_classifier and material_grade_from_classifier not in ["UNKNOWN", "UNCLASSIFIED"]: - material_spec = material_grade_from_classifier - - # materials 테이블의 material_grade도 업데이트 - db.execute(text(""" - UPDATE materials - SET material_grade = :new_material_grade - WHERE id = :material_id - """), { - "new_material_grade": material_grade_from_classifier, - "material_id": material_id - }) - print(f"PIPE material_grade 업데이트: {material_grade_from_classifier}") - else: - # 기존 파싱 결과 사용 - material_spec = material_data.get("material_grade", "") - - # 제조방법 추출 - manufacturing_method = "" - if isinstance(manufacturing_info, dict): - manufacturing_method = manufacturing_info.get("method", "UNKNOWN") - else: - manufacturing_method = str(manufacturing_info) if manufacturing_info else "UNKNOWN" - - # 스케줄 정보 추출 - schedule = "" - if isinstance(schedule_info, dict): - schedule = schedule_info.get("schedule", "UNKNOWN") - else: - schedule = str(schedule_info) if schedule_info else "UNKNOWN" - - db.execute(pipe_detail_insert_query, { - "material_id": material_id, - "file_id": file_id, - "outer_diameter": outer_diameter, - "schedule": schedule, - "material_spec": material_spec, - "manufacturing_method": manufacturing_method, - "end_preparation": end_prep, - "length_mm": material_data.get("length", 0.0) if material_data.get("length") else 0.0 - }) - - print("PIPE 상세 정보 저장 완료") - - # FITTING 분류 결과인 경우 상세 정보 저장 - elif classification_result.get("category") == "FITTING": - print("FITTING 상세 정보 저장 시작") - - # 피팅 정보 추출 - fitting_type_info = classification_result.get("fitting_type", {}) - connection_info = classification_result.get("connection_method", {}) - pressure_info = classification_result.get("pressure_rating", {}) - material_info = classification_result.get("material", {}) - - # 피팅 타입 및 서브타입 - fitting_type = fitting_type_info.get("type", "UNKNOWN") - fitting_subtype = fitting_type_info.get("subtype", "UNKNOWN") - - # 연결 방식 - connection_method = connection_info.get("method", "UNKNOWN") - - # 압력 등급 - pressure_rating = pressure_info.get("rating", "UNKNOWN") - - # 재질 정보 - material_standard = material_info.get("standard", "") - material_grade = material_info.get("grade", "") - - # 분류기에서 더 상세한 재질 정보가 나왔으면 업데이트 - if material_grade and material_grade not in ["UNKNOWN", "UNCLASSIFIED"]: - db.execute(text(""" - UPDATE materials - SET material_grade = :new_material_grade - WHERE id = :material_id - """), { - "new_material_grade": material_grade, - "material_id": material_id - }) - print(f"FITTING material_grade 업데이트: {material_grade}") - - # main_size와 reduced_size - main_size = material_data.get("main_nom") or material_data.get("size_spec", "") - reduced_size = material_data.get("red_nom", "") - - # NIPPLE인 경우 길이와 스케줄 정보 추가 - length_mm = None - schedule = "UNKNOWN" - if fitting_type == "NIPPLE": - # 길이 정보 추출 - length_mm = material_data.get("length", 0.0) if material_data.get("length") else None - - # 스케줄 정보 추출 (분류 결과에서) - schedule_info = classification_result.get("schedule_info", {}) - schedule = schedule_info.get("schedule", "UNKNOWN") - schedule_info = classification_result.get("schedule", {}) - if isinstance(schedule_info, dict): - schedule = schedule_info.get("schedule", "UNKNOWN") - else: - schedule = str(schedule_info) if schedule_info else "UNKNOWN" - - db.execute(text(""" - INSERT INTO fitting_details ( - material_id, file_id, fitting_type, fitting_subtype, - connection_method, pressure_rating, material_standard, - material_grade, main_size, reduced_size, length_mm, schedule - ) VALUES ( - :material_id, :file_id, :fitting_type, :fitting_subtype, - :connection_method, :pressure_rating, :material_standard, - :material_grade, :main_size, :reduced_size, :length_mm, :schedule - ) - """), { - "material_id": material_id, - "file_id": file_id, - "fitting_type": fitting_type, - "fitting_subtype": fitting_subtype, - "connection_method": connection_method, - "pressure_rating": pressure_rating, - "material_standard": material_standard, - "material_grade": material_grade, - "main_size": main_size, - "reduced_size": reduced_size, - "length_mm": length_mm, - "schedule": schedule - }) - - print(f"FITTING 상세 정보 저장 완료: {fitting_type} - {fitting_subtype}") - - # FLANGE 분류 결과인 경우 상세 정보 저장 - if classification_result.get("category") == "FLANGE": - print("FLANGE 상세 정보 저장 시작") - - # 플랜지 타입 정보 - flange_type_info = classification_result.get("flange_type", {}) - pressure_info = classification_result.get("pressure_rating", {}) - face_finish_info = classification_result.get("face_finish", {}) - material_info = classification_result.get("material", {}) - - # 플랜지 타입 (WN, BL, SO 등) - flange_type = "" - if isinstance(flange_type_info, dict): - flange_type = flange_type_info.get("type", "UNKNOWN") - else: - flange_type = str(flange_type_info) if flange_type_info else "UNKNOWN" - - # 압력 등급 (150LB, 300LB 등) - pressure_rating = "" - if isinstance(pressure_info, dict): - pressure_rating = pressure_info.get("rating", "UNKNOWN") - else: - pressure_rating = str(pressure_info) if pressure_info else "UNKNOWN" - - # 면 가공 (RF, FF, RTJ 등) - facing_type = "" - if isinstance(face_finish_info, dict): - facing_type = face_finish_info.get("finish", "UNKNOWN") - else: - facing_type = str(face_finish_info) if face_finish_info else "UNKNOWN" - - # 재질 정보 - material_standard = "" - material_grade = "" - if isinstance(material_info, dict): - material_standard = material_info.get("standard", "") - material_grade = material_info.get("grade", "") - - # 분류기에서 더 상세한 재질 정보가 나왔으면 업데이트 - if material_grade and material_grade not in ["UNKNOWN", "UNCLASSIFIED"]: - db.execute(text(""" - UPDATE materials - SET material_grade = :new_material_grade - WHERE id = :material_id - """), { - "new_material_grade": material_grade, - "material_id": material_id - }) - print(f"FLANGE material_grade 업데이트: {material_grade}") - - # 사이즈 정보 - size_inches = material_data.get("main_nom") or material_data.get("size_spec", "") - - db.execute(text(""" - INSERT INTO flange_details ( - material_id, file_id, flange_type, pressure_rating, - facing_type, material_standard, material_grade, size_inches - ) VALUES ( - :material_id, :file_id, :flange_type, :pressure_rating, - :facing_type, :material_standard, :material_grade, :size_inches - ) - """), { - "material_id": material_id, - "file_id": file_id, - "flange_type": flange_type, - "pressure_rating": pressure_rating, - "facing_type": facing_type, - "material_standard": material_standard, - "material_grade": material_grade, - "size_inches": size_inches - }) - - print(f"FLANGE 상세 정보 저장 완료: {flange_type} - {pressure_rating}") - - # GASKET 분류 결과인 경우 상세 정보 저장 - if classification_result.get("category") == "GASKET": - print("GASKET 상세 정보 저장 시작") - - # 가스켓 타입 정보 - gasket_type_info = classification_result.get("gasket_type", {}) - gasket_material_info = classification_result.get("gasket_material", {}) - pressure_info = classification_result.get("pressure_rating", {}) - - # 가스켓 타입 (SPIRAL_WOUND, O_RING 등) - gasket_type = "" - if isinstance(gasket_type_info, dict): - gasket_type = gasket_type_info.get("type", "UNKNOWN") - else: - gasket_type = str(gasket_type_info) if gasket_type_info else "UNKNOWN" - - # 가스켓 소재 - SWG의 경우 메탈 부분을 우선으로 - material_type = "" - if isinstance(gasket_material_info, dict): - # SWG 상세 정보가 있으면 메탈 부분을 material_type으로 사용 - swg_details = gasket_material_info.get("swg_details", {}) - if swg_details and swg_details.get("outer_ring"): - material_type = swg_details.get("outer_ring", "UNKNOWN") # SS304 - else: - material_type = gasket_material_info.get("material", "UNKNOWN") - else: - material_type = str(gasket_material_info) if gasket_material_info else "UNKNOWN" - - # 압력 등급 - pressure_rating = "" - if isinstance(pressure_info, dict): - pressure_rating = pressure_info.get("rating", "UNKNOWN") - else: - pressure_rating = str(pressure_info) if pressure_info else "UNKNOWN" - - # 사이즈 정보 - size_inches = material_data.get("main_nom") or material_data.get("size_spec", "") - - # SWG 상세 정보 추출 - swg_details = gasket_material_info.get("swg_details", {}) if isinstance(gasket_material_info, dict) else {} - thickness = swg_details.get("thickness", None) if swg_details else None - filler_material = swg_details.get("filler", "") if swg_details else "" - - # additional_info에 SWG 상세 정보 저장 - additional_info = "" - if swg_details: - face_type = swg_details.get("face_type", "") - outer_ring = swg_details.get("outer_ring", "") - inner_ring = swg_details.get("inner_ring", "") - construction = swg_details.get("detailed_construction", "") - - # JSON 형태로 additional_info 생성 - additional_info = { - "face_type": face_type, - "construction": construction, - "outer_ring": outer_ring, - "inner_ring": inner_ring, - "filler": swg_details.get("filler", ""), - "thickness": swg_details.get("thickness", None) - } - additional_info_json = json.dumps(additional_info, ensure_ascii=False) - - db.execute(text(""" - INSERT INTO gasket_details ( - material_id, file_id, gasket_type, material_type, - pressure_rating, size_inches, thickness, filler_material, additional_info - ) VALUES ( - :material_id, :file_id, :gasket_type, :material_type, - :pressure_rating, :size_inches, :thickness, :filler_material, :additional_info - ) - """), { - "material_id": material_id, - "file_id": file_id, - "gasket_type": gasket_type, - "material_type": material_type, - "pressure_rating": pressure_rating, - "size_inches": size_inches, - "thickness": thickness, - "filler_material": filler_material, - "additional_info": additional_info_json - }) - - print(f"GASKET 상세 정보 저장 완료: {gasket_type} - {material_type}") - - # BOLT 분류 결과인 경우 상세 정보 저장 - if classification_result.get("category") == "BOLT": - print("BOLT 상세 정보 저장 시작") - - # 볼트 타입 정보 - fastener_type_info = classification_result.get("fastener_type", {}) - thread_spec_info = classification_result.get("thread_specification", {}) - dimensions_info = classification_result.get("dimensions", {}) - material_info = classification_result.get("material", {}) - - print(f"🔧 fastener_type_info: {fastener_type_info}") - - # 볼트 타입 (STUD_BOLT, HEX_BOLT 등) - bolt_type = "" - if isinstance(fastener_type_info, dict): - bolt_type = fastener_type_info.get("type", "UNKNOWN") - print(f"🔧 추출된 bolt_type: {bolt_type}") - else: - bolt_type = str(fastener_type_info) if fastener_type_info else "UNKNOWN" - print(f"🔧 문자열 bolt_type: {bolt_type}") - - # 특수 용도 볼트 확인 (PSV, LT, CK 등) - special_result = classification_result.get("special_applications", {}) - print(f"🔧 special_result: {special_result}") - - # 특수 용도가 감지되면 타입 우선 적용 - if special_result and special_result.get("detected_applications"): - detected_apps = special_result.get("detected_applications", []) - if "LT" in detected_apps: - bolt_type = "LT_BOLT" - print(f"🔧 특수 용도 감지로 bolt_type 변경: {bolt_type}") - elif "PSV" in detected_apps: - bolt_type = "PSV_BOLT" - print(f"🔧 특수 용도 감지로 bolt_type 변경: {bolt_type}") - elif "CK" in detected_apps: - bolt_type = "CK_BOLT" - print(f"🔧 특수 용도 감지로 bolt_type 변경: {bolt_type}") - - print(f"🔧 최종 bolt_type: {bolt_type}") - - # 나사 타입 (METRIC, INCH 등) - thread_type = "" - if isinstance(thread_spec_info, dict): - thread_type = thread_spec_info.get("standard", "UNKNOWN") - else: - thread_type = str(thread_spec_info) if thread_spec_info else "UNKNOWN" - - # 치수 정보 (실제 볼트 사이즈 사용) - diameter = "" - length = "" - nominal_size_fraction = "" - if isinstance(dimensions_info, dict): - # 볼트 분류기에서 추출한 실제 볼트 사이즈 사용 - diameter = dimensions_info.get("nominal_size", material_data.get("main_nom", "")) - nominal_size_fraction = dimensions_info.get("nominal_size_fraction", diameter) - length = dimensions_info.get("length", "") - if not length and "70.0000 LG" in description: - # 원본 설명에서 길이 추출 - import re - length_match = re.search(r'(\d+(?:\.\d+)?)\s*LG', description.upper()) - if length_match: - length = f"{length_match.group(1)}mm" - - # 재질 정보 - material_standard = "" - material_grade = "" - if isinstance(material_info, dict): - material_standard = material_info.get("standard", "") - material_grade = material_info.get("grade", "") - - # 분류기에서 더 상세한 재질 정보가 나왔으면 업데이트 - if material_grade and material_grade not in ["UNKNOWN", "UNCLASSIFIED"]: - db.execute(text(""" - UPDATE materials - SET material_grade = :new_material_grade - WHERE id = :material_id - """), { - "new_material_grade": material_grade, - "material_id": material_id - }) - print(f"BOLT material_grade 업데이트: {material_grade}") - - # 압력 등급 (150LB 등) - pressure_rating = "" - if "150LB" in description.upper(): - pressure_rating = "150LB" - elif "300LB" in description.upper(): - pressure_rating = "300LB" - elif "600LB" in description.upper(): - pressure_rating = "600LB" - - # 코팅 타입 (ELEC.GALV 등) - coating_type = "" - if "ELEC.GALV" in description.upper() or "ELEC GALV" in description.upper(): - coating_type = "ELECTRO_GALVANIZED" - elif "HOT.GALV" in description.upper() or "HOT GALV" in description.upper(): - coating_type = "HOT_DIP_GALVANIZED" - elif "GALV" in description.upper(): - coating_type = "GALVANIZED" - elif "ZINC" in description.upper(): - coating_type = "ZINC_PLATED" - elif "DACROMET" in description.upper(): - coating_type = "DACROMET" - elif "SS" in description.upper() or "STAINLESS" in description.upper(): - coating_type = "STAINLESS" - elif "PLAIN" in description.upper() or "BLACK" in description.upper(): - coating_type = "PLAIN" - - db.execute(text(""" - INSERT INTO bolt_details ( - material_id, file_id, bolt_type, thread_type, - diameter, length, material_standard, material_grade, - coating_type, pressure_rating, classification_confidence - ) VALUES ( - :material_id, :file_id, :bolt_type, :thread_type, - :diameter, :length, :material_standard, :material_grade, - :coating_type, :pressure_rating, :classification_confidence - ) - """), { - "material_id": material_id, - "file_id": file_id, - "bolt_type": bolt_type, - "thread_type": thread_type, - "diameter": diameter, - "length": length, - "material_standard": material_standard, - "material_grade": material_grade, - "coating_type": coating_type, - "pressure_rating": pressure_rating, - "classification_confidence": classification_result.get("overall_confidence", 0.0) - }) - - print(f"BOLT 상세 정보 저장 완료: {bolt_type} - {material_standard} {material_grade}") - - # SUPPORT 분류 결과인 경우 상세 정보 저장 - if classification_result.get("category") == "SUPPORT": - print("SUPPORT 상세 정보 저장 시작") - - support_type = classification_result.get("support_type", "UNKNOWN") - support_subtype = classification_result.get("support_subtype", "") - load_rating = classification_result.get("load_rating", "") - load_capacity = classification_result.get("load_capacity", "") - - # 재질 정보 - material_info = classification_result.get("material", {}) - material_standard = material_info.get("standard", "UNKNOWN") - material_grade = material_info.get("grade", "UNKNOWN") - - # 사이즈 정보 - size_info = classification_result.get("size_info", {}) - pipe_size = size_info.get("pipe_size", "") - dimensions = size_info.get("dimensions", {}) - - length_mm = None - width_mm = None - height_mm = None - - if dimensions: - length_str = dimensions.get("length", "") - width_str = dimensions.get("width", "") - height_str = dimensions.get("height", "") - - # mm 단위 추출 - import re - if length_str: - length_match = re.search(r'(\d+(?:\.\d+)?)', length_str) - if length_match: - length_mm = float(length_match.group(1)) - - if width_str: - width_match = re.search(r'(\d+(?:\.\d+)?)', width_str) - if width_match: - width_mm = float(width_match.group(1)) - - if height_str: - height_match = re.search(r'(\d+(?:\.\d+)?)', height_str) - if height_match: - height_mm = float(height_match.group(1)) - - db.execute(text(""" - INSERT INTO support_details ( - material_id, file_id, support_type, support_subtype, - load_rating, load_capacity, material_standard, material_grade, - pipe_size, length_mm, width_mm, height_mm, classification_confidence - ) VALUES ( - :material_id, :file_id, :support_type, :support_subtype, - :load_rating, :load_capacity, :material_standard, :material_grade, - :pipe_size, :length_mm, :width_mm, :height_mm, :classification_confidence - ) - """), { - "material_id": material_id, - "file_id": file_id, - "support_type": support_type, - "support_subtype": support_subtype, - "load_rating": load_rating, - "load_capacity": load_capacity, - "material_standard": material_standard, - "material_grade": material_grade, - "pipe_size": pipe_size, - "length_mm": length_mm, - "width_mm": width_mm, - "height_mm": height_mm, - "classification_confidence": classification_result.get("overall_confidence", 0.0) - }) - - print(f"SUPPORT 상세 정보 저장 완료: {support_type} - {material_standard} {material_grade}") - - # VALVE 분류 결과인 경우 상세 정보 저장 - if classification_result.get("category") == "VALVE": - print("VALVE 상세 정보 저장 시작") - - # 밸브 타입 정보 - valve_type_info = classification_result.get("valve_type", {}) - connection_info = classification_result.get("connection_method", {}) - pressure_info = classification_result.get("pressure_rating", {}) - material_info = classification_result.get("material", {}) - - # 밸브 타입 (GATE_VALVE, BALL_VALVE 등) - valve_type = "" - if isinstance(valve_type_info, dict): - valve_type = valve_type_info.get("type", "UNKNOWN") - else: - valve_type = str(valve_type_info) if valve_type_info else "UNKNOWN" - - # 밸브 서브타입 (특수 기능) - valve_subtype = "" - special_features = classification_result.get("special_features", []) - if special_features: - valve_subtype = ", ".join(special_features) - - # 작동 방식 - actuator_type = "MANUAL" # 기본값 - actuation_info = classification_result.get("actuation", {}) - if isinstance(actuation_info, dict): - actuator_type = actuation_info.get("method", "MANUAL") - - # 연결 방식 (FLANGED, THREADED 등) - connection_method = "" - if isinstance(connection_info, dict): - connection_method = connection_info.get("method", "UNKNOWN") - else: - connection_method = str(connection_info) if connection_info else "UNKNOWN" - - # 압력 등급 (150LB, 300LB 등) - pressure_rating = "" - if isinstance(pressure_info, dict): - pressure_rating = pressure_info.get("rating", "UNKNOWN") - else: - pressure_rating = str(pressure_info) if pressure_info else "UNKNOWN" - - # 재질 정보 - body_material = "" - trim_material = "" - if isinstance(material_info, dict): - body_material = material_info.get("grade", "") - # 트림 재질은 일반적으로 바디와 동일하거나 별도 명시 - trim_material = body_material - - # 분류기에서 더 상세한 재질 정보가 나왔으면 업데이트 - if body_material and body_material != "UNKNOWN": - db.execute(text(""" - UPDATE materials - SET material_grade = :new_material_grade - WHERE id = :material_id - """), { - "new_material_grade": body_material, - "material_id": material_id - }) - print(f"VALVE material_grade 업데이트: {body_material}") - - # 사이즈 정보 - size_inches = material_data.get("main_nom") or material_data.get("size_spec", "") - - # 특수 기능 (Fire Safe, Anti-Static 등) - fire_safe = any("FIRE" in feature.upper() for feature in special_features) - low_temp_service = any("CRYO" in feature.upper() or "LOW" in feature.upper() for feature in special_features) - - # 추가 정보 JSON 생성 - additional_info = { - "characteristics": valve_type_info.get("characteristics", "") if isinstance(valve_type_info, dict) else "", - "typical_connections": valve_type_info.get("typical_connections", []) if isinstance(valve_type_info, dict) else [], - "special_features": special_features, - "manufacturing": classification_result.get("manufacturing", {}), - "evidence": valve_type_info.get("evidence", []) if isinstance(valve_type_info, dict) else [] - } - additional_info_json = json.dumps(additional_info, ensure_ascii=False) - - db.execute(text(""" - INSERT INTO valve_details ( - material_id, file_id, valve_type, valve_subtype, - actuator_type, connection_method, pressure_rating, - body_material, trim_material, size_inches, - fire_safe, low_temp_service, classification_confidence, additional_info - ) VALUES ( - :material_id, :file_id, :valve_type, :valve_subtype, - :actuator_type, :connection_method, :pressure_rating, - :body_material, :trim_material, :size_inches, - :fire_safe, :low_temp_service, :classification_confidence, :additional_info - ) - """), { - "material_id": material_id, - "file_id": file_id, - "valve_type": valve_type, - "valve_subtype": valve_subtype, - "actuator_type": actuator_type, - "connection_method": connection_method, - "pressure_rating": pressure_rating, - "body_material": body_material, - "trim_material": trim_material, - "size_inches": size_inches, - "fire_safe": fire_safe, - "low_temp_service": low_temp_service, - "classification_confidence": classification_result.get("overall_confidence", 0.0), - "additional_info": additional_info_json - }) - - print(f"VALVE 상세 정보 저장 완료: {valve_type} - {connection_method} - {pressure_rating}") + # [변경] MaterialService를 사용하여 자재 처리 및 저장 + materials_inserted = MaterialService.process_and_save_materials( + db, file_id, materials_data, + revision_comparison=revision_comparison, + parent_file_id=parent_file_id, + purchased_materials_map=purchased_materials_map + ) db.commit() print(f"자재 저장 완료: {materials_inserted}개") - # 리비전 업로드인 경우 구매신청 정보 수량 기반 상속 + # [변경] MaterialService를 사용하여 구매신청 정보 상속 if parent_file_id is not None: - try: - print(f"🔄 구매신청 정보 상속 처리 시작...") - - # 1. 이전 리비전에서 그룹별 구매신청 수량 집계 - prev_purchase_summary = text(""" - SELECT - m.original_description, - m.size_spec, - m.material_grade, - m.drawing_name, - COUNT(DISTINCT pri.material_id) as purchased_count, - SUM(pri.quantity) as total_purchased_qty, - MIN(pri.request_id) as request_id - FROM materials m - JOIN purchase_request_items pri ON m.id = pri.material_id - WHERE m.file_id = :parent_file_id - GROUP BY m.original_description, m.size_spec, m.material_grade, m.drawing_name - """) - - prev_purchases = db.execute(prev_purchase_summary, {"parent_file_id": parent_file_id}).fetchall() - - # 2. 새 리비전에서 같은 그룹의 자재에 수량만큼 구매신청 상속 - for prev_purchase in prev_purchases: - purchased_count = prev_purchase.purchased_count - - # 새 리비전에서 같은 그룹의 자재 조회 (순서대로) - new_group_materials = text(""" - SELECT id, quantity - FROM materials - WHERE file_id = :file_id - AND original_description = :description - AND COALESCE(size_spec, '') = :size_spec - AND COALESCE(material_grade, '') = :material_grade - AND COALESCE(drawing_name, '') = :drawing_name - ORDER BY id - LIMIT :limit - """) - - new_materials = db.execute(new_group_materials, { - "file_id": file_id, - "description": prev_purchase.original_description, - "size_spec": prev_purchase.size_spec or '', - "material_grade": prev_purchase.material_grade or '', - "drawing_name": prev_purchase.drawing_name or '', - "limit": purchased_count - }).fetchall() - - # 구매신청 수량만큼만 상속 - for new_mat in new_materials: - inherit_query = text(""" - INSERT INTO purchase_request_items ( - request_id, material_id, quantity, unit, user_requirement - ) VALUES ( - :request_id, :material_id, :quantity, 'EA', '' - ) - ON CONFLICT DO NOTHING - """) - db.execute(inherit_query, { - "request_id": prev_purchase.request_id, - "material_id": new_mat.id, - "quantity": new_mat.quantity - }) - - inherited_count = len(new_materials) - if inherited_count > 0: - print(f" ✅ {prev_purchase.original_description[:30]}... (도면: {prev_purchase.drawing_name or 'N/A'}) → {inherited_count}/{purchased_count}개 상속") - - db.commit() - print(f"✅ 구매신청 정보 상속 완료") - - except Exception as e: - print(f"❌ 구매신청 정보 상속 실패: {str(e)}") - db.rollback() - # 상속 실패는 업로드 성공에 영향 없음 + MaterialService.inherit_purchase_requests(db, file_id, parent_file_id) + db.commit() # 활동 로그 기록 try: diff --git a/backend/app/services/excel_parser.py b/backend/app/services/excel_parser.py new file mode 100644 index 0000000..c75165f --- /dev/null +++ b/backend/app/services/excel_parser.py @@ -0,0 +1,300 @@ + +import pandas as pd +import re +from typing import List, Dict, Optional +import uuid +from datetime import datetime +from pathlib import Path + +# 허용된 확장자 +ALLOWED_EXTENSIONS = {".xlsx", ".xls", ".csv"} + +class BOMParser: + """BOM 파일 파싱을 담당하는 클래스""" + + @staticmethod + def validate_extension(filename: str) -> bool: + return Path(filename).suffix.lower() in ALLOWED_EXTENSIONS + + @staticmethod + def generate_unique_filename(original_filename: str) -> str: + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + unique_id = str(uuid.uuid4())[:8] + stem = Path(original_filename).stem + suffix = Path(original_filename).suffix + return f"{stem}_{timestamp}_{unique_id}{suffix}" + + @staticmethod + def detect_format(df: pd.DataFrame) -> str: + """ + 엑셀 헤더를 분석하여 양식을 감지합니다. + + Returns: + 'INVENTOR': 인벤터 추출 양식 (NO., NAME, Q'ty, LENGTH & THICKNESS...) + 'STANDARD': 기존 표준/퍼지 매핑 엑셀 (DWG_NAME, LINE_NUM, MAIN_NOM...) + """ + columns = [str(c).strip().upper() for c in df.columns] + + # 인벤터 양식 특징 (오타 포함) + INVENTOR_KEYWORDS = ['LENGTH & THICKNESS', 'DESCIPTION', "Q'TY"] + + for keyword in INVENTOR_KEYWORDS: + if any(keyword in col for col in columns): + return 'INVENTOR' + + # 표준 양식 특징 + STANDARD_KEYWORDS = ['MAIN_NOM', 'DWG_NAME', 'LINE_NUM'] + for keyword in STANDARD_KEYWORDS: + if any(keyword in col for col in columns): + return 'STANDARD' + + return 'STANDARD' # 기본값 + + @classmethod + def parse_file(cls, file_path: str) -> List[Dict]: + """파일 경로를 받아 적절한 파서를 통해 데이터를 추출합니다.""" + file_extension = Path(file_path).suffix.lower() + + try: + if file_extension == ".csv": + df = pd.read_csv(file_path, encoding='utf-8') + elif file_extension in [".xlsx", ".xls"]: + # xlrd 엔진 명시 (xls 지원) + if file_extension == ".xls": + df = pd.read_excel(file_path, sheet_name=0, engine='xlrd') + else: + df = pd.read_excel(file_path, sheet_name=0, engine='openpyxl') + else: + raise ValueError("지원하지 않는 파일 형식") + + # 데이터프레임 전처리 (빈 행 제거 등) + df = df.dropna(how='all') + + # 양식 감지 + format_type = cls.detect_format(df) + print(f"📋 감지된 BOM 양식: {format_type}") + + if format_type == 'INVENTOR': + return cls._parse_inventor_bom(df) + else: + return cls._parse_standard_bom(df) + + except Exception as e: + raise ValueError(f"파일 파싱 실패: {str(e)}") + + @staticmethod + def _parse_standard_bom(df: pd.DataFrame) -> List[Dict]: + """기존의 퍼지 매핑 방식 파서 (표준 양식)""" + # 컬럼명 전처리 + df.columns = df.columns.str.strip().str.upper() + + # 대소문자 구분 없는 매핑을 위해 컬럼명 대문자화 사용 + column_mapping = { + 'description': ['DESCRIPTION', 'ITEM', 'MATERIAL', '품명', '자재명', 'SPECIFICATION'], + 'quantity': ['QTY', 'QUANTITY', 'EA', '수량', 'AMOUNT'], + 'main_size': ['MAIN_NOM', 'NOMINAL_DIAMETER', 'ND', '주배관', 'SIZE'], + 'red_size': ['RED_NOM', 'REDUCED_DIAMETER', '축소배관'], + 'length': ['LENGTH', 'LEN', '길이'], + 'weight': ['WEIGHT', 'WT', '중량'], + 'dwg_name': ['DWG_NAME', 'DRAWING', '도면명', '도면번호'], + 'line_num': ['LINE_NUM', 'LINE_NUMBER', '라인번호'] + } + + mapped_columns = {} + for standard_col, possible_names in column_mapping.items(): + for possible_name in possible_names: + # 대문자로 비교 + possible_upper = possible_name.upper() + if possible_upper in df.columns: + mapped_columns[standard_col] = possible_upper + break + + print(f"📋 [Standard] 컬럼 매핑 결과: {mapped_columns}") + + materials = [] + for index, row in df.iterrows(): + description = str(row.get(mapped_columns.get('description', ''), '')) + + # 제외 항목 처리 + description_upper = description.upper() + if ('WELD GAP' in description_upper or 'WELDING GAP' in description_upper or + '웰드갭' in description_upper or '용접갭' in description_upper): + continue + + # 수량 처리 + quantity_raw = row.get(mapped_columns.get('quantity', ''), 0) + try: + quantity = float(quantity_raw) if pd.notna(quantity_raw) else 0 + except: + quantity = 0 + + # 재질 등급 추출 (ASTM) + material_grade = "" + if "ASTM" in description_upper: + astm_match = re.search(r'ASTM\s+(A\d{3,4}[A-Z]*(?:\s+(?:GR\s+)?[A-Z0-9]+)?)', description_upper) + if astm_match: + material_grade = astm_match.group(0).strip() + + # 사이즈 처리 + main_size = str(row.get(mapped_columns.get('main_size', ''), '')) + red_size = str(row.get(mapped_columns.get('red_size', ''), '')) + + main_nom = main_size if main_size != 'nan' and main_size != '' else None + red_nom = red_size if red_size != 'nan' and red_size != '' else None + + if main_size != 'nan' and red_size != 'nan' and red_size != '': + size_spec = f"{main_size} x {red_size}" + elif main_size != 'nan' and main_size != '': + size_spec = main_size + else: + size_spec = "" + + # 길이 처리 + length_raw = row.get(mapped_columns.get('length', ''), '') + length_value = None + if pd.notna(length_raw) and str(length_raw).strip() != '': + try: + length_value = float(str(length_raw).strip()) + except: + length_value = None + + # 도면/라인 번호 + dwg_name = row.get(mapped_columns.get('dwg_name', ''), '') + dwg_name = str(dwg_name).strip() if pd.notna(dwg_name) and str(dwg_name).strip() not in ['', 'nan', 'None'] else None + + line_num = row.get(mapped_columns.get('line_num', ''), '') + line_num = str(line_num).strip() if pd.notna(line_num) and str(line_num).strip() not in ['', 'nan', 'None'] else None + + if description and description not in ['nan', 'None', '']: + materials.append({ + 'original_description': description, + 'quantity': quantity, + 'unit': "EA", + 'size_spec': size_spec, + 'main_nom': main_nom, + 'red_nom': red_nom, + 'material_grade': material_grade, + 'length': length_value, + 'dwg_name': dwg_name, + 'line_num': line_num, + 'line_number': index + 1, + 'row_number': index + 1 + }) + + return materials + + @staticmethod + def _parse_inventor_bom(df: pd.DataFrame) -> List[Dict]: + """ + [신규] 인벤터 추출 양식 파서 + 헤더: NO., NAME, Q'ty, LENGTH & THICKNESS, WEIGHT, DESCIPTION, REMARK + 특징: Size 컬럼 부재, NAME에 주요 정보 포함 + """ + print("⚠️ [Inventor] 인벤터 양식 파서를 사용합니다.") + + # 컬럼명 전처리 (좌우 공백 제거 및 대문자화) + df.columns = df.columns.str.strip().str.upper() + + # 인벤터 전용 매핑 + col_name = 'NAME' + col_qty = "Q'TY" + col_desc = 'DESCIPTION' # 오타 그대로 반영 + col_remark = 'REMARK' + col_length = 'LENGTH & THICKNESS' # 길이 정보가 여기 있을 수 있음 + + materials = [] + for index, row in df.iterrows(): + # 1. 품명 (NAME 컬럼 우선 사용) + name_val = str(row.get(col_name, '')).strip() + desc_val = str(row.get(col_desc, '')).strip() + + # NAME과 DESCIPTION 병합 (필요시) + # 보통 인벤터에서 NAME이 'ADAPTER OD 1/4" x NPT(F) 1/2"' 같이 상세 스펙 + # DESCIPTION은 'SS-400-7-8' 같은 자재 코드일 수 있음 + # 일단 NAME을 메인 description으로 사용하고, DESCIPTION을 괄호에 추가 + if desc_val and desc_val not in ['nan', 'None', '']: + full_description = f"{name_val} ({desc_val})" + else: + full_description = name_val + + if not full_description or full_description in ['nan', 'None', '']: + continue + + # 2. 수량 + qty_raw = row.get(col_qty, 0) + try: + quantity = float(qty_raw) if pd.notna(qty_raw) else 0 + except: + quantity = 0 + + # 3. 사이즈 추출 (NAME 컬럼 분석) + # 패턴: 1/2", 1/4", 100A, 50A, 10x20 등 + size_spec = "" + main_nom = None + red_nom = None + + # 인치/MM 사이즈 추출 시도 + # 예: "ADAPTER OD 1/4" x NPT(F) 1/2"" -> 1/4" x 1/2" + # 예: "ELBOW 90D 100A" -> 100A + + # 인치 패턴 (1/2", 3/4" 등) + inch_sizes = re.findall(r'(\d+(?:/\d+)?)"', name_val) + # A단위 패턴 (100A, 50A 등) + a_sizes = re.findall(r'(\d+)A', name_val) + + if inch_sizes: + if len(inch_sizes) >= 2: + main_nom = f'{inch_sizes[0]}"' + red_nom = f'{inch_sizes[1]}"' + size_spec = f'{main_nom} x {red_nom}' + else: + main_nom = f'{inch_sizes[0]}"' + size_spec = main_nom + elif a_sizes: + if len(a_sizes) >= 2: + main_nom = f'{a_sizes[0]}A' + red_nom = f'{a_sizes[1]}A' + size_spec = f'{main_nom} x {red_nom}' + else: + main_nom = f'{a_sizes[0]}A' + size_spec = main_nom + + # 4. 재질 정보 + material_grade = "" + # NAME이나 DESCIPTION에서 재질 키워드 찾기 (SS, SUS, A105 등) + combined_text = (full_description + " " + desc_val).upper() + if "SUS" in combined_text or "SS" in combined_text: + if "304" in combined_text: material_grade = "SUS304" + elif "316" in combined_text: material_grade = "SUS316" + else: material_grade = "SUS" + elif "A105" in combined_text: + material_grade = "A105" + + # 5. 길이 정보 + length_value = None + length_raw = row.get(col_length, '') + # 값이 있고 숫자로 변환 가능하면 사용 + if pd.notna(length_raw) and str(length_raw).strip(): + try: + # '100 mm' 등의 형식 처리 필요할 수 있음 + length_str = str(length_raw).lower().replace('mm', '').strip() + length_value = float(length_str) + except: + pass + + materials.append({ + 'original_description': full_description, + 'quantity': quantity, + 'unit': "EA", + 'size_spec': size_spec, + 'main_nom': main_nom, + 'red_nom': red_nom, + 'material_grade': material_grade, + 'length': length_value, + 'dwg_name': None, # 인벤터 파일엔 도면명 컬럼이 없음 + 'line_num': None, + 'line_number': index + 1, + 'row_number': index + 1 + }) + + return materials diff --git a/backend/app/services/fitting_classifier.py b/backend/app/services/fitting_classifier.py index 7c0a04d..27d9e0d 100644 --- a/backend/app/services/fitting_classifier.py +++ b/backend/app/services/fitting_classifier.py @@ -248,78 +248,23 @@ def classify_fitting(dat_file: str, description: str, main_nom: str, ) # 6. 최종 결과 조합 + # --- 계장용(Instrument/Swagelok) 피팅 감지 로직 추가 --- + instrument_keywords = ["SWAGELOK", "DK-LOK", "TUBE FITTING", "UNION", "FERRULE", "MALE CONNECTOR", "FEMALE CONNECTOR"] + is_instrument = any(kw in desc_upper for kw in instrument_keywords) + + if is_instrument: + fitting_type["category"] = "INSTRUMENT_FITTING" + if "SWAGELOK" in desc_upper: fitting_type["brand"] = "SWAGELOK" + + # Tube OD 추출 (예: 1/4", 6MM, 12MM) + tube_match = re.search(r'(\d+(?:/\d+)?)\s*(?:\"|INCH|MM)\s*(?:OD|TUBE)', desc_upper) + if tube_match: + fitting_type["tube_od"] = tube_match.group(0) + return { "category": "FITTING", - - # 재질 정보 (공통 모듈) - "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) - }, - - # 피팅 특화 정보 - "fitting_type": { - "type": fitting_type_result.get('type', 'UNKNOWN'), - "subtype": fitting_type_result.get('subtype', 'UNKNOWN'), - "confidence": fitting_type_result.get('confidence', 0.0), - "evidence": fitting_type_result.get('evidence', []) - }, - - "connection_method": { - "method": connection_result.get('method', 'UNKNOWN'), - "confidence": connection_result.get('confidence', 0.0), - "matched_code": connection_result.get('matched_code', ''), - "size_range": connection_result.get('size_range', ''), - "pressure_range": connection_result.get('pressure_range', '') - }, - - "pressure_rating": { - "rating": pressure_result.get('rating', 'UNKNOWN'), - "confidence": pressure_result.get('confidence', 0.0), - "max_pressure": pressure_result.get('max_pressure', ''), - "common_use": pressure_result.get('common_use', '') - }, - - "manufacturing": { - "method": manufacturing_result.get('method', 'UNKNOWN'), - "confidence": manufacturing_result.get('confidence', 0.0), - "evidence": manufacturing_result.get('evidence', []), - "characteristics": manufacturing_result.get('characteristics', '') - }, - - "size_info": { - "main_size": main_nom, - "reduced_size": red_nom, - "size_description": format_fitting_size(main_nom, red_nom), - "requires_two_sizes": fitting_type_result.get('requires_two_sizes', False) - }, - - "schedule_info": { - "schedule": schedule_result.get('schedule', 'UNKNOWN'), - "schedule_number": schedule_result.get('schedule_number', ''), - "wall_thickness": schedule_result.get('wall_thickness', ''), - "pressure_class": schedule_result.get('pressure_class', ''), - "confidence": schedule_result.get('confidence', 0.0) - }, - - # 전체 신뢰도 - "overall_confidence": calculate_fitting_confidence({ - "material": material_result.get('confidence', 0), - "fitting_type": fitting_type_result.get('confidence', 0), - "connection": connection_result.get('confidence', 0), - "pressure": pressure_result.get('confidence', 0) - }), - - # 통합분류기 호환성을 위한 confidence 필드 - "confidence": calculate_fitting_confidence({ - "material": material_result.get('confidence', 0), - "fitting_type": fitting_type_result.get('confidence', 0), - "connection": connection_result.get('confidence', 0), - "pressure": pressure_result.get('confidence', 0) - }) - } + "fitting_type": fitting_type, + def analyze_size_pattern_for_fitting_type(description: str, main_nom: str, red_nom: str = None) -> Dict: """ diff --git a/backend/app/services/integrated_classifier.py b/backend/app/services/integrated_classifier.py index f3e7628..fcc0214 100644 --- a/backend/app/services/integrated_classifier.py +++ b/backend/app/services/integrated_classifier.py @@ -13,10 +13,18 @@ LEVEL1_TYPE_KEYWORDS = { "VALVE": ["VALVE", "GATE", "BALL", "GLOBE", "CHECK", "BUTTERFLY", "NEEDLE", "RELIEF", "SIGHT GLASS", "STRAINER", "밸브", "게이트", "볼", "글로브", "체크", "버터플라이", "니들", "릴리프", "사이트글라스", "스트레이너"], "FLANGE": ["FLG", "FLANGE", "플랜지", "프랜지", "ORIFICE", "SPECTACLE", "PADDLE", "SPACER", "BLIND", "REDUCING FLANGE", "RED FLANGE"], "PIPE": ["PIPE", "TUBE", "파이프", "배관", "SMLS", "SEAMLESS"], - "FITTING": ["SOCK-O-LET", "WELD-O-LET", "ELL-O-LET", "THREAD-O-LET", "ELB-O-LET", "NIP-O-LET", "COUP-O-LET", "SOCKOLET", "WELDOLET", "ELLOLET", "THREADOLET", "ELBOLET", "NIPOLET", "COUPOLET", "OLET", "ELBOW", "ELL", "TEE", "REDUCER", "CAP", "COUPLING", "NIPPLE", "SWAGE", "PLUG", "엘보", "티", "리듀서", "캡", "니플", "커플링", "플러그", "CONC", "ECC"], + "FITTING": [ + "SOCK-O-LET", "WELD-O-LET", "ELL-O-LET", "THREAD-O-LET", "ELB-O-LET", "NIP-O-LET", "COUP-O-LET", + "SOCKOLET", "WELDOLET", "ELLOLET", "THREADOLET", "ELBOLET", "NIPOLET", "COUPOLET", "OLET", + "ELBOW", "ELL", "TEE", "REDUCER", "CAP", "COUPLING", "NIPPLE", "SWAGE", "PLUG", + "엘보", "티", "리듀서", "캡", "니플", "커플링", "플러그", "CONC", "ECC", + "SWAGELOK", "UNION", "CONNECTOR", "FERRULE", "NUT & FERRULE", "MALE CONNECTOR", "FEMALE CONNECTOR" + ], "GASKET": ["GASKET", "GASK", "가스켓", "SWG", "SPIRAL"], "INSTRUMENT": ["GAUGE", "TRANSMITTER", "SENSOR", "THERMOMETER", "계기", "게이지", "트랜스미터", "센서"], - "SUPPORT": ["URETHANE BLOCK", "URETHANE", "BLOCK SHOE", "CLAMP", "SUPPORT", "HANGER", "SPRING", "우레탄", "블록", "클램프", "서포트", "행거", "스프링"] + "SUPPORT": ["URETHANE BLOCK", "URETHANE", "BLOCK SHOE", "CLAMP", "SUPPORT", "HANGER", "SPRING", "우레탄", "블록", "클램프", "서포트", "행거", "스프링"], + "PLATE": ["PLATE", "PL", "CHECKER PLATE", "판재", "철판"], + "STRUCTURAL": ["H-BEAM", "BEAM", "ANGLE", "CHANNEL", "H-SECTION", "I-BEAM", "형강", "앵글", "채널"] } # Level 2: 서브타입 키워드 (구체화) @@ -171,6 +179,37 @@ def classify_material_integrated(description: str, main_nom: str = "", # 긴 키워드부터 확인 (FLANGE BOLT가 FLANGE보다 먼저 매칭되도록) sorted_keywords = sorted(keywords, key=len, reverse=True) for keyword in sorted_keywords: + # [강화된 로직] 짧은 키워드나 중의적 키워드에 대한 엄격한 검사 + is_strict_match = True + + # 1. "PL" 키워드 검사 (PLATE) + if keyword == "PL": + # 단독 단어이거나 숫자 뒤에 붙는 경우만 허용 (예: 10PL, 10 PL) + # COUPLING, NIPPLE, PLUG 등에 포함된 PL은 제외 + pl_pattern = r'(\b|\d)PL\b' + if not re.search(pl_pattern, desc_upper): + is_strict_match = False + + # 2. "ANGLE" 키워드 검사 (STRUCTURAL) + elif keyword == "ANGLE" or keyword == "앵글": + # VALVE와 함께 쓰이면 제외 (ANGLE VALVE) + if "VALVE" in desc_upper or "밸브" in desc_upper: + is_strict_match = False + + # 3. "UNION" 키워드 검사 (FITTING) + elif keyword == "UNION": + # 계장용인지 파이프용인지 구분은 fitting_classifier에서 하되, + # 여기서는 일단 FITTING으로 잡히도록 둠. + pass + + # 4. "BEAM" 키워드 검사 (STRUCTURAL) + elif keyword == "BEAM": + # "BEAM CLAMP" 같은 경우 SUPPORT로 가야 함 (SUPPORT가 우선순위 높으므로 괜찮음) + pass + + if not is_strict_match: + continue + # 전체 문자열에서 찾기 if keyword in desc_upper: detected_types.append((material_type, keyword)) diff --git a/backend/app/services/material_service.py b/backend/app/services/material_service.py new file mode 100644 index 0000000..d1ecccd --- /dev/null +++ b/backend/app/services/material_service.py @@ -0,0 +1,592 @@ + +from sqlalchemy.orm import Session +from sqlalchemy import text +from typing import List, Dict, Optional +import json +from datetime import datetime + +from app.services.integrated_classifier import classify_material_integrated, should_exclude_material +from app.services.bolt_classifier import classify_bolt +from app.services.flange_classifier import classify_flange +from app.services.fitting_classifier import classify_fitting +from app.services.gasket_classifier import classify_gasket +from app.services.instrument_classifier import classify_instrument +from app.services.valve_classifier import classify_valve +from app.services.support_classifier import classify_support +from app.services.plate_classifier import classify_plate +from app.services.structural_classifier import classify_structural +from app.services.pipe_classifier import classify_pipe_for_purchase, extract_end_preparation_info +from app.services.material_grade_extractor import extract_full_material_grade + +class MaterialService: + """자재 처리 및 저장을 담당하는 서비스""" + + @staticmethod + def process_and_save_materials( + db: Session, + file_id: int, + materials_data: List[Dict], + revision_comparison: Optional[Dict] = None, + parent_file_id: Optional[int] = None, + purchased_materials_map: Optional[Dict] = None + ) -> int: + """ + 자재 목록을 분류하고 DB에 저장합니다. + + Args: + db: DB 세션 + file_id: 파일 ID + materials_data: 파싱된 자재 데이터 목록 + revision_comparison: 리비전 비교 결과 + parent_file_id: 이전 리비전 파일 ID + purchased_materials_map: 구매 확정된 자재 매핑 정보 + + Returns: + 저장된 자재 수 + """ + materials_inserted = 0 + + # 변경/신규 자재 키 집합 (리비전 추적용) + changed_materials_keys = set() + new_materials_keys = set() + + # 리비전 업로드인 경우 변경사항 분석 + if parent_file_id is not None: + MaterialService._analyze_changes( + db, parent_file_id, materials_data, + changed_materials_keys, new_materials_keys + ) + + # 변경 없는 자재 (확정된 자재) 먼저 처리 + if revision_comparison and revision_comparison.get("has_previous_confirmation", False): + unchanged_materials = revision_comparison.get("unchanged_materials", []) + for material_data in unchanged_materials: + MaterialService._save_unchanged_material(db, file_id, material_data) + materials_inserted += 1 + + # 분류가 필요한 자재 처리 (신규 또는 변경된 자재) + # revision_comparison에서 필터링된 목록이 있으면 그것을 사용, 아니면 전체 + materials_to_classify = materials_data + if revision_comparison and revision_comparison.get("materials_to_classify"): + materials_to_classify = revision_comparison.get("materials_to_classify") + + print(f"🔧 자재 분류 및 저장 시작: {len(materials_to_classify)}개") + + for material_data in materials_to_classify: + MaterialService._classify_and_save_single_material( + db, file_id, material_data, + changed_materials_keys, new_materials_keys, + purchased_materials_map + ) + materials_inserted += 1 + + return materials_inserted + + @staticmethod + def _analyze_changes(db: Session, parent_file_id: int, materials_data: List[Dict], + changed_keys: set, new_keys: set): + """이전 리비전과 비교하여 변경/신규 자재를 식별합니다.""" + try: + prev_materials_query = text(""" + SELECT original_description, size_spec, material_grade, main_nom, + drawing_name, line_no, quantity + FROM materials + WHERE file_id = :parent_file_id + """) + prev_materials = db.execute(prev_materials_query, {"parent_file_id": parent_file_id}).fetchall() + + prev_dict = {} + for pm in prev_materials: + key = MaterialService._generate_material_key( + pm.drawing_name, pm.line_no, pm.original_description, + pm.size_spec, pm.material_grade + ) + prev_dict[key] = float(pm.quantity) if pm.quantity else 0 + + for mat in materials_data: + new_key = MaterialService._generate_material_key( + mat.get("dwg_name"), mat.get("line_num"), mat["original_description"], + mat.get("size_spec"), mat.get("material_grade") + ) + + if new_key in prev_dict: + if abs(prev_dict[new_key] - float(mat.get("quantity", 0))) > 0.001: + changed_keys.add(new_key) + else: + new_keys.add(new_key) + + except Exception as e: + print(f"❌ 변경사항 분석 실패: {e}") + + @staticmethod + def _generate_material_key(dwg, line, desc, size, grade): + """자재 고유 키 생성""" + parts = [] + if dwg: parts.append(str(dwg)) + elif line: parts.append(str(line)) + + parts.append(str(desc)) + parts.append(str(size or '')) + parts.append(str(grade or '')) + return "|".join(parts) + + @staticmethod + def _save_unchanged_material(db: Session, file_id: int, material_data: Dict): + """변경 없는(확정된) 자재 저장""" + previous_item = material_data.get("previous_item", {}) + + query = text(""" + INSERT INTO materials ( + file_id, original_description, classified_category, confidence, + quantity, unit, size_spec, material_grade, specification, + reused_from_confirmation, created_at + ) VALUES ( + :file_id, :desc, :category, 1.0, + :qty, :unit, :size, :grade, :spec, + TRUE, :created_at + ) + """) + + db.execute(query, { + "file_id": file_id, + "desc": material_data["original_description"], + "category": previous_item.get("category", "UNCLASSIFIED"), + "qty": material_data["quantity"], + "unit": material_data.get("unit", "EA"), + "size": material_data.get("size_spec", ""), + "grade": previous_item.get("material", ""), + "spec": previous_item.get("specification", ""), + "created_at": datetime.now() + }) + + @staticmethod + def _classify_and_save_single_material( + db: Session, file_id: int, material_data: Dict, + changed_keys: set, new_keys: set, purchased_map: Optional[Dict] + ): + """단일 자재 분류 및 저장 (상세 정보 포함)""" + description = material_data["original_description"] + main_nom = material_data.get("main_nom", "") + red_nom = material_data.get("red_nom", "") + length_val = material_data.get("length") + + # 1. 통합 분류 + integrated_result = classify_material_integrated(description, main_nom, red_nom, length_val) + classification_result = integrated_result + + # 2. 상세 분류 + if not should_exclude_material(description): + category = integrated_result.get('category') + if category == "PIPE": + classification_result = classify_pipe_for_purchase("", description, main_nom, length_val) + elif category == "FITTING": + classification_result = classify_fitting("", description, main_nom, red_nom) + elif category == "FLANGE": + classification_result = classify_flange("", description, main_nom, red_nom) + elif category == "VALVE": + classification_result = classify_valve("", description, main_nom) + elif category == "BOLT": + classification_result = classify_bolt("", description, main_nom) + elif category == "GASKET": + classification_result = classify_gasket("", description, main_nom) + elif category == "INSTRUMENT": + classification_result = classify_instrument("", description, main_nom) + elif category == "SUPPORT": + classification_result = classify_support("", description, main_nom) + elif category == "PLATE": + classification_result = classify_plate("", description, main_nom) + elif category == "STRUCTURAL": + classification_result = classify_structural("", description, main_nom) + + # 신뢰도 조정 + if integrated_result.get('confidence', 0) < 0.5: + classification_result['overall_confidence'] = min( + classification_result.get('overall_confidence', 1.0), + integrated_result.get('confidence', 0.0) + 0.2 + ) + else: + classification_result = {"category": "EXCLUDE", "overall_confidence": 0.95} + + # 3. 구매 확정 정보 상속 확인 + is_purchase_confirmed = False + purchase_confirmed_at = None + purchase_confirmed_by = None + + if purchased_map: + key = f"{description.strip().upper()}|{material_data.get('size_spec', '')}" + if key in purchased_map: + info = purchased_map[key] + is_purchase_confirmed = True + purchase_confirmed_at = info.get("purchase_confirmed_at") + purchase_confirmed_by = info.get("purchase_confirmed_by") + + # 4. 자재 기본 정보 저장 + full_grade = extract_full_material_grade(description) or material_data.get("material_grade", "") + + insert_query = text(""" + INSERT INTO materials ( + file_id, original_description, quantity, unit, size_spec, + main_nom, red_nom, material_grade, full_material_grade, line_number, row_number, + classified_category, classification_confidence, is_verified, + drawing_name, line_no, created_at, + purchase_confirmed, purchase_confirmed_at, purchase_confirmed_by, + revision_status + ) VALUES ( + :file_id, :desc, :qty, :unit, :size, + :main, :red, :grade, :full_grade, :line_num, :row_num, + :category, :confidence, :verified, + :dwg, :line, :created_at, + :confirmed, :confirmed_at, :confirmed_by, + :status + ) RETURNING id + """) + + # 리비전 상태 결정 + mat_key = MaterialService._generate_material_key( + material_data.get("dwg_name"), material_data.get("line_num"), description, + material_data.get("size_spec"), material_data.get("material_grade") + ) + rev_status = 'changed' if mat_key in changed_keys else ('active' if mat_key in new_keys else None) + + result = db.execute(insert_query, { + "file_id": file_id, + "desc": description, + "qty": material_data["quantity"], + "unit": material_data["unit"], + "size": material_data.get("size_spec", ""), + "main": main_nom, + "red": red_nom, + "grade": material_data.get("material_grade", ""), + "full_grade": full_grade, + "line_num": material_data.get("line_number"), + "row_num": material_data.get("row_number"), + "category": classification_result.get("category", "UNCLASSIFIED"), + "confidence": classification_result.get("overall_confidence", 0.0), + "verified": False, + "dwg": material_data.get("dwg_name"), + "line": material_data.get("line_num"), + "created_at": datetime.now(), + "confirmed": is_purchase_confirmed, + "confirmed_at": purchase_confirmed_at, + "confirmed_by": purchase_confirmed_by, + "status": rev_status + }) + + material_id = result.fetchone()[0] + + # 5. 상세 정보 저장 (별도 메서드로 분리) + MaterialService._save_material_details( + db, material_id, file_id, classification_result, material_data + ) + + @staticmethod + def _save_material_details(db: Session, material_id: int, file_id: int, + result: Dict, data: Dict): + """카테고리별 상세 정보 저장""" + category = result.get("category") + + if category == "PIPE": + MaterialService._save_pipe_details(db, material_id, file_id, result, data) + elif category == "FITTING": + MaterialService._save_fitting_details(db, material_id, file_id, result, data) + elif category == "FLANGE": + MaterialService._save_flange_details(db, material_id, file_id, result, data) + elif category == "BOLT": + MaterialService._save_bolt_details(db, material_id, file_id, result, data) + elif category == "VALVE": + MaterialService._save_valve_details(db, material_id, file_id, result, data) + elif category == "GASKET": + MaterialService._save_gasket_details(db, material_id, file_id, result, data) + elif category == "SUPPORT": + MaterialService._save_support_details(db, material_id, file_id, result, data) + elif category == "PLATE": + MaterialService._save_plate_details(db, material_id, file_id, result, data) + elif category == "STRUCTURAL": + MaterialService._save_structural_details(db, material_id, file_id, result, data) + + @staticmethod + def _save_plate_details(db: Session, mid: int, fid: int, res: Dict, data: Dict): + """판재 정보 업데이트 (현재는 materials 테이블에 요약 정보 저장)""" + details = res.get("details", {}) + spec = f"{details.get('thickness')}T x {details.get('dimensions')}" + db.execute(text(""" + UPDATE materials + SET size_spec = :size, material_grade = :mat + WHERE id = :id + """), {"size": spec, "mat": details.get("material"), "id": mid}) + + @staticmethod + def _save_structural_details(db: Session, mid: int, fid: int, res: Dict, data: Dict): + """형강 정보 업데이트 (현재는 materials 테이블에 요약 정보 저장)""" + details = res.get("details", {}) + spec = f"{details.get('type')} {details.get('dimension')}" + db.execute(text(""" + UPDATE materials + SET size_spec = :size + WHERE id = :id + """), {"size": spec, "id": mid}) + + # --- 각 카테고리별 상세 저장 메서드들 (기존 로직 이관) --- + + @staticmethod + def inherit_purchase_requests(db: Session, current_file_id: int, parent_file_id: int): + """이전 리비전의 구매신청 정보를 상속합니다.""" + try: + print(f"🔄 구매신청 정보 상속 처리 시작...") + + # 1. 이전 리비전에서 그룹별 구매신청 수량 집계 + prev_purchase_summary = text(""" + SELECT + m.original_description, + m.size_spec, + m.material_grade, + m.drawing_name, + COUNT(DISTINCT pri.material_id) as purchased_count, + SUM(pri.quantity) as total_purchased_qty, + MIN(pri.request_id) as request_id + FROM materials m + JOIN purchase_request_items pri ON m.id = pri.material_id + WHERE m.file_id = :parent_file_id + GROUP BY m.original_description, m.size_spec, m.material_grade, m.drawing_name + """) + + prev_purchases = db.execute(prev_purchase_summary, {"parent_file_id": parent_file_id}).fetchall() + + # 2. 새 리비전에서 같은 그룹의 자재에 수량만큼 구매신청 상속 + for prev_purchase in prev_purchases: + purchased_count = prev_purchase.purchased_count + + # 새 리비전에서 같은 그룹의 자재 조회 (순서대로) + new_group_materials = text(""" + SELECT id, quantity + FROM materials + WHERE file_id = :file_id + AND original_description = :description + AND COALESCE(size_spec, '') = :size_spec + AND COALESCE(material_grade, '') = :material_grade + AND COALESCE(drawing_name, '') = :drawing_name + ORDER BY id + LIMIT :limit + """) + + new_materials = db.execute(new_group_materials, { + "file_id": current_file_id, + "description": prev_purchase.original_description, + "size_spec": prev_purchase.size_spec or '', + "material_grade": prev_purchase.material_grade or '', + "drawing_name": prev_purchase.drawing_name or '', + "limit": purchased_count + }).fetchall() + + # 구매신청 수량만큼만 상속 + for new_mat in new_materials: + inherit_query = text(""" + INSERT INTO purchase_request_items ( + request_id, material_id, quantity, unit, user_requirement + ) VALUES ( + :request_id, :material_id, :quantity, 'EA', '' + ) + ON CONFLICT DO NOTHING + """) + db.execute(inherit_query, { + "request_id": prev_purchase.request_id, + "material_id": new_mat.id, + "quantity": new_mat.quantity + }) + + inherited_count = len(new_materials) + if inherited_count > 0: + print(f" ✅ {prev_purchase.original_description[:30]}... (도면: {prev_purchase.drawing_name or 'N/A'}) → {inherited_count}/{purchased_count}개 상속") + + # 커밋은 호출하는 쪽에서 일괄 처리하거나 여기서 처리 + # db.commit() + print(f"✅ 구매신청 정보 상속 완료") + + except Exception as e: + print(f"❌ 구매신청 정보 상속 실패: {str(e)}") + # 상속 실패는 전체 프로세스를 중단하지 않음 + + @staticmethod + def _save_pipe_details(db, mid, fid, res, data): + # PIPE 상세 저장 로직 + end_prep_info = extract_end_preparation_info(data["original_description"]) + + # 1. End Prep 정보 저장 + db.execute(text(""" + INSERT INTO pipe_end_preparations ( + material_id, file_id, end_preparation_type, end_preparation_code, + machining_required, cutting_note, original_description, confidence + ) VALUES ( + :mid, :fid, :type, :code, :req, :note, :desc, :conf + ) + """), { + "mid": mid, "fid": fid, + "type": end_prep_info["end_preparation_type"], + "code": end_prep_info["end_preparation_code"], + "req": end_prep_info["machining_required"], + "note": end_prep_info["cutting_note"], + "desc": end_prep_info["original_description"], + "conf": end_prep_info["confidence"] + }) + + # 2. Pipe Details 저장 + length_info = res.get("length_info", {}) + length_mm = length_info.get("length_mm") or data.get("length", 0.0) + + mat_info = res.get("material", {}) + sch_info = res.get("schedule", {}) + + # 재질 정보 업데이트 + if mat_info.get("grade"): + db.execute(text("UPDATE materials SET material_grade = :g WHERE id = :id"), + {"g": mat_info.get("grade"), "id": mid}) + + db.execute(text(""" + INSERT INTO pipe_details ( + material_id, file_id, outer_diameter, schedule, + material_spec, manufacturing_method, length_mm + ) VALUES (:mid, :fid, :od, :sch, :spec, :method, :len) + """), { + "mid": mid, "fid": fid, + "od": data.get("main_nom") or data.get("size_spec"), + "sch": sch_info.get("schedule", "UNKNOWN") if isinstance(sch_info, dict) else str(sch_info), + "spec": mat_info.get("grade", "") if isinstance(mat_info, dict) else "", + "method": res.get("manufacturing", {}).get("method", "UNKNOWN") if isinstance(res.get("manufacturing"), dict) else "UNKNOWN", + "len": length_mm or 0.0 + }) + + @staticmethod + def _save_fitting_details(db, mid, fid, res, data): + fit_type = res.get("fitting_type", {}) + mat_info = res.get("material", {}) + + if mat_info.get("grade"): + db.execute(text("UPDATE materials SET material_grade = :g WHERE id = :id"), + {"g": mat_info.get("grade"), "id": mid}) + + db.execute(text(""" + INSERT INTO fitting_details ( + material_id, file_id, fitting_type, fitting_subtype, + connection_method, pressure_rating, material_grade, + main_size, reduced_size + ) VALUES (:mid, :fid, :type, :subtype, :conn, :rating, :grade, :main, :red) + """), { + "mid": mid, "fid": fid, + "type": fit_type.get("type", "UNKNOWN") if isinstance(fit_type, dict) else str(fit_type), + "subtype": fit_type.get("subtype", "UNKNOWN") if isinstance(fit_type, dict) else "UNKNOWN", + "conn": res.get("connection_method", {}).get("method", "UNKNOWN") if isinstance(res.get("connection_method"), dict) else "UNKNOWN", + "rating": res.get("pressure_rating", {}).get("rating", "UNKNOWN") if isinstance(res.get("pressure_rating"), dict) else "UNKNOWN", + "grade": mat_info.get("grade", "") if isinstance(mat_info, dict) else "", + "main": data.get("main_nom") or data.get("size_spec"), + "red": data.get("red_nom", "") + }) + + @staticmethod + def _save_flange_details(db, mid, fid, res, data): + flg_type = res.get("flange_type", {}) + mat_info = res.get("material", {}) + + if mat_info.get("grade"): + db.execute(text("UPDATE materials SET material_grade = :g WHERE id = :id"), + {"g": mat_info.get("grade"), "id": mid}) + + db.execute(text(""" + INSERT INTO flange_details ( + material_id, file_id, flange_type, pressure_rating, + facing_type, material_grade, size_inches + ) VALUES (:mid, :fid, :type, :rating, :face, :grade, :size) + """), { + "mid": mid, "fid": fid, + "type": flg_type.get("type", "UNKNOWN") if isinstance(flg_type, dict) else str(flg_type), + "rating": res.get("pressure_rating", {}).get("rating", "UNKNOWN") if isinstance(res.get("pressure_rating"), dict) else "UNKNOWN", + "face": res.get("face_finish", {}).get("finish", "UNKNOWN") if isinstance(res.get("face_finish"), dict) else "UNKNOWN", + "grade": mat_info.get("grade", "") if isinstance(mat_info, dict) else "", + "size": data.get("main_nom") or data.get("size_spec") + }) + + @staticmethod + def _save_bolt_details(db, mid, fid, res, data): + fast_type = res.get("fastener_type", {}) + mat_info = res.get("material", {}) + dim_info = res.get("dimensions", {}) + + if mat_info.get("grade"): + db.execute(text("UPDATE materials SET material_grade = :g WHERE id = :id"), + {"g": mat_info.get("grade"), "id": mid}) + + # 볼트 타입 결정 (특수 용도 고려) + bolt_type = fast_type.get("type", "UNKNOWN") if isinstance(fast_type, dict) else str(fast_type) + special_apps = res.get("special_applications", {}).get("detected_applications", []) + if "LT" in special_apps: bolt_type = "LT_BOLT" + elif "PSV" in special_apps: bolt_type = "PSV_BOLT" + + # 코팅 타입 + desc_upper = data["original_description"].upper() + coating = "UNKNOWN" + if "GALV" in desc_upper: coating = "GALVANIZED" + elif "ZINC" in desc_upper: coating = "ZINC_PLATED" + + db.execute(text(""" + INSERT INTO bolt_details ( + material_id, file_id, bolt_type, thread_type, + diameter, length, material_grade, coating_type + ) VALUES (:mid, :fid, :type, :thread, :dia, :len, :grade, :coating) + """), { + "mid": mid, "fid": fid, + "type": bolt_type, + "thread": res.get("thread_specification", {}).get("standard", "UNKNOWN") if isinstance(res.get("thread_specification"), dict) else "UNKNOWN", + "dia": dim_info.get("nominal_size", data.get("main_nom", "")), + "len": dim_info.get("length", ""), + "grade": mat_info.get("grade", "") if isinstance(mat_info, dict) else "", + "coating": coating + }) + + @staticmethod + def _save_valve_details(db, mid, fid, res, data): + val_type = res.get("valve_type", {}) + mat_info = res.get("material", {}) + + if mat_info.get("grade"): + db.execute(text("UPDATE materials SET material_grade = :g WHERE id = :id"), + {"g": mat_info.get("grade"), "id": mid}) + + db.execute(text(""" + INSERT INTO valve_details ( + material_id, file_id, valve_type, connection_method, + pressure_rating, body_material, size_inches + ) VALUES (:mid, :fid, :type, :conn, :rating, :body, :size) + """), { + "mid": mid, "fid": fid, + "type": val_type.get("type", "UNKNOWN") if isinstance(val_type, dict) else str(val_type), + "conn": res.get("connection_method", {}).get("method", "UNKNOWN") if isinstance(res.get("connection_method"), dict) else "UNKNOWN", + "rating": res.get("pressure_rating", {}).get("rating", "UNKNOWN") if isinstance(res.get("pressure_rating"), dict) else "UNKNOWN", + "body": mat_info.get("grade", "") if isinstance(mat_info, dict) else "", + "size": data.get("main_nom") or data.get("size_spec") + }) + + @staticmethod + def _save_gasket_details(db, mid, fid, res, data): + gask_type = res.get("gasket_type", {}) + + db.execute(text(""" + INSERT INTO gasket_details ( + material_id, file_id, gasket_type, pressure_rating, size_inches + ) VALUES (:mid, :fid, :type, :rating, :size) + """), { + "mid": mid, "fid": fid, + "type": gask_type.get("type", "UNKNOWN") if isinstance(gask_type, dict) else str(gask_type), + "rating": res.get("pressure_rating", {}).get("rating", "UNKNOWN") if isinstance(res.get("pressure_rating"), dict) else "UNKNOWN", + "size": data.get("main_nom") or data.get("size_spec") + }) + + @staticmethod + def _save_support_details(db, mid, fid, res, data): + db.execute(text(""" + INSERT INTO support_details ( + material_id, file_id, support_type, pipe_size + ) VALUES (:mid, :fid, :type, :size) + """), { + "mid": mid, "fid": fid, + "type": res.get("support_type", "UNKNOWN"), + "size": res.get("size_info", {}).get("pipe_size", "") + }) diff --git a/backend/app/services/plate_classifier.py b/backend/app/services/plate_classifier.py new file mode 100644 index 0000000..098e8db --- /dev/null +++ b/backend/app/services/plate_classifier.py @@ -0,0 +1,50 @@ + +import re +from typing import Dict, Optional + +def classify_plate(tag: str, description: str, main_nom: str = "") -> Dict: + """ + 판재(PLATE) 분류기 + 규격 예: PLATE 10T x 1219 x 2438 + """ + desc_upper = description.upper() + + # 1. 두께(Thickness) 추출 + # 패턴: 10T, 10.5T, THK 10, THK. 10, t=10 + thickness = None + t_match = re.search(r'(\d+(?:\.\d+)?)\s*T\b', desc_upper) + if not t_match: + t_match = re.search(r'(?:THK\.?|t=)\s*(\d+(?:\.\d+)?)', desc_upper, re.IGNORECASE) + + if t_match: + thickness = t_match.group(1) + + # 2. 규격(Dimensions) 추출 + # 패턴: 1219x2438, 4'x8', 1000*2000 + dimensions = "" + dim_match = re.search(r'(\d+(?:\.\d+)?)\s*[X\*]\s*(\d+(?:\.\d+)?)(?:\s*[X\*]\s*(\d+(?:\.\d+)?))?', desc_upper) + if dim_match: + groups = [g for g in dim_match.groups() if g] + dimensions = " x ".join(groups) + + # 3. 재질 추출 + material = "UNKNOWN" + # 압력용기용 및 일반 구조용 강판 재질 추가 + plate_materials = [ + "SUS304", "SUS316", "SUS321", "SS400", "A36", "SM490", + "SA516", "A516", "SA283", "A283", "SA537", "A537", "POS-M" + ] + for mat in plate_materials: + if mat in desc_upper: + material = mat + break + + return { + "category": "PLATE", + "overall_confidence": 0.9, + "details": { + "thickness": thickness, + "dimensions": dimensions, + "material": material + } + } diff --git a/backend/app/services/structural_classifier.py b/backend/app/services/structural_classifier.py new file mode 100644 index 0000000..be601bc --- /dev/null +++ b/backend/app/services/structural_classifier.py @@ -0,0 +1,34 @@ + +import re +from typing import Dict + +def classify_structural(tag: str, description: str, main_nom: str = "") -> Dict: + """ + 형강(STRUCTURAL) 분류기 + 규격 예: H-BEAM 100x100x6x8 + """ + desc_upper = description.upper() + + # 1. 타입 식별 + struct_type = "UNKNOWN" + if "H-BEAM" in desc_upper or "H-SECTION" in desc_upper: struct_type = "H-BEAM" + elif "ANGLE" in desc_upper or "L-SECTION" in desc_upper: struct_type = "ANGLE" + elif "CHANNEL" in desc_upper or "C-SECTION" in desc_upper: struct_type = "CHANNEL" + elif "BEAM" in desc_upper: struct_type = "I-BEAM" + + # 2. 규격 추출 + # 패턴: 100x100, 100x100x6x8, 50x50x5T, H-200x200 + dimension = "" + # 숫자와 x 또는 *로 구성된 패턴, 앞에 H- 등이 붙을 수 있음 + dim_match = re.search(r'(?:H-|L-|C-)?(\d+(?:\.\d+)?(?:\s*[X\*]\s*\d+(?:\.\d+)?){1,3})', desc_upper) + if dim_match: + dimension = dim_match.group(1).replace("*", "x") + + return { + "category": "STRUCTURAL", + "overall_confidence": 0.9, + "details": { + "type": struct_type, + "dimension": dimension + } + }