""" 리비전 비교 서비스 - 기존 확정 자재와 신규 자재 비교 - 변경된 자재만 분류 처리 - 리비전 업로드 최적화 """ from sqlalchemy.orm import Session from sqlalchemy import text from typing import List, Dict, Tuple, Optional import hashlib import logging logger = logging.getLogger(__name__) class RevisionComparator: """리비전 비교 및 차이 분석 클래스""" def __init__(self, db: Session): self.db = db def get_previous_confirmed_materials(self, job_no: str, current_revision: str) -> Optional[Dict]: """ 이전 확정된 자재 목록 조회 Args: job_no: 프로젝트 번호 current_revision: 현재 리비전 (예: Rev.1) Returns: 확정된 자재 정보 딕셔너리 또는 None """ try: # 현재 리비전 번호 추출 current_rev_num = self._extract_revision_number(current_revision) # 이전 리비전들 중 확정된 것 찾기 (역순으로 검색) for prev_rev_num in range(current_rev_num - 1, -1, -1): prev_revision = f"Rev.{prev_rev_num}" # 해당 리비전의 확정 데이터 조회 query = text(""" SELECT pc.id, pc.revision, pc.confirmed_at, pc.confirmed_by, COUNT(cpi.id) as confirmed_items_count FROM purchase_confirmations pc LEFT JOIN confirmed_purchase_items cpi ON pc.id = cpi.confirmation_id WHERE pc.job_no = :job_no AND pc.revision = :revision AND pc.is_active = TRUE GROUP BY pc.id, pc.revision, pc.confirmed_at, pc.confirmed_by ORDER BY pc.confirmed_at DESC LIMIT 1 """) result = self.db.execute(query, { "job_no": job_no, "revision": prev_revision }).fetchone() if result and result.confirmed_items_count > 0: logger.info(f"이전 확정 자료 발견: {job_no} {prev_revision} ({result.confirmed_items_count}개 품목)") # 확정된 품목들 상세 조회 items_query = text(""" SELECT cpi.item_code, cpi.category, cpi.specification, cpi.size, cpi.material, cpi.bom_quantity, cpi.calculated_qty, cpi.unit, cpi.safety_factor FROM confirmed_purchase_items cpi WHERE cpi.confirmation_id = :confirmation_id ORDER BY cpi.category, cpi.specification """) items_result = self.db.execute(items_query, { "confirmation_id": result.id }).fetchall() return { "confirmation_id": result.id, "revision": result.revision, "confirmed_at": result.confirmed_at, "confirmed_by": result.confirmed_by, "items": [dict(item) for item in items_result], "items_count": len(items_result) } logger.info(f"이전 확정 자료 없음: {job_no} (현재: {current_revision})") return None except Exception as e: logger.error(f"이전 확정 자료 조회 실패: {str(e)}") return None def compare_materials(self, previous_confirmed: Dict, new_materials: List[Dict]) -> Dict: """ 기존 확정 자재와 신규 자재 비교 """ try: from rapidfuzz import fuzz # 이전 확정 자재 해시맵 생성 confirmed_materials = {} for item in previous_confirmed["items"]: material_hash = self._generate_material_hash( item["specification"], item["size"], item["material"] ) confirmed_materials[material_hash] = item # 해시 역참조 맵 (유사도 비교용) # 해시 -> 정규화된 설명 문자열 (비교 대상) # 여기서는 specification 자체를 비교 대상으로 사용 (가장 정보량이 많음) confirmed_specs = { h: item["specification"] for h, item in confirmed_materials.items() } # 신규 자재 분석 unchanged_materials = [] changed_materials = [] new_materials_list = [] for new_material in new_materials: description = new_material.get("description", "") size = self._extract_size_from_description(description) material = self._extract_material_from_description(description) material_hash = self._generate_material_hash(description, size, material) if material_hash in confirmed_materials: # 정확히 일치하는 자재 발견 (해시 일치) confirmed_item = confirmed_materials[material_hash] new_qty = float(new_material.get("quantity", 0)) confirmed_qty = float(confirmed_item["bom_quantity"]) if abs(new_qty - confirmed_qty) > 0.001: changed_materials.append({ **new_material, "change_type": "QUANTITY_CHANGED", "previous_quantity": confirmed_qty, "previous_item": confirmed_item }) else: unchanged_materials.append({ **new_material, "reuse_classification": True, "previous_item": confirmed_item }) else: # 해시 불일치 - 유사도 검사 (Fuzzy Matching) # 신규 자재 설명과 기존 확정 자재들의 스펙 비교 best_match_hash = None best_match_score = 0 # 성능을 위해 간단한 필터링 후 정밀 비교 권장되나, # 현재는 전체 비교 (데이터량이 많지 않다고 가정) for h, spec in confirmed_specs.items(): score = fuzz.ratio(description.lower(), spec.lower()) if score > 85: # 85점 이상이면 매우 유사 if score > best_match_score: best_match_score = score best_match_hash = h if best_match_hash: # 유사한 자재 발견 (오타 또는 미세 변경 가능성) similar_item = confirmed_materials[best_match_hash] new_materials_list.append({ **new_material, "change_type": "NEW_BUT_SIMILAR", "similarity_score": best_match_score, "similar_to": similar_item }) else: # 완전히 새로운 자재 new_materials_list.append({ **new_material, "change_type": "NEW_MATERIAL" }) # 삭제된 자재 찾기 new_material_hashes = set() for material in new_materials: d = material.get("description", "") s = self._extract_size_from_description(d) m = self._extract_material_from_description(d) new_material_hashes.add(self._generate_material_hash(d, s, m)) removed_materials = [] for hash_key, confirmed_item in confirmed_materials.items(): if hash_key not in new_material_hashes: removed_materials.append({ "change_type": "REMOVED", "previous_item": confirmed_item }) comparison_result = { "has_previous_confirmation": True, "previous_revision": previous_confirmed["revision"], "previous_confirmed_at": previous_confirmed["confirmed_at"], "unchanged_count": len(unchanged_materials), "changed_count": len(changed_materials), "new_count": len(new_materials_list), "removed_count": len(removed_materials), "total_materials": len(new_materials), "classification_needed": len(changed_materials) + len(new_materials_list), "unchanged_materials": unchanged_materials, "changed_materials": changed_materials, "new_materials": new_materials_list, "removed_materials": removed_materials } logger.info(f"리비전 비교 완료 (Fuzzy 적용): 변경없음 {len(unchanged_materials)}, " f"변경됨 {len(changed_materials)}, 신규 {len(new_materials_list)}, " f"삭제됨 {len(removed_materials)}") return comparison_result except Exception as e: logger.error(f"자재 비교 실패: {str(e)}") raise def _extract_revision_number(self, revision: str) -> int: """리비전 문자열에서 숫자 추출 (Rev.1 → 1)""" try: if revision.startswith("Rev."): return int(revision.replace("Rev.", "")) return 0 except ValueError: return 0 def _generate_material_hash(self, description: str, size: str, material: str) -> str: """ 자재 고유성 판단을 위한 해시 생성 Args: description: 자재 설명 size: 자재 규격/크기 material: 자재 재질 Returns: MD5 해시 문자열 """ import re def normalize(s: Optional[str]) -> str: if s is None: return "" # 다중 공백을 단일 공백으로 치환하고 앞뒤 공백 제거 s = re.sub(r'\s+', ' ', str(s)) return s.strip().lower() # 각 컴포넌트 정규화 d_norm = normalize(description) s_norm = normalize(size) m_norm = normalize(material) # RULES.md의 코딩 컨벤션 준수 (pipe separator 사용) # 값이 없는 경우에도 구분자를 포함하여 구조 유지 (예: "desc||mat") hash_input = f"{d_norm}|{s_norm}|{m_norm}" return hashlib.md5(hash_input.encode()).hexdigest() def _extract_size_from_description(self, description: str) -> str: """ 자재 설명에서 사이즈 정보 추출 지원하는 패턴 (단어 경계 \b 추가하여 정확도 향상): - 1/2" (인치) - 100A (A단위) - 50mm (밀리미터) - 10x20 (가로x세로) - DN100 (DN단위) """ if not description: return "" import re size_patterns = [ # 인치 패턴 (분수 포함): 1/2", 1.5", 1-1/2" r'\b(\d+(?:[-/.]\d+)?)\s*(?:inch|인치|")', # 밀리미터 패턴: 100mm, 100.5 MM r'\b(\d+(?:\.\d+)?)\s*(?:mm|MM)\b', # A단위 패턴: 100A, 100 A r'\b(\d+)\s*A\b', # DN단위 패턴: DN100, DN 100 r'DN\s*(\d+)\b', # 치수 패턴: 10x20, 10*20 r'\b(\d+(?:\.\d+)?)\s*[xX*]\s*(\d+(?:\.\d+)?)\b' ] for pattern in size_patterns: match = re.search(pattern, description, re.IGNORECASE) if match: return match.group(0).strip() return "" def _load_materials_from_db(self) -> List[str]: """DB에서 자재 목록 동적 로딩 (캐싱 적용 고려 가능)""" try: # MaterialSpecification 및 SpecialMaterial 테이블에서 자재 코드 조회 query = text(""" SELECT spec_code FROM material_specifications WHERE is_active = TRUE UNION SELECT grade_code FROM material_grades WHERE is_active = TRUE UNION SELECT material_name FROM special_materials WHERE is_active = TRUE """) result = self.db.execute(query).fetchall() db_materials = [row[0] for row in result] # 기본 하드코딩 리스트 (DB 조회 실패 시 또는 보완용) default_materials = [ "SUS316L", "SUS316", "SUS304L", "SUS304", "SS316L", "SS316", "SS304L", "SS304", "A105N", "A105", "A234 WPB", "A234", "A106 Gr.B", "A106", "WCB", "CF8M", "CF8", "CS", "STS", "PVC", "PP", "PE" ] # 합치고 중복 제거 후 길이 역순 정렬 (긴 단어 우선 매칭) combined = list(set(db_materials + default_materials)) combined.sort(key=len, reverse=True) return combined except Exception as e: logger.warning(f"DB 자재 로딩 실패 (기본값 사용): {str(e)}") materials = [ "SUS316L", "SUS316", "SUS304L", "SUS304", "SS316L", "SS316", "SS304L", "SS304", "A105N", "A105", "A234 WPB", "A234", "A106 Gr.B", "A106", "WCB", "CF8M", "CF8", "CS", "STS", "PVC", "PP", "PE" ] return materials def _extract_material_from_description(self, description: str) -> str: """ 자재 설명에서 재질 정보 추출 우선순위에 따라 매칭 (구체적인 재질 먼저) """ if not description: return "" # 자재 목록 로딩 (메모리 캐싱을 위해 클래스 속성으로 저장 고려 가능하지만 여기선 매번 호출로 단순화) # 성능이 중요하다면 __init__ 시점에 로드하거나 lru_cache 사용 권장 materials = self._load_materials_from_db() description_upper = description.upper() for material in materials: # 단어 매칭을 위해 간단한 검사 수행 (부분 문자열이 다른 단어의 일부가 아닌지) if material.upper() in description_upper: return material return "" def get_revision_comparison(db: Session, job_no: str, current_revision: str, new_materials: List[Dict]) -> Dict: """ 리비전 비교 수행 (편의 함수) Args: db: 데이터베이스 세션 job_no: 프로젝트 번호 current_revision: 현재 리비전 new_materials: 신규 자재 목록 Returns: 비교 결과 또는 전체 분류 필요 정보 """ comparator = RevisionComparator(db) # 이전 확정 자료 조회 previous_confirmed = comparator.get_previous_confirmed_materials(job_no, current_revision) if previous_confirmed is None: # 이전 확정 자료가 없으면 전체 분류 필요 return { "has_previous_confirmation": False, "classification_needed": len(new_materials), "all_materials_need_classification": True, "materials_to_classify": new_materials, "message": "이전 확정 자료가 없어 전체 자재를 분류합니다." } # 이전 확정 자료가 있으면 비교 수행 return comparator.compare_materials(previous_confirmed, new_materials)