backend/scripts/generate_test_account.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
CarrotSkin 测试账户生成器
此脚本创建新用户、角色，并输出所有测试信息
"""

import json
import random
import sys
from datetime import datetime
from pathlib import Path

try:
    import requests
except ImportError:
    print("错误: 需要安装 requests 库")
    print("请运行: pip install requests")
    sys.exit(1)

# 颜色输出支持（Windows 和 Unix 都支持）
try:
    from colorama import init, Fore, Style
    init(autoreset=True)
    HAS_COLORAMA = True
    # colorama 没有 GRAY，使用 WHITE 代替
    Fore.GRAY = Fore.WHITE
except ImportError:
    # 如果没有 colorama，使用空字符串
    class Fore:
        CYAN = YELLOW = GREEN = RED = GRAY = WHITE = RESET = ""
    class Style:
        RESET_ALL = ""
    HAS_COLORAMA = False

# API 基础 URL
BASE_URL = "http://localhost:8080/api/v1"


def print_colored(text, color=Fore.RESET):
    """打印彩色文本"""
    print(f"{color}{text}{Style.RESET_ALL}")


def print_header(text):
    """打印标题"""
    print_colored(f"\n{'=' * 40}", Fore.CYAN)
    print_colored(f"  {text}", Fore.CYAN)
    print_colored(f"{'=' * 40}\n", Fore.CYAN)


def print_step(text):
    """打印步骤标题"""
    print_colored(f"\n=== {text} ===", Fore.YELLOW)


def print_success(text):
    """打印成功消息"""
    print_colored(f"✓ {text}", Fore.GREEN)


def print_error(text):
    """打印错误消息"""
    print_colored(f"✗ {text}", Fore.RED)


def print_info(text):
    """打印信息消息"""
    print_colored(f"  {text}", Fore.GRAY)


def register_user():
    """步骤1: 注册新用户"""
    print_step("步骤1: 注册新用户")
    
    random_num = random.randint(10000, 99999)
    username = f"testuser{random_num}"
    email = f"test{random_num}@example.com"
    login_password = "password123"
    verification_code = "123456"
    
    print_info(f"用户名: {username}")
    print_info(f"邮箱: {email}")
    print_info(f"密码: {login_password}")
    
    register_data = {
        "username": username,
        "email": email,
        "password": login_password,
        "verification_code": verification_code
    }
    
    try:
        response = requests.post(
            f"{BASE_URL}/auth/register",
            json=register_data,
            headers={"Content-Type": "application/json"},
            timeout=10
        )
        response.raise_for_status()
        result = response.json()
        
        if result.get("code") == 200:
            jwt_token = result["data"]["token"]
            user_id = result["data"]["user_info"]["id"]
            user_role = result["data"]["user_info"]["role"]
            
            print_success("注册成功!")
            print_info(f"用户ID: {user_id}")
            print_info(f"角色: {user_role}")
            
            return {
                "username": username,
                "email": email,
                "login_password": login_password,
                "jwt_token": jwt_token,
                "user_id": user_id,
                "user_role": user_role
            }
        else:
            print_error(f"注册失败: {result.get('message', '未知错误')}")
            sys.exit(1)
    except requests.exceptions.RequestException as e:
        print_error(f"注册失败: {str(e)}")
        sys.exit(1)


def create_profile(jwt_token):
    """步骤2: 创建Minecraft角色"""
    print_step("步骤2: 创建Minecraft角色")
    
    profile_name = f"TestPlayer{random.randint(100, 999)}"
    print_info(f"角色名: {profile_name}")
    
    profile_data = {
        "name": profile_name
    }
    
    headers = {
        "Authorization": f"Bearer {jwt_token}",
        "Content-Type": "application/json"
    }
    
    try:
        response = requests.post(
            f"{BASE_URL}/profile/",
            json=profile_data,
            headers=headers,
            timeout=10
        )
        response.raise_for_status()
        result = response.json()
        
        if result.get("code") == 200:
            profile_uuid = result["data"]["uuid"]
            print_success("角色创建成功!")
            print_info(f"角色UUID: {profile_uuid}")
            print_info(f"角色名: {result['data']['name']}")
            
            return {
                "profile_name": profile_name,
                "profile_uuid": profile_uuid
            }
        else:
            print_error(f"角色创建失败: {result.get('message', '未知错误')}")
            sys.exit(1)
    except requests.exceptions.RequestException as e:
        print_error(f"角色创建失败: {str(e)}")
        sys.exit(1)


def reset_yggdrasil_password(jwt_token):
    """步骤3: 重置并获取Yggdrasil密码"""
    print_step("步骤3: 重置并获取Yggdrasil密码")
    
    headers = {
        "Authorization": f"Bearer {jwt_token}",
        "Content-Type": "application/json"
    }
    
    try:
        response = requests.post(
            f"{BASE_URL}/user/yggdrasil-password/reset",
            headers=headers,
            timeout=10
        )
        response.raise_for_status()
        result = response.json()
        
        if result.get("code") == 200:
            yggdrasil_password = result["data"]["password"]
            print_success("Yggdrasil密码重置成功!")
            print_info(f"Yggdrasil密码: {yggdrasil_password}")
            return yggdrasil_password
        else:
            print_error(f"Yggdrasil密码重置失败: {result.get('message', '未知错误')}")
            return ""
    except requests.exceptions.RequestException as e:
        print_error(f"Yggdrasil密码重置失败: {str(e)}")
        return ""


def generate_output(user_info, profile_info, yggdrasil_password):
    """生成输出信息"""
    output = f"""========================================
  CarrotSkin 测试账户信息
