301 lines
12 KiB
Python
301 lines
12 KiB
Python
"""
|
||
文件上传网站 - 后端服务
|
||
基于 Flask，提供文件上传、图形验证码、Rate Limiting 等功能
|
||
"""
|
||
import os
|
||
import re
|
||
import time
|
||
import random
|
||
import string
|
||
from io import BytesIO
|
||
from datetime import datetime
|
||
|
||
from flask import Flask, request, jsonify, render_template, session, send_file
|
||
from flask_limiter import Limiter
|
||
from flask_limiter.util import get_remote_address
|
||
from PIL import Image, ImageDraw, ImageFont
|
||
import magic
|
||
|
||
from config import Config
|
||
|
||
# ── 初始化 Flask 应用 ──────────────────────────────────────────────
|
||
app = Flask(__name__)
|
||
app.config.from_object(Config)
|
||
app.secret_key = Config.SECRET_KEY
|
||
|
||
# ── Rate Limiter ───────────────────────────────────────────────────
|
||
limiter = Limiter(
|
||
app=app,
|
||
key_func=get_remote_address,
|
||
default_limits=[],
|
||
storage_uri="memory://"
|
||
)
|
||
|
||
# ── 常量 ───────────────────────────────────────────────────────────
|
||
# 姓名校验：中文、英文字母、数字、下划线，2-20 字符
|
||
NAME_REGEX = re.compile(r'^[\u4e00-\u9fa5a-zA-Z0-9_]{2,20}$')
|
||
|
||
# 文件名中不安全的字符
|
||
UNSAFE_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
|
||
|
||
|
||
# ── 工具函数 ────────────────────────────────────────────────────────
|
||
def sanitize_name(name: str) -> str | None:
|
||
"""校验并清洗姓名，防止路径穿越攻击"""
|
||
name = name.strip()
|
||
# 移除路径穿越字符
|
||
name = name.replace('..', '').replace('/', '').replace('\\', '')
|
||
# 校验格式
|
||
if not NAME_REGEX.match(name):
|
||
return None
|
||
return name
|
||
|
||
|
||
def sanitize_filename(filename: str) -> str:
|
||
"""清洗文件名，移除不安全字符"""
|
||
# 只保留文件名部分（去掉路径）
|
||
filename = os.path.basename(filename)
|
||
# 移除不安全字符
|
||
filename = UNSAFE_FILENAME_CHARS.sub('', filename)
|
||
# 去掉首尾空格和点
|
||
filename = filename.strip(' .')
|
||
# 限制长度（保留扩展名）
|
||
if len(filename) > 200:
|
||
name_part, ext = os.path.splitext(filename)
|
||
filename = name_part[:200 - len(ext)] + ext
|
||
# 如果文件名为空，使用默认名
|
||
return filename if filename else 'unnamed.txt'
|
||
|
||
|
||
def verify_file_type(file_data: bytes, filename: str) -> tuple[bool, str | None]:
|
||
"""通过 Magic Bytes 校验文件真实类型"""
|
||
try:
|
||
mime = magic.from_buffer(file_data, mime=True)
|
||
except Exception:
|
||
return False, '无法识别文件类型'
|
||
|
||
if mime not in Config.ALLOWED_MIME_TYPES:
|
||
return False, f'不支持的文件类型 ({mime})，请上传文字类文档'
|
||
|
||
return True, None
|
||
|
||
|
||
def generate_captcha_text(length: int = 4) -> str:
|
||
"""生成随机验证码文本（去掉易混淆字符）"""
|
||
chars = string.ascii_uppercase + string.digits
|
||
for c in 'O0I1L':
|
||
chars = chars.replace(c, '')
|
||
return ''.join(random.choices(chars, k=length))
|
||
|
||
|
||
def generate_captcha_image(text: str) -> Image.Image:
|
||
"""生成验证码图片（含干扰线和噪点）"""
|
||
width, height = 220, 70
|
||
# 浅色随机背景
|
||
bg_color = (random.randint(235, 255), random.randint(235, 255), random.randint(235, 255))
|
||
image = Image.new('RGB', (width, height), bg_color)
|
||
draw = ImageDraw.Draw(image)
|
||
|
||
# 干扰线
|
||
for _ in range(6):
|
||
x1, y1 = random.randint(0, width), random.randint(0, height)
|
||
x2, y2 = random.randint(0, width), random.randint(0, height)
|
||
line_color = (random.randint(160, 210), random.randint(160, 210), random.randint(160, 210))
|
||
draw.line([(x1, y1), (x2, y2)], fill=line_color, width=1)
|
||
|
||
# 噪点
|
||
for _ in range(120):
|
||
x, y = random.randint(0, width - 1), random.randint(0, height - 1)
|
||
dot_color = (random.randint(140, 200), random.randint(140, 200), random.randint(140, 200))
|
||
draw.point((x, y), fill=dot_color)
|
||
|
||
# 绘制文字
|
||
try:
|
||
font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 50)
|
||
except OSError:
|
||
font = ImageFont.load_default()
|
||
|
||
for i, char in enumerate(text):
|
||
x = 16 + i * 50
|
||
y = random.randint(4, 12)
|
||
color = (random.randint(0, 80), random.randint(0, 80), random.randint(0, 80))
|
||
draw.text((x, y), char, font=font, fill=color)
|
||
|
||
return image
|
||
|
||
|
||
def update_upload_log(user_dir: str, name: str, results: list, timestamp: str):
|
||
"""更新用户文件夹中的 Markdown 上传日志"""
|
||
log_file = os.path.join(user_dir, '00_上传日志.md')
|
||
|
||
# 统计累计文件数（排除日志文件）
|
||
total_files = len([f for f in os.listdir(user_dir) if not f.startswith('00_')])
|
||
|
||
# 构建日志条目
|
||
now_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
lines = [
|
||
f"\n### {now_str}",
|
||
f"- 上传者：{name}",
|
||
f"- 本次上传文件数：{len(results)}",
|
||
"| 原文件名 | 存储文件名 |",
|
||
"|---------|----------|",
|
||
]
|
||
for r in results:
|
||
lines.append(f"| {r['filename']} | {r['storage_filename']} |")
|
||
lines.append(f"- 累计总文件数：{total_files}")
|
||
entry = '\n'.join(lines) + '\n'
|
||
|
||
# 写入或追加
|
||
if os.path.exists(log_file):
|
||
with open(log_file, 'a', encoding='utf-8') as f:
|
||
f.write(entry)
|
||
else:
|
||
with open(log_file, 'w', encoding='utf-8') as f:
|
||
f.write("## 上传记录\n")
|
||
f.write(entry)
|
||
|
||
|
||
# ── 路由 ────────────────────────────────────────────────────────────
|
||
@app.route('/')
|
||
def index():
|
||
"""渲染上传页面"""
|
||
return render_template('index.html')
|
||
|
||
|
||
@app.route('/api/captcha')
|
||
def get_captcha():
|
||
"""生成并返回新的验证码图片"""
|
||
text = generate_captcha_text()
|
||
session['captcha'] = text.lower()
|
||
session['captcha_time'] = time.time()
|
||
|
||
image = generate_captcha_image(text)
|
||
buf = BytesIO()
|
||
image.save(buf, format='PNG')
|
||
buf.seek(0)
|
||
|
||
response = send_file(buf, mimetype='image/png')
|
||
response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0'
|
||
return response
|
||
|
||
|
||
@app.route('/api/upload', methods=['POST'])
|
||
@limiter.limit(f"{Config.RATE_LIMIT_PER_HOUR}/hour")
|
||
def upload_file():
|
||
"""处理文件上传"""
|
||
|
||
# ── 1. 校验验证码 ──────────────────────────────────────────
|
||
captcha_input = request.form.get('captcha', '').strip().lower()
|
||
captcha_expected = session.get('captcha')
|
||
captcha_time = session.get('captcha_time', 0)
|
||
|
||
if not captcha_expected:
|
||
return jsonify({'error': '请先获取验证码'}), 400
|
||
if time.time() - captcha_time > Config.CAPTCHA_EXPIRY:
|
||
session.pop('captcha', None)
|
||
session.pop('captcha_time', None)
|
||
return jsonify({'error': '验证码已过期，请刷新'}), 400
|
||
if captcha_input != captcha_expected:
|
||
return jsonify({'error': '验证码错误'}), 400
|
||
|
||
# 验证码使用后立即清除（一次性）
|
||
session.pop('captcha', None)
|
||
session.pop('captcha_time', None)
|
||
|
||
# ── 2. 校验姓名 ────────────────────────────────────────────
|
||
raw_name = request.form.get('name', '')
|
||
name = sanitize_name(raw_name)
|
||
if not name:
|
||
return jsonify({'error': '姓名格式不正确（2-20个字符，仅支持中文、英文字母、数字和下划线）'}), 400
|
||
|
||
# ── 3. 校验文件列表 ────────────────────────────────────────
|
||
files = request.files.getlist('files')
|
||
if not files or all(f.filename == '' for f in files):
|
||
return jsonify({'error': '请选择要上传的文件'}), 400
|
||
valid_files = [f for f in files if f.filename]
|
||
if len(valid_files) > Config.MAX_FILES_PER_UPLOAD:
|
||
return jsonify({'error': f'单次最多上传 {Config.MAX_FILES_PER_UPLOAD} 个文件'}), 400
|
||
|
||
# ── 4. 创建用户目录 ────────────────────────────────────────
|
||
user_dir = os.path.join(Config.UPLOAD_ROOT, name)
|
||
os.makedirs(user_dir, exist_ok=True)
|
||
|
||
# ── 5. 逐个处理文件 ────────────────────────────────────────
|
||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||
results = []
|
||
|
||
for file in valid_files:
|
||
original_filename = file.filename
|
||
file_data = file.read()
|
||
|
||
# 5a. 文件大小校验
|
||
if len(file_data) > Config.MAX_FILE_SIZE:
|
||
results.append({
|
||
'filename': original_filename,
|
||
'success': False,
|
||
'error': f'文件超过 {Config.MAX_FILE_SIZE // (1024 * 1024)}MB 限制'
|
||
})
|
||
continue
|
||
|
||
# 5b. 文件类型校验（Magic Bytes）
|
||
valid, error_msg = verify_file_type(file_data, original_filename)
|
||
if not valid:
|
||
results.append({
|
||
'filename': original_filename,
|
||
'success': False,
|
||
'error': error_msg
|
||
})
|
||
continue
|
||
|
||
# 5c. 清洗文件名 + 生成存储文件名
|
||
safe_name = sanitize_filename(original_filename)
|
||
storage_filename = f"{timestamp}_{safe_name}"
|
||
filepath = os.path.join(user_dir, storage_filename)
|
||
|
||
# 处理同名文件（同一秒内上传的）
|
||
counter = 1
|
||
while os.path.exists(filepath):
|
||
name_part, ext = os.path.splitext(safe_name)
|
||
storage_filename = f"{timestamp}_{name_part}_{counter}{ext}"
|
||
filepath = os.path.join(user_dir, storage_filename)
|
||
counter += 1
|
||
|
||
# 5d. 写入文件
|
||
with open(filepath, 'wb') as f:
|
||
f.write(file_data)
|
||
|
||
results.append({
|
||
'filename': original_filename,
|
||
'storage_filename': storage_filename,
|
||
'success': True
|
||
})
|
||
|
||
# ── 6. 更新上传日志 ────────────────────────────────────────
|
||
success_results = [r for r in results if r['success']]
|
||
if success_results:
|
||
update_upload_log(user_dir, name, success_results, timestamp)
|
||
|
||
# ── 7. 返回结果 ────────────────────────────────────────────
|
||
total_files = len([f for f in os.listdir(user_dir) if not f.startswith('00_')])
|
||
|
||
return jsonify({
|
||
'success': len(success_results) > 0,
|
||
'results': results,
|
||
'success_count': len(success_results),
|
||
'fail_count': len(results) - len(success_results),
|
||
'total_files': total_files,
|
||
'storage_path': f"文字稿/{name}/"
|
||
})
|
||
|
||
|
||
# ── 自定义 Rate Limit 错误处理 ──────────────────────────────────────
|
||
@app.errorhandler(429)
|
||
def ratelimit_handler(e):
|
||
return jsonify({'error': '上传过于频繁，请稍后再试（每小时最多 20 次上传）'}), 429
|
||
|
||
|
||
# ── 启动 ────────────────────────────────────────────────────────────
|
||
if __name__ == '__main__':
|
||
os.makedirs(Config.UPLOAD_ROOT, exist_ok=True)
|
||
app.run(host='0.0.0.0', port=Config.PORT, debug=Config.DEBUG)
|