|
|
@@ -1,5 +1,11 @@
|
|
|
-from flask import Blueprint, render_template, send_from_directory, current_app, redirect, url_for, request, jsonify
|
|
|
-from api.utils import login_required, load_led_config
|
|
|
+import os
|
|
|
+import time
|
|
|
+import uuid
|
|
|
+
|
|
|
+from flask import Blueprint, render_template, send_from_directory, current_app, redirect, url_for, request, jsonify, session
|
|
|
+from werkzeug.utils import secure_filename
|
|
|
+
|
|
|
+from api.utils import login_required, load_led_config, get_server_ip
|
|
|
from application.scheduler_service import scheduler_service
|
|
|
from application.self_check_service import run_all_checks
|
|
|
from application.uap_message_service import send_self_check_notification
|
|
|
@@ -7,6 +13,54 @@ from utils.logger_config import logger
|
|
|
|
|
|
main_bp = Blueprint('main', __name__)
|
|
|
|
|
|
+ATTACHMENT_TEST_SESSION_KEY = 'attachment_test_filename'
|
|
|
+# 附件测试页允许的类型(单附件,新上传会替换)
|
|
|
+ATTACHMENT_TEST_EXTENSIONS = {
|
|
|
+ 'png', 'jpg', 'jpeg', 'gif', 'bmp', 'webp', 'heic', 'heif',
|
|
|
+ 'mp4', 'mov', 'webm', 'mkv', 'avi',
|
|
|
+ 'pdf', 'doc', 'docx', 'xls', 'xlsx', 'ppt', 'pptx',
|
|
|
+ 'txt', 'csv', 'zip', 'rar', '7z',
|
|
|
+ 'mp3', 'wav', 'm4a',
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+def _attachment_test_public_url(filename):
|
|
|
+ if not filename:
|
|
|
+ return None
|
|
|
+ host = request.host
|
|
|
+ if 'localhost' in host or '127.0.0.1' in host:
|
|
|
+ server_ip = get_server_ip()
|
|
|
+ port = request.environ.get('SERVER_PORT', '5000')
|
|
|
+ host = f"{server_ip}:{port}"
|
|
|
+ scheme = request.headers.get('X-Forwarded-Proto', request.scheme)
|
|
|
+ return f"{scheme}://{host}/uploads/{filename}"
|
|
|
+
|
|
|
+
|
|
|
+def _attachment_test_remove_stored():
|
|
|
+ """删除 session 中记录的文件(若存在)。"""
|
|
|
+ name = session.pop(ATTACHMENT_TEST_SESSION_KEY, None)
|
|
|
+ if not name:
|
|
|
+ return
|
|
|
+ path = os.path.join(current_app.config['UPLOAD_FOLDER'], name)
|
|
|
+ try:
|
|
|
+ if os.path.isfile(path):
|
|
|
+ os.remove(path)
|
|
|
+ except OSError as e:
|
|
|
+ logger.warning(f"删除附件测试文件失败 {path}: {e}")
|
|
|
+
|
|
|
+
|
|
|
+def _guess_ext_from_content_type(content_type):
|
|
|
+ if not content_type:
|
|
|
+ return None
|
|
|
+ ct = content_type.lower()
|
|
|
+ mapping = {
|
|
|
+ 'image/png': 'png', 'image/jpeg': 'jpg', 'image/jpg': 'jpg', 'image/gif': 'gif',
|
|
|
+ 'image/webp': 'webp', 'image/bmp': 'bmp', 'image/heic': 'heic', 'image/heif': 'heif',
|
|
|
+ 'video/mp4': 'mp4', 'video/quicktime': 'mov', 'video/webm': 'webm',
|
|
|
+ 'application/pdf': 'pdf',
|
|
|
+ }
|
|
|
+ return mapping.get(ct.split(';')[0].strip())
|
|
|
+
|
|
|
@main_bp.route('/')
|
|
|
@login_required
|
|
|
def index():
|
|
|
@@ -44,6 +98,83 @@ def self_check_page():
|
|
|
"""设备自检页面"""
|
|
|
return render_template('self_check.html', active_page='self_check')
|
|
|
|
|
|
+
|
|
|
+@main_bp.route('/attachment_upload_test')
|
|
|
+@login_required
|
|
|
+def attachment_test_page():
|
|
|
+ """附件上传测试页面(单附件,会话内替换)"""
|
|
|
+ return render_template('attachment_test/index.html', active_page='attachment_test')
|
|
|
+
|
|
|
+
|
|
|
+@main_bp.route('/api/attachment_test/status', methods=['GET'])
|
|
|
+@login_required
|
|
|
+def attachment_test_status():
|
|
|
+ """当前会话中的附件信息"""
|
|
|
+ fn = session.get(ATTACHMENT_TEST_SESSION_KEY)
|
|
|
+ if not fn:
|
|
|
+ return jsonify({'success': True, 'has_file': False})
|
|
|
+ folder = current_app.config['UPLOAD_FOLDER']
|
|
|
+ path = os.path.join(folder, fn)
|
|
|
+ if not os.path.isfile(path):
|
|
|
+ session.pop(ATTACHMENT_TEST_SESSION_KEY, None)
|
|
|
+ return jsonify({'success': True, 'has_file': False})
|
|
|
+ return jsonify({
|
|
|
+ 'success': True,
|
|
|
+ 'has_file': True,
|
|
|
+ 'filename': fn,
|
|
|
+ 'url': _attachment_test_public_url(fn),
|
|
|
+ })
|
|
|
+
|
|
|
+
|
|
|
+@main_bp.route('/api/attachment_test/upload', methods=['POST'])
|
|
|
+@login_required
|
|
|
+def attachment_test_upload():
|
|
|
+ """上传单个附件,覆盖会话中已有文件"""
|
|
|
+ if 'file' not in request.files:
|
|
|
+ return jsonify({'success': False, 'message': '缺少文件字段 file'}), 400
|
|
|
+ file = request.files['file']
|
|
|
+ if not file or file.filename == '':
|
|
|
+ return jsonify({'success': False, 'message': '未选择文件'}), 400
|
|
|
+
|
|
|
+ original = file.filename
|
|
|
+ base = secure_filename(original)
|
|
|
+ ext = None
|
|
|
+ if base and '.' in base:
|
|
|
+ ext = base.rsplit('.', 1)[1].lower()
|
|
|
+ if not ext or ext not in ATTACHMENT_TEST_EXTENSIONS:
|
|
|
+ ext = _guess_ext_from_content_type(file.content_type or '')
|
|
|
+ if not ext or ext not in ATTACHMENT_TEST_EXTENSIONS:
|
|
|
+ return jsonify({
|
|
|
+ 'success': False,
|
|
|
+ 'message': f'不支持的文件类型,允许: {", ".join(sorted(ATTACHMENT_TEST_EXTENSIONS))}',
|
|
|
+ }), 400
|
|
|
+
|
|
|
+ _attachment_test_remove_stored()
|
|
|
+
|
|
|
+ new_name = f"att_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}.{ext}"
|
|
|
+ folder = current_app.config['UPLOAD_FOLDER']
|
|
|
+ os.makedirs(folder, exist_ok=True)
|
|
|
+ filepath = os.path.join(folder, new_name)
|
|
|
+ file.save(filepath)
|
|
|
+ session[ATTACHMENT_TEST_SESSION_KEY] = new_name
|
|
|
+
|
|
|
+ return jsonify({
|
|
|
+ 'success': True,
|
|
|
+ 'message': '上传成功',
|
|
|
+ 'filename': new_name,
|
|
|
+ 'url': _attachment_test_public_url(new_name),
|
|
|
+ })
|
|
|
+
|
|
|
+
|
|
|
+@main_bp.route('/api/attachment_test', methods=['DELETE'])
|
|
|
+@login_required
|
|
|
+def attachment_test_delete():
|
|
|
+ """删除当前会话中的附件"""
|
|
|
+ if not session.get(ATTACHMENT_TEST_SESSION_KEY):
|
|
|
+ return jsonify({'success': True, 'message': '没有可删除的附件'})
|
|
|
+ _attachment_test_remove_stored()
|
|
|
+ return jsonify({'success': True, 'message': '已删除'})
|
|
|
+
|
|
|
@main_bp.route('/api/send_report', methods=['POST'])
|
|
|
@login_required
|
|
|
def send_report_api():
|