========================================

=== 账户信息 ===
用户名: {user_info['username']}
邮箱: {user_info['email']}
登录密码: {user_info['login_password']}
用户ID: {user_info['user_id']}

=== JWT Token (API认证) ===
Token: {user_info['jwt_token']}

=== 角色信息 ===
角色名: {profile_info['profile_name']}
角色UUID: {profile_info['profile_uuid']}

=== Yggdrasil信息 ===
Yggdrasil密码: {yggdrasil_password}

=== 测试命令 ===

# 1. API登录
curl -X POST http://localhost:8080/api/v1/auth/login \\
  -H 'Content-Type: application/json' \\
  -d '{{"username":"{user_info['username']}","password":"{user_info['login_password']}"}}'

# 2. 创建角色
curl -X POST http://localhost:8080/api/v1/profile/ \\
  -H 'Content-Type: application/json' \\
  -H 'Authorization: Bearer {user_info['jwt_token']}' \\
  -d '{{"name":"NewProfile"}}'

# 3. 重置Yggdrasil密码
curl -X POST http://localhost:8080/api/v1/user/yggdrasil-password/reset \\
  -H 'Content-Type: application/json' \\
  -H 'Authorization: Bearer {user_info['jwt_token']}'

# 4. Yggdrasil认证
curl -X POST http://localhost:8080/api/v1/yggdrasil/authserver/authenticate \\
  -H 'Content-Type: application/json' \\
  -d '{{
    "username": "{user_info['username']}",
    "password": "{yggdrasil_password}",
    "requestUser": true
  }}'

# 5. 获取角色信息
curl -X GET http://localhost:8080/api/v1/profile/{profile_info['profile_uuid']}

========================================
"""
    return output


def set_user_role(admin_token, user_id, role):
    """设置用户角色"""
    headers = {
        "Authorization": f"Bearer {admin_token}",
        "Content-Type": "application/json"
    }
    
    try:
        response = requests.put(
            f"{BASE_URL}/admin/users/role",
            json={"user_id": user_id, "role": role},
            headers=headers,
            timeout=10
        )
        response.raise_for_status()
        result = response.json()
        
        if result.get("code") == 200:
            print_success(f"用户角色设置为: {role}")
            return True
        else:
            print_error(f"设置角色失败: {result.get('message', '未知错误')}")
            return False
    except requests.exceptions.RequestException as e:
        print_error(f"设置角色失败: {str(e)}")
        return False


def login_user(username, password):
    """登录用户"""
    try:
        response = requests.post(
            f"{BASE_URL}/auth/login",
            json={"username": username, "password": password},
            headers={"Content-Type": "application/json"},
            timeout=10
        )
        response.raise_for_status()
        result = response.json()
        
        if result.get("code") == 200:
            return result["data"]["token"]
        return None
    except requests.exceptions.RequestException:
        return None


def create_admin_user():
    """创建管理员用户"""
    print_step("创建管理员用户")
    
    random_num = random.randint(10000, 99999)
    username = f"admin{random_num}"
    email = f"admin{random_num}@example.com"
    login_password = "admin123456"
    verification_code = "123456"
    
    print_info(f"用户名: {username}")
    print_info(f"邮箱: {email}")
    print_info(f"密码: {login_password}")
    
    register_data = {
        "username": username,
        "email": email,
        "password": login_password,
        "verification_code": verification_code
    }
    
    try:
        response = requests.post(
            f"{BASE_URL}/auth/register",
            json=register_data,
            headers={"Content-Type": "application/json"},
            timeout=10
        )
        response.raise_for_status()
        result = response.json()
        
        if result.get("code") == 200:
            jwt_token = result["data"]["token"]
            user_id = result["data"]["user_info"]["id"]
            
            print_success("用户注册成功!")
            print_info(f"用户ID: {user_id}")
            
            # 使用默认管理员账户提升权限
            print_info("尝试使用默认管理员账户提升权限...")
            default_admin_token = login_user("admin", "admin123456")
            
            if default_admin_token:
                if set_user_role(default_admin_token, user_id, "admin"):
                    print_success("管理员权限设置成功!")
                    return {
                        "username": username,
                        "email": email,
                        "password": login_password,
                        "jwt_token": jwt_token,
                        "user_id": user_id,
                        "role": "admin"
                    }
            
            print_error("无法提升权限，请手动设置")
            return {
                "username": username,
                "email": email,
                "password": login_password,
                "jwt_token": jwt_token,
                "user_id": user_id,
                "role": "user"
            }
        else:
            print_error(f"注册失败: {result.get('message', '未知错误')}")
            return None
    except requests.exceptions.RequestException as e:
        print_error(f"注册失败: {str(e)}")
        return None


def generate_admin_output(admin_info):
    """生成管理员账户输出信息"""
    output = f"""========================================
  CarrotSkin 管理员账户信息
