242 lines
7.2 KiB
Python
242 lines
7.2 KiB
Python
#!/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) |