""" 간단한 리비전 관리 서비스 복잡한 dependency 없이 핵심 리비전 로직만 구현 """ from sqlalchemy.orm import Session from sqlalchemy import and_, or_ from typing import Dict, List, Any, Optional, Tuple import json from datetime import datetime from ..models import Material, File, SimpleRevisionComparison, SimpleRevisionMaterial from ..utils.logger import get_logger logger = get_logger(__name__) class SimpleRevisionService: """간단한 리비전 관리 서비스""" def __init__(self, db: Session): self.db = db def compare_revisions(self, current_file_id: int, previous_file_id: int, category: str, username: str = "system") -> Dict[str, Any]: """ 두 리비전 간의 자재 비교 Args: current_file_id: 현재 파일 ID previous_file_id: 이전 파일 ID category: 비교할 카테고리 (PIPE, FITTING, FLANGE 등) username: 비교 수행자 Returns: 비교 결과 딕셔너리 """ try: # 현재 파일과 이전 파일의 자재 조회 current_materials = self._get_materials_by_category(current_file_id, category) previous_materials = self._get_materials_by_category(previous_file_id, category) # 자재 비교 수행 comparison_result = self._perform_material_comparison( current_materials, previous_materials, category ) # 비교 결과 저장 comparison_record = self._save_comparison_result( current_file_id, previous_file_id, category, comparison_result, username ) return { 'comparison_id': comparison_record.id, 'category': category, 'summary': comparison_result['summary'], 'materials': comparison_result['materials'], 'revision_actions': comparison_result['revision_actions'] } except Exception as e: logger.error(f"리비전 비교 실패: {e}") raise def _get_materials_by_category(self, file_id: int, category: str) -> List[Material]: """파일 ID와 카테고리로 자재 조회""" return self.db.query(Material).filter( and_( Material.file_id == file_id, or_( Material.classified_category == category, Material.category == category ) ) ).all() def _perform_material_comparison(self, current_materials: List[Material], previous_materials: List[Material], category: str) -> Dict[str, Any]: """자재 비교 로직 수행""" # 자재를 description + size로 그룹화 current_dict = self._group_materials_by_key(current_materials) previous_dict = self._group_materials_by_key(previous_materials) # 비교 결과 초기화 added_materials = [] removed_materials = [] changed_materials = [] unchanged_materials = [] revision_actions = [] # 현재 자재 기준으로 비교 (개선된 버전) for key, current_group in current_dict.items(): if key in previous_dict: previous_group = previous_dict[key] # 수량 비교 current_qty = current_group['quantity'] previous_qty = previous_group['quantity'] if current_qty != previous_qty: # 수량 변경됨 changed_materials.append({ 'material': current_group['representative_material'], 'previous_quantity': previous_qty, 'current_quantity': current_qty, 'quantity_difference': current_qty - previous_qty, 'purchase_confirmed': previous_group['purchase_confirmed'] }) # 리비전 액션 결정 (구매 상태 고려) action = self._determine_revision_action_enhanced( current_group, previous_group, category ) revision_actions.append(action) else: # 수량 동일 unchanged_materials.append(current_group['representative_material']) else: # 새로 추가된 자재 added_materials.append(current_group['representative_material']) revision_actions.append({ 'material_id': current_group['representative_material'].id, 'action': 'added', 'description': current_group['representative_material'].original_description, 'quantity': current_group['quantity'], 'purchase_status': 'not_purchased', 'revision_action': 'new_material' }) # 제거된 자재 확인 for key, previous_group in previous_dict.items(): if key not in current_dict: removed_materials.append(previous_group['representative_material']) revision_actions.append({ 'material_id': previous_group['representative_material'].id, 'action': 'removed', 'description': previous_group['representative_material'].original_description, 'quantity': previous_group['quantity'], 'purchase_status': 'purchased' if previous_group['purchase_confirmed'] else 'not_purchased', 'revision_action': 'deleted_material' }) # 통계 계산 summary = { 'added_count': len(added_materials), 'removed_count': len(removed_materials), 'changed_count': len(changed_materials), 'unchanged_count': len(unchanged_materials), 'total_current': len(current_materials), 'total_previous': len(previous_materials) } return { 'summary': summary, 'materials': { 'added': added_materials, 'removed': removed_materials, 'changed': changed_materials, 'unchanged': unchanged_materials }, 'revision_actions': revision_actions } def _group_materials_by_key(self, materials: List[Material]) -> Dict[str, Dict]: """자재를 고유 키로 그룹화 (개선된 버전)""" grouped = {} for material in materials: # 더 정교한 고유 키 생성 # description + drawing + main_nom + red_nom + material_grade key_parts = [ material.original_description.strip().upper(), material.drawing_name or '', material.main_nom or '', material.red_nom or '', material.material_grade or '' ] key = "|".join(key_parts) if key in grouped: # 동일한 자재가 있으면 수량 합산 grouped[key]['quantity'] += float(material.quantity) grouped[key]['materials'].append(material) else: grouped[key] = { 'key': key, 'representative_material': material, 'quantity': float(material.quantity), 'materials': [material], 'purchase_confirmed': getattr(material, 'purchase_confirmed', False) } return grouped def _determine_revision_action_enhanced(self, current_group: Dict, previous_group: Dict, category: str) -> Dict[str, Any]: """개선된 리비전 액션 결정 로직""" material = current_group['representative_material'] current_qty = current_group['quantity'] previous_qty = previous_group['quantity'] quantity_diff = current_qty - previous_qty is_purchased = previous_group['purchase_confirmed'] # 리비전 규칙 적용 if is_purchased: # 구매된 자재 if quantity_diff > 0: action = 'additional_purchase' description = f"추가 구매 필요: +{quantity_diff}" status = 'needs_additional_purchase' elif quantity_diff < 0: action = 'inventory' description = f"재고품으로 분류: {abs(quantity_diff)}" status = 'excess_inventory' else: action = 'maintain' description = "상황 유지" status = 'no_change' else: # 구매 안된 자재 if quantity_diff > 0: action = 'quantity_increase' description = f"수량 증가: +{quantity_diff}" status = 'quantity_updated' elif quantity_diff < 0: action = 'quantity_decrease' description = f"수량 감소: {quantity_diff}" status = 'quantity_reduced' else: action = 'maintain' description = "구매 표시 유지" status = 'purchase_pending' return { 'material_id': material.id, 'action': action, 'description': material.original_description, 'drawing_name': material.drawing_name, 'previous_quantity': previous_qty, 'current_quantity': current_qty, 'quantity_difference': quantity_diff, 'purchase_status': 'purchased' if is_purchased else 'not_purchased', 'revision_action': status, 'action_description': description, 'category': category } def _determine_revision_action(self, material: Material, previous_qty: float, current_qty: float, category: str) -> Dict[str, Any]: """리비전 액션 결정 로직""" # 구매 상태 확인 (간단하게 purchase_confirmed 필드 사용) is_purchased = getattr(material, 'purchase_confirmed', False) quantity_diff = current_qty - previous_qty if is_purchased: # 구매된 자재 if quantity_diff > 0: action = 'additional_purchase' description = f"추가 구매 필요: {quantity_diff}" elif quantity_diff < 0: action = 'inventory' description = f"재고품으로 분류: {abs(quantity_diff)}" else: action = 'maintain' description = "상황 유지" else: # 구매 안된 자재 if quantity_diff > 0: action = 'quantity_update' description = f"수량 증가: {quantity_diff}" elif quantity_diff < 0: action = 'quantity_decrease' description = f"수량 감소: {abs(quantity_diff)}" else: action = 'maintain' description = "구매 표시 유지" return { 'material_id': material.id, 'action': action, 'description': material.original_description, 'previous_quantity': previous_qty, 'current_quantity': current_qty, 'quantity_difference': quantity_diff, 'purchase_status': 'purchased' if is_purchased else 'not_purchased', 'action_description': description } def _save_comparison_result(self, current_file_id: int, previous_file_id: int, category: str, comparison_result: Dict[str, Any], username: str) -> SimpleRevisionComparison: """비교 결과를 데이터베이스에 저장""" # Job No 조회 current_file = self.db.query(File).filter(File.id == current_file_id).first() job_no = getattr(current_file, 'job_no', 'unknown') # 비교 결과 레코드 생성 comparison = SimpleRevisionComparison( job_no=job_no, current_file_id=current_file_id, previous_file_id=previous_file_id, category=category, added_count=comparison_result['summary']['added_count'], removed_count=comparison_result['summary']['removed_count'], changed_count=comparison_result['summary']['changed_count'], unchanged_count=comparison_result['summary']['unchanged_count'], comparison_data=comparison_result, created_by_username=username ) self.db.add(comparison) self.db.flush() # ID 생성을 위해 flush # 개별 자재 변경 로그 저장 for action in comparison_result['revision_actions']: material_log = SimpleRevisionMaterial( comparison_id=comparison.id, material_id=action['material_id'], change_type=action['action'], revision_action=action['action'], quantity_before=action.get('previous_quantity', 0), quantity_after=action.get('current_quantity', 0), quantity_difference=action.get('quantity_difference', 0), purchase_status=action.get('purchase_status', 'not_purchased') ) self.db.add(material_log) self.db.commit() return comparison def get_changed_materials_only(self, current_file_id: int, previous_file_id: int, categories: List[str] = None, username: str = "system") -> Dict[str, Any]: """변경된 자재만 필터링하여 반환""" if not categories: categories = ['PIPE', 'FITTING', 'FLANGE', 'VALVE', 'GASKET', 'BOLT', 'SUPPORT', 'SPECIAL', 'UNCLASSIFIED'] all_changes = {} total_changes = 0 for category in categories: try: comparison_result = self.compare_revisions( current_file_id, previous_file_id, category, username ) # 변경된 자재만 필터링 changed_materials = [] for action in comparison_result['revision_actions']: if action['action'] in ['added', 'removed'] or action.get('quantity_difference', 0) != 0: changed_materials.append(action) if changed_materials: all_changes[category] = { 'category': category, 'changed_count': len(changed_materials), 'changes': changed_materials, 'summary': comparison_result['summary'] } total_changes += len(changed_materials) except Exception as e: logger.warning(f"Failed to compare {category}: {e}") continue return { 'total_changed_materials': total_changes, 'categories_with_changes': list(all_changes.keys()), 'changes_by_category': all_changes, 'has_changes': total_changes > 0 } def get_comparison_result(self, comparison_id: int) -> Optional[Dict[str, Any]]: """저장된 비교 결과 조회""" comparison = self.db.query(SimpleRevisionComparison).filter( SimpleRevisionComparison.id == comparison_id ).first() if not comparison: return None # 관련 자재 변경 로그 조회 material_logs = self.db.query(SimpleRevisionMaterial).filter( SimpleRevisionMaterial.comparison_id == comparison_id ).all() return { 'comparison': comparison, 'material_logs': material_logs, 'comparison_data': comparison.comparison_data }