""" 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"}, )