85 lines
2.7 KiB
Python
85 lines
2.7 KiB
Python
import asyncio
|
|
from sqlalchemy import select
|
|
from app.db.session import AsyncSession
|
|
from app.models.user import User, Role
|
|
from app.core.security import get_password_hash
|
|
from datetime import datetime
|
|
|
|
async def create_admin_user():
|
|
"""创建初始管理员账号"""
|
|
from app.db.session import async_session_maker
|
|
|
|
async with async_session_maker() as db:
|
|
try:
|
|
# 检查管理员是否已存在
|
|
result = await db.execute(
|
|
select(User).where(User.username == "admin")
|
|
)
|
|
existing_admin = result.scalar_one_or_none()
|
|
|
|
if existing_admin:
|
|
print("管理员账号已存在，跳过创建")
|
|
print(f"用户名: {existing_admin.username}")
|
|
print(f"邮箱: {existing_admin.email}")
|
|
return
|
|
|
|
# 创建管理员角色
|
|
admin_role = Role(
|
|
role_name="超级管理员",
|
|
role_code="admin",
|
|
description="系统超级管理员，拥有所有权限",
|
|
is_system=True,
|
|
status="active",
|
|
created_by=0,
|
|
updated_by=0,
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow()
|
|
)
|
|
db.add(admin_role)
|
|
await db.flush()
|
|
|
|
# 创建管理员账号
|
|
admin_user = User(
|
|
username="admin",
|
|
password_hash=get_password_hash("admin123"),
|
|
real_name="系统管理员",
|
|
email="admin@workyai.cn",
|
|
phone="13800138000",
|
|
is_admin=True,
|
|
status="active",
|
|
created_by=0,
|
|
updated_by=0,
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow()
|
|
)
|
|
db.add(admin_user)
|
|
await db.flush()
|
|
|
|
# 分配管理员角色
|
|
from app.models.user import UserRole
|
|
user_role = UserRole(
|
|
user_id=admin_user.id,
|
|
role_id=admin_role.id,
|
|
created_by=0,
|
|
created_at=datetime.utcnow()
|
|
)
|
|
db.add(user_role)
|
|
|
|
await db.commit()
|
|
|
|
print("✅ 管理员账号创建成功！")
|
|
print(f"用户名: admin")
|
|
print(f"密码: admin123")
|
|
print(f"邮箱: admin@workyai.cn")
|
|
print(f"手机: 13800138000")
|
|
print("")
|
|
print("⚠️ 请登录后立即修改默认密码！")
|
|
|
|
except Exception as e:
|
|
await db.rollback()
|
|
print(f"❌ 创建管理员账号失败: {str(e)}")
|
|
raise
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(create_admin_user())
|