117 lines
3.8 KiB
Python
117 lines
3.8 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_test_data, get_all_test_data_for_export, delete_test_data, insert_log
|
||
|
||
bp = Blueprint("test_data", __name__)
|
||
|
||
|
||
@bp.route("/test-data")
|
||
def test_data_page():
|
||
"""测试信息页"""
|
||
return render_template("test_data.html")
|
||
|
||
|
||
@bp.route("/api/test-data")
|
||
def api_test_data():
|
||
"""分页查询测试数据"""
|
||
page = request.args.get("page", 1, type=int)
|
||
per_page = request.args.get("per_page", 20, type=int)
|
||
serial = request.args.get("serial", "", type=str)
|
||
date_from = request.args.get("date_from", "", type=str)
|
||
date_to = request.args.get("date_to", "", type=str)
|
||
test_mode = request.args.get("test_mode", "", type=str)
|
||
data_source = request.args.get("data_source", "", type=str)
|
||
|
||
records, total = get_test_data(page, per_page, serial, date_from, date_to,
|
||
test_mode, data_source)
|
||
return jsonify({
|
||
"records": records,
|
||
"total": total,
|
||
"page": page,
|
||
"per_page": per_page,
|
||
"pages": (total + per_page - 1) // per_page if total > 0 else 1,
|
||
})
|
||
|
||
|
||
@bp.route("/api/test-data/chart")
|
||
def api_chart_data():
|
||
"""返回图表所需全部数据（不分页）"""
|
||
serial = request.args.get("serial", "", type=str)
|
||
date_from = request.args.get("date_from", "", type=str)
|
||
date_to = request.args.get("date_to", "", type=str)
|
||
test_mode = request.args.get("test_mode", "", type=str)
|
||
data_source = request.args.get("data_source", "", type=str)
|
||
|
||
records = get_all_test_data_for_export(serial, date_from, date_to,
|
||
test_mode, data_source)
|
||
return jsonify({"records": records, "total": len(records)})
|
||
|
||
@bp.route("/api/test-data/export")
|
||
def api_export():
|
||
"""导出测试数据为 CSV"""
|
||
serial = request.args.get("serial", "", type=str)
|
||
date_from = request.args.get("date_from", "", type=str)
|
||
date_to = request.args.get("date_to", "", type=str)
|
||
test_mode = request.args.get("test_mode", "", type=str)
|
||
data_source = request.args.get("data_source", "", type=str)
|
||
|
||
records = get_all_test_data_for_export(serial, date_from, date_to,
|
||
test_mode, data_source)
|
||
|
||
output = io.StringIO()
|
||
writer = csv.writer(output)
|
||
|
||
# 表头
|
||
if records:
|
||
headers = [k for k in 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=test_data.csv"},
|
||
)
|
||
|
||
|
||
@bp.route("/api/test-data/delete", methods=["POST"])
|
||
@login_required
|
||
def api_delete():
|
||
"""删除测试数据（仅 admin）"""
|
||
if current_user.role != "admin":
|
||
return jsonify({"ok": False, "error": "无权限"}), 403
|
||
|
||
data = request.get_json() or {}
|
||
serial = data.get("serial", "")
|
||
date_from = data.get("date_from", "")
|
||
date_to = data.get("date_to", "")
|
||
data_source = data.get("data_source", "")
|
||
|
||
cnt = delete_test_data(serial, date_from, date_to, data_source)
|
||
|
||
detail_parts = [f"删除 {cnt} 条测试数据"]
|
||
if serial:
|
||
detail_parts.append(f"设备={serial}")
|
||
if date_from:
|
||
detail_parts.append(f"从{date_from}")
|
||
if date_to:
|
||
detail_parts.append(f"至{date_to}")
|
||
if data_source:
|
||
detail_parts.append(f"来源={data_source}")
|
||
|
||
insert_log(
|
||
current_user.id, current_user.username, "delete",
|
||
target="test_data",
|
||
detail=", ".join(detail_parts),
|
||
result="ok",
|
||
ip=request.remote_addr or "",
|
||
)
|
||
|
||
return jsonify({"ok": True, "deleted": cnt})
|