75 lines
2.9 KiB
Python
75 lines
2.9 KiB
Python
from datetime import datetime
|
|
|
|
from flask import Blueprint, render_template, request, jsonify, current_app
|
|
from flask_login import login_required
|
|
|
|
from app.forms import LogFilterForm
|
|
from app.models import ApiCallLog, ApiConfig
|
|
|
|
logs_bp = Blueprint("logs", __name__, url_prefix="/logs")
|
|
|
|
|
|
@logs_bp.route("", methods=["GET"])
|
|
@login_required
|
|
def list_logs():
|
|
form = LogFilterForm(request.args)
|
|
form.api_id.choices = [(-1, "All APIs")] + [(api.id, api.name) for api in ApiConfig.query.order_by(ApiConfig.name)]
|
|
|
|
query = ApiCallLog.query.join(ApiConfig)
|
|
if form.api_id.data and form.api_id.data != -1:
|
|
query = query.filter(ApiCallLog.api_id == form.api_id.data)
|
|
if form.success.data in {"1", "0"}:
|
|
query = query.filter(ApiCallLog.success == (form.success.data == "1"))
|
|
start = form.parse_date(form.start_date.data or "")
|
|
end = form.parse_date(form.end_date.data or "")
|
|
if start:
|
|
query = query.filter(ApiCallLog.request_time >= start)
|
|
if end:
|
|
end_dt = datetime(end.year, end.month, end.day, 23, 59, 59)
|
|
query = query.filter(ApiCallLog.request_time <= end_dt)
|
|
|
|
page = int(request.args.get("page", 1))
|
|
per_page = int(request.args.get("per_page", 10))
|
|
pagination = query.order_by(ApiCallLog.request_time.desc()).paginate(page=page, per_page=per_page, error_out=False)
|
|
# 组装分页参数时避免重复 page/per_page
|
|
query_args = request.args.to_dict(flat=True)
|
|
query_args.pop("page", None)
|
|
query_args.pop("per_page", None)
|
|
return render_template("logs/list.html", form=form, pagination=pagination, logs=pagination.items, query_args=query_args)
|
|
|
|
|
|
@logs_bp.route("/<int:log_id>", methods=["GET"])
|
|
@login_required
|
|
def log_detail(log_id: int):
|
|
log_entry = ApiCallLog.query.get_or_404(log_id)
|
|
return render_template("logs/detail.html", log=log_entry)
|
|
|
|
|
|
@logs_bp.route("/<int:log_id>/stream", methods=["GET"])
|
|
@login_required
|
|
def log_stream(log_id: int):
|
|
log_entry = ApiCallLog.query.get_or_404(log_id)
|
|
to_cst = current_app.jinja_env.filters.get("to_cst")
|
|
|
|
def fmt(dt):
|
|
return to_cst(dt) if to_cst else (dt.isoformat() if dt else "")
|
|
|
|
duration_ms = log_entry.duration_ms
|
|
if duration_ms is None and log_entry.request_time:
|
|
duration_ms = int((datetime.utcnow() - log_entry.request_time).total_seconds() * 1000)
|
|
|
|
return jsonify(
|
|
{
|
|
"id": log_entry.id,
|
|
"api_name": log_entry.api.name if log_entry.api else "",
|
|
"success": log_entry.success,
|
|
"http_status_code": log_entry.http_status_code,
|
|
"error_message": log_entry.error_message,
|
|
"response_body": log_entry.response_body or "",
|
|
"request_time": fmt(log_entry.request_time),
|
|
"response_time": fmt(log_entry.response_time) or "",
|
|
"duration_ms": duration_ms,
|
|
"finished": log_entry.success is not None,
|
|
}
|
|
)
|