74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from ninja import Router, Schema
|
|
|
|
from apps.accounts.api import require_permission
|
|
from apps.audit.models import OperationAuditLog
|
|
|
|
router = Router(tags=["audit"])
|
|
|
|
|
|
class OperationAuditLogOut(Schema):
|
|
id: int
|
|
user_id: int | None
|
|
username: str | None
|
|
module: str
|
|
action: str
|
|
target_type: str
|
|
target_id: str
|
|
request_snapshot: dict[str, Any]
|
|
result: str
|
|
ip_address: str | None
|
|
user_agent: str
|
|
created_at: datetime
|
|
|
|
|
|
@router.get("/operation-logs", response=list[OperationAuditLogOut])
|
|
def list_operation_logs(
|
|
request,
|
|
module: str | None = None,
|
|
action: str | None = None,
|
|
target_type: str | None = None,
|
|
target_id: str | None = None,
|
|
result: str | None = None,
|
|
user_id: int | None = None,
|
|
created_from: datetime | None = None,
|
|
created_to: datetime | None = None,
|
|
limit: int = 100,
|
|
):
|
|
require_permission(request, "system.audit.view")
|
|
queryset = OperationAuditLog.objects.select_related("user")
|
|
filters = {
|
|
"module": module,
|
|
"action": action,
|
|
"target_type": target_type,
|
|
"target_id": target_id,
|
|
"result": result,
|
|
"user_id": user_id,
|
|
}
|
|
for field, value in filters.items():
|
|
if value not in (None, ""):
|
|
queryset = queryset.filter(**{field: value})
|
|
if created_from:
|
|
queryset = queryset.filter(created_at__gte=created_from)
|
|
if created_to:
|
|
queryset = queryset.filter(created_at__lte=created_to)
|
|
return [
|
|
{
|
|
"id": item.id,
|
|
"user_id": item.user_id,
|
|
"username": item.user.username if item.user_id else None,
|
|
"module": item.module,
|
|
"action": item.action,
|
|
"target_type": item.target_type,
|
|
"target_id": item.target_id,
|
|
"request_snapshot": item.request_snapshot,
|
|
"result": item.result,
|
|
"ip_address": item.ip_address,
|
|
"user_agent": item.user_agent,
|
|
"created_at": item.created_at,
|
|
}
|
|
for item in queryset[: min(max(limit, 1), 500)]
|
|
]
|