288 lines
8.7 KiB
Python
288 lines
8.7 KiB
Python
import os
|
|
import re
|
|
import uuid
|
|
import shutil
|
|
from datetime import datetime
|
|
from flask import Flask, request, jsonify, render_template
|
|
from config import (
|
|
NAS_DATA_PATH, PORT, ALLOWED_EXTENSIONS,
|
|
MAX_FILES_COUNT, MAX_FILE_SIZE_MB, ALLOWED_MIME_TYPES
|
|
)
|
|
|
|
app = Flask(__name__)
|
|
|
|
|
|
# ========== 工具函数 ==========
|
|
|
|
def get_camp_folders():
|
|
"""获取所有训练营文件夹列表，按期数排序"""
|
|
if not os.path.exists(NAS_DATA_PATH):
|
|
os.makedirs(NAS_DATA_PATH, exist_ok=True)
|
|
return []
|
|
|
|
folders = []
|
|
for name in os.listdir(NAS_DATA_PATH):
|
|
full_path = os.path.join(NAS_DATA_PATH, name)
|
|
if os.path.isdir(full_path) and re.match(r'^训练营\d+期$', name):
|
|
folders.append(name)
|
|
|
|
# 按期数数字排序
|
|
folders.sort(key=lambda x: int(re.search(r'\d+', x).group()) if re.search(r'\d+', x) else 0)
|
|
return folders
|
|
|
|
|
|
def get_class_folders(camp_name):
|
|
"""获取某个训练营下的所有班级文件夹，按班号排序"""
|
|
camp_path = os.path.join(NAS_DATA_PATH, camp_name)
|
|
if not os.path.exists(camp_path):
|
|
return []
|
|
|
|
classes = []
|
|
for name in os.listdir(camp_path):
|
|
full_path = os.path.join(camp_path, name)
|
|
if os.path.isdir(full_path) and re.match(r'^\d+班$', name):
|
|
classes.append(name)
|
|
|
|
# 按班号数字排序
|
|
classes.sort(key=lambda x: int(re.search(r'\d+', x).group()) if re.search(r'\d+', x) else 0)
|
|
return classes
|
|
|
|
|
|
def validate_file(file):
|
|
"""验证上传文件是否合法，返回 (是否合法, 错误原因)"""
|
|
if not file or not file.filename:
|
|
return False, "空文件"
|
|
|
|
# 检查文件大小
|
|
file.seek(0, os.SEEK_END)
|
|
size = file.tell()
|
|
file.seek(0)
|
|
if size > MAX_FILE_SIZE_MB * 1024 * 1024:
|
|
return False, f"文件超过 {MAX_FILE_SIZE_MB}MB 限制"
|
|
|
|
if size == 0:
|
|
return False, "空文件"
|
|
|
|
# 检查扩展名
|
|
ext = file.filename.rsplit('.', 1)[-1].lower() if '.' in file.filename else ''
|
|
if ext not in ALLOWED_EXTENSIONS:
|
|
return False, f"不支持的格式: {ext}"
|
|
|
|
# 检查MIME类型
|
|
mime = file.mimetype
|
|
expected_mime = ALLOWED_MIME_TYPES.get(ext)
|
|
if expected_mime and mime and not mime.startswith(expected_mime.split('/')[0]):
|
|
return False, f"MIME类型不匹配: {mime}"
|
|
|
|
return True, None
|
|
|
|
|
|
def generate_filename(original_name, camp_name, class_name):
|
|
"""生成新文件名：训练营{n}期-{x}班-{时间戳}-{随机数}.{扩展名}"""
|
|
camp_match = re.search(r'(\d+)', camp_name)
|
|
class_match = re.search(r'(\d+)', class_name)
|
|
camp_num = camp_match.group(1) if camp_match else '?'
|
|
class_num = class_match.group(1) if class_match else '?'
|
|
|
|
now = datetime.now()
|
|
timestamp = now.strftime('%Y%m%d%H%M%S')
|
|
random_str = uuid.uuid4().hex[:6]
|
|
ext = original_name.rsplit('.', 1)[-1].lower() if '.' in original_name else 'jpg'
|
|
|
|
return f"训练营{camp_num}期-{class_num}班-{timestamp}-{random_str}.{ext}"
|
|
|
|
|
|
def get_unique_filepath(dir_path, filename):
|
|
"""获取不冲突的文件路径，若存在则追加序号"""
|
|
filepath = os.path.join(dir_path, filename)
|
|
if not os.path.exists(filepath):
|
|
return filepath
|
|
|
|
# 分离文件名和扩展名
|
|
name, ext = os.path.splitext(filename)
|
|
counter = 1
|
|
while True:
|
|
new_name = f"{name}-{counter}{ext}"
|
|
new_path = os.path.join(dir_path, new_name)
|
|
if not os.path.exists(new_path):
|
|
return new_path
|
|
counter += 1
|
|
|
|
|
|
def ensure_folders(camp_name, class_name):
|
|
"""确保训练营和班级文件夹存在，不存在则创建"""
|
|
camp_path = os.path.join(NAS_DATA_PATH, camp_name)
|
|
class_path = os.path.join(camp_path, class_name)
|
|
stat_file = os.path.join(camp_path, '统计.md')
|
|
|
|
created_camp = False
|
|
if not os.path.exists(camp_path):
|
|
os.makedirs(camp_path, exist_ok=True)
|
|
created_camp = True
|
|
|
|
if not os.path.exists(class_path):
|
|
os.makedirs(class_path, exist_ok=True)
|
|
|
|
# 如果训练营是新创建的，初始化统计文件
|
|
if created_camp and not os.path.exists(stat_file):
|
|
update_statistics(camp_name)
|
|
|
|
return camp_path, class_path
|
|
|
|
|
|
def update_statistics(camp_name):
|
|
"""更新某个训练营的统计.md文件"""
|
|
camp_path = os.path.join(NAS_DATA_PATH, camp_name)
|
|
stat_file = os.path.join(camp_path, '统计.md')
|
|
|
|
if not os.path.exists(camp_path):
|
|
return
|
|
|
|
# 扫描所有班级文件夹
|
|
class_stats = []
|
|
total = 0
|
|
|
|
for name in os.listdir(camp_path):
|
|
full_path = os.path.join(camp_path, name)
|
|
if not os.path.isdir(full_path):
|
|
continue
|
|
if not re.match(r'^\d+班$', name):
|
|
continue
|
|
|
|
# 统计该班级下的图片数量
|
|
count = 0
|
|
for fname in os.listdir(full_path):
|
|
fpath = os.path.join(full_path, fname)
|
|
if os.path.isfile(fpath):
|
|
ext = fname.rsplit('.', 1)[-1].lower() if '.' in fname else ''
|
|
if ext in ALLOWED_EXTENSIONS:
|
|
count += 1
|
|
|
|
class_stats.append((name, count))
|
|
total += count
|
|
|
|
# 按班号排序
|
|
class_stats.sort(key=lambda x: int(re.search(r'\d+', x[0]).group()) if re.search(r'\d+', x[0]) else 0)
|
|
|
|
# 写入统计文件
|
|
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
lines = [
|
|
f"# {camp_name} 图片统计",
|
|
"",
|
|
f"更新时间：{now}",
|
|
f"图片总数：{total}",
|
|
"",
|
|
"| 班级 | 图片数量 |",
|
|
"|------|----------|",
|
|
]
|
|
for class_name, count in class_stats:
|
|
lines.append(f"| {class_name} | {count} |")
|
|
|
|
with open(stat_file, 'w', encoding='utf-8') as f:
|
|
f.write('\n'.join(lines) + '\n')
|
|
|
|
|
|
# ========== 页面路由 ==========
|
|
|
|
@app.route('/')
|
|
def index():
|
|
return render_template('index.html')
|
|
|
|
|
|
# ========== API 路由 ==========
|
|
|
|
@app.route('/api/folders', methods=['GET'])
|
|
def api_folders():
|
|
"""获取所有训练营及其下的班级文件夹"""
|
|
folders = []
|
|
for camp_name in get_camp_folders():
|
|
classes = get_class_folders(camp_name)
|
|
folders.append({
|
|
'name': camp_name,
|
|
'classes': classes,
|
|
})
|
|
return jsonify({'folders': folders})
|
|
|
|
|
|
@app.route('/api/upload', methods=['POST'])
|
|
def api_upload():
|
|
"""上传图片"""
|
|
camp_name = request.form.get('folder', '').strip()
|
|
class_name = request.form.get('class', '').strip()
|
|
files = request.files.getlist('files')
|
|
|
|
# 参数校验
|
|
if not camp_name or not re.match(r'^训练营\d+期$', camp_name):
|
|
return jsonify({'success': False, 'error': '无效的训练营名称'}), 400
|
|
|
|
if not class_name or not re.match(r'^\d+班$', class_name):
|
|
return jsonify({'success': False, 'error': '无效的班级名称'}), 400
|
|
|
|
if not files:
|
|
return jsonify({'success': False, 'error': '未选择文件'}), 400
|
|
|
|
if len(files) > MAX_FILES_COUNT:
|
|
return jsonify({'success': False, 'error': f'单次最多上传 {MAX_FILES_COUNT} 张'}), 400
|
|
|
|
# 确保文件夹存在
|
|
camp_path, class_path = ensure_folders(camp_name, class_name)
|
|
|
|
# 逐个处理文件
|
|
uploaded = []
|
|
failed = []
|
|
has_new_upload = False
|
|
|
|
for file in files:
|
|
# 验证文件
|
|
valid, reason = validate_file(file)
|
|
if not valid:
|
|
failed.append({
|
|
'filename': file.filename if file.filename else '未知文件',
|
|
'status': 'failed',
|
|
'reason': reason,
|
|
})
|
|
continue
|
|
|
|
# 生成文件名并保存
|
|
new_filename = generate_filename(file.filename, camp_name, class_name)
|
|
filepath = get_unique_filepath(class_path, new_filename)
|
|
|
|
try:
|
|
file.save(filepath)
|
|
uploaded.append({
|
|
'filename': os.path.basename(filepath),
|
|
'status': 'success',
|
|
})
|
|
has_new_upload = True
|
|
except Exception as e:
|
|
failed.append({
|
|
'filename': file.filename,
|
|
'status': 'failed',
|
|
'reason': str(e),
|
|
})
|
|
|
|
# 如果有新文件上传成功，更新统计
|
|
if has_new_upload:
|
|
try:
|
|
update_statistics(camp_name)
|
|
except Exception:
|
|
pass # 统计更新失败不影响上传结果
|
|
|
|
return jsonify({
|
|
'success': len(uploaded) > 0,
|
|
'uploaded': uploaded,
|
|
'failed': failed,
|
|
})
|
|
|
|
|
|
# ========== 启动 ==========
|
|
|
|
if __name__ == '__main__':
|
|
# 确保存储目录存在
|
|
if not os.path.exists(NAS_DATA_PATH):
|
|
os.makedirs(NAS_DATA_PATH, exist_ok=True)
|
|
print(f"已创建存储目录: {NAS_DATA_PATH}")
|
|
|
|
print(f"服务启动在 http://0.0.0.0:{PORT}")
|
|
app.run(host='0.0.0.0', port=PORT, debug=False)
|