LQAdminPlatform/test_db_connection.py

#!/usr/bin/env python3
"""
测试数据库连接脚本
"""
import os
import pymysql
from dotenv import load_dotenv

def test_mysql_connection():
    """测试MySQL连接"""
    load_dotenv()
    
    # 从环境变量获取数据库配置
    database_url = os.getenv('DATABASE_URL', '')
    print(f"数据库URL: {database_url}")
    
    if not database_url:
        print("❌ DATABASE_URL未设置")
        return False
    
    # 解析数据库URL
    from urllib.parse import urlparse
    parsed = urlparse(database_url)
    
    config = {
        'host': parsed.hostname or 'localhost',
        'port': parsed.port or 3306,
        'user': parsed.username or 'root',
        'password': parsed.password or '',
        'charset': 'utf8mb4'
    }
    
    print(f"连接配置: host={config['host']}, port={config['port']}, user={config['user']}")
    
    try:
        # 测试连接（不指定数据库）
        connection = pymysql.connect(**config)
        print("✅ MySQL连接成功")
        
        # 检查数据库是否存在
        db_name = parsed.path[1:] if parsed.path else 'sso_db'
        cursor = connection.cursor()
        cursor.execute("SHOW DATABASES")
        databases = [db[0] for db in cursor.fetchall()]
        
        if db_name in databases:
            print(f"✅ 数据库 '{db_name}' 已存在")
        else:
            print(f"⚠️  数据库 '{db_name}' 不存在，需要创建")
            
            # 尝试创建数据库
            try:
                cursor.execute(f"CREATE DATABASE {db_name} CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
                print(f"✅ 数据库 '{db_name}' 创建成功")
            except Exception as e:
                print(f"❌ 创建数据库失败: {e}")
        
        cursor.close()
        connection.close()
        return True
        
    except Exception as e:
        print(f"❌ MySQL连接失败: {e}")
        print("\n可能的解决方案:")
        print("1. 检查MySQL服务是否启动")
        print("2. 检查用户名和密码是否正确")
        print("3. 检查主机和端口是否正确")
        print("4. 尝试在命令行连接: mysql -h localhost -u root -p")
        return False

if __name__ == "__main__":
    print("=" * 50)
    print("MySQL连接测试")
    print("=" * 50)
    
    if test_mysql_connection():
        print("\n🎉 数据库连接测试通过！")
        print("现在可以运行: python scripts/init_db.py")
    else:
        print("\n❌ 数据库连接测试失败")
        print("请修复数据库连接问题后重试")