LQAdminPlatform/scripts/init_db.py

"""
数据库初始化脚本
"""
import asyncio
import sys
import os

# 添加项目根目录到Python路径
sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'src'))

# 加载环境变量
from dotenv import load_dotenv
# 加载配置文件 - 从 src/app/config/.env
env_path = os.path.join(os.path.dirname(__file__), "..", "src", "app", "config", ".env")
load_dotenv(dotenv_path=env_path)

from app.base import init_db, get_engine
engine = get_engine()
from app.models import *  # 导入所有模型
from app.utils.security import hash_password, generate_app_key, generate_app_secret
from sqlalchemy.ext.asyncio import AsyncSession
from datetime import datetime
import uuid


async def create_default_data():
    """创建默认数据"""
    async with AsyncSession(engine) as session:
        try:
            # 创建默认超级管理员
            admin_password = "Admin123456"  # 使用较短的密码
            admin_user = User(
                id=str(uuid.uuid4()),
                username="admin",
                email="admin@example.com",
                password_hash=hash_password(admin_password),
                is_active=True,
                is_superuser=True,
                created_at=datetime.utcnow(),
                updated_at=datetime.utcnow()
            )
            session.add(admin_user)
            
            # 创建默认角色
            admin_role = Role(
                id=str(uuid.uuid4()),
                name="超级管理员",
                code="super_admin",
                description="系统超级管理员角色",
                is_active=True,
                created_at=datetime.utcnow(),
                updated_at=datetime.utcnow()
            )
            session.add(admin_role)
            
            user_role = Role(
                id=str(uuid.uuid4()),
                name="普通用户",
                code="user",
                description="普通用户角色",
                is_active=True,
                created_at=datetime.utcnow(),
                updated_at=datetime.utcnow()
            )
            session.add(user_role)
            
            # 创建默认权限
            permissions = [
                ("用户管理", "user:manage", "管理用户", "user", "manage"),
                ("用户查看", "user:view", "查看用户", "user", "view"),
                ("应用管理", "app:manage", "管理应用", "app", "manage"),
                ("应用查看", "app:view", "查看应用", "app", "view"),
                ("系统管理", "system:manage", "系统管理", "system", "manage"),
            ]
            
            permission_objects = []
            for name, code, desc, resource, action in permissions:
                permission = Permission(
                    id=str(uuid.uuid4()),
                    name=name,
                    code=code,
                    description=desc,
                    resource=resource,
                    action=action,
                    is_active=True,
                    created_at=datetime.utcnow(),
                    updated_at=datetime.utcnow()
                )
                session.add(permission)
                permission_objects.append(permission)
            
            await session.flush()  # 获取ID
            
            # 分配用户角色
            user_role_rel = UserRole(
                id=str(uuid.uuid4()),
                user_id=admin_user.id,
                role_id=admin_role.id,
                created_at=datetime.utcnow(),
                updated_at=datetime.utcnow()
            )
            session.add(user_role_rel)
            
            # 分配角色权限
            for permission in permission_objects:
                role_permission = RolePermission(
                    id=str(uuid.uuid4()),
                    role_id=admin_role.id,
                    permission_id=permission.id,
                    created_at=datetime.utcnow(),
                    updated_at=datetime.utcnow()
                )
                session.add(role_permission)
            
            # 创建默认测试应用
            test_app = App(
                id=str(uuid.uuid4()),
                name="测试应用",
                app_key=generate_app_key(),
                app_secret=generate_app_secret(),
                description="用于测试的默认应用",
                redirect_uris=["http://localhost:3000/callback", "http://localhost:8080/callback"],
                scope=["profile", "email"],
                is_active=True,
                is_trusted=True,
                access_token_expires=7200,
                refresh_token_expires=2592000,
                created_by=admin_user.id,
                created_at=datetime.utcnow(),
                updated_at=datetime.utcnow()
            )
            session.add(test_app)
            
            await session.commit()
            print("✅ 默认数据创建成功")
            print(f"管理员账号: admin")
            print(f"管理员密码: {admin_password}")
            print(f"测试应用Key: {test_app.app_key}")
            print(f"测试应用Secret: {test_app.app_secret}")
            
        except Exception as e:
            await session.rollback()
            print(f"❌ 创建默认数据失败: {e}")
            raise


async def main():
    """主函数"""
    print("🚀 开始初始化数据库...")
    
    try:
        # 创建所有表
        await init_db()
        print("✅ 数据库表创建成功")
        
        # 创建默认数据
        await create_default_data()
        
        print("🎉 数据库初始化完成!")
        
    except Exception as e:
        print(f"❌ 数据库初始化失败: {e}")
        sys.exit(1)


if __name__ == "__main__":
    asyncio.run(main())