1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
|
import os
import uuid
from flask import Flask, request, redirect, url_for, send_file, render_template, session, send_from_directory, abort, Response
# 创建Flask应用实例，__name__用于确定应用根目录以便查找资源
app = Flask(__name__)
# 设置应用的密钥，用于加密会话（session）。优先从环境变量FLASK_SECRET_KEY获取，若未设置则使用默认值'test_key'（生产环境应避免使用默认值）
app.secret_key = os.environ.get("FLASK_SECRET_KEY", "test_key")
# 定义文件上传的目录路径，这里设置为当前工作目录下的"uploads"文件夹
UPLOAD_FOLDER = os.path.join(os.getcwd(), "uploads")
# 确保上传目录存在，exist_ok=True表示如果目录已存在也不会引发错误
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
# 用一个简单的字典模拟用户数据库，存储用户名和用户信息（密码和角色）
users = {}
@app.route("/")
def index():
"""
根路径路由。
检查用户是否已登录（即session中是否有username）。如果未登录，重定向到登录页面；如果已登录，重定向到上传页面。
"""
if "username" not in session:
return redirect(url_for("login"))
return redirect(url_for("upload"))
@app.route("/register", methods=["GET", "POST"])
def register():
if request.method == "POST":
username = request.form["username"]
password = request.form["password"]
if username in users:
return "用户名已存在"
users[username] = {"password": password, "role": "user"}
return redirect(url_for("login"))
return render_template("register.html")
@app.route("/login", methods=["GET", "POST"])
def login():
if request.method == "POST":
username = request.form["username"]
password = request.form["password"]
if username in users and users[username]["password"] == password:
session["username"] = username
session["role"] = users[username]["role"]
return redirect(url_for("upload"))
else:
return "用户名或密码错误"
return render_template("login.html")
@app.route("/logout")
def logout():
"""
用户登出路由。
清除session中的所有数据，重定向到登录页面。
"""
session.clear()
return redirect(url_for("login"))
@app.route("/upload", methods=["GET", "POST"])
def upload():
"""
* 管理员（admin）可以选择使用表单提供的dirname或自动生成一个UUID作为目录名。
* 普通用户（user）使用自动生成的UUID作为目录名。
- 在上传目录下创建目标子目录。
- 将上传的zip文件保存到目标目录中，命名为'upload.zip'。
- 尝试使用系统命令解压该zip文件到目标目录。
- 删除原始的zip文件。
- 返回解压成功信息和一个指向下载页面的链接。
"""
if "username" not in session:
return redirect(url_for("login"))
if request.method == "POST":
file = request.files["file"]
if not file:
return "未选择文件"
role = session["role"]
if role == "admin":
dirname = request.form.get("dirname") or str(uuid.uuid4())
else:
dirname = str(uuid.uuid4())
target_dir = os.path.join(UPLOAD_FOLDER, dirname)
os.makedirs(target_dir, exist_ok=True)
zip_path = os.path.join(target_dir, "upload.zip")
file.save(zip_path)
try:
os.system(f"unzip -o {zip_path} -d {target_dir}")
except:
return "解压失败，请检查文件格式"
os.remove(zip_path)
return f"解压完成！<br>下载地址: <a href='{url_for('download', folder=dirname)}'>{request.host_url}download/{dirname}</a>"
return render_template("upload.html")
@app.route("/download/<folder>")
def download(folder):
"""
文件下载列表页面路由。
根据URL中的folder参数，定位到上传目录下的具体子目录。
检查该子目录是否存在，不存在则返回404错误。
获取该子目录下的文件列表，并渲染下载页面模板，将文件夹名和文件列表传递给模板。
"""
target_dir = os.path.join(UPLOAD_FOLDER, folder)
if not os.path.exists(target_dir):
abort(404)
files = os.listdir(target_dir)
return render_template("download.html", folder=folder, files=files)
@app.route("/download/<folder>/<filename>")
def download_file(folder, filename):
"""
单个文件下载路由。
根据URL中的folder和filename参数，构建文件的完整路径。
尝试打开文件并读取内容，然后返回一个Response对象，设置适当的MIME类型和Content-Disposition头部以触发浏览器下载。
如果文件不存在，返回404错误；其他错误返回500错误。
"""
file_path = os.path.join(UPLOAD_FOLDER, folder, filename)
try:
with open(file_path, 'r') as file: # 注意：对于二进制文件（如图片），应使用'rb'模式
content = file.read()
return Response(
content,
mimetype="application/octet-stream",
headers={
"Content-Disposition": f"attachment; filename={filename}"
}
)
except FileNotFoundError:
return "File not found", 404
except Exception as e:
return f"Error: {str(e)}", 500
if __name__ == "__main__":
# 启动Flask开发服务器，监听所有公共IP（0.0.0.0）
app.run(host="0.0.0.0")
|