131 lines
4.3 KiB
Python
131 lines
4.3 KiB
Python
#!/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()
|
|
|