========================================

=== 账户信息 ===
用户名: {admin_info['username']}
邮箱: {admin_info['email']}
密码: {admin_info['password']}
用户ID: {admin_info['user_id']}
角色: {admin_info['role']}

=== JWT Token (API认证) ===
Token: {admin_info['jwt_token']}

=== 管理员API示例 ===

# 1. 获取用户列表
curl -X GET "{BASE_URL}/admin/users" \\
  -H "Authorization: Bearer {admin_info['jwt_token']}"

# 2. 设置用户角色为管理员
curl -X PUT "{BASE_URL}/admin/users/role" \\
  -H "Content-Type: application/json" \\
  -H "Authorization: Bearer {admin_info['jwt_token']}" \\
  -d '{{"user_id": 1, "role": "admin"}}'

# 3. 获取材质列表（审核）
curl -X GET "{BASE_URL}/admin/textures" \\
  -H "Authorization: Bearer {admin_info['jwt_token']}"

# 4. 删除材质
curl -X DELETE "{BASE_URL}/admin/textures/1" \\
  -H "Authorization: Bearer {admin_info['jwt_token']}"

========================================
"""
    return output


def main():
    """主函数"""
    print_header("CarrotSkin 测试账户生成器")
    
    # 选择模式
    print("请选择操作:")
    print("  1. 创建普通测试用户")
    print("  2. 创建管理员用户")
    print("  3. 创建两者")
    
    choice = input("\n请输入选项 (1/2/3) [默认: 1]: ").strip() or "1"
    
    if choice in ["1", "3"]:
        # 创建普通用户
        print_header("创建普通测试用户")
        
        # 步骤1: 注册用户
        user_info = register_user()
        
        # 步骤2: 创建角色
        profile_info = create_profile(user_info["jwt_token"])
        
        # 步骤3: 重置Yggdrasil密码
        yggdrasil_password = reset_yggdrasil_password(user_info["jwt_token"])
        
        # 输出信息
        print_header("普通用户信息汇总")
        output = generate_output(user_info, profile_info, yggdrasil_password)
        print(output)
        
        # 保存到文件
        output_file = f"test_account_{user_info['username']}.txt"
        try:
            with open(output_file, "w", encoding="utf-8") as f:
                f.write(output)
            print_success(f"信息已保存到文件: {output_file}")
        except Exception as e:
            print_error(f"保存文件失败: {str(e)}")
    
    if choice in ["2", "3"]:
        # 创建管理员用户
        print_header("创建管理员用户")
        admin_info = create_admin_user()
        
        if admin_info:
            print_header("管理员账户信息汇总")
            admin_output = generate_admin_output(admin_info)
            print(admin_output)
            
            # 保存到文件
            admin_output_file = f"admin_account_{admin_info['username']}.txt"
            try:
                with open(admin_output_file, "w", encoding="utf-8") as f:
                    f.write(admin_output)
                print_success(f"管理员信息已保存到文件: {admin_output_file}")
            except Exception as e:
                print_error(f"保存文件失败: {str(e)}")
    
    print_header("测试完成!")


if __name__ == "__main__":
    main()