From 54c4f64a499bfb56af4dad7f8f73379894b63617 Mon Sep 17 00:00:00 2001
From: zzlgreat <zzlgreat@gmail.com>
Date: Thu, 11 Dec 2025 21:34:20 +0800
Subject: [PATCH] update pay ui

---
 app.py                 |  41 ++++++++-
 gunicorn_app_config.py |  28 +++---
 test_wechat_api.py     | 196 +++++++++++++++++++++++++++++++++++++++++
 3 files changed, 246 insertions(+), 19 deletions(-)
 create mode 100644 test_wechat_api.py

diff --git a/app.py b/app.py
index db5cddb2..5c5833ac 100755
--- a/app.py
+++ b/app.py
@@ -1,13 +1,27 @@
 # ============ Gevent Monkey Patching（必须放在所有 import 之前！）============
 # 用于支持 Gunicorn + gevent 异步模式，使 requests 等阻塞调用变为非阻塞
+# 注意：如果使用 GeventWebSocketWorker，worker 会自动 patch，这里检测避免重复
 import os
-if os.environ.get('GEVENT_SUPPORT', 'true').lower() == 'true':
+import sys
+
+def _is_already_patched():
+    """检测是否已经被 gevent patch"""
     try:
         from gevent import monkey
-        monkey.patch_all()
-        print("✅ Gevent monkey patching 已启用")
+        return monkey.is_module_patched('socket')
+    except:
+        return False
+
+# 只在未被 patch 且明确启用时才 patch
+if os.environ.get('GEVENT_SUPPORT', 'false').lower() == 'true' and not _is_already_patched():
+    try:
+        from gevent import monkey
+        monkey.patch_all(thread=False, subprocess=False)  # 避免 patch 某些可能有问题的模块
+        print("✅ Gevent monkey patching 已启用 (手动)")
     except ImportError:
         print("⚠️ Gevent 未安装，跳过 monkey patching")
+elif _is_already_patched():
+    print("✅ Gevent monkey patching 已由 Worker 启用")
 # ============ Gevent Monkey Patching 结束 ============
 
 import base64
@@ -362,11 +376,30 @@ db = SQLAlchemy(app)
 mail = Mail(app)
 
 # 初始化 Flask-SocketIO（用于实时事件推送）
+# 自动检测可用的异步模式，优先级：eventlet > gevent > threading
+def _detect_async_mode():
+    """检测可用的异步模式"""
+    try:
+        import eventlet
+        return 'eventlet'
+    except ImportError:
+        pass
+    try:
+        from gevent import monkey
+        if monkey.is_module_patched('socket'):
+            return 'gevent'
+    except ImportError:
+        pass
+    return 'gevent'  # 默认使用 gevent，Gunicorn 会 patch
+
+_async_mode = _detect_async_mode()
+print(f"📡 Flask-SocketIO async_mode: {_async_mode}")
+
 socketio = SocketIO(
     app,
     cors_allowed_origins=["http://localhost:3000", "http://127.0.0.1:3000", "http://localhost:5173",
                           "https://valuefrontier.cn", "http://valuefrontier.cn"],
-    async_mode='gevent',
+    async_mode=_async_mode,
     logger=True,
     engineio_logger=False,
     ping_timeout=120,  # 心跳超时时间（秒），客户端120秒内无响应才断开
diff --git a/gunicorn_app_config.py b/gunicorn_app_config.py
index b108aab7..c09efaba 100644
--- a/gunicorn_app_config.py
+++ b/gunicorn_app_config.py
@@ -1,16 +1,14 @@
 # -*- coding: utf-8 -*-
 """
-Gunicorn 配置文件 - app.py 生产环境配置（支持 Flask-SocketIO）
+Gunicorn 配置文件 - app.py 生产环境配置（支持 Flask-SocketIO + WebSocket）
 
 使用方式:
-    # 方式1: 推荐 - 使用 geventwebsocket 支持 WebSocket + 高并发
+    # 推荐方式: 使用此配置文件启动
     gunicorn -c gunicorn_app_config.py app:app
 
-    # 方式2: 如果方式1有问题，使用纯 gevent（无 WebSocket）
-    gunicorn -c gunicorn_app_config.py -k gevent app:app
-
-    # 方式3: 最简单的多进程模式（不支持 WebSocket，但最稳定）
-    gunicorn -w 4 -b 0.0.0.0:5001 --timeout 120 app:app
+    # 如果遇到 502 错误，可以尝试安装 eventlet 后使用:
+    pip install eventlet
+    gunicorn -k eventlet -w 1 -b 0.0.0.0:5001 --timeout 300 app:app
 """
 
 import os
@@ -20,20 +18,19 @@ import os
 # 绑定地址和端口
 bind = '0.0.0.0:5001'
 
-# Worker 进程数
-# 注意：geventwebsocket 要求单 worker，如果用多 worker 需要用 sync 或 gevent
+# Worker 进程数（WebSocket 需要单 worker）
 workers = 1
 
 # Worker 类型 - 使用 geventwebsocket 支持 WebSocket
-# 如果遇到问题，可以命令行添加 -k gevent 覆盖此设置
+# Flask-SocketIO 需要异步 worker 来支持 WebSocket
 worker_class = 'geventwebsocket.gunicorn.workers.GeventWebSocketWorker'
 
 # Worker 连接数（gevent 异步模式下可以处理大量并发连接）
-worker_connections = 1000
+worker_connections = 2000
 
-# 每个 worker 处理的最大请求数，超过后重启（防止内存泄漏）
-max_requests = 5000
-max_requests_jitter = 500
+# 每个 worker 处理的最大请求数
+max_requests = 0  # 禁用自动重启，避免 WebSocket 连接中断
+max_requests_jitter = 0
 
 # ==================== 超时配置 ====================
 
@@ -59,7 +56,8 @@ if os.path.exists(cert_file) and os.path.exists(key_file):
 
 accesslog = '-'
 errorlog = '-'
-loglevel = 'info'
+loglevel = 'debug'  # 调试时用 debug，正常运行用 info
+capture_output = True  # 捕获 print 输出到日志
 access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)sμs'
 
 # ==================== 进程管理 ====================
diff --git a/test_wechat_api.py b/test_wechat_api.py
new file mode 100644
index 00000000..f286c025
--- /dev/null
+++ b/test_wechat_api.py
@@ -0,0 +1,196 @@
+#!/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()
+