LQAdminPlatform/create_test_app.py

#!/usr/bin/env python3
"""
创建测试应用
"""
import sys
import os
import pymysql
from urllib.parse import urlparse
import uuid
import json
from dotenv import load_dotenv

load_dotenv()

def get_db_connection():
    """获取数据库连接"""
    try:
        database_url = os.getenv('DATABASE_URL', '')
        if not database_url:
            return None
            
        parsed = urlparse(database_url)
        config = {
            'host': parsed.hostname or 'localhost',
            'port': parsed.port or 3306,
            'user': parsed.username or 'root',
            'password': parsed.password or '',
            'database': parsed.path[1:] if parsed.path else 'sso_db',
            'charset': 'utf8mb4'
        }
        
        return pymysql.connect(**config)
    except Exception as e:
        print(f"数据库连接失败: {e}")
        return None

def create_test_app():
    """创建测试应用"""
    conn = get_db_connection()
    if not conn:
        print("❌ 数据库连接失败")
        return
    
    cursor = conn.cursor()
    
    # 检查是否已存在测试应用
    cursor.execute("SELECT id FROM apps WHERE app_key = %s", ("eqhoIdAyAWbA8MsYHsNqQqNLJbCayTjY",))
    if cursor.fetchone():
        print("✅ 测试应用已存在")
        cursor.close()
        conn.close()
        return
    
    # 获取admin用户ID
    cursor.execute("SELECT id FROM users WHERE username = 'admin'")
    admin_result = cursor.fetchone()
    if not admin_result:
        print("❌ 未找到admin用户")
        cursor.close()
        conn.close()
        return
    
    admin_id = admin_result[0]
    
    # 创建测试应用
    app_id = str(uuid.uuid4())
    app_key = "eqhoIdAyAWbA8MsYHsNqQqNLJbCayTjY"
    app_secret = "LKJm5XHJFhhgxSv9nQhoQNNI3wrKyWGZCaPQ4qc43Lf5qfXdLAHoGAHhCYqApEpr"
    
    cursor.execute("""
        INSERT INTO apps (
            id, name, app_key, app_secret, description, icon_url,
            redirect_uris, scope, is_active, is_trusted,
            access_token_expires, refresh_token_expires, created_by,
            created_at, updated_at, is_deleted
        ) VALUES (
            %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), %s
        )
    """, (
        app_id,
        "子系统演示应用",
        app_key,
        app_secret,
        "用于演示SSO集成的子系统应用",
        "",
        json.dumps([
            "http://localhost:8001/auth/callback",
            "http://localhost:3001/auth/callback"
        ]),
        json.dumps(["profile", "email", "phone"]),
        True,  # is_active
        True,  # is_trusted
        7200,  # access_token_expires (2小时)
        2592000,  # refresh_token_expires (30天)
        admin_id,
        False  # is_deleted
    ))
    
    conn.commit()
    cursor.close()
    conn.close()
    
    print("✅ 测试应用创建成功")
    print(f"应用名称: 子系统演示应用")
    print(f"App Key: {app_key}")
    print(f"App Secret: {app_secret}")
    print(f"回调URL: http://localhost:8001/auth/callback")
    print(f"权限范围: profile, email, phone")
    print(f"受信任应用: 是")

if __name__ == "__main__":
    create_test_app()