wangfq
000e4f8d3a
- auth.py: 新增 privileged_required 装饰器 (admin+manager)，admin_required 仅限用户管理 - 路由权限: fixture/logs/device_logs/test_data 的 admin 检查改为 admin+manager - 前端: 导航栏/删除按钮/配置按钮扩展为 admin+manager 可见 - 用户管理: 角色下拉增加 manager 选项，仍仅 admin 可访问 - 新增 /change-password 路由+模板，所有登录用户可自行修改密码 - edc_server models.py: role COMMENT 更新 + ALTER TABLE 迁移
96 lines
2.9 KiB
Python
96 lines
2.9 KiB
Python
"""设备事件日志 API"""
|
|
|
|
import csv
|
|
import io
|
|
|
|
from flask import Blueprint, jsonify, render_template, request, Response
|
|
from flask_login import login_required, current_user
|
|
from app.models import get_device_logs, export_device_logs, delete_device_logs, insert_log
|
|
|
|
bp = Blueprint("device_logs", __name__)
|
|
|
|
|
|
@bp.route("/device-logs")
|
|
@login_required
|
|
def index():
|
|
"""设备日志页面"""
|
|
return render_template("device_logs.html")
|
|
|
|
|
|
@bp.route("/api/device-logs")
|
|
@login_required
|
|
def api_device_logs():
|
|
"""查询设备事件日志"""
|
|
page = request.args.get("page", 1, type=int)
|
|
per_page = request.args.get("per_page", 30, type=int)
|
|
serial = request.args.get("serial", "", type=str)
|
|
event_type = request.args.get("event_type", "", type=str)
|
|
date_from = request.args.get("date_from", "", type=str)
|
|
date_to = request.args.get("date_to", "", type=str)
|
|
|
|
records, total = get_device_logs(
|
|
page=page, per_page=per_page,
|
|
serial=serial, event_type=event_type,
|
|
date_from=date_from, date_to=date_to,
|
|
)
|
|
pages = max(1, (total + per_page - 1) // per_page)
|
|
return jsonify({"records": records, "total": total, "pages": pages})
|
|
|
|
|
|
@bp.route("/api/device-logs/export")
|
|
@login_required
|
|
def api_export():
|
|
"""导出设备事件日志为 CSV"""
|
|
serial = request.args.get("serial", "", type=str)
|
|
event_type = request.args.get("event_type", "", type=str)
|
|
date_from = request.args.get("date_from", "", type=str)
|
|
date_to = request.args.get("date_to", "", type=str)
|
|
|
|
records = export_device_logs(
|
|
serial=serial, event_type=event_type,
|
|
date_from=date_from, date_to=date_to,
|
|
)
|
|
|
|
output = io.StringIO()
|
|
writer = csv.writer(output)
|
|
|
|
if records:
|
|
headers = list(records[0].keys())
|
|
writer.writerow(headers)
|
|
for r in records:
|
|
writer.writerow(r.values())
|
|
|
|
output.seek(0)
|
|
return Response(
|
|
output.getvalue(),
|
|
mimetype="text/csv",
|
|
headers={"Content-Disposition": "attachment; filename=device_logs.csv"},
|
|
)
|
|
|
|
|
|
@bp.route("/api/device-logs/delete", methods=["POST"])
|
|
@login_required
|
|
def api_device_logs_delete():
|
|
"""删除设备日志（admin 权限）"""
|
|
if current_user.role not in ("admin", "manager"):
|
|
return jsonify({"ok": False, "error": "无权限"}), 403
|
|
|
|
data = request.get_json()
|
|
serial = data.get("serial", "")
|
|
event_type = data.get("event_type", "")
|
|
date_from = data.get("date_from", "")
|
|
date_to = data.get("date_to", "")
|
|
|
|
deleted = delete_device_logs(
|
|
serial=serial, event_type=event_type,
|
|
date_from=date_from, date_to=date_to,
|
|
)
|
|
insert_log(
|
|
current_user.id, current_user.username, "delete",
|
|
target="device_log",
|
|
detail=f"删除 {deleted} 条设备日志 serial={serial} type={event_type}",
|
|
result="ok",
|
|
ip=request.remote_addr or "",
|
|
)
|
|
return jsonify({"ok": True, "deleted": deleted})
|