197 lines
5.7 KiB
Python
197 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
微信 API 连接测试脚本
|
||
用于诊断微信登录 502 问题
|
||
"""
|
||
|
||
import sys
|
||
import time
|
||
|
||
def test_requests():
|
||
"""测试 requests 库是否正常工作"""
|
||
print("=" * 50)
|
||
print("🔍 测试 1: requests 库基本功能")
|
||
print("=" * 50)
|
||
|
||
try:
|
||
import requests
|
||
|
||
# 测试普通 HTTPS 请求
|
||
print(" 测试 HTTPS 请求...")
|
||
start = time.time()
|
||
response = requests.get("https://httpbin.org/get", timeout=10)
|
||
elapsed = time.time() - start
|
||
print(f" ✅ httpbin.org 请求成功: {response.status_code} ({elapsed:.2f}s)")
|
||
|
||
# 测试微信 API 连通性
|
||
print(" 测试微信 API 连通性...")
|
||
start = time.time()
|
||
response = requests.get("https://api.weixin.qq.com/", timeout=10)
|
||
elapsed = time.time() - start
|
||
print(f" ✅ 微信 API 可达: {response.status_code} ({elapsed:.2f}s)")
|
||
|
||
return True
|
||
except Exception as e:
|
||
print(f" ❌ 请求失败: {e}")
|
||
return False
|
||
|
||
|
||
def test_redis():
|
||
"""测试 Redis 连接"""
|
||
print("\n" + "=" * 50)
|
||
print("🔍 测试 2: Redis 连接")
|
||
print("=" * 50)
|
||
|
||
try:
|
||
import redis
|
||
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
|
||
|
||
# 测试连接
|
||
r.ping()
|
||
print(" ✅ Redis 连接成功")
|
||
|
||
# 测试读写
|
||
r.setex("test_key", 10, "test_value")
|
||
value = r.get("test_key")
|
||
r.delete("test_key")
|
||
print(f" ✅ Redis 读写测试成功: {value}")
|
||
|
||
return True
|
||
except Exception as e:
|
||
print(f" ❌ Redis 连接失败: {e}")
|
||
return False
|
||
|
||
|
||
def test_gevent():
|
||
"""测试 gevent 环境"""
|
||
print("\n" + "=" * 50)
|
||
print("🔍 测试 3: Gevent 环境")
|
||
print("=" * 50)
|
||
|
||
try:
|
||
from gevent import monkey
|
||
is_patched = monkey.is_module_patched('socket')
|
||
print(f" Gevent monkey patched: {is_patched}")
|
||
|
||
if is_patched:
|
||
print(" ✅ Gevent 已正确 patch")
|
||
else:
|
||
print(" ⚠️ Gevent 未 patch，在 Gunicorn 下应该会自动 patch")
|
||
|
||
return True
|
||
except ImportError:
|
||
print(" ⚠️ Gevent 未安装")
|
||
return False
|
||
|
||
|
||
def test_wechat_access_token():
|
||
"""测试微信 access_token 获取（模拟）"""
|
||
print("\n" + "=" * 50)
|
||
print("🔍 测试 4: 微信 API 调用（模拟）")
|
||
print("=" * 50)
|
||
|
||
try:
|
||
import requests
|
||
|
||
# 使用一个无效的 code 测试 API 是否可达
|
||
url = "https://api.weixin.qq.com/sns/oauth2/access_token"
|
||
params = {
|
||
'appid': 'test_appid',
|
||
'secret': 'test_secret',
|
||
'code': 'test_code',
|
||
'grant_type': 'authorization_code'
|
||
}
|
||
|
||
print(" 发送测试请求到微信 API...")
|
||
start = time.time()
|
||
response = requests.get(url, params=params, timeout=15)
|
||
elapsed = time.time() - start
|
||
|
||
data = response.json()
|
||
print(f" 响应时间: {elapsed:.2f}s")
|
||
print(f" 响应内容: {data}")
|
||
|
||
# 预期会返回错误（因为参数无效），但这说明 API 可达
|
||
if 'errcode' in data:
|
||
print(f" ✅ 微信 API 可达（预期的错误响应: {data.get('errmsg', '')}）")
|
||
return True
|
||
else:
|
||
print(" ⚠️ 意外的响应格式")
|
||
return True
|
||
|
||
except requests.exceptions.Timeout:
|
||
print(" ❌ 请求超时 - 网络可能有问题")
|
||
return False
|
||
except requests.exceptions.SSLError as e:
|
||
print(f" ❌ SSL 错误: {e}")
|
||
print(" 可能需要更新 CA 证书或检查网络代理设置")
|
||
return False
|
||
except Exception as e:
|
||
print(f" ❌ 请求失败: {e}")
|
||
return False
|
||
|
||
|
||
def test_flask_app():
|
||
"""测试 Flask 应用是否能正常导入"""
|
||
print("\n" + "=" * 50)
|
||
print("🔍 测试 5: Flask 应用导入")
|
||
print("=" * 50)
|
||
|
||
try:
|
||
print(" 正在导入 app...")
|
||
start = time.time()
|
||
from app import app
|
||
elapsed = time.time() - start
|
||
print(f" ✅ Flask 应用导入成功 ({elapsed:.2f}s)")
|
||
return True
|
||
except Exception as e:
|
||
print(f" ❌ Flask 应用导入失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
def main():
|
||
print("\n" + "=" * 60)
|
||
print(" 微信登录问题诊断工具")
|
||
print("=" * 60)
|
||
|
||
results = []
|
||
|
||
results.append(("requests 库", test_requests()))
|
||
results.append(("Redis 连接", test_redis()))
|
||
results.append(("Gevent 环境", test_gevent()))
|
||
results.append(("微信 API", test_wechat_access_token()))
|
||
|
||
# Flask 导入测试可能会比较慢，放最后
|
||
if '--skip-flask' not in sys.argv:
|
||
results.append(("Flask 应用", test_flask_app()))
|
||
|
||
print("\n" + "=" * 60)
|
||
print(" 测试结果汇总")
|
||
print("=" * 60)
|
||
|
||
all_passed = True
|
||
for name, passed in results:
|
||
status = "✅ 通过" if passed else "❌ 失败"
|
||
print(f" {name}: {status}")
|
||
if not passed:
|
||
all_passed = False
|
||
|
||
print("\n" + "=" * 60)
|
||
if all_passed:
|
||
print("✅ 所有测试通过！")
|
||
print("\n如果微信登录仍然 502，请检查:")
|
||
print(" 1. Gunicorn 日志中的具体错误信息")
|
||
print(" 2. 微信开放平台的回调 URL 配置")
|
||
print(" 3. 服务器防火墙是否允许访问微信 API")
|
||
else:
|
||
print("❌ 部分测试失败，请根据上述信息排查问题")
|
||
print("=" * 60)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|
||
|