assist/test_mysql_connection.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
MySQL连接诊断工具
"""

import socket
import sys
import os
from urllib.parse import urlparse

# 添加项目路径以导入编码工具
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

try:
    from src.utils.encoding_helper import setup_utf8_output, safe_print
    setup_utf8_output()
except ImportError:
    # 如果导入失败，使用本地实现
    import io
    if sys.platform == 'win32':
        try:
            sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
            sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace')
            os.system('chcp 65001 >nul 2>&1')
        except:
            pass
    safe_print = print

def test_port(host, port, timeout=5):
    """测试端口是否开放"""
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        result = sock.connect_ex((host, port))
        sock.close()
        return result == 0
    except Exception as e:
        print(f"端口测试异常: {e}")
        return False

def test_mysql_connection():
    """测试MySQL连接"""
    # 从配置中提取连接信息
    db_url = "mysql+pymysql://tsp_assistant:123456@jeason.online/tsp_assistant?charset=utf8mb4"
    
    parsed = urlparse(db_url.replace("mysql+pymysql://", "http://"))
    host = parsed.hostname
    port = parsed.port or 3306
    
    # 使用安全的UTF-8输出
    safe_print("=" * 60)
    safe_print("MySQL连接诊断工具")
    safe_print("=" * 60)
    safe_print(f"主机: {host}")
    safe_print(f"端口: {port}")
    safe_print()
    
    # 1. 测试网络连通性
    safe_print("[1] 测试网络连通性 (Ping)...")
    try:
        import subprocess
        result = subprocess.run(
            ["ping", "-n", "2", host],
            capture_output=True,
            text=True,
            timeout=10
        )
        if "TTL" in result.stdout or "TTL" in result.stderr:
            safe_print("[OK] 网络连通正常")
        else:
            safe_print("[X] 网络不通")
    except Exception as e:
        safe_print(f"[!] Ping测试失败: {e}")
    safe_print()
    
    # 2. 测试端口是否开放
    safe_print(f"[2] 测试端口 {port} 是否开放...")
    if test_port(host, port, timeout=10):
        safe_print(f"[OK] 端口 {port} 开放")
    else:
        safe_print(f"[X] 端口 {port} 无法连接")
        safe_print()
        safe_print("可能的原因:")
        safe_print("  1. MySQL服务器防火墙未开放3306端口")
        safe_print("  2. MySQL配置只允许localhost连接")
        safe_print("  3. 云服务商安全组规则阻止了3306端口")
        safe_print("  4. MySQL服务未启动")
        safe_print()
        safe_print("解决方案:")
        safe_print("  1. 检查MySQL服务器防火墙配置:")
        safe_print("     - Linux: sudo ufw allow 3306/tcp")
        safe_print("     - Windows: 在防火墙中添加入站规则")
        safe_print("  2. 检查MySQL配置文件 (my.cnf 或 my.ini):")
        safe_print("     - 确保 bind-address = 0.0.0.0 (不是127.0.0.1)")
        safe_print("  3. 检查云服务商安全组:")
        safe_print("     - 添加规则允许3306端口入站")
        safe_print("  4. 使用SSH隧道连接:")
        safe_print("     ssh -L 3306:localhost:3306 user@jeason.online")
    safe_print()
    
    # 3. 尝试使用PyMySQL连接
    safe_print("[3] 尝试使用PyMySQL连接...")
    try:
        import pymysql
        connection = pymysql.connect(
            host=host,
            port=port,
            user="tsp_assistant",
            password="123456",
            database="tsp_assistant",
            connect_timeout=10,
            read_timeout=10,
            write_timeout=10
        )
        safe_print("[OK] MySQL连接成功!")
        connection.close()
    except Exception as e:
        safe_print(f"[X] MySQL连接失败: {e}")
        safe_print()
        if "timed out" in str(e).lower() or "can't connect" in str(e).lower():
            safe_print("这是连接超时错误，说明端口无法访问。")
            safe_print("请按照上面的解决方案检查服务器配置。")
    safe_print()
    
    safe_print("=" * 60)

if __name__ == "__main__":
    test_mysql_connection()