56 lines
2.0 KiB
Python
56 lines
2.0 KiB
Python
|
||
from flask import Flask, request, jsonify
|
||
import os
|
||
from flask import send_from_directory
|
||
|
||
app = Flask(__name__)
|
||
UPLOAD_FOLDER = 'uploaded_firmware'
|
||
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
|
||
|
||
# 临时保存每个上传会话的分包
|
||
chunk_cache = {}
|
||
|
||
@app.route('/api/upload_firmware', methods=['POST'])
|
||
def upload_firmware():
|
||
chunk = request.files.get('chunk')
|
||
index = int(request.form.get('index', -1))
|
||
total = int(request.form.get('total', -1))
|
||
session_id = request.remote_addr # 简单用IP区分会话，实际可用token等
|
||
if chunk is None or index < 0 or total < 1:
|
||
return '参数错误', 400
|
||
|
||
# 缓存分包
|
||
if session_id not in chunk_cache:
|
||
chunk_cache[session_id] = [None] * total
|
||
chunk_cache[session_id][index] = chunk.read()
|
||
|
||
# 如果全部收到，合并保存
|
||
if all(part is not None for part in chunk_cache[session_id]):
|
||
firmware_data = b''.join(chunk_cache[session_id])
|
||
save_path = os.path.join(UPLOAD_FOLDER, f'firmware_{session_id}.bin')
|
||
with open(save_path, 'wb') as f:
|
||
f.write(firmware_data)
|
||
del chunk_cache[session_id]
|
||
return jsonify({'status': 'ok', 'msg': '全部分包已接收并保存'})
|
||
return jsonify({'status': 'ok', 'msg': f'分包{index+1}/{total}已接收'})
|
||
|
||
# 版本信息接口
|
||
@app.route('/api/version_info')
|
||
def version_info():
|
||
# 实际可从设备、数据库或文件读取，这里演示写死
|
||
verstr = 'Bootloader版本：V1.0\n固件版本：V1.3\n编译日期：2025.09.06\n固件日志：修复若干bug，提升稳定性。'
|
||
return verstr, 200, {'Content-Type': 'text/plain; charset=utf-8'}
|
||
|
||
# 静态页面路由
|
||
@app.route('/')
|
||
def index():
|
||
return send_from_directory('.', 'firmware_upgrade.html')
|
||
|
||
@app.route('/<path:filename>')
|
||
def static_files(filename):
|
||
# 允许访问同目录下的静态资源（如js/css等）
|
||
return send_from_directory('.', filename)
|
||
|
||
if __name__ == '__main__':
|
||
app.run(host='0.0.0.0', port=5000, debug=True)
|