LQAdminPlatform/check_config.py

#!/usr/bin/env python3
"""
配置检查脚本
"""
import os
import sys
from urllib.parse import urlparse

def check_database_config():
    """检查数据库配置"""
    print("🔍 检查数据库配置...")
    
    # 检查.env文件是否存在
    if not os.path.exists('.env'):
        print("❌ .env文件不存在，请复制.env.example并修改配置")
        return False
    
    # 读取环境变量
    from dotenv import load_dotenv
    load_dotenv()
    
    database_url = os.getenv('DATABASE_URL')
    if not database_url:
        print("❌ DATABASE_URL未配置")
        return False
    
    # 解析数据库URL
    try:
        parsed = urlparse(database_url)
        print(f"✅ 数据库类型: {parsed.scheme}")
        print(f"✅ 数据库主机: {parsed.hostname}")
        print(f"✅ 数据库端口: {parsed.port}")
        print(f"✅ 数据库名称: {parsed.path[1:]}")  # 去掉开头的/
        print(f"✅ 数据库用户: {parsed.username}")
    except Exception as e:
        print(f"❌ 数据库URL格式错误: {e}")
        return False
    
    return True

def check_required_env():
    """检查必需的环境变量"""
    print("\n🔍 检查必需的环境变量...")
    
    required_vars = [
        'SECRET_KEY',
        'JWT_SECRET_KEY',
        'DATABASE_URL'
    ]
    
    missing_vars = []
    for var in required_vars:
        value = os.getenv(var)
        if not value:
            missing_vars.append(var)
        else:
            print(f"✅ {var}: {'*' * min(len(value), 10)}")
    
    if missing_vars:
        print(f"❌ 缺少必需的环境变量: {', '.join(missing_vars)}")
        return False
    
    return True

def test_database_connection():
    """测试数据库连接"""
    print("\n🔍 测试数据库连接...")
    
    try:
        import aiomysql
        import asyncio
        from urllib.parse import urlparse
        
        database_url = os.getenv('DATABASE_URL')
        parsed = urlparse(database_url)
        
        async def test_connection():
            try:
                conn = await aiomysql.connect(
                    host=parsed.hostname,
                    port=parsed.port or 3306,
                    user=parsed.username,
                    password=parsed.password,
                    db=parsed.path[1:] if parsed.path else None,
                    autocommit=True
                )
                await conn.ensure_closed()
                return True
            except Exception as e:
                print(f"❌ 数据库连接失败: {e}")
                return False
        
        result = asyncio.run(test_connection())
        if result:
            print("✅ 数据库连接成功")
            return True
        else:
            return False
            
    except ImportError:
        print("⚠️  aiomysql未安装，跳过数据库连接测试")
        return True
    except Exception as e:
        print(f"❌ 数据库连接测试失败: {e}")
        return False

def main():
    """主函数"""
    print("=" * 50)
    print("SSO后端配置检查")
    print("=" * 50)
    
    # 加载环境变量
    try:
        from dotenv import load_dotenv
        load_dotenv()
    except ImportError:
        print("❌ python-dotenv未安装，请先运行: pip install python-dotenv")
        sys.exit(1)
    
    checks = [
        check_database_config,
        check_required_env,
        test_database_connection
    ]
    
    all_passed = True
    for check in checks:
        if not check():
            all_passed = False
    
    print("\n" + "=" * 50)
    if all_passed:
        print("🎉 所有检查通过！可以运行初始化脚本了")
        print("\n下一步: python scripts/init_db.py")
    else:
        print("❌ 部分检查失败，请修复配置后重试")
        sys.exit(1)

if __name__ == "__main__":
    main()