144 lines
5.4 KiB
Python
144 lines
5.4 KiB
Python
# 文件路径: inventory-backend/app/api/v1/common/upload.py
|
||
|
||
import os
|
||
import uuid
|
||
from flask import Blueprint, request, jsonify, send_from_directory
|
||
|
||
# 定义蓝图
|
||
# 注意：在 app/__init__.py 或类似入口文件中，注册此蓝图时 url_prefix 通常应为 '/api/v1/common'
|
||
upload_bp = Blueprint('upload', __name__)
|
||
|
||
|
||
# =========================================================
|
||
# 配置上传路径
|
||
# =========================================================
|
||
def get_project_root():
|
||
"""获取项目根目录 inventory-backend"""
|
||
current_path = os.path.abspath(__file__)
|
||
# 向上回退直到找到根目录，根据你的目录结构可能需要调整层级
|
||
# 假设结构: inventory-backend/app/api/v1/common/upload.py (回退5层)
|
||
base = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(current_path)))))
|
||
return base
|
||
|
||
|
||
BASE_DIR = get_project_root()
|
||
UPLOAD_FOLDER = os.path.join(BASE_DIR, 'uploads')
|
||
|
||
# 允许上传的文件后缀
|
||
ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'bmp', 'webp', 'pdf', 'doc', 'docx', 'xls', 'xlsx'}
|
||
|
||
# ★ 文件上传安全加固：限制最大文件大小 (10MB)
|
||
MAX_CONTENT_LENGTH = 10 * 1024 * 1024 # 10MB
|
||
|
||
|
||
def allowed_file(filename):
|
||
return '.' in filename and \
|
||
filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
|
||
|
||
|
||
def ensure_upload_folder_exists():
|
||
if not os.path.exists(UPLOAD_FOLDER):
|
||
try:
|
||
os.makedirs(UPLOAD_FOLDER)
|
||
print(f"✅ [Upload] 目录创建成功: {UPLOAD_FOLDER}")
|
||
except Exception as e:
|
||
print(f"❌ [Upload] 目录创建失败: {e}")
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# 1. 文件上传接口
|
||
# 完整 URL: /api/v1/common/upload (POST)
|
||
# ------------------------------------------------------------------
|
||
@upload_bp.route('/upload', methods=['POST'])
|
||
def upload_file():
|
||
ensure_upload_folder_exists()
|
||
|
||
if 'file' not in request.files:
|
||
return jsonify({"code": 400, "msg": "未找到文件部分"}), 400
|
||
|
||
file = request.files['file']
|
||
|
||
if file.filename == '':
|
||
return jsonify({"code": 400, "msg": "未选择文件"}), 400
|
||
|
||
# ★ 文件上传安全加固：检查文件大小
|
||
file.seek(0, os.SEEK_END)
|
||
file_size = file.tell()
|
||
file.seek(0) # 重置文件指针到开头
|
||
if file_size > MAX_CONTENT_LENGTH:
|
||
return jsonify({
|
||
"code": 400,
|
||
"msg": f"文件大小超过限制 ({MAX_CONTENT_LENGTH // (1024*1024)}MB)"
|
||
}), 400
|
||
|
||
if file and allowed_file(file.filename):
|
||
try:
|
||
# 获取后缀并生成唯一文件名
|
||
ext = file.filename.rsplit('.', 1)[1].lower()
|
||
new_filename = f"{uuid.uuid4().hex}.{ext}"
|
||
|
||
save_path = os.path.join(UPLOAD_FOLDER, new_filename)
|
||
file.save(save_path)
|
||
|
||
print(f"💾 [Upload] 文件已保存: {save_path}")
|
||
|
||
# 生成访问 URL (返回给前端的相对路径)
|
||
# 前端展示时通常拼接 baseURL，或者直接使用此路径访问
|
||
# 这里的 /api/v1/common 需与蓝图注册路径一致
|
||
file_url = f"/api/v1/common/files/{new_filename}"
|
||
|
||
return jsonify({
|
||
"code": 200,
|
||
"msg": "上传成功",
|
||
"data": {
|
||
"url": file_url,
|
||
"filename": new_filename
|
||
}
|
||
})
|
||
except Exception as e:
|
||
print(f"❌ [Upload] 保存异常: {e}")
|
||
return jsonify({"code": 500, "msg": "文件保存失败"}), 500
|
||
|
||
return jsonify({"code": 400, "msg": "不支持的文件格式"}), 400
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# 2. 静态文件访问接口 (回显)
|
||
# 完整 URL: /api/v1/common/files/<filename> (GET)
|
||
# ------------------------------------------------------------------
|
||
@upload_bp.route('/files/<filename>', methods=['GET'])
|
||
def uploaded_file(filename):
|
||
full_path = os.path.join(UPLOAD_FOLDER, filename)
|
||
if not os.path.exists(full_path):
|
||
# 尝试调试路径问题
|
||
print(f"❌ [File Access] 文件未找到: {full_path}")
|
||
return jsonify({"code": 404, "msg": "文件不存在"}), 404
|
||
|
||
return send_from_directory(UPLOAD_FOLDER, filename)
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# 3. 文件删除接口 (同步删除物理文件)
|
||
# 完整 URL: /api/v1/common/files/<filename> (DELETE)
|
||
# ------------------------------------------------------------------
|
||
@upload_bp.route('/files/<filename>', methods=['DELETE'])
|
||
def delete_file(filename):
|
||
try:
|
||
# 安全处理文件名 (防止路径遍历)
|
||
safe_filename = os.path.basename(filename)
|
||
file_path = os.path.join(UPLOAD_FOLDER, safe_filename)
|
||
|
||
print(f"🗑️ [Delete] 尝试删除文件: {file_path}")
|
||
|
||
if os.path.exists(file_path):
|
||
os.remove(file_path)
|
||
print(f"✅ [Delete] 文件删除成功")
|
||
return jsonify({"code": 200, "msg": "文件已删除"})
|
||
else:
|
||
print(f"⚠️ [Delete] 文件不存在，无需删除")
|
||
# 即使文件不存在也返回成功，保证前端逻辑闭环
|
||
return jsonify({"code": 200, "msg": "文件不存在或已删除"})
|
||
|
||
except Exception as e:
|
||
print(f"❌ [Delete] 删除异常: {e}")
|
||
return jsonify({"code": 500, "msg": f"删除失败: {str(e)}"}), 500 |