012-kaopeilian/backend/create_system_users.py

#!/usr/bin/env python3
"""
创建系统测试账户脚本
"""

import asyncio
import sys
from datetime import datetime
from pathlib import Path

# 添加项目根目录到 Python 路径
sys.path.insert(0, str(Path(__file__).parent))

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

from app.core.config import settings
from app.core.security import get_password_hash
from app.models.user import User
from app.models.base import Base

# 要创建的系统账户
SYSTEM_USERS = [
    {
        "username": "superadmin",
        "password": "Superadmin123!",
        "email": "superadmin@kaopeilian.com",
        "full_name": "超级管理员",
        "role": "super_admin",
        "phone": "13800000001",
        "is_active": True,
        "department": "系统管理部",
        "position": "超级管理员",
    },
    {
        "username": "admin",
        "password": "Admin123!",
        "email": "admin@kaopeilian.com",
        "full_name": "系统管理员",
        "role": "admin",
        "phone": "13800000002",
        "is_active": True,
        "department": "系统管理部",
        "position": "系统管理员",
    },
    {
        "username": "testuser",
        "password": "TestPass123!",
        "email": "testuser@kaopeilian.com",
        "full_name": "测试学员",
        "role": "student",
        "phone": "13800000004",
        "is_active": True,
        "department": "学员部",
        "position": "学员",
    },
]


async def create_system_users():
    """创建系统测试账户"""
    # 创建数据库引擎
    engine = create_async_engine(settings.DATABASE_URL, echo=False)
    
    # 创建会话
    async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    
    async with async_session() as session:
        created_users = []
        updated_users = []
        
        for user_data in SYSTEM_USERS:
            try:
                # 检查用户是否已存在
                result = await session.execute(
                    select(User).where(User.username == user_data["username"])
                )
                existing_user = result.scalar_one_or_none()
                
                if existing_user:
                    # 更新现有用户的密码和信息
                    existing_user.hashed_password = get_password_hash(user_data["password"])
                    existing_user.email = user_data["email"]
                    existing_user.full_name = user_data["full_name"]
                    existing_user.role = user_data["role"]
                    existing_user.phone = user_data["phone"]
                    existing_user.is_active = user_data["is_active"]
                    existing_user.department = user_data["department"]
                    existing_user.position = user_data["position"]
                    existing_user.is_deleted = False
                    existing_user.updated_at = datetime.utcnow()
                    
                    await session.commit()
                    updated_users.append(user_data["username"])
                    print(f"✓ 更新用户: {user_data['username']} ({user_data['full_name']})")
                else:
                    # 创建新用户
                    new_user = User(
                        username=user_data["username"],
                        hashed_password=get_password_hash(user_data["password"]),
                        email=user_data["email"],
                        full_name=user_data["full_name"],
                        role=user_data["role"],
                        phone=user_data["phone"],
                        is_active=user_data["is_active"],
                        department=user_data["department"],
                        position=user_data["position"],
                        created_at=datetime.utcnow(),
                        updated_at=datetime.utcnow(),
                    )
                    
                    session.add(new_user)
                    await session.commit()
                    created_users.append(user_data["username"])
                    print(f"✓ 创建用户: {user_data['username']} ({user_data['full_name']})")
                    
            except Exception as e:
                print(f"✗ 处理用户 {user_data['username']} 时出错: {str(e)}")
                await session.rollback()
        
        # 打印总结
        print("\n" + "="*50)
        print("系统用户创建/更新完成！")
        print(f"新创建用户数: {len(created_users)}")
        print(f"更新用户数: {len(updated_users)}")
        print("\n系统账户信息：")
        print("-"*50)
        print("| 角色       | 用户名     | 密码           | 权限说明                     |")
        print("| ---------- | ---------- | -------------- | ---------------------------- |")
        print("| 超级管理员 | superadmin | Superadmin123! | 系统最高权限，可管理所有功能 |")
        print("| 系统管理员 | admin      | Admin123!      | 可管理除“系统管理”模块外的全部功能（管理员仪表盘、用户管理、岗位管理、系统日志） |")
        print("| 测试学员   | testuser   | TestPass123!   | 可学习课程、参加考试和训练   |")
        print("-"*50)
        
    await engine.dispose()


if __name__ == "__main__":
    print("开始创建系统测试账户...")
    asyncio.run(create_system_users())