ThbTech/BodyBalanceEvaluation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修改了debug_server.py

Browse Source
This commit is contained in:
root 2025-08-13 14:17:50 +08:00
parent 625e372c11
commit 0a3f741dae
32 changed files with 12897 additions and 241 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9968
.gitignore vendored
View File

File diff suppressed because it is too large Load Diff

BIN
backend/__pycache__/app.cpython-311.pyc
View File

Binary file not shown.

BIN
backend/__pycache__/data_processor.cpython-311.pyc
View File

Binary file not shown.

BIN
backend/__pycache__/database.cpython-311.pyc
View File

Binary file not shown.

BIN
backend/__pycache__/detection_engine.cpython-311.pyc
View File

Binary file not shown.

BIN
backend/__pycache__/device_manager.cpython-311.pyc
View File

Binary file not shown.

BIN
backend/__pycache__/utils.cpython-311.pyc
View File

Binary file not shown.

203
backend/app.py
View File

@ -30,12 +30,24 @@ from database import DatabaseManager
from device_manager import DeviceManager, VideoStreamManager
from utils import config as app_config
# 确定日志文件路径
if getattr(sys, 'frozen', False):
# 如果是打包后的exe，日志文件在exe同目录下的logs文件夹
log_dir = os.path.join(os.path.dirname(sys.executable), 'logs')
else:
# 如果是开发环境，使用当前目录的logs文件夹
log_dir = 'logs'
# 确保日志目录存在
os.makedirs(log_dir, exist_ok=True)
log_file = os.path.join(log_dir, 'backend.log')
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('logs/backend.log', encoding='utf-8'),
logging.FileHandler(log_file, encoding='utf-8'),
logging.StreamHandler()
]
)
@ -44,11 +56,22 @@ logger = logging.getLogger(__name__)
# 创建Flask应用
app = Flask(__name__)
app.config['SECRET_KEY'] = 'body-balance-detection-system-2024'
socketio = SocketIO(app,
cors_allowed_origins='*',
async_mode='threading',
logger=False,
engineio_logger=False)
# 初始化SocketIO
try:
socketio = SocketIO(
app,
cors_allowed_origins='*',
async_mode='threading',
logger=False,
engineio_logger=False,
ping_timeout=60,
ping_interval=25
)
logger.info('SocketIO初始化成功')
except Exception as e:
logger.error(f'SocketIO初始化失败: {e}')
socketio = None
import logging
logging.getLogger('socketio').setLevel(logging.WARNING)
@ -61,7 +84,16 @@ CORS(app, origins='*', supports_credentials=True, allow_headers=['Content-Type',
# 读取RTSP配置
config = configparser.ConfigParser()
config.read(os.path.join(os.path.dirname(__file__), '..', 'config.ini'), encoding='utf-8')
# 确定配置文件路径
if getattr(sys, 'frozen', False):
# 如果是打包后的exe，配置文件在exe同目录下
config_path = os.path.join(os.path.dirname(sys.executable), 'config.ini')
else:
# 如果是开发环境，配置文件在上级目录
config_path = os.path.join(os.path.dirname(__file__), '..', 'config.ini')
config.read(config_path, encoding='utf-8')
device_index = config.get('CAMERA', 'device_index', fallback=None)
# 全局变量
@ -78,20 +110,28 @@ def init_app():
global db_manager, device_manager, video_stream_manager
try:
# 确定基础目录
if getattr(sys, 'frozen', False):
# 如果是打包后的exe，使用exe同目录
base_dir = os.path.dirname(sys.executable)
else:
# 如果是开发环境，使用当前脚本目录
base_dir = os.path.dirname(os.path.abspath(__file__))
# 创建必要的目录
os.makedirs('logs', exist_ok=True)
os.makedirs('data', exist_ok=True)
logs_dir = os.path.join(base_dir, 'logs')
data_dir = os.path.join(base_dir, 'data')
os.makedirs(logs_dir, exist_ok=True)
os.makedirs(data_dir, exist_ok=True)
# 从配置文件读取数据库路径
db_path_config = app_config.get('DATABASE', 'path', 'backend/data/body_balance.db')
# 如果是相对路径，基于当前脚本目录解析
db_path_config = app_config.get('DATABASE', 'path', 'data/body_balance.db')
# 如果是相对路径，基于基础目录解析
if not os.path.isabs(db_path_config):
# 获取当前脚本所在目录（backend目录）
current_dir = os.path.dirname(os.path.abspath(__file__))
# 如果配置路径以 'backend/' 开头，去掉这个前缀
if db_path_config.startswith('backend/'):
db_path_config = db_path_config[8:] # 去掉 'backend/' 前缀
db_path = os.path.join(current_dir, db_path_config)
db_path = os.path.join(base_dir, db_path_config)
else:
db_path = db_path_config
@ -99,15 +139,28 @@ def init_app():
db_dir = os.path.dirname(db_path)
os.makedirs(db_dir, exist_ok=True)
# 输出数据库路径信息
print(f"\n=== 系统初始化 ===")
print(f"数据库配置路径: {db_path_config}")
print(f"数据库实际路径: {db_path}")
print(f"数据库目录: {db_dir}")
print(f"当前工作目录: {os.getcwd()}")
print(f"数据库文件存在: {'是' if os.path.exists(db_path) else '否'}")
print(f"==================\n")
# 初始化数据库
db_manager = DatabaseManager(db_path)
db_manager.init_database()
# 初始化设备管理器
device_manager = DeviceManager(db_manager)
device_manager.set_socketio(socketio) # 设置WebSocket连接
# 初始化视频流管理器
video_stream_manager = VideoStreamManager(socketio, device_manager)
if socketio is not None:
device_manager.set_socketio(socketio) # 设置WebSocket连接
# 初始化视频流管理器
video_stream_manager = VideoStreamManager(socketio, device_manager)
else:
logger.info('SocketIO未启用，跳过WebSocket相关初始化')
video_stream_manager = None
logger.info('应用初始化完成')
@ -1044,66 +1097,68 @@ if __name__ == '__main__':
# ==================== WebSocket 事件处理 ====================
# 简单的测试事件处理器
@socketio.on('connect')
def handle_connect():
# print('CLIENT CONNECTED!!!', flush=True) # 控制台打印测试已关闭
logger.info('客户端已连接')
# 只有当socketio不为None时才注册事件处理器
if socketio is not None:
# 简单的测试事件处理器
@socketio.on('connect')
def handle_connect():
# print('CLIENT CONNECTED!!!', flush=True) # 控制台打印测试已关闭
logger.info('客户端已连接')
@socketio.on('disconnect')
def handle_disconnect():
# print('CLIENT DISCONNECTED!!!', flush=True) # 控制台打印测试已关闭
logger.info('客户端已断开连接')
@socketio.on('disconnect')
def handle_disconnect():
# print('CLIENT DISCONNECTED!!!', flush=True) # 控制台打印测试已关闭
logger.info('客户端已断开连接')
# 原始的start_video处理逻辑（暂时注释）
@socketio.on('start_video_stream')
def handle_start_video(data=None):
try:
results = {'status': 'success', 'cameras': {}}
logger.info(f'video_stream_manager状态: {video_stream_manager is not None}')
logger.info(f'device_manager状态: {device_manager is not None}')
# 启动视频流管理器（普通摄像头）
if video_stream_manager:
logger.info('正在启动视频流管理器...')
video_result = video_stream_manager.start_video_stream()
logger.info(f'视频流管理器启动结果: {video_result}')
results['cameras']['normal'] = video_result
else:
logger.error('视频流管理器未初始化')
results['cameras']['normal'] = {'status': 'error', 'message': '视频流管理器未初始化'}
# 启动FemtoBolt深度相机
if device_manager:
logger.info('正在启动FemtoBolt深度相机...')
femtobolt_result = device_manager.start_femtobolt_stream()
logger.info(f'FemtoBolt启动结果: {femtobolt_result}')
if femtobolt_result:
results['cameras']['femtobolt'] = {'status': 'success', 'message': 'FemtoBolt深度相机推流已启动'}
# 原始的start_video处理逻辑（暂时注释）
@socketio.on('start_video_stream')
def handle_start_video(data=None):
try:
results = {'status': 'success', 'cameras': {}}
logger.info(f'video_stream_manager状态: {video_stream_manager is not None}')
logger.info(f'device_manager状态: {device_manager is not None}')
# 启动视频流管理器（普通摄像头）
if video_stream_manager:
logger.info('正在启动视频流管理器...')
video_result = video_stream_manager.start_video_stream()
logger.info(f'视频流管理器启动结果: {video_result}')
results['cameras']['normal'] = video_result
else:
results['cameras']['femtobolt'] = {'status': 'error', 'message': 'FemtoBolt深度相机启动失败'}
else:
logger.error('设备管理器未初始化')
results['cameras']['femtobolt'] = {'status': 'error', 'message': '设备管理器未初始化'}
# 检查是否有任何相机启动成功
success_count = sum(1 for cam_result in results['cameras'].values() if cam_result.get('status') == 'success')
if success_count == 0:
results['status'] = 'error'
results['message'] = '所有相机启动失败'
elif success_count < len(results['cameras']):
results['status'] = 'partial'
results['message'] = f'{success_count}/{len(results["cameras"])}个相机启动成功'
else:
results['message'] = '所有相机启动成功'
logger.info(f'发送video_status事件: {results}')
emit('video_status', results)
except Exception as e:
logger.error(f'启动视频流失败: {e}', exc_info=True)
emit('video_status', {'status': 'error', 'message': f'启动失败: {str(e)}'})
logger.error('视频流管理器未初始化')
results['cameras']['normal'] = {'status': 'error', 'message': '视频流管理器未初始化'}
# 启动FemtoBolt深度相机
if device_manager:
logger.info('正在启动FemtoBolt深度相机...')
femtobolt_result = device_manager.start_femtobolt_stream()
logger.info(f'FemtoBolt启动结果: {femtobolt_result}')
if femtobolt_result:
results['cameras']['femtobolt'] = {'status': 'success', 'message': 'FemtoBolt深度相机推流已启动'}
else:
results['cameras']['femtobolt'] = {'status': 'error', 'message': 'FemtoBolt深度相机启动失败'}
else:
logger.error('设备管理器未初始化')
results['cameras']['femtobolt'] = {'status': 'error', 'message': '设备管理器未初始化'}
# 检查是否有任何相机启动成功
success_count = sum(1 for cam_result in results['cameras'].values() if cam_result.get('status') == 'success')
if success_count == 0:
results['status'] = 'error'
results['message'] = '所有相机启动失败'
elif success_count < len(results['cameras']):
results['status'] = 'partial'
results['message'] = f'{success_count}/{len(results["cameras"])}个相机启动成功'
else:
results['message'] = '所有相机启动成功'
logger.info(f'发送video_status事件: {results}')
emit('video_status', results)
except Exception as e:
logger.error(f'启动视频流失败: {e}', exc_info=True)
emit('video_status', {'status': 'error', 'message': f'启动失败: {str(e)}'})
@socketio.on('stop_video_stream')
def handle_stop_video(data=None):

102
backend/app.spec Normal file
View File

@ -0,0 +1,102 @@
# -*- mode: python ; coding: utf-8 -*-
block_cipher = None
a = Analysis(
['app.py'],
pathex=[],
binaries=[
('dll/femtobolt/bin/k4a.dll', '.'), # K4A动态库
('dll/femtobolt/bin/k4arecord.dll', '.'), # K4A录制库
('dll/femtobolt/bin/depthengine_2_0.dll', '.'), # 深度引擎
('dll/femtobolt/bin/OrbbecSDK.dll', '.'), # Orbbec SDK
('dll/femtobolt/bin/ob_usb.dll', '.'), # Orbbec USB库
('dll/femtobolt/bin/live555.dll', '.'), # Live555库
('dll/smitsense/SMiTSenseUsb-F3.0.dll', '.'), # SMiTSense传感器库
('dll/femtobolt/bin/OrbbecSDKConfig_v1.0.xml', '.'), # Orbbec配置文件
],
datas=[
('config.ini', '.'), # 配置文件
('data', 'data'), # 数据文件夹
],
hiddenimports=[
'flask',
'flask_socketio',
'flask_cors',
'cv2',
'numpy',
'pandas',
'scipy',
'matplotlib',
'seaborn',
'sklearn',
'PIL',
'reportlab',
'sqlite3',
'configparser',
'logging',
'threading',
'queue',
'base64',
'psutil',
'pykinect_azure',
'pyserial',
'requests',
'yaml',
'click',
'colorama',
'tqdm',
'database',
'device_manager',
'utils',
'eventlet',
'socketio',
'engineio',
'engineio.async_drivers.threading',
'engineio.async_drivers.eventlet',
'engineio.async_eventlet',
'socketio.async_eventlet',
'greenlet',
'gevent',
'gevent.socket',
'gevent.select',
'dns',
'dns.resolver',
'dns.reversename',
'dns.e164',
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
noarchive=False,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
[],
name='BodyBalanceBackend_Full',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon=None
)

242
backend/app_minimal.py Normal file
View File

@ -0,0 +1,242 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
平衡体态检测系统 - 后端服务（最小版本）
基于Flask的本地API服务，暂时移除SocketIO功能
"""
import os
import sys
import json
import time
import threading
from datetime import datetime
from flask import Flask, jsonify, send_file
from flask import request as flask_request
from flask_cors import CORS
import sqlite3
import logging
from pathlib import Path
import cv2
import configparser
# 添加当前目录到Python路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
# 导入自定义模块
from database import DatabaseManager
from device_manager import DeviceManager
from utils import config as app_config
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('logs/backend.log', encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# 创建Flask应用
app = Flask(__name__)
app.config['SECRET_KEY'] = 'body-balance-detection-system-2024'
# 启用CORS
CORS(app, origins=['http://localhost:3000', 'http://127.0.0.1:3000', 'http://192.168.1.58:3000'])
# 全局变量
db_manager = None
device_manager = None
socketio = None # 暂时禁用SocketIO
def init_app():
"""初始化应用"""
global db_manager, device_manager
try:
# 创建必要的目录
os.makedirs('logs', exist_ok=True)
os.makedirs('data', exist_ok=True)
# 从配置文件读取数据库路径
db_path_config = app_config.get('DATABASE', 'path', 'data/body_balance.db')
# 对于打包后的exe，确保数据库路径基于exe所在目录
if getattr(sys, 'frozen', False):
# 如果是打包后的exe，使用exe所在目录
exe_dir = os.path.dirname(sys.executable)
if not os.path.isabs(db_path_config):
db_path = os.path.join(exe_dir, db_path_config)
else:
db_path = db_path_config
else:
# 如果是开发环境，使用脚本所在目录
if not os.path.isabs(db_path_config):
current_dir = os.path.dirname(os.path.abspath(__file__))
if db_path_config.startswith('backend/'):
db_path_config = db_path_config[8:] # 去掉 'backend/' 前缀
db_path = os.path.join(current_dir, db_path_config)
else:
db_path = db_path_config
# 确保数据库目录存在
db_dir = os.path.dirname(db_path)
os.makedirs(db_dir, exist_ok=True)
# 打印数据库路径信息用于调试
logger.info(f'数据库配置路径: {db_path_config}')
logger.info(f'数据库实际路径: {db_path}')
logger.info(f'数据库目录: {db_dir}')
logger.info(f'当前工作目录: {os.getcwd()}')
# 初始化数据库
db_manager = DatabaseManager(db_path)
db_manager.init_database()
# 初始化设备管理器（不使用SocketIO）
device_manager = DeviceManager(db_manager)
logger.info('应用初始化完成（最小版本，无SocketIO）')
except Exception as e:
logger.error(f'应用初始化失败: {e}')
logger.warning('部分功能可能不可用，但服务将继续运行')
# ==================== 基础API ====================
@app.route('/health', methods=['GET'])
def health_check():
"""健康检查"""
return jsonify({
'status': 'ok',
'timestamp': datetime.now().isoformat(),
'version': '1.0.0-minimal',
'socketio_enabled': False
})
@app.route('/api/health', methods=['GET'])
def api_health_check():
"""API健康检查"""
return jsonify({
'status': 'ok',
'timestamp': datetime.now().isoformat(),
'version': '1.0.0-minimal',
'socketio_enabled': False
})
@app.route('/api/test', methods=['GET'])
def test_endpoint():
"""测试端点"""
return jsonify({
'message': 'Backend is running!',
'timestamp': datetime.now().isoformat(),
'database_status': 'connected' if db_manager else 'not_connected',
'device_manager_status': 'initialized' if device_manager else 'not_initialized'
})
# ==================== 认证API ====================
@app.route('/api/auth/login', methods=['POST'])
def login():
"""用户登录"""
try:
data = flask_request.get_json()
username = data.get('username')
password = data.get('password')
remember = data.get('remember', False)
if not username or not password:
return jsonify({
'success': False,
'message': '用户名或密码不能为空'
}), 400
# 简单的测试登录逻辑
if username == 'admin' and password == 'admin':
# 生成token（实际项目中应使用JWT等安全token）
token = f"token_{username}_{int(time.time())}"
return jsonify({
'success': True,
'message': '登录成功',
'data': {
'token': token,
'user': {
'id': 1,
'username': username,
'role': 'admin'
}
}
})
else:
return jsonify({
'success': False,
'message': '用户名或密码错误'
}), 401
except Exception as e:
logger.error(f'登录失败: {e}')
return jsonify({
'success': False,
'message': '登录过程中发生错误'
}), 500
@app.route('/api/auth/logout', methods=['POST'])
def logout():
"""用户登出"""
return jsonify({
'success': True,
'message': '登出成功'
})
# ==================== 设备API ====================
@app.route('/api/devices/status', methods=['GET'])
def get_device_status():
"""获取设备状态"""
try:
if device_manager:
status = {
'cameras': device_manager.get_camera_status(),
'sensors': device_manager.get_sensor_status()
}
else:
status = {
'cameras': {},
'sensors': {},
'error': '设备管理器未初始化'
}
return jsonify({
'success': True,
'data': status
})
except Exception as e:
logger.error(f'获取设备状态失败: {e}')
return jsonify({
'success': False,
'message': f'获取设备状态失败: {str(e)}'
}), 500
# ==================== 错误处理 ====================
@app.errorhandler(404)
def not_found(error):
return jsonify({
'success': False,
'message': '接口不存在'
}), 404
@app.errorhandler(500)
def internal_error(error):
return jsonify({
'success': False,
'message': '服务器内部错误'
}), 500
if __name__ == '__main__':
init_app()
app.run(host='0.0.0.0', port=5000, debug=True)

71
backend/app_minimal.spec Normal file
View File

@ -0,0 +1,71 @@
# -*- mode: python ; coding: utf-8 -*-
block_cipher = None
a = Analysis(
['app_minimal.py'],
pathex=[],
binaries=[
('dll/smitsense/SMiTSenseUsb-F3.0.dll', '.'), # SMiTSense传感器库
],
datas=[
('config.ini', '.'), # 配置文件
('data', 'data'), # 数据文件夹
],
hiddenimports=[
'flask',
'flask_cors',
'sqlite3',
'configparser',
'logging',
'threading',
'queue',
'base64',
'psutil',
'database',
'device_manager',
'utils',
'cv2',
'numpy',
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[
'flask_socketio',
'socketio',
'engineio',
'eventlet',
'gevent',
],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
noarchive=False,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
[],
name='BodyBalanceBackend',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon=None
)

46
backend/backend.spec
View File

@ -1,62 +1,62 @@
# -*- mode: python ; coding: utf-8 -*-
block_cipher = None
a = Analysis(
['app_simple.py'],
['main_exe.py'],
pathex=[],
binaries=[
('dll/femtobolt/bin/k4a.dll', '.'), # K4A动态库
('dll/femtobolt/bin/k4arecord.dll', '.'), # K4A录制库
('dll/femtobolt/bin/depthengine_2_0.dll', '.'), # 深度引擎
('dll/femtobolt/bin/OrbbecSDK.dll', '.'), # Orbbec SDK
('dll/femtobolt/bin/ob_usb.dll', '.'), # Orbbec USB库
('dll/femtobolt/bin/live555.dll', '.'), # Live555库
('dll/smitsense/SMiTSenseUsb-F3.0.dll', '.'), # SMiTSense传感器库
('dll/femtobolt/bin/OrbbecSDKConfig_v1.0.xml', '.'), # Orbbec配置文件
# 只包含确实存在的DLL文件
],
datas=[
('..\config.ini', '.'), # 配置文件
('..\config.json', '.'), # JSON配置文件
('tests', 'tests'), # 测试文件夹
('config.ini', '.'), # 配置文件
('data', 'data'), # 数据文件夹
('tests', 'tests'), # 测试文件夹
('app_minimal.py', '.'), # 应用入口文件
],
hiddenimports=[
'flask',
'flask_socketio',
'flask_cors',
'socketio',
'engineio',
'engineio.async_drivers.threading',
'socketio.namespace',
'cv2',
'numpy',
'psutil',
'sqlite3',
'configparser',
'logging',
'threading',
'queue',
'base64',
'psutil',
'pykinect_azure',
'requests',
'click',
'colorama',
'database',
'device_manager',
'utils',
'dns',
'dns.resolver',
'dns.asyncresolver'
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
excludes=[
'eventlet',
'gevent',
'gevent_uwsgi'
],
noarchive=False,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
[],
name='BodyBalanceBackend',
@ -72,5 +72,5 @@ exe = EXE(
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon='..\document\icon.ico' if os.path.exists('..\document\icon.ico') else None,
icon=None
)

367
backend/build_app.py Normal file
View File

@ -0,0 +1,367 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
app.py 完整版打包脚本
使用PyInstaller将完整版Flask应用程序（包含SocketIO）打包成独立的exe文件
"""
import os
import sys
import shutil
import subprocess
from pathlib import Path
def clean_build_dirs():
"""清理构建目录"""
print("清理构建目录...")
dirs_to_clean = ['build', 'dist', '__pycache__']
for dir_name in dirs_to_clean:
if os.path.exists(dir_name):
try:
shutil.rmtree(dir_name)
print(f"✓ 已删除 {dir_name}")
except Exception as e:
print(f"⚠️ 删除 {dir_name} 失败: {e}")
def create_app_spec():
"""创建app.py的PyInstaller spec文件"""
spec_content = '''
# -*- mode: python ; coding: utf-8 -*-
block_cipher = None
a = Analysis(
['app.py'],
pathex=[],
binaries=[
('dll/femtobolt/bin/k4a.dll', '.'), # K4A动态库
('dll/femtobolt/bin/k4arecord.dll', '.'), # K4A录制库
('dll/femtobolt/bin/depthengine_2_0.dll', '.'), # 深度引擎
('dll/femtobolt/bin/OrbbecSDK.dll', '.'), # Orbbec SDK
('dll/femtobolt/bin/ob_usb.dll', '.'), # Orbbec USB库
('dll/femtobolt/bin/live555.dll', '.'), # Live555库
('dll/smitsense/SMiTSenseUsb-F3.0.dll', '.'), # SMiTSense传感器库
('dll/femtobolt/bin/OrbbecSDKConfig_v1.0.xml', '.'), # Orbbec配置文件
],
datas=[
('config.ini', '.'), # 配置文件
('data', 'data'), # 数据文件夹
],
hiddenimports=[
'flask',
'flask_socketio',
'flask_cors',
'cv2',
'numpy',
'pandas',
'scipy',
'matplotlib',
'seaborn',
'sklearn',
'PIL',
'reportlab',
'sqlite3',
'configparser',
'logging',
'threading',
'queue',
'base64',
'psutil',
'pykinect_azure',
'pyserial',
'requests',
'yaml',
'click',
'colorama',
'tqdm',
'database',
'device_manager',
'utils',
'eventlet',
'socketio',
'engineio',
'engineio.async_drivers.threading',
'engineio.async_drivers.eventlet',
'engineio.async_eventlet',
'socketio.async_eventlet',
'greenlet',
'gevent',
'gevent.socket',
'gevent.select',
'dns',
'dns.resolver',
'dns.reversename',
'dns.e164',
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
noarchive=False,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
[],
name='BodyBalanceBackend_Full',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon=None
)
'''
with open('app.spec', 'w', encoding='utf-8') as f:
f.write(spec_content)
print("✓ 已创建 app.spec 文件")
def build_exe():
"""构建exe文件"""
print("\n开始构建exe文件...")
try:
# 使用PyInstaller构建
cmd = [sys.executable, '-m', 'PyInstaller', 'app.spec', '--clean', '--noconfirm']
print(f"执行命令: {' '.join(cmd)}")
# 修改编码处理，避免UTF-8错误
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8', errors='ignore')
if result.returncode == 0:
print("✓ 构建成功！")
return True
else:
print(f"✗ 构建失败")
print(f"错误输出: {result.stderr}")
return False
except Exception as e:
print(f"✗ 构建过程出错: {e}")
return False
def create_startup_script():
"""创建启动脚本"""
startup_script = '''@echo off
echo 启动身体平衡评估系统后端服务（完整版）...
echo.
echo 服务信息:
echo - HTTP API: http://localhost:5000
echo - SocketIO: ws://localhost:5000
echo - 管理界面: http://localhost:5000
echo.
echo 按Ctrl+C停止服务
echo.
"BodyBalanceBackend_Full.exe"
if %errorlevel% neq 0 (
echo.
echo 服务启动失败，请检查错误信息
pause
)
'''
dist_dir = 'dist'
if not os.path.exists(dist_dir):
os.makedirs(dist_dir)
with open(os.path.join(dist_dir, 'start_backend_full.bat'), 'w', encoding='utf-8') as f:
f.write(startup_script)
print("✓ 创建启动脚本: dist/start_backend_full.bat")
def create_directories():
"""创建必要的目录结构"""
print("创建目录结构...")
dist_dir = 'dist'
directories = ['dll', 'data', 'logs']
for directory in directories:
dir_path = os.path.join(dist_dir, directory)
if not os.path.exists(dir_path):
try:
os.makedirs(dir_path)
print(f"✓ 创建目录: {directory}/")
except Exception as e:
print(f"⚠️ 创建目录 {directory} 失败: {e}")
else:
print(f"✓ 目录已存在: {directory}/")
def copy_dll_files():
"""复制DLL文件到dll目录"""
print("复制DLL文件...")
dll_dir = os.path.join('dist', 'dll')
# 查找所有DLL文件
dll_files = []
# 从当前目录查找DLL文件
for file in os.listdir('.'):
if file.endswith('.dll'):
dll_files.append(file)
# 从子目录查找DLL文件
for root, dirs, files in os.walk('.'):
if 'dist' in root or 'build' in root or '__pycache__' in root:
continue
for file in files:
if file.endswith('.dll'):
dll_files.append(os.path.join(root, file))
if not dll_files:
print("⚠️ 未找到DLL文件")
return
for dll_file in dll_files:
if os.path.exists(dll_file):
try:
shutil.copy2(dll_file, dll_dir)
print(f"✓ 已复制 {os.path.basename(dll_file)}")
except Exception as e:
print(f"⚠️ 复制 {dll_file} 失败: {e}")
def copy_data_files():
"""复制数据库文件到data目录"""
print("复制数据库文件...")
data_dir = os.path.join('dist', 'data')
# 数据库文件列表
db_files = ['body_balance.db', 'database.db']
for db_file in db_files:
if os.path.exists(db_file):
try:
shutil.copy2(db_file, data_dir)
print(f"✓ 已复制 {db_file}")
except Exception as e:
print(f"⚠️ 复制 {db_file} 失败: {e}")
else:
print(f"⚠️ 数据库文件不存在: {db_file}")
def copy_config_files():
"""复制配置文件到dist目录"""
print("复制配置文件...")
config_files = ['config.ini']
dist_dir = 'dist'
if not os.path.exists(dist_dir):
os.makedirs(dist_dir)
for config_file in config_files:
if os.path.exists(config_file):
try:
shutil.copy2(config_file, dist_dir)
print(f"✓ 已复制 {config_file}")
except Exception as e:
print(f"⚠️ 复制 {config_file} 失败: {e}")
else:
print(f"⚠️ 配置文件不存在: {config_file}")
def main():
"""主函数"""
print("=" * 60)
print("身体平衡评估系统 - app.py 完整版打包工具")
print("=" * 60)
print()
# 检查当前目录
if not os.path.exists('app.py'):
print("✗ 错误: 找不到 app.py 文件")
print("请确保在backend目录下运行此脚本")
input("按回车键退出...")
return
try:
# 清理构建目录
clean_build_dirs()
print()
# 创建spec文件
print("创建PyInstaller配置...")
create_app_spec()
print()
# 构建exe
if build_exe():
print()
print("后处理...")
# 检查生成的exe文件
exe_path = 'dist/BodyBalanceBackend_Full.exe'
if os.path.exists(exe_path):
print(f"✓ exe文件位置: {exe_path}")
# 创建目录结构
create_directories()
print()
# 复制DLL文件
copy_dll_files()
print()
# 复制数据库文件
copy_data_files()
print()
# 复制配置文件
copy_config_files()
print()
# 创建启动脚本
create_startup_script()
print()
print("🎉 打包完成！")
print()
print("输出文件:")
print(f"- 可执行文件: {exe_path}")
print("- 启动脚本: dist/start_backend_full.bat")
print("- 配置文件: dist/config.ini")
print()
print("目录结构:")
print("- dll/ - DLL文件")
print("- data/ - 数据库文件")
print("- logs/ - 日志文件")
print()
print("使用方式:")
print("1. 直接运行: dist/BodyBalanceBackend_Full.exe")
print("2. 使用脚本: dist/start_backend_full.bat")
print()
print("服务地址: http://localhost:5000")
print("SocketIO: ws://localhost:5000")
else:
print("✗ 错误: 未找到生成的exe文件")
else:
print("\n✗ 打包失败")
except Exception as e:
print(f"\n✗ 打包过程出错: {e}")
print()
input("按回车键退出...")
if __name__ == '__main__':
main()

227
backend/build_exe.py
View File

@ -1,76 +1,108 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
后端应用程序打包脚本
使用PyInstaller将Flask应用程序打包成独立的exe文件
身体平衡评估系统后端打包脚本
基于最小测试框架的成功经验，简化打包配置
"""
import os
import sys
import shutil
import subprocess
from pathlib import Path
def check_dependencies():
"""检查必需的依赖"""
print("检查依赖模块...")
required_modules = [
'flask',
'flask_socketio',
'flask_cors',
'socketio',
'engineio',
'numpy',
'cv2',
'psutil'
]
missing_modules = []
for module in required_modules:
try:
__import__(module)
print(f"✓ {module}")
except ImportError:
print(f"✗ {module} (缺失)")
missing_modules.append(module)
if missing_modules:
print(f"\n缺失模块: {', '.join(missing_modules)}")
print("请运行: pip install -r requirements_build.txt")
return False
print("✓ 所有依赖模块检查通过")
return True
def create_spec_file():
"""创建PyInstaller spec文件"""
spec_content = '''
# -*- mode: python ; coding: utf-8 -*-
block_cipher = None
a = Analysis(
['app_simple.py'],
['main_exe.py'],
pathex=[],
binaries=[
('dll/femtobolt/bin/k4a.dll', '.'), # K4A动态库
('dll/femtobolt/bin/k4arecord.dll', '.'), # K4A录制库
('dll/femtobolt/bin/depthengine_2_0.dll', '.'), # 深度引擎
('dll/femtobolt/bin/OrbbecSDK.dll', '.'), # Orbbec SDK
('dll/femtobolt/bin/ob_usb.dll', '.'), # Orbbec USB库
('dll/femtobolt/bin/live555.dll', '.'), # Live555库
('dll/smitsense/SMiTSenseUsb-F3.0.dll', '.'), # SMiTSense传感器库
('dll/femtobolt/bin/OrbbecSDKConfig_v1.0.xml', '.'), # Orbbec配置文件
# 只包含确实存在的DLL文件
],
datas=[
('..\\config.ini', '.'), # 配置文件
('..\\config.json', '.'), # JSON配置文件
('tests', 'tests'), # 测试文件夹
('config.ini', '.'), # 配置文件
('data', 'data'), # 数据文件夹
('tests', 'tests'), # 测试文件夹
('app_minimal.py', '.'), # 应用入口文件
],
hiddenimports=[
'flask',
'flask_socketio',
'flask_cors',
'socketio',
'engineio',
'engineio.async_drivers.threading',
'socketio.namespace',
'cv2',
'numpy',
'psutil',
'sqlite3',
'configparser',
'logging',
'threading',
'queue',
'base64',
'psutil',
'pykinect_azure',
'requests',
'click',
'colorama',
'database',
'device_manager',
'utils',
'dns',
'dns.resolver',
'dns.asyncresolver'
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
excludes=[
'eventlet',
'gevent',
'gevent_uwsgi'
],
noarchive=False,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
[],
name='BodyBalanceBackend',
@ -86,106 +118,35 @@ exe = EXE(
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon='..\\document\\icon.ico' if os.path.exists('..\\document\\icon.ico') else None,
icon=None
)
'''
with open('backend.spec', 'w', encoding='utf-8') as f:
f.write(spec_content)
print("✓ 已创建 backend.spec 文件")
def create_main_entry():
"""创建主入口文件"""
main_content = '''
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
身体平衡评估系统 - 后端服务入口
独立运行的exe版本
"""
import os
import sys
import logging
from pathlib import Path
# 设置工作目录
if getattr(sys, 'frozen', False):
# 如果是打包后的exe
application_path = os.path.dirname(sys.executable)
else:
# 如果是开发环境
application_path = os.path.dirname(os.path.abspath(__file__))
os.chdir(application_path)
sys.path.insert(0, application_path)
# 创建必要的目录
os.makedirs('logs', exist_ok=True)
os.makedirs('data', exist_ok=True)
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('logs/backend.log', encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
def main():
"""主函数"""
try:
logger.info("启动身体平衡评估系统后端服务...")
logger.info(f"工作目录: {os.getcwd()}")
# 导入并启动Flask应用
from app import app, socketio, init_app
# 初始化应用
init_app()
# 启动服务器
logger.info("后端服务器启动在 http://localhost:5000")
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()
'''
with open('main_exe.py', 'w', encoding='utf-8') as f:
f.write(main_content)
print("✓ 已创建 main_exe.py 入口文件")
print("✓ 已创建 backend.spec 文件")
def build_exe():
"""构建exe文件"""
print("开始构建exe文件...")
# 检查依赖模块
print("\n检查依赖模块...")
if not check_dependencies():
print("\n❌ 依赖检查失败，请先安装缺失的模块")
return False
# 检查PyInstaller是否安装
try:
import PyInstaller
print(f"✓ PyInstaller版本: {PyInstaller.__version__}")
print(f"\n✓ PyInstaller版本: {PyInstaller.__version__}")
except ImportError:
print("❌ PyInstaller未安装，正在安装...")
os.system('pip install pyinstaller')
subprocess.run([sys.executable, '-m', 'pip', 'install', 'pyinstaller>=6.10.0'])
# 清理之前的构建
print("\n清理构建目录...")
if os.path.exists('build'):
shutil.rmtree('build')
print("✓ 清理build目录")
@ -194,20 +155,25 @@ def build_exe():
shutil.rmtree('dist')
print("✓ 清理dist目录")
# 使用spec文件构建
cmd = 'python -m PyInstaller backend.spec --clean --noconfirm'
print(f"执行命令: {cmd}")
result = os.system(cmd)
# 创建spec文件
create_spec_file()
if result == 0:
# 使用spec文件构建
print("\n开始打包...")
cmd = [sys.executable, '-m', 'PyInstaller', 'backend.spec', '--clean', '--noconfirm']
print(f"执行命令: {' '.join(cmd)}")
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8')
if result.returncode == 0:
print("\n✓ 构建成功！")
print("exe文件位置: dist/BodyBalanceBackend.exe")
# 复制额外的文件
dist_path = Path('dist')
if dist_path.exists():
# 复制配置文件
config_files = ['../config.ini', '../config.json']
# 复制配置文件（如果存在）
config_files = ['config.ini', 'config.json']
for config_file in config_files:
if os.path.exists(config_file):
shutil.copy2(config_file, dist_path)
@ -217,37 +183,38 @@ def build_exe():
start_script = dist_path / 'start_backend.bat'
with open(start_script, 'w', encoding='utf-8') as f:
f.write('@echo off\n')
f.write('echo 启动身体平衡评估系统后端服务...\n')
f.write('chcp 65001 >nul\n')
f.write('echo Starting Body Balance Backend...\n')
f.write('BodyBalanceBackend.exe\n')
f.write('pause\n')
print("✓ 创建启动脚本: start_backend.bat")
print("\n🎉 打包完成！")
print("运行方式:")
print("\n测试方法:")
print("1. 直接运行: dist/BodyBalanceBackend.exe")
print("2. 使用脚本: dist/start_backend.bat")
print("3. 在浏览器中访问: http://localhost:5000")
return True
else:
print("❌ 构建失败")
print("\n❌ 构建失败")
print(f"错误信息: {result.stderr}")
return False
return True
def main():
"""主函数"""
print("=" * 50)
print("身体平衡评估系统 - 后端打包工具")
print("=" * 50)
print("================================================")
print("身体平衡评估系统 - 后端打包工具 (简化版)")
print("基于最小测试框架的成功经验")
print("================================================")
print()
# 检查当前目录
if not os.path.exists('app.py'):
print("❌ 请在backend目录下运行此脚本")
return
try:
# 创建配置文件
create_spec_file()
create_main_entry()
# 检查是否在正确的目录
if not os.path.exists('main_exe.py'):
print("❌ 错误：请在backend目录下运行此脚本")
print("当前目录应包含 main_exe.py 文件")
return
# 构建exe
if build_exe():
@ -257,6 +224,8 @@ def main():
except Exception as e:
print(f"❌ 错误: {e}")
import traceback
traceback.print_exc()
input("\n按回车键退出...")

524
backend/build_exe_original.py Normal file
View File

@ -0,0 +1,524 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
后端应用程序打包脚本
使用PyInstaller将Flask应用程序打包成独立的exe文件
"""
import os
import sys
import shutil
import subprocess
from pathlib import Path
def create_spec_file():
"""创建PyInstaller spec文件"""
spec_content = '''
# -*- mode: python ; coding: utf-8 -*-
block_cipher = None
a = Analysis(
['main_exe.py'],
pathex=[],
binaries=[
('dll/femtobolt/bin/k4a.dll', '.'), # K4A动态库
('dll/femtobolt/bin/k4arecord.dll', '.'), # K4A录制库
('dll/femtobolt/bin/depthengine_2_0.dll', '.'), # 深度引擎
('dll/femtobolt/bin/OrbbecSDK.dll', '.'), # Orbbec SDK
('dll/femtobolt/bin/ob_usb.dll', '.'), # Orbbec USB库
('dll/femtobolt/bin/live555.dll', '.'), # Live555库
('dll/smitsense/SMiTSenseUsb-F3.0.dll', '.'), # SMiTSense传感器库
('dll/femtobolt/bin/OrbbecSDKConfig_v1.0.xml', '.'), # Orbbec配置文件
],
datas=[
('..\\config.ini', '.'), # 配置文件
('..\\config.json', '.'), # JSON配置文件
('tests', 'tests'), # 测试文件夹
('data', 'data'), # 数据文件夹
],
hiddenimports=[
'flask',
'flask_socketio',
'flask_cors',
'cv2',
'numpy',
'pandas',
'scipy',
'matplotlib',
'seaborn',
'sklearn',
'PIL',
'reportlab',
'sqlite3',
'configparser',
'logging',
'threading',
'queue',
'base64',
'psutil',
'pykinect_azure',
'pyserial',
'requests',
'yaml',
'click',
'colorama',
'tqdm',
'database',
'device_manager',
'utils',
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
noarchive=False,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
[],
name='BodyBalanceBackend',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon=None, # 图标文件不存在，暂时禁用
)
'''
with open('backend.spec', 'w', encoding='utf-8') as f:
f.write(spec_content)
print("✓ 已创建 backend.spec 文件")
def create_debug_spec():
"""创建诊断脚本的spec文件"""
import subprocess
spec_content = f'''# -*- mode: python ; coding: utf-8 -*-
block_cipher = None
a = Analysis(
['test_socketio_debug.py'],
pathex=['{os.getcwd()}'],
binaries=[],
datas=[],
hiddenimports=[
'flask',
'flask_socketio',
'eventlet',
'threading',
'engineio.async_drivers.threading',
'engineio.async_drivers.eventlet',
'engineio.async_eventlet',
'socketio.async_eventlet',
'greenlet',
],
hookspath=[],
hooksconfig={{}},
runtime_hooks=[],
excludes=[],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
noarchive=False,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
[],
name='test_socketio_debug',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
)
'''
with open('debug.spec', 'w', encoding='utf-8') as f:
f.write(spec_content)
print("✓ 已创建 debug.spec 文件")
# 直接构建诊断exe
print("开始构建诊断exe文件...")
result = subprocess.run([sys.executable, '-m', 'PyInstaller', 'debug.spec'],
capture_output=True, text=True, encoding='utf-8')
if result.returncode == 0:
print("✓ 诊断exe构建成功！")
print("运行诊断: dist/test_socketio_debug.exe")
else:
print(f"❌ 诊断exe构建失败: {result.stderr}")
def create_minimal_spec():
"""创建最小化测试脚本的spec文件"""
print("创建最小化测试脚本的spec文件...")
spec_content = '''# -*- mode: python ; coding: utf-8 -*-
block_cipher = None
a = Analysis(
['test_socketio_minimal.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=[
'flask',
'flask_socketio',
'eventlet',
'eventlet.hubs',
'eventlet.hubs.epolls',
'eventlet.hubs.kqueue',
'eventlet.hubs.selects',
'threading',
'engineio',
'engineio.async_drivers',
'engineio.async_drivers.threading',
'engineio.async_drivers.eventlet',
'engineio.async_drivers.gevent',
'socketio',
'socketio.namespace'
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
noarchive=False,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
[],
name='test_socketio_minimal',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
)
'''
with open('minimal.spec', 'w', encoding='utf-8') as f:
f.write(spec_content)
print("✓ 已创建 minimal.spec 文件")
# 直接构建
print("开始构建最小化测试exe...")
result = subprocess.run([sys.executable, '-m', 'PyInstaller', 'minimal.spec', '--clean'],
capture_output=True, text=True, encoding='utf-8')
if result.returncode == 0:
print("✓ 最小化测试exe构建完成: dist/test_socketio_minimal.exe")
else:
print(f"❌ 最小化测试exe构建失败: {result.stderr}")
def create_main_entry():
"""创建主入口文件"""
main_content = '''
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
身体平衡评估系统 - 后端服务入口
独立运行的exe版本
"""
import os
import sys
import logging
from pathlib import Path
# 设置工作目录
if getattr(sys, 'frozen', False):
# 如果是打包后的exe
application_path = os.path.dirname(sys.executable)
else:
# 如果是开发环境
application_path = os.path.dirname(os.path.abspath(__file__))
os.chdir(application_path)
sys.path.insert(0, application_path)
# 创建必要的目录
os.makedirs('logs', exist_ok=True)
os.makedirs('data', exist_ok=True)
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('logs/backend.log', encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
def main():
"""主函数"""
try:
logger.info("启动身体平衡评估系统后端服务...")
logger.info(f"工作目录: {os.getcwd()}")
# 导入并启动Flask应用
from app import app, socketio, init_app
# 初始化应用
init_app()
# 启动服务器
logger.info("后端服务器启动在 http://localhost:5000")
socketio.run(
app,
host='0.0.0.0',
port=5000,
debug=False
)
except KeyboardInterrupt:
logger.info("用户中断，正在关闭服务器...")
except Exception as e:
logger.error(f"启动失败: {e}")
input("按回车键退出...")
sys.exit(1)
if __name__ == '__main__':
main()
'''
with open('main_exe.py', 'w', encoding='utf-8') as f:
f.write(main_content)
print("✓ 已创建 main_exe.py 入口文件")
def check_dependencies():
"""检查必需的依赖模块"""
required_modules = [
'flask', 'flask_cors', 'flask_socketio', 'numpy', 'pandas',
'cv2', 'matplotlib', 'seaborn', 'PIL', 'reportlab',
'serial', 'requests', 'sqlite3', 'configparser', 'scipy', 'eventlet'
]
missing_modules = []
for module in required_modules:
try:
__import__(module)
print(f"✓ {module}")
except ImportError:
missing_modules.append(module)
print(f"❌ {module} - 未安装")
if missing_modules:
print(f"\n警告: 发现 {len(missing_modules)} 个缺失的依赖模块")
print("建议运行: pip install -r requirements.txt")
return False
print("✓ 所有必需依赖模块检查通过")
return True
def build_exe():
"""构建exe文件"""
print("开始构建exe文件...")
# 检查依赖模块
print("\n检查依赖模块...")
if not check_dependencies():
print("\n❌ 依赖检查失败，请先安装缺失的模块")
return False
# 检查PyInstaller是否安装
try:
import PyInstaller
print(f"\n✓ PyInstaller版本: {PyInstaller.__version__}")
except ImportError:
print("❌ PyInstaller未安装，正在安装...")
os.system('pip install pyinstaller')
# 清理之前的构建
if os.path.exists('build'):
shutil.rmtree('build')
print("✓ 清理build目录")
if os.path.exists('dist'):
shutil.rmtree('dist')
print("✓ 清理dist目录")
# 使用spec文件构建
cmd = 'python -m PyInstaller backend.spec --clean --noconfirm'
print(f"执行命令: {cmd}")
result = os.system(cmd)
if result == 0:
print("\n✓ 构建成功！")
print("exe文件位置: dist/BodyBalanceBackend.exe")
# 复制额外的文件
dist_path = Path('dist')
if dist_path.exists():
# 复制配置文件
config_files = ['../config.ini', '../config.json']
for config_file in config_files:
if os.path.exists(config_file):
shutil.copy2(config_file, dist_path)
print(f"✓ 复制配置文件: {config_file}")
# 创建启动脚本
start_script = dist_path / 'start_backend.bat'
with open(start_script, 'w', encoding='utf-8') as f:
f.write('@echo off\n')
f.write('echo 启动身体平衡评估系统后端服务...\n')
f.write('BodyBalanceBackend.exe\n')
f.write('pause\n')
print("✓ 创建启动脚本: start_backend.bat")
print("\n🎉 打包完成！")
print("运行方式:")
print("1. 直接运行: dist/BodyBalanceBackend.exe")
print("2. 使用脚本: dist/start_backend.bat")
else:
print("❌ 构建失败")
return False
return True
def check_required_files():
"""检查必需的文件是否存在"""
required_files = {
'app.py': '主应用文件',
'database.py': '数据库模块',
'device_manager.py': '设备管理模块',
'utils.py': '工具模块',
'../config.ini': '配置文件',
'../config.json': 'JSON配置文件'
}
required_dlls = {
'dll/femtobolt/bin/k4a.dll': 'K4A动态库',
'dll/femtobolt/bin/k4arecord.dll': 'K4A录制库',
'dll/femtobolt/bin/depthengine_2_0.dll': '深度引擎',
'dll/femtobolt/bin/OrbbecSDK.dll': 'Orbbec SDK',
'dll/femtobolt/bin/ob_usb.dll': 'Orbbec USB库',
'dll/femtobolt/bin/live555.dll': 'Live555库',
'dll/smitsense/SMiTSenseUsb-F3.0.dll': 'SMiTSense传感器库',
'dll/femtobolt/bin/OrbbecSDKConfig_v1.0.xml': 'Orbbec配置文件'
}
missing_files = []
print("检查必需文件...")
for file_path, description in required_files.items():
if os.path.exists(file_path):
print(f"✓ {description}: {file_path}")
else:
missing_files.append((file_path, description))
print(f"❌ {description}: {file_path} - 文件不存在")
print("\n检查DLL文件...")
for dll_path, description in required_dlls.items():
if os.path.exists(dll_path):
print(f"✓ {description}: {dll_path}")
else:
missing_files.append((dll_path, description))
print(f"❌ {description}: {dll_path} - 文件不存在")
if missing_files:
print(f"\n警告: 发现 {len(missing_files)} 个缺失文件")
print("缺失的文件可能导致打包失败或运行时错误")
return False
print("\n✓ 所有必需文件检查通过")
return True
def main():
"""主函数"""
print("=" * 50)
print("身体平衡评估系统 - 后端打包工具")
print("=" * 50)
print()
# 检查当前目录
if not os.path.exists('app.py'):
print("❌ 请在backend目录下运行此脚本")
return
# 检查必需文件
print("\n检查必需文件...")
if not check_required_files():
print("\n⚠️ 文件检查发现问题，但将继续执行打包")
print("如果打包失败，请检查上述缺失的文件")
input("按回车键继续...")
try:
# 创建配置文件
create_spec_file()
# create_main_entry() # 注释掉以保留现有的main_exe.py修改
# 临时：为诊断脚本创建spec文件
if len(sys.argv) > 1 and sys.argv[1] == 'debug':
create_debug_spec()
return
elif len(sys.argv) > 1 and sys.argv[1] == 'minimal':
create_minimal_spec()
return
# 构建exe
if build_exe():
print("\n✅ 所有操作完成！")
else:
print("\n❌ 构建过程中出现错误")
except Exception as e:
print(f"❌ 错误: {e}")
input("\n按回车键退出...")
if __name__ == '__main__':
main()

244
backend/build_minimal.py Normal file
View File

@ -0,0 +1,244 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
app_minimal.py 打包脚本
使用PyInstaller将最小版本Flask应用程序打包成独立的exe文件
"""
import os
import sys
import shutil
import subprocess
from pathlib import Path
def clean_build_dirs():
"""清理构建目录"""
print("清理构建目录...")
dirs_to_clean = ['build', 'dist', '__pycache__']
for dir_name in dirs_to_clean:
if os.path.exists(dir_name):
try:
shutil.rmtree(dir_name)
print(f"✓ 已删除 {dir_name}")
except Exception as e:
print(f"⚠️ 删除 {dir_name} 失败: {e}")
def create_minimal_spec():
"""创建app_minimal.py的PyInstaller spec文件"""
spec_content = '''
# -*- mode: python ; coding: utf-8 -*-
block_cipher = None
a = Analysis(
['app_minimal.py'],
pathex=[],
binaries=[
('dll/smitsense/SMiTSenseUsb-F3.0.dll', '.'), # SMiTSense传感器库
],
datas=[
('config.ini', '.'), # 配置文件
('data', 'data'), # 数据文件夹
],
hiddenimports=[
'flask',
'flask_cors',
'sqlite3',
'configparser',
'logging',
'threading',
'queue',
'base64',
'psutil',
'database',
'device_manager',
'utils',
'cv2',
'numpy',
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[
'flask_socketio',
'socketio',
'engineio',
'eventlet',
'gevent',
],
win_no_prefer_redirects=False,
win_private_assemblies=False,
cipher=block_cipher,
noarchive=False,
)
pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.zipfiles,
a.datas,
[],
name='BodyBalanceBackend',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon=None
)
'''
with open('app_minimal.spec', 'w', encoding='utf-8') as f:
f.write(spec_content)
print("✓ 已创建 app_minimal.spec 文件")
def build_exe():
"""构建exe文件"""
print("\n开始构建exe文件...")
try:
# 使用PyInstaller构建
cmd = [sys.executable, '-m', 'PyInstaller', 'app_minimal.spec', '--clean', '--noconfirm']
print(f"执行命令: {' '.join(cmd)}")
result = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8')
if result.returncode == 0:
print("✓ 构建成功！")
return True
else:
print(f"✗ 构建失败")
print(f"错误输出: {result.stderr}")
return False
except Exception as e:
print(f"✗ 构建过程出错: {e}")
return False
def create_startup_script():
"""创建启动脚本"""
startup_script = '''@echo off
echo 启动身体平衡评估系统后端服务（最小版本）...
echo.
echo 服务信息:
echo - HTTP API: http://localhost:5000
echo - 管理界面: http://localhost:5000
echo.
echo 按Ctrl+C停止服务
echo.
"BodyBalanceBackend.exe"
if %errorlevel% neq 0 (
echo.
echo 服务启动失败，请检查错误信息
pause
)
'''
dist_dir = 'dist'
if not os.path.exists(dist_dir):
os.makedirs(dist_dir)
with open(os.path.join(dist_dir, 'start_backend.bat'), 'w', encoding='utf-8') as f:
f.write(startup_script)
print("✓ 创建启动脚本: dist/start_backend.bat")
def copy_config_files():
"""复制配置文件到dist目录"""
print("复制配置文件...")
config_files = ['config.ini']
dist_dir = 'dist'
if not os.path.exists(dist_dir):
os.makedirs(dist_dir)
for config_file in config_files:
if os.path.exists(config_file):
try:
shutil.copy2(config_file, dist_dir)
print(f"✓ 已复制 {config_file}")
except Exception as e:
print(f"⚠️ 复制 {config_file} 失败: {e}")
else:
print(f"⚠️ 配置文件不存在: {config_file}")
def main():
"""主函数"""
print("=" * 60)
print("身体平衡评估系统 - app_minimal.py 打包工具")
print("=" * 60)
print()
# 检查当前目录
if not os.path.exists('app_minimal.py'):
print("✗ 错误: 找不到 app_minimal.py 文件")
print("请确保在backend目录下运行此脚本")
input("按回车键退出...")
return
try:
# 清理构建目录
clean_build_dirs()
print()
# 创建spec文件
print("创建PyInstaller配置...")
create_minimal_spec()
print()
# 构建exe
if build_exe():
print()
print("后处理...")
# 检查生成的exe文件
exe_path = 'dist/BodyBalanceBackend.exe'
if os.path.exists(exe_path):
print(f"✓ exe文件位置: {exe_path}")
# 复制配置文件
copy_config_files()
# 创建启动脚本
create_startup_script()
print()
print("🎉 打包完成！")
print()
print("输出文件:")
print(f"- 可执行文件: {exe_path}")
print("- 启动脚本: dist/start_backend.bat")
print("- 配置文件: dist/config.ini")
print()
print("使用方式:")
print("1. 直接运行: dist/BodyBalanceBackend.exe")
print("2. 使用脚本: dist/start_backend.bat")
print()
print("服务地址: http://localhost:5000")
else:
print("✗ 错误: 未找到生成的exe文件")
else:
print("\n✗ 打包失败")
except Exception as e:
print(f"\n✗ 打包过程出错: {e}")
print()
input("按回车键退出...")
if __name__ == '__main__':
main()

BIN
backend/data/patients/202508030001/20250813134852/feet.mp4 Normal file
View File

Binary file not shown.

BIN
backend/data/patients/202508030001/20250813134852/screen.webm Normal file
View File

Binary file not shown.

23
backend/main_exe.py
View File

@ -45,20 +45,27 @@ def main():
logger.info(f"工作目录: {os.getcwd()}")
# 导入并启动Flask应用
from app import app, socketio, init_app
from app_minimal import app, socketio, init_app
# 初始化应用
init_app()
# 启动服务器
logger.info("后端服务器启动在 http://localhost:5000")
socketio.run(
app,
host='0.0.0.0',
port=5000,
debug=False,
allow_unsafe_werkzeug=True
)
if socketio is not None:
socketio.run(
app,
host='0.0.0.0',
port=5000,
debug=False
)
else:
logger.info("使用基本Flask模式启动（无SocketIO）")
app.run(
host='0.0.0.0',
port=5000,
debug=False
)
except KeyboardInterrupt:
logger.info("用户中断，正在关闭服务器...")

1
backend/requirements.txt
View File

@ -2,6 +2,7 @@
Flask
Flask-CORS
Flask-SocketIO
eventlet
# Data processing and scientific computing
numpy

6
backend/requirements_build.txt
View File

@ -7,14 +7,14 @@ Flask-CORS==4.0.0
Flask-SocketIO==5.3.6
# Core dependencies
numpy==1.24.3
opencv-python==4.8.1.78
numpy>=1.24.0
opencv-python>=4.8.0
psutil==5.9.5
requests==2.31.0
python-dateutil==2.8.2
# PyInstaller for building
PyInstaller==6.2.0
PyInstaller>=6.10.0
# Optional - only if available
# pykinect_azure # Comment out if not available

117
backend/test_socketio_debug.py Normal file
View File

@ -0,0 +1,117 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
测试SocketIO异步模式问题的诊断脚本
"""
import sys
import os
# 添加当前目录到Python路径
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
print("=== SocketIO 异步模式诊断 ===")
print(f"Python版本: {sys.version}")
print(f"工作目录: {os.getcwd()}")
print()
# 检查可用的异步模式
print("检查可用的异步驱动模块:")
# 检查 threading
try:
import threading
print("✓ threading 模块可用")
except ImportError as e:
print(f"✗ threading 模块不可用: {e}")
# 检查 eventlet
try:
import eventlet
print("✓ eventlet 模块可用")
except ImportError as e:
print(f"✗ eventlet 模块不可用: {e}")
# 检查 gevent
try:
import gevent
print("✓ gevent 模块可用")
except ImportError as e:
print(f"✗ gevent 模块不可用: {e}")
print()
print("检查 engineio 异步驱动:")
# 检查 engineio.async_drivers.threading
try:
import engineio.async_drivers.threading
print("✓ engineio.async_drivers.threading 可用")
except ImportError as e:
print(f"✗ engineio.async_drivers.threading 不可用: {e}")
# 检查 engineio.async_drivers.eventlet
try:
import engineio.async_drivers.eventlet
print("✓ engineio.async_drivers.eventlet 可用")
except ImportError as e:
print(f"✗ engineio.async_drivers.eventlet 不可用: {e}")
# 检查 engineio.async_eventlet
try:
import engineio.async_eventlet
print("✓ engineio.async_eventlet 可用")
except ImportError as e:
print(f"✗ engineio.async_eventlet 不可用: {e}")
# 检查 engineio.async_threading
try:
import engineio.async_threading
print("✓ engineio.async_threading 可用")
except ImportError as e:
print(f"✗ engineio.async_threading 不可用: {e}")
print()
print("测试 SocketIO 初始化:")
try:
from flask import Flask
from flask_socketio import SocketIO
app = Flask(__name__)
app.config['SECRET_KEY'] = 'test-key'
# 测试不指定 async_mode
print("测试 SocketIO() 不指定 async_mode...")
try:
socketio = SocketIO(app, cors_allowed_origins='*')
print(f"✓ SocketIO 初始化成功，自动选择的异步模式: {socketio.async_mode}")
except Exception as e:
print(f"✗ SocketIO 初始化失败: {e}")
# 测试指定 threading
print("测试 SocketIO(async_mode='threading')...")
try:
app2 = Flask(__name__)
app2.config['SECRET_KEY'] = 'test-key-2'
socketio2 = SocketIO(app2, cors_allowed_origins='*', async_mode='threading')
print(f"✓ SocketIO threading 模式初始化成功")
except Exception as e:
print(f"✗ SocketIO threading 模式初始化失败: {e}")
# 测试指定 eventlet
print("测试 SocketIO(async_mode='eventlet')...")
try:
app3 = Flask(__name__)
app3.config['SECRET_KEY'] = 'test-key-3'
socketio3 = SocketIO(app3, cors_allowed_origins='*', async_mode='eventlet')
print(f"✓ SocketIO eventlet 模式初始化成功")
except Exception as e:
print(f"✗ SocketIO eventlet 模式初始化失败: {e}")
except Exception as e:
print(f"✗ Flask/SocketIO 导入失败: {e}")
print()
print("=== 诊断完成 ===")
print("按回车键退出...")
input()

112
backend/test_socketio_minimal.py Normal file
View File

@ -0,0 +1,112 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
最小化SocketIO测试脚本
用于诊断PyInstaller打包后的SocketIO初始化问题
"""
import sys
import os
import traceback
print(f"Python版本: {sys.version}")
print(f"工作目录: {os.getcwd()}")
print(f"Python路径: {sys.path[:3]}...") # 只显示前3个路径
print("="*50)
# 测试基础模块导入
try:
import threading
print("✓ threading模块可用")
except ImportError as e:
print(f"✗ threading模块不可用: {e}")
try:
import eventlet
print("✓ eventlet模块可用")
except ImportError as e:
print(f"✗ eventlet模块不可用: {e}")
try:
import gevent
print("✓ gevent模块可用")
except ImportError as e:
print(f"✗ gevent模块不可用: {e}")
print("="*50)
# 测试Flask导入
try:
from flask import Flask
print("✓ Flask导入成功")
except ImportError as e:
print(f"✗ Flask导入失败: {e}")
sys.exit(1)
# 测试SocketIO导入
try:
from flask_socketio import SocketIO
print("✓ SocketIO导入成功")
except ImportError as e:
print(f"✗ SocketIO导入失败: {e}")
sys.exit(1)
print("="*50)
# 测试engineio异步驱动
try:
import engineio.async_drivers.threading
print("✓ engineio.async_drivers.threading可用")
except ImportError as e:
print(f"✗ engineio.async_drivers.threading不可用: {e}")
try:
import engineio.async_drivers.eventlet
print("✓ engineio.async_drivers.eventlet可用")
except ImportError as e:
print(f"✗ engineio.async_drivers.eventlet不可用: {e}")
try:
import engineio.async_drivers.gevent
print("✓ engineio.async_drivers.gevent可用")
except ImportError as e:
print(f"✗ engineio.async_drivers.gevent不可用: {e}")
print("="*50)
# 创建Flask应用
app = Flask(__name__)
app.config['SECRET_KEY'] = 'test-key'
print("测试SocketIO初始化...")
# 测试1: 不指定async_mode
print("\n1. 测试自动检测async_mode:")
try:
socketio_auto = SocketIO(app, cors_allowed_origins='*')
print(f"✓ 自动检测成功，使用模式: {socketio_auto.async_mode}")
except Exception as e:
print(f"✗ 自动检测失败: {e}")
print(f"错误详情: {traceback.format_exc()}")
# 测试2: 明确指定threading模式
print("\n2. 测试threading模式:")
try:
socketio_threading = SocketIO(app, cors_allowed_origins='*', async_mode='threading')
print(f"✓ threading模式成功，使用模式: {socketio_threading.async_mode}")
except Exception as e:
print(f"✗ threading模式失败: {e}")
print(f"错误详情: {traceback.format_exc()}")
# 测试3: 明确指定eventlet模式
print("\n3. 测试eventlet模式:")
try:
socketio_eventlet = SocketIO(app, cors_allowed_origins='*', async_mode='eventlet')
print(f"✓ eventlet模式成功，使用模式: {socketio_eventlet.async_mode}")
except Exception as e:
print(f"✗ eventlet模式失败: {e}")
print(f"错误详情: {traceback.format_exc()}")
print("\n" + "="*50)
print("测试完成！")
input("按回车键退出...")

7
debug_server.py
View File

@ -95,7 +95,12 @@ def start_debug_server():
# 初始化应用
logger.info('初始化应用...')
init_app()
try:
init_app()
logger.info('应用初始化成功')
except Exception as init_e:
logger.error(f'应用初始化失败: {init_e}', exc_info=True)
raise
# 获取本机IP地址
local_ip = get_local_ip()

2
frontend/src/main/main.js
View File

@ -7,7 +7,7 @@ const log = require('electron-log');
let mainWindow;
let pythonProcess;
const BACKEND_PORT = 5000;
const BACKEND_HOST = process.env.BACKEND_HOST || '0.0.0.0';
const BACKEND_HOST = process.env.BACKEND_HOST || '127.0.0.1';
const BACKEND_URL = `http://${BACKEND_HOST}:${BACKEND_PORT}`;
// 配置日志

4
frontend/src/renderer/src/services/api.js
View File

@ -16,7 +16,7 @@ api.interceptors.request.use(
if (window.electronAPI) {
config.baseURL = window.electronAPI.getBackendUrl()
} else {
config.baseURL = 'http://192.168.1.173:5000'
config.baseURL = 'http://localhost:5000'
}
@ -600,7 +600,7 @@ export const getBackendUrl = () => {
if (window.electronAPI) {
return window.electronAPI.getBackendUrl()
} else {
return 'http://192.168.1.173:5000'
return 'http://localhost:5000'
}
}

200
install/README.md Normal file
View File

@ -0,0 +1,200 @@
# 最小功能测试框架
## 概述
这是一个最小功能测试框架，用于验证 Flask + SocketIO + threading 技术栈在打包成 exe 后的可用性。
## 功能特性
- ✅ Flask HTTP 服务
- ✅ SocketIO WebSocket 服务
- ✅ 强制使用 threading 异步模式
- ✅ 完整的前端测试界面
- ✅ HTTP API 测试
- ✅ WebSocket 连接测试
- ✅ 实时消息收发测试
- ✅ 系统信息显示
## 文件结构
```
install/
├── minimal_test_app.py # 主应用文件
├── build_minimal.py # 打包脚本
├── requirements_minimal.txt # 最小依赖列表
└── README.md # 说明文档
```
## 快速开始
### 1. 安装依赖
```bash
pip install -r requirements_minimal.txt
```
### 2. 测试运行
```bash
python minimal_test_app.py
```
然后在浏览器中访问 http://localhost:5000 进行测试。
### 3. 打包成 exe
```bash
python build_minimal.py
```
### 4. 测试 exe
打包完成后，运行：
```bash
# 方式1：直接运行
dist/MinimalTestApp.exe
# 方式2：使用脚本
dist/start_test.bat
```
## 测试步骤
### HTTP API 测试
1. 点击「测试 HTTP API」按钮
2. 确认返回成功响应
3. 检查响应数据包含服务器信息
### WebSocket 测试
1. 页面加载时会自动连接 WebSocket
2. 确认连接状态显示为「已连接」
3. 点击「发送测试消息」
4. 确认能收到服务器响应
5. 测试断开和重连功能
### 系统信息检查
- 服务器时间：确认时间正确
- SocketIO模式：确认显示为 "threading"
- Flask版本：确认版本信息
## 技术要点
### 异步模式选择
```python
# 强制使用 threading 模式，避免 eventlet/gevent 依赖问题
socketio = SocketIO(
app,
cors_allowed_origins='*',
async_mode='threading', # 关键配置
logger=False,
engineio_logger=False
)
```
### 打包配置
```python
# PyInstaller 隐藏导入配置
hiddenimports=[
'flask',
'flask_socketio',
'socketio',
'engineio',
'engineio.async_drivers.threading', # 关键：threading 驱动
'socketio.namespace',
'dns',
'dns.resolver',
'dns.asyncresolver'
],
# 排除不需要的异步模式
excludes=[
'eventlet',
'gevent',
'gevent_uwsgi'
]
```
## 故障排除
### 常见问题
1. **"Invalid async_mode specified" 错误**
- 确认已安装所有依赖
- 检查 PyInstaller 隐藏导入配置
- 确认排除了不需要的异步模式
2. **WebSocket 连接失败**
- 检查防火墙设置
- 确认端口 5000 未被占用
- 查看控制台错误信息
3. **打包失败**
- 确认 PyInstaller 版本兼容
- 检查依赖版本冲突
- 查看详细错误输出
### 调试模式
如需调试，可以修改 `minimal_test_app.py`：
```python
# 启用调试模式
socketio.run(
app,
host='0.0.0.0',
port=5000,
debug=True, # 启用调试
allow_unsafe_werkzeug=True
)
```
## 扩展指南
### 添加新功能
1. **新增 HTTP 路由**
```python
@app.route('/api/new-endpoint')
def new_endpoint():
return jsonify({'message': 'New endpoint'})
```
2. **新增 SocketIO 事件**
```python
@socketio.on('new_event')
def handle_new_event(data):
emit('response', {'status': 'received'})
```
3. **添加依赖**
- 更新 `requirements_minimal.txt`
- 更新 `build_minimal.py` 中的 `hiddenimports`
- 重新测试打包
### 渐进式集成
按照以下顺序逐步添加业务功能：
1. ✅ 基础 Flask + SocketIO（当前阶段）
2. 🔄 添加数据库支持
3. 🔄 添加文件操作
4. 🔄 添加外部库依赖
5. 🔄 添加硬件设备支持
每个阶段都要确保打包和运行正常，出现问题时更容易定位。
## 版本信息
- Flask: 2.3.3
- Flask-SocketIO: 5.3.6
- PyInstaller: 6.1.0
- Python: 3.8+
## 许可证
MIT License

224
install/build_minimal.py Normal file
View File

@ -0,0 +1,224 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
最小功能测试框架打包脚本
用于将Flask + SocketIO应用打包成exe文件
"""
import os
import sys
import shutil
import subprocess
from pathlib import Path
def check_dependencies():
"""检查必需的依赖"""
print("检查依赖模块...")
required_modules = [
'flask',
'flask_socketio',
'socketio',
'engineio'
]
missing_modules = []
for module in required_modules:
try:
__import__(module)
print(f"✓ {module}")
except ImportError:
print(f"✗ {module} (缺失)")
missing_modules.append(module)
if missing_modules:
print(f"\n缺失模块: {', '.join(missing_modules)}")
print("请运行: pip install -r requirements_minimal.txt")
return False
print("✓ 所有依赖模块检查通过")
return True
def create_spec_file():
"""创建PyInstaller spec文件"""
spec_content = '''
# -*- mode: python ; coding: utf-8 -*-
a = Analysis(
['minimal_test_app.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=[
'flask',
'flask_socketio',
'socketio',
'engineio',
'engineio.async_drivers.threading',
'socketio.namespace',
'dns',
'dns.resolver',
'dns.asyncresolver'
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[
'eventlet',
'gevent',
'gevent_uwsgi'
],
noarchive=False,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='MinimalTestApp',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon=None
)
'''
with open('minimal_test.spec', 'w', encoding='utf-8') as f:
f.write(spec_content)
print("✓ 已创建 minimal_test.spec 文件")
def clean_build_dirs():
"""清理构建目录"""
dirs_to_clean = ['build', 'dist', '__pycache__']
for dir_name in dirs_to_clean:
if os.path.exists(dir_name):
try:
shutil.rmtree(dir_name)
print(f"✓ 清理目录: {dir_name}")
except Exception as e:
print(f"警告: 无法清理目录 {dir_name}: {e}")
def build_exe():
"""构建exe文件"""
print("\n开始构建exe文件...")
try:
# 使用PyInstaller构建
cmd = [sys.executable, '-m', 'PyInstaller', 'minimal_test.spec', '--clean']
print(f"执行命令: {' '.join(cmd)}")
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
print("✓ 构建成功！")
return True
else:
print(f"✗ 构建失败")
print(f"错误输出: {result.stderr}")
return False
except Exception as e:
print(f"✗ 构建过程出错: {e}")
return False
def create_test_script():
"""创建测试脚本"""
test_script = '''
@echo off
echo 启动最小功能测试应用...
echo.
echo 测试说明:
echo 1. 应用启动后，在浏览器中访问 http://localhost:5000
echo 2. 测试HTTP API和WebSocket功能
echo 3. 按Ctrl+C停止应用
echo.
"MinimalTestApp.exe"
pause
'''
with open('dist/start_test.bat', 'w', encoding='utf-8') as f:
f.write(test_script)
print("✓ 创建测试脚本: dist/start_test.bat")
def main():
"""主函数"""
print("="*60)
print("最小功能测试框架 - 打包工具")
print("="*60)
print()
# 检查当前目录
if not os.path.exists('minimal_test_app.py'):
print("✗ 错误: 找不到 minimal_test_app.py 文件")
print("请确保在正确的目录下运行此脚本")
input("按回车键退出...")
return
# 检查依赖
if not check_dependencies():
input("按回车键退出...")
return
print()
# 清理构建目录
print("清理构建目录...")
clean_build_dirs()
print()
# 创建spec文件
print("创建PyInstaller配置...")
create_spec_file()
print()
# 构建exe
if build_exe():
print()
print("后处理...")
# 检查生成的exe文件
exe_path = 'dist/MinimalTestApp.exe'
if os.path.exists(exe_path):
print(f"✓ exe文件位置: {exe_path}")
# 创建测试脚本
create_test_script()
print()
print("🎉 打包完成！")
print()
print("测试方式:")
print("1. 直接运行: dist/MinimalTestApp.exe")
print("2. 使用脚本: dist/start_test.bat")
print()
print("测试步骤:")
print("1. 启动应用")
print("2. 浏览器访问 http://localhost:5000")
print("3. 测试HTTP API和WebSocket功能")
print("4. 确认所有功能正常工作")
else:
print("✗ 错误: 未找到生成的exe文件")
else:
print("\n✗ 打包失败")
print()
input("按回车键退出...")
if __name__ == '__main__':
main()

53
install/minimal_test.spec Normal file
View File

@ -0,0 +1,53 @@
# -*- mode: python ; coding: utf-8 -*-
a = Analysis(
['minimal_test_app.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=[
'flask',
'flask_socketio',
'socketio',
'engineio',
'engineio.async_drivers.threading',
'socketio.namespace',
'dns',
'dns.resolver',
'dns.asyncresolver'
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
excludes=[
'eventlet',
'gevent',
'gevent_uwsgi'
],
noarchive=False,
)
pyz = PYZ(a.pure)
exe = EXE(
pyz,
a.scripts,
a.binaries,
a.datas,
[],
name='MinimalTestApp',
debug=False,
bootloader_ignore_signals=False,
strip=False,
upx=True,
upx_exclude=[],
runtime_tmpdir=None,
console=True,
disable_windowed_traceback=False,
argv_emulation=False,
target_arch=None,
codesign_identity=None,
entitlements_file=None,
icon=None
)

274
install/minimal_test_app.py Normal file
View File

@ -0,0 +1,274 @@
#!/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 = """
<!DOCTYPE html>
<html>
<head>
<title>最小功能测试 - Flask + SocketIO</title>
<meta charset="utf-8">
<style>
body { font-family: Arial, sans-serif; margin: 40px; }
.container { max-width: 800px; margin: 0 auto; }
.section { margin: 20px 0; padding: 20px; border: 1px solid #ddd; border-radius: 5px; }
.success { color: green; }
.error { color: red; }
.info { color: blue; }
button { padding: 10px 20px; margin: 5px; cursor: pointer; }
#messages { height: 200px; overflow-y: auto; border: 1px solid #ccc; padding: 10px; background: #f9f9f9; }
.message { margin: 5px 0; }
</style>
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
</head>
<body>
<div class="container">
<h1>最小功能测试框架</h1>
<p class="info">测试Flask HTTP服务和SocketIO WebSocket服务</p>
<!-- HTTP API测试 -->
<div class="section">
<h2>HTTP API 测试</h2>
<button onclick="testHttpApi()">测试 HTTP API</button>
<div id="http-result"></div>
</div>
<!-- WebSocket测试 -->
<div class="section">
<h2>WebSocket 测试</h2>
<button onclick="connectSocket()">连接 WebSocket</button>
<button onclick="disconnectSocket()">断开连接</button>
<button onclick="sendTestMessage()">发送测试消息</button>
<div>连接状态: <span id="connection-status" class="error">未连接</span></div>
<div id="messages"></div>
</div>
<!-- 系统信息 -->
<div class="section">
<h2>系统信息</h2>
<div id="system-info">
<p>服务器时间: {{ server_time }}</p>
<p>SocketIO模式: {{ socketio_mode }}</p>
<p>Flask版本: {{ flask_version }}</p>
</div>
</div>
</div>
<script>
let socket = null;
function addMessage(message, type = 'info') {
const messages = document.getElementById('messages');
const div = document.createElement('div');
div.className = `message ${type}`;
div.innerHTML = `[${new Date().toLocaleTimeString()}] ${message}`;
messages.appendChild(div);
messages.scrollTop = messages.scrollHeight;
}
function testHttpApi() {
const resultDiv = document.getElementById('http-result');
resultDiv.innerHTML = '测试中...';
fetch('/api/test')
.then(response => response.json())
.then(data => {
resultDiv.innerHTML = `<span class="success">✓ HTTP API 正常</span><br>响应: ${JSON.stringify(data, null, 2)}`;
})
.catch(error => {
resultDiv.innerHTML = `<span class="error">✗ HTTP API 错误: ${error}</span>`;
});
}
function connectSocket() {
if (socket && socket.connected) {
addMessage('WebSocket已连接', 'info');
return;
}
socket = io();
socket.on('connect', function() {
document.getElementById('connection-status').innerHTML = '<span class="success">已连接</span>';
addMessage('WebSocket连接成功', 'success');
});
socket.on('disconnect', function() {
document.getElementById('connection-status').innerHTML = '<span class="error">未连接</span>';
addMessage('WebSocket连接断开', 'error');
});
socket.on('test_response', function(data) {
addMessage(`收到服务器响应: ${JSON.stringify(data)}`, 'success');
});
socket.on('server_message', function(data) {
addMessage(`服务器消息: ${data.message}`, 'info');
});
}
function disconnectSocket() {
if (socket) {
socket.disconnect();
socket = null;
}
}
function sendTestMessage() {
if (!socket || !socket.connected) {
addMessage('请先连接WebSocket', 'error');
return;
}
const testData = {
message: 'Hello from client',
timestamp: new Date().toISOString()
};
socket.emit('test_message', testData);
addMessage(`发送消息: ${JSON.stringify(testData)}`, 'info');
}
// 页面加载时自动连接
window.onload = function() {
connectSocket();
};
</script>
</body>
</html>
"""
# 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()

110
install/quick_start.bat Normal file
View File

@ -0,0 +1,110 @@
@echo off
echo ========================================
echo Minimal Test Framework - Quick Start
echo ========================================
echo.
:menu
echo Please select an option:
echo 1. Install dependencies
echo 2. Test run (development mode)
echo 3. Build exe
echo 4. Run built exe
echo 5. Show help
echo 0. Exit
echo.
set /p choice=Enter option (0-5):
if "%choice%"=="1" goto install_deps
if "%choice%"=="2" goto test_run
if "%choice%"=="3" goto build_exe
if "%choice%"=="4" goto run_exe
if "%choice%"=="5" goto show_help
if "%choice%"=="0" goto exit
echo Invalid option, please try again
echo.
goto menu
:install_deps
echo.
echo Installing dependencies...
pip install -r requirements_minimal.txt
if %errorlevel% equ 0 (
echo Dependencies installed successfully
) else (
echo Failed to install dependencies
)
echo.
pause
goto menu
:test_run
echo.
echo Starting test server...
echo After server starts, visit http://localhost:5000 in your browser
echo Press Ctrl+C to stop the server
echo.
python minimal_test_app.py
echo.
pause
goto menu
:build_exe
echo.
echo Starting build...
python build_minimal.py
echo.
pause
goto menu
:run_exe
echo.
if exist "dist\MinimalTestApp.exe" (
echo Starting built application...
echo After app starts, visit http://localhost:5000 in your browser
echo.
cd dist
MinimalTestApp.exe
cd ..
) else (
echo dist\MinimalTestApp.exe not found
echo Please run option 3 to build first
)
echo.
pause
goto menu
:show_help
echo.
echo ========================================
echo Usage Instructions
echo ========================================
echo.
echo 1. First time: select "1. Install dependencies"
echo 2. After installation: select "2. Test run" to verify
echo 3. After testing: select "3. Build exe"
echo 4. After building: select "4. Run built exe" to verify
echo.
echo Test Steps:
echo - HTTP API test: Click "Test HTTP API" button on page
echo - WebSocket test: Page auto-connects, can send test messages
echo - System info: Check if SocketIO mode shows "threading"
echo.
echo Files:
echo - minimal_test_app.py: Main application
echo - build_minimal.py: Build script
echo - requirements_minimal.txt: Dependencies
echo - README.md: Detailed documentation
echo.
echo Troubleshooting:
echo - If "Invalid async_mode specified" error occurs
echo check if dependencies are fully installed
echo - If WebSocket connection fails, check firewall settings
echo - See README.md for detailed instructions
echo.
pause
goto menu
:exit
echo Thank you for using!
exit /b 0

11
install/requirements_minimal.txt Normal file
View File

@ -0,0 +1,11 @@
# 最小功能测试框架依赖
# 只包含Flask + SocketIO + threading的核心依赖
Flask==2.3.3
Flask-SocketIO==5.3.6
Werkzeug==2.3.7
python-socketio==5.8.0
python-engineio==4.7.1
# 打包工具
PyInstaller>=6.10.0