#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 最小功能测试框架 - Flask + SocketIO + threading 用于验证打包exe后HTTP和WebSocket服务的可用性 """ import os import sys from flask import Flask, render_template_string, jsonify from flask_socketio import SocketIO, emit import logging from datetime import datetime # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # 创建Flask应用 app = Flask(__name__) app.config['SECRET_KEY'] = 'minimal-test-secret-key-2024' # 初始化SocketIO，强制使用threading模式 try: logger.info("初始化SocketIO（threading模式）...") socketio = SocketIO( app, cors_allowed_origins='*', async_mode='threading', logger=False, engineio_logger=False ) logger.info(f"SocketIO初始化成功，异步模式: {socketio.async_mode}") except Exception as e: logger.error(f"SocketIO初始化失败: {e}") sys.exit(1) # HTML测试页面模板 HTML_TEMPLATE = """ 最小功能测试 - Flask + SocketIO

最小功能测试框架

测试Flask HTTP服务和SocketIO WebSocket服务

HTTP API 测试

WebSocket 测试

连接状态: 未连接

系统信息

服务器时间: {{ server_time }}

SocketIO模式: {{ socketio_mode }}

Flask版本: {{ flask_version }}

""" # HTTP路由 @app.route('/') def index(): """主页面""" import flask return render_template_string(HTML_TEMPLATE, server_time=datetime.now().strftime('%Y-%m-%d %H:%M:%S'), socketio_mode=socketio.async_mode, flask_version=flask.__version__ ) @app.route('/api/test') def api_test(): """HTTP API测试接口""" return jsonify({ 'status': 'success', 'message': 'HTTP API工作正常', 'timestamp': datetime.now().isoformat(), 'server_info': { 'socketio_mode': socketio.async_mode, 'working_directory': os.getcwd() } }) @app.route('/health') def health_check(): """健康检查接口""" return jsonify({ 'status': 'healthy', 'services': { 'http': 'running', 'websocket': 'running', 'socketio_mode': socketio.async_mode }, 'timestamp': datetime.now().isoformat() }) # SocketIO事件处理 @socketio.on('connect') def handle_connect(): """客户端连接事件""" logger.info(f"客户端连接: {request.sid if 'request' in globals() else 'unknown'}") emit('server_message', { 'message': 'WebSocket连接成功', 'timestamp': datetime.now().isoformat(), 'socketio_mode': socketio.async_mode }) @socketio.on('disconnect') def handle_disconnect(): """客户端断开连接事件""" logger.info(f"客户端断开连接: {request.sid if 'request' in globals() else 'unknown'}") @socketio.on('test_message') def handle_test_message(data): """处理测试消息""" logger.info(f"收到测试消息: {data}") # 回复客户端 emit('test_response', { 'status': 'received', 'original_message': data, 'server_timestamp': datetime.now().isoformat(), 'socketio_mode': socketio.async_mode }) def main(): """主函数""" logger.info("="*50) logger.info("最小功能测试框架启动") logger.info("="*50) logger.info(f"工作目录: {os.getcwd()}") logger.info(f"Python版本: {sys.version}") logger.info(f"SocketIO异步模式: {socketio.async_mode}") try: # 启动服务器 logger.info("启动服务器 http://localhost:5000") logger.info("按 Ctrl+C 停止服务器") socketio.run( app, host='0.0.0.0', port=5000, debug=False, allow_unsafe_werkzeug=True ) except KeyboardInterrupt: logger.info("用户中断，正在关闭服务器...") except Exception as e: logger.error(f"服务器启动失败: {e}") input("按回车键退出...") sys.exit(1) if __name__ == '__main__': main()