feat: Audit Log

Author: shaohuzhang1

apps/application/views/application_views.py

@@ -24,6 +24,7 @@
 from common.constants.permission_constants import CompareConstants, PermissionConstants, Permission, Group, Operate, \
     ViewPermission, RoleConstants
 from common.exception.app_exception import AppAuthenticationFailed
+from common.log.log import log
 from common.response import result
 from common.swagger_api.common_api import CommonApi
 from common.util.common import query_params_to_single_dict
@@ -603,6 +604,7 @@ class Page(APIView):
                              responses=result.get_page_api_response(ApplicationApi.get_response_body_api()),
                              tags=[_('Application')])
         @has_permissions(PermissionConstants.APPLICATION_READ, compare=CompareConstants.AND)
+        @log(menu=_('Application'), operate=_("Get the application list by page"))
         def get(self, request: Request, current_page: int, page_size: int):
             return result.success(
                 ApplicationSerializer.Query(

apps/common/log/log.py

@@ -0,0 +1,85 @@
+# coding=utf-8
+"""
+    @project: MaxKB
+    @Author：虎
+    @file： log.py
+    @date：2025/3/14 16:09
+    @desc:
+"""
+
+from setting.models.log_management import Log
+
+
+def _get_ip_address(request):
+    """
+    获取ip地址
+    @param request:
+    @return:
+    """
+    x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
+    if x_forwarded_for:
+        ip = x_forwarded_for.split(',')[0]
+    else:
+        ip = request.META.get('REMOTE_ADDR')
+    return ip
+
+
+def _get_user(request):
+    """
+    获取用户
+    @param request:
+    @return:
+    """
+    user = request.user
+    return {
+        "id": str(user.id),
+        "email": user.email,
+        "phone": user.phone,
+        "nick_name": user.nick_name,
+        "username": user.username,
+        "role": user.role,
+    }
+
+
+def _get_details(request):
+    path = request.path
+    body = request.data
+    query = request.query_params
+    return {
+        'path': path,
+        'body': body,
+        'query': query
+    }
+
+
+def log(menu: str, operate, get_user=_get_user, get_ip_address=_get_ip_address, get_details=_get_details):
+    """
+    记录审计日志
+    @param menu: 操作菜单 str
+    @param operate: 操作 str|func 如果是一个函数 入参将是一个request 响应为str def operate(request): return "操作菜单"
+    @param get_user: 获取用户
+    @param get_ip_address:获取IP地址
+    @param get_details: 获取执行详情
+    @return:
+    """
+
+    def inner(func):
+        def run(view, request, **kwargs):
+            status = 200
+            try:
+                return func(view, request, **kwargs)
+            except Exception as e:
+                status = 500
+            finally:
+                ip = get_ip_address(request)
+                user = get_user(request)
+                details = get_details(request)
+                _operate = operate
+                if callable(operate):
+                    _operate = operate(request)
+                # 插入审计日志
+                Log(menu=menu, operate=_operate, user=user, status=status, ip_address=ip, details=details).save()
+
+        return run
+
+    return inner

apps/setting/migrations/0010_log.py

@@ -0,0 +1,31 @@
+# Generated by Django 4.2.18 on 2025-03-17 02:50
+
+from django.db import migrations, models
+import uuid
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('setting', '0009_set_default_model_params_form'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Log',
+            fields=[
+                ('create_time', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
+                ('update_time', models.DateTimeField(auto_now=True, verbose_name='修改时间')),
+                ('id', models.UUIDField(default=uuid.uuid1, editable=False, primary_key=True, serialize=False, verbose_name='主键id')),
+                ('menu', models.CharField(max_length=128, verbose_name='操作菜单')),
+                ('operate', models.CharField(max_length=128, verbose_name='操作')),
+                ('user', models.JSONField(default=dict, verbose_name='用户信息')),
+                ('status', models.IntegerField(max_length=20, verbose_name='状态')),
+                ('ip_address', models.CharField(max_length=128, verbose_name='ip地址')),
+                ('details', models.JSONField(default=dict, verbose_name='详情')),
+            ],
+            options={
+                'db_table': 'log',
+            },
+        ),
+    ]

apps/setting/models/log_management.py

@@ -0,0 +1,35 @@
+# coding=utf-8
+"""
+    @project: MaxKB
+    @Author：虎
+    @file： log_management.py
+    @date：2025/3/17 9:54
+    @desc:
+"""
+import uuid
+
+from django.db import models
+
+from common.mixins.app_model_mixin import AppModelMixin
+
+
+class Log(AppModelMixin):
+    """
+    审计日志
+    """
+    id = models.UUIDField(primary_key=True, max_length=128, default=uuid.uuid1, editable=False, verbose_name="主键id")
+
+    menu = models.CharField(max_length=128, verbose_name="操作菜单")
+
+    operate = models.CharField(max_length=128, verbose_name="操作")
+
+    user = models.JSONField(verbose_name="用户信息", default=dict)
+
+    status = models.IntegerField(max_length=20, verbose_name="状态")
+
+    ip_address = models.CharField(max_length=128, verbose_name="ip地址")
+
+    details = models.JSONField(verbose_name="详情", default=dict)
+
+    class Meta:
+        db_table = "log"