diff --git a/tools/db_init.py b/tools/db_init.py new file mode 100644 index 0000000..01b33bc --- /dev/null +++ b/tools/db_init.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +""" +GUARDiA ITSM — DB 초기화 헬퍼 (설치 스크립트용) + +스키마 불일치 감지 시 자동 백업 후 재초기화. +설치·업그레이드 시 setup 스크립트에서 호출한다. + +사용법: + python tools/db_init.py [--force] + --force: 기존 DB 강제 삭제 후 재초기화 (신규 설치) +""" +import asyncio +import os +import shutil +import sys +from datetime import datetime +from pathlib import Path + +# itsm/ 디렉토리를 Python 경로에 추가 +ITSM_DIR = Path(__file__).parent.parent +sys.path.insert(0, str(ITSM_DIR)) +os.chdir(str(ITSM_DIR)) + +# .env 로드 (있으면) +try: + from dotenv import load_dotenv + load_dotenv(".env") +except ImportError: + pass + + +FORCE = "--force" in sys.argv +DB_PATH = Path(os.getenv("DATABASE_URL", "sqlite+aiosqlite:///./guardia_itsm.db") + .replace("sqlite+aiosqlite:///", "").replace("./", "")) + + +def _backup_db(): + if DB_PATH.exists(): + backup = DB_PATH.with_suffix( + f".backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}.db" + ) + shutil.copy2(DB_PATH, backup) + print(f"[BACKUP] {DB_PATH} → {backup}") + return backup + return None + + +async def _verify_schema(max_retries: int = 1) -> bool: + """ + 현재 DB 스키마가 모델과 일치하는지 확인. + 주요 테이블의 컬럼 존재 여부를 점검한다. + """ + from database import SessionLocal + from models import User + from sqlalchemy import select, text + + for attempt in range(max_retries + 1): + try: + async with SessionLocal() as db: + # User 테이블 전체 컬럼 조회 (가장 자주 변경되는 테이블) + await db.execute(select(User).limit(1)) + return True + except Exception as e: + if attempt == 0: + print(f"[SCHEMA] 스키마 불일치 감지: {e}") + return False + return False + + +async def main(): + from database import init_db, engine + from core.seed import seed_all + from database import SessionLocal + + db_is_sqlite = not os.getenv("DATABASE_URL", "").startswith("postgresql") + + # ── 신규 설치 모드 --force ────────────────────────────────────────────── + if FORCE and db_is_sqlite and DB_PATH.exists(): + print("[FORCE] 기존 DB 삭제 후 재초기화...") + _backup_db() + DB_PATH.unlink() + + # ── DB 존재 시 스키마 검증 ────────────────────────────────────────────── + if db_is_sqlite and DB_PATH.exists(): + print(f"[CHECK] 기존 DB 스키마 검증: {DB_PATH}") + schema_ok = await _verify_schema() + if not schema_ok: + print("[MIGRATE] 스키마 불일치 — 백업 후 재초기화합니다.") + _backup_db() + DB_PATH.unlink() + print("[INFO] 새 스키마로 DB를 생성합니다.") + else: + print("[OK] 스키마 정상") + + # ── 테이블 생성 ──────────────────────────────────────────────────────── + try: + await init_db() + print("[OK] DB 테이블 초기화 완료") + except Exception as e: + print(f"[ERROR] DB 초기화 실패: {e}") + sys.exit(1) + + # ── 시드 데이터 삽입 ────────────────────────────────────────────────── + try: + async with SessionLocal() as db: + await seed_all(db) + print("[OK] 시드 데이터 삽입 완료") + except Exception as e: + print(f"[WARN] 시드 데이터 삽입 오류 (무시): {e}") + + await engine.dispose() + print("[OK] DB 초기화 완료") + + +if __name__ == "__main__": + asyncio.run(main())