3398f71b80
Some checks failed
SonarQube Analysis / SonarQube Scan (push) Has been cancelled
✅ 백엔드 구조 개선: - DatabaseService: 공통 DB 쿼리 로직 통합 - FileUploadService: 파일 업로드 로직 모듈화 및 트랜잭션 관리 개선 - 서비스 레이어 패턴 도입으로 코드 재사용성 향상 ✅ 프론트엔드 컴포넌트 개선: - LoadingSpinner, ErrorMessage, ConfirmDialog 공통 컴포넌트 생성 - 재사용 가능한 컴포넌트 라이브러리 구축 - deprecated/backup 파일들 완전 제거 ✅ 성능 최적화: - optimize_database.py: 핵심 DB 인덱스 자동 생성 - 쿼리 최적화 및 통계 업데이트 자동화 - VACUUM ANALYZE 자동 실행 ✅ 코드 정리: - 개별 SQL 마이그레이션 파일들을 legacy/ 폴더로 정리 - 중복된 마이그레이션 스크립트 정리 - 깔끔하고 체계적인 프로젝트 구조 완성 ✅ 자동 마이그레이션 시스템 강화: - complete_migrate.py: SQLAlchemy 기반 완전한 마이그레이션 - analyze_and_fix_schema.py: 백엔드 코드 분석 기반 스키마 수정 - fix_missing_tables.py: 누락된 테이블/컬럼 자동 생성 - start.sh: 배포 시 자동 실행 순서 최적화
302 lines
12 KiB
Python
302 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
TK-MP-Project 완전한 자동 DB 마이그레이션 시스템
|
||
- 모든 SQLAlchemy 모델을 기반으로 테이블 생성/업데이트
|
||
- 누락된 컬럼 자동 추가
|
||
- 인덱스 자동 생성
|
||
- 초기 데이터 삽입
|
||
- macOS Docker와 Synology Container Manager 모두 지원
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
import psycopg2
|
||
from psycopg2 import OperationalError, sql
|
||
from sqlalchemy import create_engine, text, inspect
|
||
from sqlalchemy.orm import sessionmaker
|
||
from sqlalchemy.exc import OperationalError as SQLAlchemyOperationalError
|
||
from datetime import datetime
|
||
import bcrypt
|
||
|
||
# 현재 디렉토리를 Python 경로에 추가
|
||
sys.path.insert(0, '/app')
|
||
|
||
from app.database import Base, get_db
|
||
from app.auth.models import User
|
||
from app.models import * # 모든 모델을 임포트하여 Base.metadata에 등록
|
||
|
||
# 환경 변수 로드
|
||
DB_HOST = os.getenv("DB_HOST", "postgres")
|
||
DB_PORT = os.getenv("DB_PORT", "5432")
|
||
DB_NAME = os.getenv("DB_NAME", "tk_mp_bom")
|
||
DB_USER = os.getenv("DB_USER", "tkmp_user")
|
||
DB_PASSWORD = os.getenv("DB_PASSWORD", "tkmp_password_2025")
|
||
|
||
DATABASE_URL = f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
|
||
|
||
ADMIN_USERNAME = os.getenv("ADMIN_USERNAME", "admin")
|
||
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "admin123")
|
||
ADMIN_NAME = os.getenv("ADMIN_NAME", "시스템 관리자")
|
||
ADMIN_EMAIL = os.getenv("ADMIN_EMAIL", "admin@tkmp.com")
|
||
|
||
SYSTEM_USERNAME = os.getenv("SYSTEM_USERNAME", "system")
|
||
SYSTEM_PASSWORD = os.getenv("SYSTEM_PASSWORD", "admin123")
|
||
SYSTEM_NAME = os.getenv("SYSTEM_NAME", "시스템 계정")
|
||
SYSTEM_EMAIL = os.getenv("SYSTEM_EMAIL", "system@tkmp.com")
|
||
|
||
def wait_for_db(max_attempts=120, delay=2):
|
||
"""데이터베이스 연결 대기 - 더 긴 대기 시간과 상세한 로그"""
|
||
print(f"Waiting for database connection...")
|
||
print(f" Host: {DB_HOST}:{DB_PORT}")
|
||
print(f" Database: {DB_NAME}")
|
||
print(f" User: {DB_USER}")
|
||
|
||
for i in range(1, max_attempts + 1):
|
||
try:
|
||
conn = psycopg2.connect(
|
||
host=DB_HOST,
|
||
port=DB_PORT,
|
||
database=DB_NAME,
|
||
user=DB_USER,
|
||
password=DB_PASSWORD,
|
||
connect_timeout=5
|
||
)
|
||
conn.close()
|
||
print(f"SUCCESS: Database connection established! ({i}/{max_attempts})")
|
||
return True
|
||
except OperationalError as e:
|
||
if i <= 5 or i % 10 == 0 or i == max_attempts:
|
||
print(f"Waiting for database... ({i}/{max_attempts}) - {str(e)[:100]}")
|
||
if i == max_attempts:
|
||
print(f"FAILED: Database connection timeout after {max_attempts * delay} seconds")
|
||
print(f"Error: {e}")
|
||
time.sleep(delay)
|
||
return False
|
||
|
||
def get_existing_columns(engine, table_name):
|
||
"""기존 테이블의 컬럼 목록 조회"""
|
||
inspector = inspect(engine)
|
||
try:
|
||
columns = inspector.get_columns(table_name)
|
||
return {col['name']: col for col in columns}
|
||
except Exception:
|
||
return {}
|
||
|
||
def get_model_columns(model):
|
||
"""SQLAlchemy 모델의 컬럼 정보 추출"""
|
||
columns = {}
|
||
for column in model.__table__.columns:
|
||
columns[column.name] = {
|
||
'name': column.name,
|
||
'type': str(column.type),
|
||
'nullable': column.nullable,
|
||
'default': column.default,
|
||
'primary_key': column.primary_key
|
||
}
|
||
return columns
|
||
|
||
def add_missing_columns(engine, model):
|
||
"""누락된 컬럼을 기존 테이블에 추가"""
|
||
table_name = model.__tablename__
|
||
existing_columns = get_existing_columns(engine, table_name)
|
||
model_columns = get_model_columns(model)
|
||
|
||
missing_columns = []
|
||
for col_name, col_info in model_columns.items():
|
||
if col_name not in existing_columns:
|
||
missing_columns.append((col_name, col_info))
|
||
|
||
if not missing_columns:
|
||
return True
|
||
|
||
print(f"📝 테이블 '{table_name}'에 누락된 컬럼 {len(missing_columns)}개 추가 중...")
|
||
|
||
try:
|
||
with engine.connect() as connection:
|
||
for col_name, col_info in missing_columns:
|
||
# 컬럼 타입 매핑
|
||
col_type = col_info['type']
|
||
if 'VARCHAR' in col_type:
|
||
sql_type = col_type
|
||
elif 'INTEGER' in col_type:
|
||
sql_type = 'INTEGER'
|
||
elif 'BOOLEAN' in col_type:
|
||
sql_type = 'BOOLEAN'
|
||
elif 'DATETIME' in col_type:
|
||
sql_type = 'TIMESTAMP'
|
||
elif 'TEXT' in col_type:
|
||
sql_type = 'TEXT'
|
||
elif 'NUMERIC' in col_type:
|
||
sql_type = col_type
|
||
elif 'JSON' in col_type:
|
||
sql_type = 'JSON'
|
||
else:
|
||
sql_type = 'TEXT' # 기본값
|
||
|
||
# NULL 허용 여부
|
||
nullable = "NULL" if col_info['nullable'] else "NOT NULL"
|
||
|
||
# 기본값 설정
|
||
default_clause = ""
|
||
if col_info['default'] is not None:
|
||
if col_info['type'] == 'BOOLEAN':
|
||
default_value = 'TRUE' if str(col_info['default']).lower() in ['true', '1'] else 'FALSE'
|
||
default_clause = f" DEFAULT {default_value}"
|
||
elif 'VARCHAR' in col_info['type'] or 'TEXT' in col_info['type']:
|
||
default_clause = f" DEFAULT '{col_info['default']}'"
|
||
else:
|
||
default_clause = f" DEFAULT {col_info['default']}"
|
||
|
||
alter_sql = f"ALTER TABLE {table_name} ADD COLUMN IF NOT EXISTS {col_name} {sql_type}{default_clause} {nullable};"
|
||
|
||
try:
|
||
connection.execute(text(alter_sql))
|
||
print(f" ✅ 컬럼 '{col_name}' ({sql_type}) 추가 완료")
|
||
except Exception as e:
|
||
print(f" ⚠️ 컬럼 '{col_name}' 추가 실패: {e}")
|
||
|
||
connection.commit()
|
||
return True
|
||
except Exception as e:
|
||
print(f"❌ 테이블 '{table_name}' 컬럼 추가 실패: {e}")
|
||
return False
|
||
|
||
def create_tables_and_migrate():
|
||
"""테이블 생성 및 마이그레이션"""
|
||
print("🔄 완전한 스키마 동기화 및 마이그레이션 시작...")
|
||
engine = create_engine(DATABASE_URL)
|
||
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||
|
||
try:
|
||
# 1. 모든 테이블 생성 (존재하지 않으면 생성)
|
||
print("📋 새 테이블 생성 중...")
|
||
Base.metadata.create_all(bind=engine)
|
||
print("✅ 새 테이블 생성 완료")
|
||
|
||
# 2. 기존 테이블에 누락된 컬럼 추가
|
||
print("🔧 기존 테이블 컬럼 동기화 중...")
|
||
|
||
# 모든 모델에 대해 컬럼 동기화
|
||
models_to_check = [
|
||
User, Project, File, Material, MaterialStandard,
|
||
MaterialCategory, MaterialSpecification, MaterialGrade,
|
||
MaterialPattern, SpecialMaterial, SpecialMaterialGrade,
|
||
SpecialMaterialPattern, PipeDetail, RequirementType,
|
||
UserRequirement, TubingCategory, TubingSpecification,
|
||
TubingManufacturer, TubingProduct, MaterialTubingMapping,
|
||
SupportDetails, PurchaseRequestItems, FittingDetails,
|
||
FlangeDetails, ValveDetails, GasketDetails, BoltDetails,
|
||
InstrumentDetails, PurchaseRequests, Jobs, PipeEndPreparations,
|
||
MaterialPurchaseTracking, ExcelExports, UserActivityLogs,
|
||
ExcelExportHistory, ExportedMaterials, PurchaseStatusHistory
|
||
]
|
||
|
||
for model in models_to_check:
|
||
if hasattr(model, '__tablename__'):
|
||
add_missing_columns(engine, model)
|
||
|
||
print("✅ 모든 테이블 컬럼 동기화 완료")
|
||
|
||
# 3. 초기 사용자 데이터 생성
|
||
with SessionLocal() as db:
|
||
# 관리자 계정 확인 및 생성
|
||
admin_user = db.query(User).filter(User.username == ADMIN_USERNAME).first()
|
||
if not admin_user:
|
||
hashed_password = bcrypt.hashpw(ADMIN_PASSWORD.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
||
admin_user = User(
|
||
username=ADMIN_USERNAME,
|
||
password=hashed_password,
|
||
name=ADMIN_NAME,
|
||
email=ADMIN_EMAIL,
|
||
role='admin',
|
||
access_level='admin',
|
||
department='IT',
|
||
position='시스템 관리자',
|
||
status='active'
|
||
)
|
||
db.add(admin_user)
|
||
print(f"➕ 관리자 계정 '{ADMIN_USERNAME}' 생성 완료")
|
||
else:
|
||
print(f"☑️ 관리자 계정 '{ADMIN_USERNAME}' 이미 존재")
|
||
|
||
# 시스템 계정 확인 및 생성
|
||
system_user = db.query(User).filter(User.username == SYSTEM_USERNAME).first()
|
||
if not system_user:
|
||
hashed_password = bcrypt.hashpw(SYSTEM_PASSWORD.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
|
||
system_user = User(
|
||
username=SYSTEM_USERNAME,
|
||
password=hashed_password,
|
||
name=SYSTEM_NAME,
|
||
email=SYSTEM_EMAIL,
|
||
role='system',
|
||
access_level='system',
|
||
department='IT',
|
||
position='시스템 계정',
|
||
status='active'
|
||
)
|
||
db.add(system_user)
|
||
print(f"➕ 시스템 계정 '{SYSTEM_USERNAME}' 생성 완료")
|
||
else:
|
||
print(f"☑️ 시스템 계정 '{SYSTEM_USERNAME}' 이미 존재")
|
||
|
||
db.commit()
|
||
print("✅ 초기 사용자 계정 확인 및 생성 완료")
|
||
|
||
# 4. 성능 인덱스 추가
|
||
print("🚀 성능 인덱스 생성 중...")
|
||
with engine.connect() as connection:
|
||
indexes = [
|
||
"CREATE INDEX IF NOT EXISTS idx_materials_main_nom ON materials(main_nom);",
|
||
"CREATE INDEX IF NOT EXISTS idx_materials_red_nom ON materials(red_nom);",
|
||
"CREATE INDEX IF NOT EXISTS idx_materials_full_material_grade ON materials(full_material_grade);",
|
||
"CREATE INDEX IF NOT EXISTS idx_materials_revision_status ON materials(revision_status);",
|
||
"CREATE INDEX IF NOT EXISTS idx_materials_file_id ON materials(file_id);",
|
||
"CREATE INDEX IF NOT EXISTS idx_materials_drawing_name ON materials(drawing_name);",
|
||
"CREATE INDEX IF NOT EXISTS idx_materials_classified_category ON materials(classified_category);",
|
||
"CREATE INDEX IF NOT EXISTS idx_files_job_no ON files(job_no);",
|
||
"CREATE INDEX IF NOT EXISTS idx_files_uploaded_by ON files(uploaded_by);",
|
||
"CREATE INDEX IF NOT EXISTS idx_users_username ON users(username);",
|
||
"CREATE INDEX IF NOT EXISTS idx_users_status ON users(status);",
|
||
]
|
||
|
||
for index_sql in indexes:
|
||
try:
|
||
connection.execute(text(index_sql))
|
||
except Exception as e:
|
||
print(f" ⚠️ 인덱스 생성 실패: {e}")
|
||
|
||
connection.commit()
|
||
print("✅ 성능 인덱스 생성 완료")
|
||
|
||
return True
|
||
except SQLAlchemyOperationalError as e:
|
||
print(f"❌ 데이터베이스 작업 실패: {e}")
|
||
return False
|
||
except Exception as e:
|
||
print(f"❌ 예상치 못한 오류 발생: {e}")
|
||
return False
|
||
|
||
if __name__ == "__main__":
|
||
print("🚀 TK-MP-Project 완전한 자동 DB 마이그레이션 시작")
|
||
print(f"⏰ 시작 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
print(f"🖥️ 환경: {os.uname().sysname} {os.uname().machine}")
|
||
|
||
print("🔧 DB 설정 확인:")
|
||
print(f" - DB_HOST: {DB_HOST}")
|
||
print(f" - DB_PORT: {DB_PORT}")
|
||
print(f" - DB_NAME: {DB_NAME}")
|
||
print(f" - DB_USER: {DB_USER}")
|
||
|
||
if not wait_for_db():
|
||
print("❌ DB 마이그레이션 실패. 서버 시작을 중단합니다.")
|
||
exit(1)
|
||
|
||
if not create_tables_and_migrate():
|
||
print("⚠️ DB 마이그레이션에서 일부 오류가 발생했지만 서버를 시작합니다.")
|
||
print(" (기존 스키마가 있거나 부분적으로 성공했을 수 있습니다)")
|
||
else:
|
||
print("✅ 완전한 DB 마이그레이션 성공")
|
||
|
||
print(f"⏰ 완료 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|