mirror of https://gitee.com/anolis/sysom.git
50 lines
1.9 KiB
Python
50 lines
1.9 KiB
Python
import time
|
|
from django.core.handlers.asgi import ASGIRequest
|
|
from django.conf import settings
|
|
from rest_framework.renderers import JSONRenderer
|
|
from rest_framework.request import Request
|
|
|
|
from apps.accounts.models import HandlerOptionEnum, User
|
|
from sysom_utils import SysomFramework
|
|
|
|
# User = get_user_model()
|
|
|
|
|
|
class SysomJsonRender(JSONRenderer):
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
self.product = SysomFramework.cec_producer()
|
|
|
|
def render(self, data, accepted_media_type=None, renderer_context=None):
|
|
if renderer_context:
|
|
request = renderer_context.get('request', None)
|
|
view = renderer_context.get('view', None)
|
|
response = renderer_context.get('response', None)
|
|
self.before_response_save_log(request, view, response)
|
|
return super().render(data, accepted_media_type, renderer_context)
|
|
|
|
def before_response_save_log(self, request: Request, view, response):
|
|
request: ASGIRequest = getattr(request, '_request', None)
|
|
if request.path in settings.API_INTERFACE_NO_LOG_LIST: return
|
|
kwargs = {
|
|
'ip': request.META.get('REMOTE_ADDR', None),
|
|
'path': request.path,
|
|
'browser_agent': request.headers.get('User-Agent', ''),
|
|
'methond': request.method,
|
|
'handler': view.__class__.__name__,
|
|
'status': getattr(response, 'status_code', 200),
|
|
'ts': int(time.time()),
|
|
}
|
|
if 'auth' in request.path:
|
|
kwargs['request_type'] = HandlerOptionEnum.LOGIN.label
|
|
elif 'logout' in request.path:
|
|
kwargs['request_type'] = HandlerOptionEnum.LOGOUT.label
|
|
else:
|
|
kwargs['request_type'] = HandlerOptionEnum.ACTION.label
|
|
user: User = getattr(request, 'user')
|
|
if user:
|
|
kwargs['user'] = user.pk
|
|
|
|
self.product.produce(settings.SYSOM_CEC_LOG_DISPATCH_TOPIC, kwargs)
|
|
self.product.flush()
|