103 lines
2.6 KiB
Python
103 lines
2.6 KiB
Python
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
SKIP_PATHS = (
|
|
"/static/",
|
|
"/media/",
|
|
"/admin/",
|
|
"/accounts/token/",
|
|
"/accounts/token/refresh/",
|
|
"/accounts/login/",
|
|
"/accounts/logout/",
|
|
"/api/",
|
|
"/health/",
|
|
"/__debug__/",
|
|
"/favicon.ico",
|
|
"/robots.txt",
|
|
)
|
|
|
|
METHOD_TO_ACTIVITY = {
|
|
"POST": "create",
|
|
"PUT": "update",
|
|
"PATCH": "update",
|
|
"DELETE": "delete",
|
|
}
|
|
|
|
|
|
class ActivityLoggingMiddleware:
|
|
def __init__(self, get_response):
|
|
self.get_response = get_response
|
|
|
|
def __call__(self, request):
|
|
response = self.get_response(request)
|
|
self._log_activity(request, response)
|
|
return response
|
|
|
|
def _log_activity(self, request, response):
|
|
user = getattr(request, "user", None)
|
|
if not user or not user.is_authenticated:
|
|
return
|
|
|
|
method = request.method
|
|
if method not in METHOD_TO_ACTIVITY:
|
|
return
|
|
|
|
path = request.path
|
|
for skip in SKIP_PATHS:
|
|
if path.startswith(skip):
|
|
return
|
|
|
|
if response.status_code >= 400:
|
|
return
|
|
|
|
try:
|
|
from django.urls import resolve
|
|
|
|
match = resolve(path)
|
|
except Exception:
|
|
return
|
|
|
|
app_name = match.app_name or ""
|
|
url_name = match.url_name or ""
|
|
kwargs = match.kwargs or {}
|
|
|
|
if not app_name and not url_name:
|
|
return
|
|
|
|
module = self._derive_module(app_name, url_name)
|
|
description = self._build_description(method, url_name, kwargs)
|
|
|
|
try:
|
|
from apps.accounts.services import StaffActivityService
|
|
|
|
StaffActivityService.log(
|
|
user=user,
|
|
activity_type=METHOD_TO_ACTIVITY[method],
|
|
description=description,
|
|
module=module,
|
|
action=url_name or path,
|
|
request=request,
|
|
)
|
|
except Exception:
|
|
logger.debug("ActivityLoggingMiddleware: failed to log", exc_info=True)
|
|
|
|
def _derive_module(self, app_name, url_name):
|
|
if app_name:
|
|
return app_name
|
|
if url_name and ":" in url_name:
|
|
return url_name.split(":")[0]
|
|
return ""
|
|
|
|
def _build_description(self, method, url_name, kwargs):
|
|
parts = [f"{method}"]
|
|
|
|
if url_name:
|
|
parts.append(url_name.replace("-", " ").replace("_", " "))
|
|
|
|
obj_id = kwargs.get("pk") or kwargs.get("id") or kwargs.get("uuid")
|
|
if obj_id:
|
|
parts.append(f"(#{str(obj_id)[:8]})")
|
|
|
|
return " ".join(parts)
|