57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
import os
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from app.core.config import get_settings
|
|
from app.core.security import hash_password
|
|
from app.models import User, Role, UserRole
|
|
|
|
|
|
def build_db_url() -> str:
|
|
settings = get_settings()
|
|
return (
|
|
f"mysql+pymysql://{settings.db_user}:{settings.db_password}"
|
|
f"@{settings.db_host}:{settings.db_port}/{settings.db_name}"
|
|
"?charset=utf8mb4"
|
|
)
|
|
|
|
|
|
def main():
|
|
username = os.getenv("INIT_ADMIN_USERNAME", "admin")
|
|
password = os.getenv("INIT_ADMIN_PASSWORD", "123456")
|
|
|
|
engine = create_engine(build_db_url(), pool_pre_ping=True)
|
|
with Session(engine) as db:
|
|
role = db.query(Role).filter(Role.role_code == "admin").first()
|
|
if not role:
|
|
role = Role(role_code="admin", role_name="管理员")
|
|
db.add(role)
|
|
db.flush()
|
|
|
|
user = db.query(User).filter(User.username == username).first()
|
|
if not user:
|
|
user = User(
|
|
username=username,
|
|
display_name="Administrator",
|
|
email=None,
|
|
phone=None,
|
|
password_hash=hash_password(password),
|
|
status=1,
|
|
is_deleted=False,
|
|
)
|
|
db.add(user)
|
|
db.flush()
|
|
else:
|
|
user.password_hash = hash_password(password)
|
|
user.status = 1
|
|
|
|
exist_link = db.query(UserRole).filter(UserRole.user_id == user.user_id, UserRole.role_id == role.role_id).first()
|
|
if not exist_link:
|
|
db.add(UserRole(user_id=user.user_id, role_id=role.role_id))
|
|
|
|
db.commit()
|
|
print(f"Admin ready: {username}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|