jxgzcd/test_db_connection.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
数据库连接测试脚本
连接到MySQL数据库并测试连接是否成功
支持Windows和Linux (CentOS) 环境
"""

import pymysql
import sys
import os

# 设置控制台输出编码（Windows需要，Linux默认UTF-8）
if sys.platform == 'win32':
    try:
        os.system('chcp 65001 >nul 2>&1')
    except:
        pass
else:
    # Linux环境下设置UTF-8编码
    import locale
    try:
        locale.setlocale(locale.LC_ALL, 'en_US.UTF-8')
    except:
        try:
            locale.setlocale(locale.LC_ALL, 'C.UTF-8')
        except:
            pass

def test_db_connection():
    """测试数据库连接"""
    # 数据库连接配置
    db_config = {
        'host': '127.0.0.1',
        'port': 3306,
        'user': 'prod',
        'password': 'hmdmxjIvfIjIoflL',
        'database': 'iot',
        'charset': 'utf8mb4',
        'connect_timeout': 10  # 连接超时时间（秒）
    }
    
    connection = None
    try:
        # 尝试连接数据库
        print("正在尝试连接到数据库...")
        print(f"主机: {db_config['host']}")
        print(f"端口: {db_config['port']}")
        print(f"用户名: {db_config['user']}")
        print("-" * 50)
        
        connection = pymysql.connect(**db_config)
        
        # 如果连接成功，执行一个简单查询来验证
        with connection.cursor() as cursor:
            cursor.execute("SELECT VERSION()")
            version = cursor.fetchone()
            print("[成功] 数据库连接成功！")
            print(f"[成功] MySQL版本: {version[0]}")
            
            # 获取当前数据库名称
            cursor.execute("SELECT DATABASE()")
            db_name = cursor.fetchone()
            if db_name[0]:
                print(f"[成功] 当前数据库: {db_name[0]}")
            else:
                print("[成功] 当前未选择数据库")
            
            # 获取连接ID
            cursor.execute("SELECT CONNECTION_ID()")
            conn_id = cursor.fetchone()
            print(f"[成功] 连接ID: {conn_id[0]}")
        
        return True
        
    except pymysql.Error as e:
        error_code, error_msg = e.args
        print("[失败] 数据库连接失败！")
        print(f"[失败] 错误代码: {error_code}")
        print(f"[失败] 错误信息: {error_msg}")
        return False
        
    except Exception as e:
        print("[失败] 发生未知错误！")
        print(f"[失败] 错误信息: {e}")
        return False
        
    finally:
        # 关闭数据库连接
        if connection:
            connection.close()
            print("-" * 50)
            print("[成功] 数据库连接已关闭")

if __name__ == "__main__":
    print("=" * 50)
    print("数据库连接测试")
    print("=" * 50)
    
    success = test_db_connection()
    
    print("=" * 50)
    if success:
        print("测试结果: 成功")
        sys.exit(0)
    else:
        print("测试结果: 失败")
        sys.exit(1)