65 lines
1.8 KiB
Python
65 lines
1.8 KiB
Python
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker, declarative_base
|
|
from app.config import DATABASE_URL
|
|
import os
|
|
|
|
# MySQL 連線引擎設定
|
|
engine = create_engine(
|
|
DATABASE_URL,
|
|
pool_size=10,
|
|
max_overflow=20,
|
|
pool_pre_ping=True,
|
|
pool_recycle=3600,
|
|
echo=False
|
|
)
|
|
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
Base = declarative_base()
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
# Import models to register them
|
|
from app.models.user import User, UserRole
|
|
from app.models.dit import DitRecord
|
|
from app.models.sample import SampleRecord
|
|
from app.models.order import OrderRecord
|
|
from app.models.match import MatchResult, ReviewLog
|
|
|
|
def init_db():
|
|
"""初始化資料庫並建立預設管理員"""
|
|
from app.utils.security import get_password_hash
|
|
|
|
# 建立所有資料表
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
# 建立預設管理員帳號
|
|
db = SessionLocal()
|
|
try:
|
|
admin_email = os.getenv("ADMIN_EMAIL", "admin@example.com")
|
|
admin_password = os.getenv("ADMIN_PASSWORD", "admin123")
|
|
|
|
existing = db.query(User).filter(User.email == admin_email).first()
|
|
if not existing:
|
|
admin = User(
|
|
email=admin_email,
|
|
password_hash=get_password_hash(admin_password),
|
|
display_name="Administrator",
|
|
language="zh-TW",
|
|
role=UserRole.admin
|
|
)
|
|
db.add(admin)
|
|
db.commit()
|
|
print(f"[Init] Admin user created: {admin_email}")
|
|
else:
|
|
print(f"[Init] Admin user exists: {admin_email}")
|
|
except Exception as e:
|
|
print(f"[Init] Error creating admin: {e}")
|
|
db.rollback()
|
|
finally:
|
|
db.close()
|