ai_member_xiaoban/test_mysql_pg.py

#!/usr/bin/env python3
"""
MySQL 和 PostgreSQL 连接测试脚本
仅用于测试连接和读取基本信息，不进行任何写入操作
"""

import warnings
warnings.filterwarnings('ignore')

def test_mysql_connection(host, port, user, password, description):
    """测试 MySQL 连接"""
    try:
        import pymysql
        
        print(f"\n{'='*60}")
        print(f"测试: {description}")
        print(f"地址: {host}:{port}")
        print(f"{'='*60}")
        
        # 尝试连接
        connection = pymysql.connect(
            host=host,
            port=port,
            user=user,
            password=password,
            connect_timeout=10,
            read_timeout=10
        )
        
        print(f"✅ 连接成功!")
        
        # 获取服务器信息
        with connection.cursor() as cursor:
            cursor.execute("SELECT VERSION()")
            version = cursor.fetchone()
            print(f"   版本: {version[0] if version else 'N/A'}")
            
            # 获取数据库列表
            cursor.execute("SHOW DATABASES")
            databases = cursor.fetchall()
            print(f"   数据库数量: {len(databases)}")
            if databases:
                print(f"   数据库示例: {', '.join([db[0] for db in databases[:5]])}")
        
        connection.close()
        return True
        
    except Exception as e:
        print(f"❌ 连接异常: {str(e)[:200]}")
        return False

def test_postgresql_connection(host, port, user, password, description):
    """测试 PostgreSQL 连接"""
    try:
        import psycopg2
        
        print(f"\n{'='*60}")
        print(f"测试: {description}")
        print(f"地址: {host}:{port}")
        print(f"{'='*60}")
        
        # 尝试连接 - 先尝试连接 postgres 数据库
        try:
            connection = psycopg2.connect(
                host=host,
                port=port,
                user=user,
                password=password,
                dbname='postgres',
                connect_timeout=10
            )
        except:
            # 如果 postgres 数据库连接失败，尝试不指定数据库
            print(f"   尝试不指定数据库连接...")
            connection = psycopg2.connect(
                host=host,
                port=port,
                user=user,
                password=password,
                connect_timeout=10
            )
        
        print(f"✅ 连接成功!")
        
        # 获取服务器信息
        with connection.cursor() as cursor:
            cursor.execute("SELECT version()")
            version = cursor.fetchone()
            print(f"   版本: {version[0].split()[0] if version else 'N/A'}")
            
            # 获取数据库列表
            try:
                cursor.execute("SELECT datname FROM pg_database WHERE datistemplate = false")
                databases = cursor.fetchall()
                print(f"   数据库数量: {len(databases)}")
                if databases:
                    print(f"   数据库示例: {', '.join([db[0] for db in databases[:5]])}")
            except:
                print(f"   无法获取数据库列表（权限限制）")
        
        connection.close()
        return True
        
    except Exception as e:
        print(f"❌ 连接异常: {str(e)[:200]}")
        return False

def main():
    print("="*60)
    print("MySQL 和 PostgreSQL 数据库连接测试")
    print("注意: 仅进行连接测试和只读操作")
    print("="*60)
    
    results = {}
    
    # MySQL 配置
    mysql_configs = [
        {
            "description": "Online MySQL (线上版本)",
            "host": "bj-cdb-dh2fkqa0.sql.tencentcdb.com",
            "port": 27751,
            "user": "read_only",
            "password": "fsdo45ijfmfmuu77$%^&"
        },
        {
            "description": "Test MySQL (测试环境)",
            "host": "bj-cdb-8frbdwju.sql.tencentcdb.com",
            "port": 25413,
            "user": "read_only",
            "password": "fdsfiidier^$*hjfdijjd232"
        }
    ]
    
    # PostgreSQL 配置（更新后的配置）
    pg_configs = [
        {
            "description": "Online PostgreSQL (正式环境用户行为数据)",
            "host": "bj-postgres-16pob4sg.sql.tencentcdb.com",
            "port": 28591,
            "user": "ai_member",
            "password": "LdfjdjL83h3h3^$&**YGG*"
        },
        {
            "description": "Test PostgreSQL (测试环境行为数据)",
            "host": "bj-postgres-642mcico.sql.tencentcdb.com",
            "port": 21531,
            "user": "ai_member",
            "password": "dsjsLGU&%$%FG*((yy9y8"
        }
    ]
    
    # 测试 MySQL 连接
    print("\n" + "="*60)
    print("测试 MySQL 数据库")
    print("="*60)
    for config in mysql_configs:
        result = test_mysql_connection(**config)
        results[config["description"]] = result
    
    # 测试 PostgreSQL 连接
    print("\n" + "="*60)
    print("测试 PostgreSQL 数据库")
    print("="*60)
    for config in pg_configs:
        result = test_postgresql_connection(**config)
        results[config["description"]] = result
    
    # 总结
    print("\n" + "="*60)
    print("测试总结")
    print("="*60)
    for name, result in results.items():
        status = "✅ 成功" if result else "❌ 失败"
        print(f"{name}: {status}")

if __name__ == "__main__":
    main()