""" PIT3000 Database Setup SQLAlchemy 기반 데이터베이스 연결 및 스키마 정의 Windows/Linux 크로스 플랫폼 지원 """ from sqlalchemy import create_engine, MetaData, text from sqlalchemy.orm import declarative_base from sqlalchemy.orm import sessionmaker from sqlalchemy import Column, String, Integer, DateTime, Text import urllib.parse import platform import pyodbc # 기본 설정 Base = declarative_base() def get_available_odbc_driver(): """ 시스템에서 사용 가능한 SQL Server ODBC 드라이버를 자동 감지 Windows/Linux 환경에서 모두 동작하도록 설계 """ try: # 사용 가능한 ODBC 드라이버 목록 조회 available_drivers = pyodbc.drivers() # SQL Server 드라이버 우선순위 (최신 버전 우선) preferred_drivers = [ "ODBC Driver 18 for SQL Server", # 최신 "ODBC Driver 17 for SQL Server", # 일반적 "ODBC Driver 13 for SQL Server", # 구버전 "ODBC Driver 11 for SQL Server", # 구버전 "SQL Server Native Client 11.0", # Windows 레거시 "SQL Server", # 기본 ] # 시스템별 특화 드라이버 체크 os_name = platform.system().lower() if os_name == "linux": # Linux에서 주로 사용되는 드라이버들 linux_drivers = [ "ODBC Driver 18 for SQL Server", "ODBC Driver 17 for SQL Server", "FreeTDS", # Linux 오픈소스 드라이버 ] # Linux 드라이버를 우선순위에 추가 for driver in linux_drivers: if driver not in preferred_drivers: preferred_drivers.insert(0, driver) # 우선순위에 따라 사용 가능한 드라이버 찾기 for driver in preferred_drivers: if driver in available_drivers: print(f"[DBSETUP] 감지된 ODBC 드라이버: {driver} (OS: {platform.system()})") return driver # 사용 가능한 드라이버가 없으면 기본값 반환 if available_drivers: fallback_driver = available_drivers[0] print(f"[DBSETUP] 기본 드라이버 사용: {fallback_driver}") return fallback_driver else: # 최후의 수단으로 기본 드라이버명 반환 default_driver = "ODBC Driver 17 for SQL Server" print(f"[DBSETUP] 경고: ODBC 드라이버를 찾을 수 없어 기본값 사용: {default_driver}") return default_driver except Exception as e: print(f"[DBSETUP] 드라이버 감지 중 오류 발생: {e}") return "ODBC Driver 17 for SQL Server" # 기본값 반환 class DatabaseConfig: """PIT3000 데이터베이스 연결 설정""" SERVER = "192.168.0.4\\PM2014" USERNAME = "sa" PASSWORD = "tmddls214!%(" # 원본 비밀번호 # 동적 ODBC 드라이버 감지 DRIVER = get_available_odbc_driver() # URL 인코딩된 비밀번호 PASSWORD_ENCODED = urllib.parse.quote_plus(PASSWORD) # URL 인코딩된 드라이버 DRIVER_ENCODED = urllib.parse.quote_plus(DRIVER) # 데이터베이스별 연결 문자열 (동적 드라이버 사용) @classmethod def get_database_urls(cls): """동적으로 생성된 데이터베이스 연결 URL 딕셔너리 반환""" base_url = f"mssql+pyodbc://{cls.USERNAME}:{cls.PASSWORD_ENCODED}@{cls.SERVER}" # Connection Timeout을 60초로 증가, Login Timeout 추가 driver_params = f"driver={cls.DRIVER_ENCODED}&Encrypt=no&TrustServerCertificate=yes&Connection+Timeout=60&Login+Timeout=30" return { # 핵심 업무 데이터베이스 'PM_BASE': f"{base_url}/PM_BASE?{driver_params}", # 환자 정보, 개인정보 관리 'PM_PRES': f"{base_url}/PM_PRES?{driver_params}", # 처방전, 실제 판매 데이터 (SALE_sub) 'PM_DRUG': f"{base_url}/PM_DRUG?{driver_params}", # 약품 마스터 데이터 (CD_GOODS), 창고 거래 (WH_sub) # 재고 관리 시스템 (2025-09-20 추가) ⭐ 핵심 'PM_DUMS': f"{base_url}/PM_DUMS?{driver_params}", # 실제 재고 관리 (INVENTORY, NIMS_REALTIME_INVENTORY) # 알림 및 통신 시스템 'PM_ALIMI': f"{base_url}/PM_ALIMI?{driver_params}", # 알림톡, SMS 관리 'PM_ALDB': f"{base_url}/PM_ALDB?{driver_params}", # 알림 데이터베이스 # EDI 전자문서교환 시스템 'PM_EDIRECE': f"{base_url}/PM_EDIRECE?{driver_params}", # EDI 수신 데이터 'PM_EDISEND': f"{base_url}/PM_EDISEND?{driver_params}", # EDI 발송 데이터 # 부가 시스템 'PM_IMAGE': f"{base_url}/PM_IMAGE?{driver_params}", # 약품 이미지, 사진 관리 'PM_JOBLOG': f"{base_url}/PM_JOBLOG?{driver_params}", # 작업 로그, 시스템 로그 } # 하위 호환성을 위한 DATABASES 속성 @property def DATABASES(self): """하위 호환성을 위한 DATABASES 속성""" return self.get_database_urls() class DatabaseManager: """데이터베이스 연결 관리자""" def __init__(self): self.engines = {} self.sessions = {} self.database_urls = DatabaseConfig.get_database_urls() def get_engine(self, database='PM_BASE'): """특정 데이터베이스 엔진 반환""" if database not in self.engines: self.engines[database] = create_engine( self.database_urls[database], pool_size=5, # 커넥션 수 감소 (불필요한 연결 방지) max_overflow=10, # 최대 15개까지 pool_timeout=60, # 풀 대기 시간 60초로 증가 pool_recycle=1800, # 30분마다 재활용 (끊어진 연결 방지) pool_pre_ping=True, # 🔥 사용 전 연결 체크 (가장 중요!) echo=False, # True로 설정하면 SQL 쿼리 로깅 connect_args={ 'timeout': 60, # pyodbc 레벨 타임아웃 } ) return self.engines[database] def get_session(self, database='PM_BASE'): """특정 데이터베이스 세션 반환""" if database not in self.sessions: engine = self.get_engine(database) Session = sessionmaker(bind=engine) self.sessions[database] = Session() return self.sessions[database] def rollback_session(self, database='PM_BASE'): """세션 롤백 (트랜잭션 에러 복구용)""" if database in self.sessions: try: self.sessions[database].rollback() print(f"[DB Manager] {database} 세션 롤백 완료") return True except Exception as e: print(f"[DB Manager] {database} 세션 롤백 실패: {e}") return False return False def reset_session(self, database='PM_BASE'): """세션 재생성 (복구 불가능한 경우)""" if database in self.sessions: try: self.sessions[database].close() del self.sessions[database] print(f"[DB Manager] {database} 세션 삭제 완료") except Exception as e: print(f"[DB Manager] {database} 세션 삭제 실패: {e}") # 새 세션 생성 return self.get_session(database) def test_connection(self, database='PM_BASE'): """연결 테스트""" try: engine = self.get_engine(database) with engine.connect() as conn: result = conn.execute(text("SELECT 1")) return True, f"{database} 연결 성공" except Exception as e: return False, f"{database} 연결 실패: {e}" def close_all(self): """모든 연결 종료""" for session in self.sessions.values(): session.close() for engine in self.engines.values(): engine.dispose() # 전역 데이터베이스 매니저 인스턴스 db_manager = DatabaseManager() def get_db_session(database='PM_BASE'): """데이터베이스 세션 획득""" return db_manager.get_session(database) def test_all_connections(): """모든 데이터베이스 연결 테스트""" print("=== PIT3000 데이터베이스 연결 테스트 ===") print(f"감지된 ODBC 드라이버: {DatabaseConfig.DRIVER}") print(f"운영체제: {platform.system()} {platform.release()}") print("-" * 50) database_urls = DatabaseConfig.get_database_urls() for db_name in database_urls.keys(): success, message = db_manager.test_connection(db_name) status = "[OK]" if success else "[FAIL]" print(f"{status} {message}") print("\n연결 테스트 완료!") if __name__ == "__main__": test_all_connections()