#!/usr/bin/env python3 """ 简化的数据库初始化脚本 """ import sys import os # 添加项目根目录到Python路径 sys.path.append(os.path.join(os.path.dirname(__file__), 'src')) # 加载环境变量 from dotenv import load_dotenv load_dotenv() import pymysql from urllib.parse import urlparse import uuid from datetime import datetime def get_db_config(): """获取数据库配置""" database_url = os.getenv('DATABASE_URL', '') parsed = urlparse(database_url) return { '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' } def hash_password_simple(password): """简单的密码哈希""" import hashlib import secrets # 生成盐值 salt = secrets.token_hex(16) # 使用SHA256哈希 password_hash = hashlib.sha256((password + salt).encode()).hexdigest() return f"sha256${salt}${password_hash}" def generate_random_string(length=32): """生成随机字符串""" import secrets import string alphabet = string.ascii_letters + string.digits return ''.join(secrets.choice(alphabet) for _ in range(length)) def create_tables(connection): """创建数据库表""" cursor = connection.cursor() # 用户表 cursor.execute(""" CREATE TABLE IF NOT EXISTS users ( id VARCHAR(36) PRIMARY KEY COMMENT '用户ID', username VARCHAR(50) UNIQUE NOT NULL COMMENT '用户名', email VARCHAR(100) UNIQUE NOT NULL COMMENT '邮箱', phone VARCHAR(20) UNIQUE COMMENT '手机号', password_hash VARCHAR(255) NOT NULL COMMENT '密码哈希', avatar_url VARCHAR(500) COMMENT '头像URL', is_active BOOLEAN DEFAULT TRUE COMMENT '是否激活', is_superuser BOOLEAN DEFAULT FALSE COMMENT '是否超级管理员', last_login_at TIMESTAMP NULL COMMENT '最后登录时间', last_login_ip VARCHAR(45) COMMENT '最后登录IP', failed_login_attempts INT DEFAULT 0 COMMENT '失败登录次数', locked_until TIMESTAMP NULL COMMENT '锁定直到时间', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', is_deleted BOOLEAN DEFAULT FALSE COMMENT '是否删除', INDEX idx_username (username), INDEX idx_email (email), INDEX idx_phone (phone), INDEX idx_created_at (created_at) ) COMMENT='用户表' """) # 应用表 cursor.execute(""" CREATE TABLE IF NOT EXISTS apps ( id VARCHAR(36) PRIMARY KEY COMMENT '应用ID', name VARCHAR(100) NOT NULL COMMENT '应用名称', app_key VARCHAR(100) UNIQUE NOT NULL COMMENT '应用Key', app_secret VARCHAR(255) NOT NULL COMMENT '应用Secret', description TEXT COMMENT '应用描述', icon_url VARCHAR(500) COMMENT '应用图标', redirect_uris JSON NOT NULL COMMENT '回调URL列表', scope JSON COMMENT '权限范围', is_active BOOLEAN DEFAULT TRUE COMMENT '是否激活', is_trusted BOOLEAN DEFAULT FALSE COMMENT '是否受信任应用', access_token_expires INT DEFAULT 7200 COMMENT '访问令牌过期时间(秒)', refresh_token_expires INT DEFAULT 2592000 COMMENT '刷新令牌过期时间(秒)', created_by VARCHAR(36) COMMENT '创建者ID', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', is_deleted BOOLEAN DEFAULT FALSE COMMENT '是否删除', INDEX idx_app_key (app_key), INDEX idx_created_at (created_at) ) COMMENT='应用表' """) # 访问令牌表 cursor.execute(""" CREATE TABLE IF NOT EXISTS oauth_access_tokens ( id VARCHAR(36) PRIMARY KEY COMMENT '令牌ID', user_id VARCHAR(36) NOT NULL COMMENT '用户ID', app_id VARCHAR(36) COMMENT '应用ID', token VARCHAR(512) UNIQUE NOT NULL COMMENT '访问令牌', refresh_token VARCHAR(512) UNIQUE COMMENT '刷新令牌', token_type VARCHAR(50) DEFAULT 'Bearer' COMMENT '令牌类型', scope VARCHAR(500) COMMENT '权限范围', expires_at TIMESTAMP NOT NULL COMMENT '过期时间', revoked BOOLEAN DEFAULT FALSE COMMENT '是否撤销', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', last_used_at TIMESTAMP NULL COMMENT '最后使用时间', is_deleted BOOLEAN DEFAULT FALSE COMMENT '是否删除', INDEX idx_token (token), INDEX idx_refresh_token (refresh_token), INDEX idx_user_app (user_id, app_id), INDEX idx_expires_at (expires_at) ) COMMENT='访问令牌表' """) # 授权码表 cursor.execute(""" CREATE TABLE IF NOT EXISTS oauth_authorization_codes ( id VARCHAR(36) PRIMARY KEY COMMENT '授权码ID', user_id VARCHAR(36) NOT NULL COMMENT '用户ID', app_id VARCHAR(36) NOT NULL COMMENT '应用ID', code VARCHAR(100) UNIQUE NOT NULL COMMENT '授权码', redirect_uri VARCHAR(500) NOT NULL COMMENT '回调URL', scope VARCHAR(500) COMMENT '权限范围', expires_at TIMESTAMP NOT NULL COMMENT '过期时间', used BOOLEAN DEFAULT FALSE COMMENT '是否已使用', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', is_deleted BOOLEAN DEFAULT FALSE COMMENT '是否删除', INDEX idx_code (code), INDEX idx_user_app (user_id, app_id), INDEX idx_expires_at (expires_at) ) COMMENT='授权码表' """) # 登录日志表 cursor.execute(""" CREATE TABLE IF NOT EXISTS login_logs ( id VARCHAR(36) PRIMARY KEY COMMENT '日志ID', user_id VARCHAR(36) COMMENT '用户ID', username VARCHAR(50) NOT NULL COMMENT '用户名', login_type VARCHAR(20) COMMENT '登录方式', ip_address VARCHAR(45) COMMENT 'IP地址', user_agent TEXT COMMENT '用户代理', success BOOLEAN DEFAULT FALSE COMMENT '是否成功', failure_reason VARCHAR(200) COMMENT '失败原因', login_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '登录时间', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', is_deleted BOOLEAN DEFAULT FALSE COMMENT '是否删除', INDEX idx_user_id (user_id), INDEX idx_username (username), INDEX idx_login_at (login_at), INDEX idx_ip_address (ip_address) ) COMMENT='登录日志表' """) connection.commit() print("✅ 数据库表创建成功") def create_default_data(connection): """创建默认数据""" cursor = connection.cursor() # 检查是否已有管理员用户 cursor.execute("SELECT COUNT(*) FROM users WHERE username = 'admin'") if cursor.fetchone()[0] > 0: print("⚠️ 管理员用户已存在，跳过创建") # 获取现有的测试应用信息 cursor.execute("SELECT app_key, app_secret FROM apps WHERE name = '测试应用' LIMIT 1") result = cursor.fetchone() if result: return result[0], result[1] else: # 如果没有测试应用，创建一个 app_key = generate_random_string(32) app_secret = generate_random_string(64) return app_key, app_secret # 创建管理员用户 admin_id = str(uuid.uuid4()) admin_password = "Admin123456" password_hash = hash_password_simple(admin_password) cursor.execute(""" INSERT INTO users (id, username, email, password_hash, is_active, is_superuser, created_at, updated_at, is_deleted) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """, (admin_id, "admin", "admin@example.com", password_hash, True, True, datetime.now(), datetime.now(), False)) # 创建测试应用 app_id = str(uuid.uuid4()) app_key = generate_random_string(32) app_secret = generate_random_string(64) cursor.execute(""" INSERT INTO apps (id, name, app_key, app_secret, description, 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, %s, %s) """, ( app_id, "测试应用", app_key, app_secret, "用于测试的默认应用", '["http://localhost:3000/callback", "http://localhost:8080/callback", "http://localhost:8001/auth/callback"]', '["profile", "email"]', True, True, 7200, 2592000, admin_id, datetime.now(), datetime.now(), False )) connection.commit() print("✅ 默认数据创建成功") print(f"管理员账号: admin") print(f"管理员密码: {admin_password}") print(f"测试应用Key: {app_key}") print(f"测试应用Secret: {app_secret}") return app_key, app_secret def main(): """主函数""" print("=" * 50) print("SSO数据库初始化") print("=" * 50) try: # 获取数据库配置 config = get_db_config() print(f"连接数据库: {config['host']}:{config['port']}/{config['database']}") # 连接数据库 connection = pymysql.connect(**config) print("✅ 数据库连接成功") # 创建表 create_tables(connection) # 创建默认数据 result = create_default_data(connection) if result: app_key, app_secret = result else: app_key, app_secret = "demo_key", "demo_secret" # 关闭连接 connection.close() print("\n" + "=" * 50) print("🎉 数据库初始化完成!") print("\n下一步:") print("1. 启动后端服务: cd src && python -m app.main") print("2. 启动前端服务: cd ../sso-frontend && npm run dev") print(f"3. 配置子系统: CLIENT_ID={app_key}") print(f" CLIENT_SECRET={app_secret}") except Exception as e: print(f"❌ 数据库初始化失败: {e}") sys.exit(1) if __name__ == "__main__": main()