119 lines
4.1 KiB
Python
119 lines
4.1 KiB
Python
from flask import Blueprint, send_file, jsonify, request
|
||
import os
|
||
import zipfile
|
||
import tempfile
|
||
import shutil
|
||
from datetime import datetime
|
||
import hashlib
|
||
|
||
# 创建Blueprint
|
||
common_bp = Blueprint('common', __name__)
|
||
|
||
# 存储最近生成的压缩包信息
|
||
latest_zip_info = {
|
||
"filename": None,
|
||
"md5": None,
|
||
"timestamp": None
|
||
}
|
||
|
||
def calculate_file_md5(file_path):
|
||
"""计算文件的MD5值"""
|
||
md5_hash = hashlib.md5()
|
||
with open(file_path, "rb") as f:
|
||
# 分块读取文件以处理大文件
|
||
for chunk in iter(lambda: f.read(4096), b""):
|
||
md5_hash.update(chunk)
|
||
return md5_hash.hexdigest()
|
||
|
||
@common_bp.route("/download_static_resources", methods=["GET"])
|
||
def download_static_resources():
|
||
try:
|
||
# 创建临时目录用于存放zip文件
|
||
with tempfile.TemporaryDirectory() as temp_dir:
|
||
zip_filename = f"static_resources_{datetime.now().strftime('%Y%m%d_%H%M%S')}.zip"
|
||
zip_path = os.path.join(temp_dir, zip_filename)
|
||
|
||
# 获取UI_next目录的绝对路径
|
||
ui_next_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
static_dir = os.path.join(ui_next_dir, 'static')
|
||
|
||
# 创建zip文件
|
||
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
|
||
# 遍历static目录
|
||
for root, dirs, files in os.walk(static_dir):
|
||
if 'tmp_images' in dirs and root.endswith(os.path.join('static', 'images')):
|
||
dirs.remove('tmp_images') # 从遍历列表中移除，避免进入该目录
|
||
|
||
for file in files:
|
||
file_path = os.path.join(root, file)
|
||
# 计算相对路径，确保解压后的目录结构正确
|
||
arcname = os.path.relpath(file_path, static_dir)
|
||
zipf.write(file_path, arcname)
|
||
|
||
# 计算MD5值
|
||
md5_value = calculate_file_md5(zip_path)
|
||
|
||
# 更新最新压缩包信息
|
||
global latest_zip_info
|
||
latest_zip_info = {
|
||
"filename": zip_filename,
|
||
"md5": md5_value,
|
||
"timestamp": datetime.now().isoformat()
|
||
}
|
||
|
||
# 发送zip文件
|
||
return send_file(
|
||
zip_path,
|
||
as_attachment=True,
|
||
download_name=zip_filename,
|
||
mimetype='application/zip'
|
||
)
|
||
|
||
except Exception as e:
|
||
return jsonify({
|
||
"status": "error",
|
||
"message": f"打包静态资源失败: {str(e)}"
|
||
}), 500
|
||
|
||
@common_bp.route("/verify_static_resources", methods=["POST"])
|
||
def verify_static_resources():
|
||
try:
|
||
data = request.get_json()
|
||
if not data or "md5" not in data or "filename" not in data:
|
||
return jsonify({
|
||
"status": "error",
|
||
"message": "请提供文件名和MD5值"
|
||
}), 400
|
||
|
||
client_md5 = data["md5"]
|
||
client_filename = data["filename"]
|
||
|
||
# 验证文件名和MD5值是否匹配最近生成的压缩包
|
||
if latest_zip_info["filename"] != client_filename:
|
||
return jsonify({
|
||
"status": "error",
|
||
"message": "文件名不匹配，请重新下载"
|
||
}), 400
|
||
|
||
if latest_zip_info["md5"] != client_md5:
|
||
return jsonify({
|
||
"status": "error",
|
||
"message": "MD5值不匹配，文件可能已损坏，请重新下载",
|
||
"expected_md5": latest_zip_info["md5"]
|
||
}), 400
|
||
|
||
return jsonify({
|
||
"status": "success",
|
||
"message": "文件验证成功",
|
||
"verified": True
|
||
})
|
||
|
||
except Exception as e:
|
||
return jsonify({
|
||
"status": "error",
|
||
"message": f"验证文件失败: {str(e)}"
|
||
}), 500
|
||
|
||
# # 初始化函数
|
||
# def init_common_routes(app):
|
||
# app.register_blueprint(common_bp, url_prefix='/api/common') |