1cebb02ec6
- pets 테이블 추가 (이름, 종류, 품종, 사진 등) - 반려동물 CRUD API (/api/pets) - 확장 마이페이지 (/mypage) - 카카오 로그인 기반 - 기존 마이페이지에 퀵 메뉴 추가 (반려동물/쿠폰/구매내역/내정보) - 카카오 로그인 시 세션에 user_id 저장 - 동물약 APC 매핑 가이드 문서 추가
445 lines
18 KiB
Python
445 lines
18 KiB
Python
"""
|
|
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
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
# 기본 설정
|
|
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()
|
|
|
|
# SQLite 연결 추가
|
|
self.sqlite_conn = None
|
|
self.sqlite_db_path = Path(__file__).parent / 'mileage.db'
|
|
|
|
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()
|
|
else:
|
|
# 🔥 기존 세션 상태 체크 및 자동 복구
|
|
session = self.sessions[database]
|
|
try:
|
|
# 세션이 유효한지 간단한 쿼리로 테스트
|
|
session.execute(text("SELECT 1"))
|
|
except Exception as e:
|
|
error_msg = str(e).lower()
|
|
# 연결 끊김 또는 트랜잭션 에러 감지
|
|
if any(keyword in error_msg for keyword in [
|
|
'invalid transaction', 'rollback', 'connection',
|
|
'closed', 'lost', 'timeout', 'network', 'disconnect'
|
|
]):
|
|
print(f"[DB Manager] {database} 세션 복구 시도: {e}")
|
|
try:
|
|
session.rollback()
|
|
print(f"[DB Manager] {database} 롤백 성공, 세션 재사용")
|
|
except Exception as rollback_err:
|
|
print(f"[DB Manager] {database} 롤백 실패, 세션 재생성: {rollback_err}")
|
|
try:
|
|
session.close()
|
|
except:
|
|
pass
|
|
del self.sessions[database]
|
|
# 새 세션 생성
|
|
engine = self.get_engine(database)
|
|
Session = sessionmaker(bind=engine)
|
|
self.sessions[database] = Session()
|
|
print(f"[DB Manager] {database} 새 세션 생성 완료")
|
|
else:
|
|
# 다른 종류의 에러면 롤백만 시도
|
|
try:
|
|
session.rollback()
|
|
except:
|
|
pass
|
|
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 get_sqlite_connection(self, new_connection=False):
|
|
"""
|
|
SQLite mileage.db 연결 반환
|
|
|
|
Args:
|
|
new_connection: True면 항상 새 연결 생성 (멀티스레드 안전)
|
|
|
|
Returns:
|
|
sqlite3.Connection: SQLite 연결 객체
|
|
"""
|
|
# 새 연결 요청 시 항상 새로 생성
|
|
if new_connection:
|
|
return self._create_sqlite_connection()
|
|
|
|
# 기존 싱글톤 방식 (하위 호환)
|
|
if self.sqlite_conn is not None:
|
|
try:
|
|
cursor = self.sqlite_conn.cursor()
|
|
cursor.execute("SELECT 1")
|
|
cursor.fetchone()
|
|
cursor.close()
|
|
except Exception as e:
|
|
print(f"[DB Manager] SQLite 연결 체크 실패, 재연결: {e}")
|
|
try:
|
|
self.sqlite_conn.close()
|
|
except:
|
|
pass
|
|
self.sqlite_conn = None
|
|
|
|
if self.sqlite_conn is None:
|
|
self.sqlite_conn = self._create_sqlite_connection()
|
|
|
|
return self.sqlite_conn
|
|
|
|
def _create_sqlite_connection(self):
|
|
"""새 SQLite 연결 생성"""
|
|
is_new_db = not self.sqlite_db_path.exists()
|
|
|
|
conn = sqlite3.connect(
|
|
str(self.sqlite_db_path),
|
|
check_same_thread=False,
|
|
timeout=10.0
|
|
)
|
|
conn.row_factory = sqlite3.Row
|
|
|
|
if is_new_db:
|
|
# 스키마 초기화 (임시로 self.sqlite_conn 설정)
|
|
old_conn = self.sqlite_conn
|
|
self.sqlite_conn = conn
|
|
self.init_sqlite_schema()
|
|
self.sqlite_conn = old_conn
|
|
print(f"[DB Manager] SQLite 신규 DB 생성 완료: {self.sqlite_db_path}")
|
|
else:
|
|
# 기존 DB: 마이그레이션 실행
|
|
old_conn = self.sqlite_conn
|
|
self.sqlite_conn = conn
|
|
self._migrate_sqlite()
|
|
self.sqlite_conn = old_conn
|
|
|
|
return conn
|
|
|
|
def init_sqlite_schema(self):
|
|
"""
|
|
mileage_schema.sql 실행하여 테이블 생성
|
|
"""
|
|
schema_path = Path(__file__).parent / 'mileage_schema.sql'
|
|
|
|
if not schema_path.exists():
|
|
raise FileNotFoundError(f"Schema file not found: {schema_path}")
|
|
|
|
with open(schema_path, 'r', encoding='utf-8') as f:
|
|
schema_sql = f.read()
|
|
|
|
# 스키마 실행
|
|
cursor = self.sqlite_conn.cursor()
|
|
cursor.executescript(schema_sql)
|
|
self.sqlite_conn.commit()
|
|
|
|
print(f"[DB Manager] SQLite 스키마 초기화 완료")
|
|
|
|
def _migrate_sqlite(self):
|
|
"""기존 DB에 새 컬럼/테이블 추가 (마이그레이션)"""
|
|
cursor = self.sqlite_conn.cursor()
|
|
cursor.execute("PRAGMA table_info(users)")
|
|
columns = [row[1] for row in cursor.fetchall()]
|
|
if 'birthday' not in columns:
|
|
cursor.execute("ALTER TABLE users ADD COLUMN birthday VARCHAR(10)")
|
|
self.sqlite_conn.commit()
|
|
print("[DB Manager] SQLite 마이그레이션: users.birthday 컬럼 추가")
|
|
|
|
# alimtalk_logs 테이블 생성
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='alimtalk_logs'")
|
|
if not cursor.fetchone():
|
|
cursor.executescript("""
|
|
CREATE TABLE IF NOT EXISTS alimtalk_logs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
template_code VARCHAR(50) NOT NULL,
|
|
recipient_no VARCHAR(20) NOT NULL,
|
|
user_id INTEGER,
|
|
trigger_source VARCHAR(20) NOT NULL,
|
|
template_params TEXT,
|
|
success BOOLEAN NOT NULL,
|
|
result_message TEXT,
|
|
transaction_id VARCHAR(20),
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (user_id) REFERENCES users(id)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_alimtalk_created ON alimtalk_logs(created_at);
|
|
CREATE INDEX IF NOT EXISTS idx_alimtalk_recipient ON alimtalk_logs(recipient_no);
|
|
""")
|
|
self.sqlite_conn.commit()
|
|
print("[DB Manager] SQLite 마이그레이션: alimtalk_logs 테이블 생성")
|
|
|
|
# ai_recommendations 테이블 생성
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='ai_recommendations'")
|
|
if not cursor.fetchone():
|
|
cursor.executescript("""
|
|
CREATE TABLE IF NOT EXISTS ai_recommendations (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
transaction_id VARCHAR(20),
|
|
recommended_product TEXT NOT NULL,
|
|
recommendation_message TEXT NOT NULL,
|
|
recommendation_reason TEXT,
|
|
trigger_products TEXT,
|
|
ai_raw_response TEXT,
|
|
status VARCHAR(20) DEFAULT 'active',
|
|
displayed_count INTEGER DEFAULT 0,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
expires_at DATETIME,
|
|
displayed_at DATETIME,
|
|
dismissed_at DATETIME,
|
|
FOREIGN KEY (user_id) REFERENCES users(id)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_rec_user_status ON ai_recommendations(user_id, status);
|
|
CREATE INDEX IF NOT EXISTS idx_rec_expires ON ai_recommendations(expires_at);
|
|
""")
|
|
self.sqlite_conn.commit()
|
|
print("[DB Manager] SQLite 마이그레이션: ai_recommendations 테이블 생성")
|
|
|
|
# customer_identities 토큰 저장 컬럼 추가
|
|
cursor.execute("PRAGMA table_info(customer_identities)")
|
|
ci_columns = [row[1] for row in cursor.fetchall()]
|
|
if 'access_token' not in ci_columns:
|
|
cursor.execute("ALTER TABLE customer_identities ADD COLUMN access_token TEXT")
|
|
cursor.execute("ALTER TABLE customer_identities ADD COLUMN refresh_token TEXT")
|
|
cursor.execute("ALTER TABLE customer_identities ADD COLUMN token_expires_at DATETIME")
|
|
self.sqlite_conn.commit()
|
|
print("[DB Manager] SQLite 마이그레이션: customer_identities 토큰 컬럼 추가")
|
|
|
|
# pets 테이블 생성 (반려동물)
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='pets'")
|
|
if not cursor.fetchone():
|
|
cursor.executescript("""
|
|
CREATE TABLE IF NOT EXISTS pets (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
name VARCHAR(50) NOT NULL,
|
|
species VARCHAR(20) NOT NULL,
|
|
breed VARCHAR(50),
|
|
gender VARCHAR(10),
|
|
birth_date DATE,
|
|
age_months INTEGER,
|
|
weight DECIMAL(5,2),
|
|
photo_url TEXT,
|
|
notes TEXT,
|
|
is_active BOOLEAN DEFAULT TRUE,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (user_id) REFERENCES users(id)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_pets_user ON pets(user_id);
|
|
CREATE INDEX IF NOT EXISTS idx_pets_species ON pets(species);
|
|
""")
|
|
self.sqlite_conn.commit()
|
|
print("[DB Manager] SQLite 마이그레이션: pets 테이블 생성")
|
|
|
|
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()
|
|
|
|
# SQLite 연결 종료
|
|
if self.sqlite_conn:
|
|
self.sqlite_conn.close()
|
|
self.sqlite_conn = None
|
|
|
|
# 전역 데이터베이스 매니저 인스턴스
|
|
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() |