118 lines
3.7 KiB
Python
118 lines
3.7 KiB
Python
"""
|
|
app/admin/audit_log/routes.py
|
|
Audit log viewer: filterable by actor, action prefix, date range.
|
|
CSV export. Append-only — no modifications permitted.
|
|
"""
|
|
|
|
import csv
|
|
import io
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from flask import Blueprint, render_template, request, Response
|
|
from flask_login import login_required
|
|
|
|
from app.models.platform import AuditLog, SystemUser
|
|
from app.admin.utils import superadmin_required
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
audit_log_bp = Blueprint("audit_log", __name__, url_prefix="/audit-log")
|
|
_PAGE_SIZE = 50
|
|
|
|
|
|
@audit_log_bp.route("/")
|
|
@login_required
|
|
@superadmin_required
|
|
def index():
|
|
actor_id = request.args.get("actor_id", type=int)
|
|
action_prefix = request.args.get("action", "").strip()
|
|
target_type = request.args.get("target_type", "").strip()
|
|
date_from_raw = request.args.get("date_from", "").strip()
|
|
date_to_raw = request.args.get("date_to", "").strip()
|
|
page = request.args.get("page", 1, type=int)
|
|
|
|
q = AuditLog.query.order_by(AuditLog.created_at.desc())
|
|
|
|
if actor_id:
|
|
q = q.filter_by(actor_id=actor_id)
|
|
if action_prefix:
|
|
q = q.filter(AuditLog.action.startswith(action_prefix))
|
|
if target_type:
|
|
q = q.filter_by(target_type=target_type)
|
|
if date_from_raw:
|
|
try:
|
|
q = q.filter(AuditLog.created_at >= datetime.fromisoformat(date_from_raw))
|
|
except ValueError:
|
|
pass
|
|
if date_to_raw:
|
|
try:
|
|
q = q.filter(AuditLog.created_at <= datetime.fromisoformat(date_to_raw))
|
|
except ValueError:
|
|
pass
|
|
|
|
pagination = q.paginate(page=page, per_page=_PAGE_SIZE, error_out=False)
|
|
system_users = SystemUser.query.order_by(SystemUser.name).all()
|
|
|
|
return render_template(
|
|
"admin/audit_log/index.html",
|
|
pagination=pagination,
|
|
entries=pagination.items,
|
|
system_users=system_users,
|
|
filters={
|
|
"actor_id": actor_id, "action": action_prefix,
|
|
"target_type": target_type, "date_from": date_from_raw, "date_to": date_to_raw,
|
|
},
|
|
)
|
|
|
|
|
|
@audit_log_bp.route("/export.csv")
|
|
@login_required
|
|
@superadmin_required
|
|
def export_csv():
|
|
"""Export filtered audit log as CSV (max 10,000 rows)."""
|
|
actor_id = request.args.get("actor_id", type=int)
|
|
action_prefix = request.args.get("action", "").strip()
|
|
target_type = request.args.get("target_type", "").strip()
|
|
date_from_raw = request.args.get("date_from", "").strip()
|
|
date_to_raw = request.args.get("date_to", "").strip()
|
|
|
|
q = AuditLog.query.order_by(AuditLog.created_at.desc())
|
|
if actor_id:
|
|
q = q.filter_by(actor_id=actor_id)
|
|
if action_prefix:
|
|
q = q.filter(AuditLog.action.startswith(action_prefix))
|
|
if target_type:
|
|
q = q.filter_by(target_type=target_type)
|
|
if date_from_raw:
|
|
try:
|
|
q = q.filter(AuditLog.created_at >= datetime.fromisoformat(date_from_raw))
|
|
except ValueError:
|
|
pass
|
|
if date_to_raw:
|
|
try:
|
|
q = q.filter(AuditLog.created_at <= datetime.fromisoformat(date_to_raw))
|
|
except ValueError:
|
|
pass
|
|
|
|
entries = q.limit(10000).all()
|
|
|
|
output = io.StringIO()
|
|
writer = csv.writer(output)
|
|
writer.writerow([
|
|
"id", "created_at", "actor_id", "actor_type", "action",
|
|
"target_type", "target_id", "ip_address",
|
|
])
|
|
for e in entries:
|
|
writer.writerow([
|
|
e.id, e.created_at, e.actor_id, e.actor_type, e.action,
|
|
e.target_type, e.target_id, e.ip_address,
|
|
])
|
|
|
|
logger.info("Audit log exported: %d rows", len(entries))
|
|
return Response(
|
|
output.getvalue(),
|
|
mimetype="text/csv",
|
|
headers={"Content-Disposition": "attachment; filename=audit_log.csv"},
|
|
)
|