50 lines
1.3 KiB
Python
50 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
def _audit_path() -> Path:
|
|
return Path(settings.admin_audit_log_path)
|
|
|
|
|
|
def write_audit_event(
|
|
action: str,
|
|
actor_user_id: str,
|
|
actor_email: str,
|
|
details: dict[str, Any] | None = None,
|
|
) -> None:
|
|
payload = {
|
|
"timestamp": datetime.now(UTC).isoformat(),
|
|
"action": action,
|
|
"actor_user_id": actor_user_id,
|
|
"actor_email": actor_email,
|
|
"details": details or {},
|
|
}
|
|
path = _audit_path()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("a", encoding="utf-8") as file:
|
|
file.write(json.dumps(payload, ensure_ascii=False))
|
|
file.write("\n")
|
|
|
|
|
|
def read_audit_events(limit: int = 200) -> list[dict[str, Any]]:
|
|
path = _audit_path()
|
|
if not path.exists():
|
|
return []
|
|
lines = path.read_text(encoding="utf-8").splitlines()
|
|
tail = lines[-limit:]
|
|
events: list[dict[str, Any]] = []
|
|
for line in tail:
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
events.append(json.loads(line))
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return list(reversed(events))
|