""" 리비전 세션 관리 서비스 - 리비전 세션 생성, 관리, 완료 처리 - 자재 변경 사항 추적 및 처리 """ import logging from typing import Dict, List, Optional, Any from datetime import datetime from sqlalchemy.orm import Session from sqlalchemy import text, and_, or_ from ..models import File, Material from ..database import get_db logger = logging.getLogger(__name__) class RevisionSessionService: """리비전 세션 관리 서비스""" def __init__(self, db: Session): self.db = db def create_revision_session( self, job_no: str, current_file_id: int, previous_file_id: int, username: str ) -> Dict[str, Any]: """새로운 리비전 세션 생성""" try: # 파일 정보 조회 current_file = self.db.query(File).filter(File.id == current_file_id).first() previous_file = self.db.query(File).filter(File.id == previous_file_id).first() if not current_file or not previous_file: raise ValueError("파일 정보를 찾을 수 없습니다") # 기존 진행 중인 세션이 있는지 확인 existing_session = self.db.execute(text(""" SELECT id FROM revision_sessions WHERE job_no = :job_no AND status = 'processing' """), {"job_no": job_no}).fetchone() if existing_session: logger.warning(f"기존 진행 중인 리비전 세션이 있습니다: {existing_session[0]}") return {"session_id": existing_session[0], "status": "existing"} # 새 세션 생성 session_data = { "job_no": job_no, "current_file_id": current_file_id, "previous_file_id": previous_file_id, "current_revision": current_file.revision, "previous_revision": previous_file.revision, "status": "processing", "created_by": username } result = self.db.execute(text(""" INSERT INTO revision_sessions ( job_no, current_file_id, previous_file_id, current_revision, previous_revision, status, created_by ) VALUES ( :job_no, :current_file_id, :previous_file_id, :current_revision, :previous_revision, :status, :created_by ) RETURNING id """), session_data) session_id = result.fetchone()[0] self.db.commit() logger.info(f"새 리비전 세션 생성: {session_id} (Job: {job_no})") return { "session_id": session_id, "status": "created", "job_no": job_no, "current_revision": current_file.revision, "previous_revision": previous_file.revision } except Exception as e: self.db.rollback() logger.error(f"리비전 세션 생성 실패: {e}") raise def get_session_status(self, session_id: int) -> Dict[str, Any]: """리비전 세션 상태 조회""" try: session_info = self.db.execute(text(""" SELECT id, job_no, current_file_id, previous_file_id, current_revision, previous_revision, status, total_materials, processed_materials, added_count, removed_count, changed_count, unchanged_count, purchase_cancel_count, inventory_transfer_count, additional_purchase_count, created_by, created_at, completed_at FROM revision_sessions WHERE id = :session_id """), {"session_id": session_id}).fetchone() if not session_info: raise ValueError(f"리비전 세션을 찾을 수 없습니다: {session_id}") # 변경 사항 상세 조회 changes = self.db.execute(text(""" SELECT category, change_type, revision_action, action_status, COUNT(*) as count, SUM(CASE WHEN purchase_status = 'purchased' THEN 1 ELSE 0 END) as purchased_count, SUM(CASE WHEN purchase_status = 'not_purchased' THEN 1 ELSE 0 END) as unpurchased_count FROM revision_material_changes WHERE session_id = :session_id GROUP BY category, change_type, revision_action, action_status ORDER BY category, change_type """), {"session_id": session_id}).fetchall() return { "session_info": dict(session_info._mapping), "changes_summary": [dict(change._mapping) for change in changes], "progress_percentage": ( (session_info.processed_materials / session_info.total_materials * 100) if session_info.total_materials > 0 else 0 ) } except Exception as e: logger.error(f"리비전 세션 상태 조회 실패: {e}") raise def update_session_progress( self, session_id: int, total_materials: int = None, processed_materials: int = None, **counts ) -> bool: """리비전 세션 진행 상황 업데이트""" try: update_fields = [] update_values = {"session_id": session_id} if total_materials is not None: update_fields.append("total_materials = :total_materials") update_values["total_materials"] = total_materials if processed_materials is not None: update_fields.append("processed_materials = :processed_materials") update_values["processed_materials"] = processed_materials # 카운트 필드들 업데이트 count_fields = [ "added_count", "removed_count", "changed_count", "unchanged_count", "purchase_cancel_count", "inventory_transfer_count", "additional_purchase_count" ] for field in count_fields: if field in counts: update_fields.append(f"{field} = :{field}") update_values[field] = counts[field] if not update_fields: return True # 업데이트할 내용이 없음 query = f""" UPDATE revision_sessions SET {', '.join(update_fields)} WHERE id = :session_id """ self.db.execute(text(query), update_values) self.db.commit() logger.info(f"리비전 세션 진행 상황 업데이트: {session_id}") return True except Exception as e: self.db.rollback() logger.error(f"리비전 세션 진행 상황 업데이트 실패: {e}") raise def complete_session(self, session_id: int, username: str) -> Dict[str, Any]: """리비전 세션 완료 처리""" try: # 세션 상태를 완료로 변경 self.db.execute(text(""" UPDATE revision_sessions SET status = 'completed', completed_at = CURRENT_TIMESTAMP WHERE id = :session_id AND status = 'processing' """), {"session_id": session_id}) # 완료 로그 기록 self.db.execute(text(""" INSERT INTO revision_action_logs ( session_id, action_type, action_description, executed_by, result, result_message ) VALUES ( :session_id, 'session_complete', '리비전 세션 완료', :username, 'success', '모든 리비전 처리 완료' ) """), { "session_id": session_id, "username": username }) self.db.commit() # 최종 상태 조회 final_status = self.get_session_status(session_id) logger.info(f"리비전 세션 완료: {session_id}") return { "status": "completed", "session_id": session_id, "final_status": final_status } except Exception as e: self.db.rollback() logger.error(f"리비전 세션 완료 처리 실패: {e}") raise def cancel_session(self, session_id: int, username: str, reason: str = None) -> bool: """리비전 세션 취소""" try: # 세션 상태를 취소로 변경 self.db.execute(text(""" UPDATE revision_sessions SET status = 'cancelled', completed_at = CURRENT_TIMESTAMP WHERE id = :session_id AND status = 'processing' """), {"session_id": session_id}) # 취소 로그 기록 self.db.execute(text(""" INSERT INTO revision_action_logs ( session_id, action_type, action_description, executed_by, result, result_message ) VALUES ( :session_id, 'session_cancel', '리비전 세션 취소', :username, 'cancelled', :reason ) """), { "session_id": session_id, "username": username, "reason": reason or "사용자 요청에 의한 취소" }) self.db.commit() logger.info(f"리비전 세션 취소: {session_id}, 사유: {reason}") return True except Exception as e: self.db.rollback() logger.error(f"리비전 세션 취소 실패: {e}") raise def get_job_revision_history(self, job_no: str) -> List[Dict[str, Any]]: """Job의 리비전 히스토리 조회""" try: sessions = self.db.execute(text(""" SELECT rs.id, rs.current_revision, rs.previous_revision, rs.status, rs.created_by, rs.created_at, rs.completed_at, rs.added_count, rs.removed_count, rs.changed_count, cf.filename as current_filename, pf.filename as previous_filename FROM revision_sessions rs LEFT JOIN files cf ON rs.current_file_id = cf.id LEFT JOIN files pf ON rs.previous_file_id = pf.id WHERE rs.job_no = :job_no ORDER BY rs.created_at DESC """), {"job_no": job_no}).fetchall() return [dict(session._mapping) for session in sessions] except Exception as e: logger.error(f"리비전 히스토리 조회 실패: {e}") raise