115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
import json
|
|
import logging
|
|
from django.utils import timezone
|
|
from django.core.serializers.json import DjangoJSONEncoder
|
|
from . import utils
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _json_safe(value):
|
|
"""
|
|
Convierte cualquier valor a un tipo seguro para JSONField.
|
|
Usa DjangoJSONEncoder para manejar date, datetime, Decimal, UUID, etc.
|
|
"""
|
|
if value is None or value == '':
|
|
return value
|
|
try:
|
|
return json.loads(json.dumps(value, cls=DjangoJSONEncoder))
|
|
except Exception:
|
|
return str(value)
|
|
|
|
|
|
class LogService:
|
|
"""
|
|
Servicio centralizado de gestión de logs.
|
|
Única fuente de inserción en la tabla audit_logs.
|
|
|
|
Uso en vistas:
|
|
log_id = LogService.gestionar_log(self, request, path='/mi/path/')
|
|
LogService.gestionar_log(self, request, log_id=log_id, body_request=data, status_code=200)
|
|
|
|
Si la BD no está disponible, el log falla silenciosamente para no
|
|
interrumpir la petición del usuario.
|
|
"""
|
|
|
|
def gestionar_log(
|
|
self,
|
|
request,
|
|
log_id=None,
|
|
path=None,
|
|
user=None,
|
|
body_request=None,
|
|
body_response=None,
|
|
status_code=None,
|
|
):
|
|
try:
|
|
return LogService._ejecutar_log(
|
|
self, request, log_id, path, user, body_request, body_response, status_code
|
|
)
|
|
except Exception as e:
|
|
logger.warning('LogService.gestionar_log falló (log omitido): %s', str(e))
|
|
return log_id # Devuelve el log_id existente o None si era creación
|
|
|
|
@staticmethod
|
|
def _ejecutar_log(caller, request, log_id, path, user, body_request, body_response, status_code):
|
|
# Importación diferida para evitar problemas de arranque si la BD no está lista
|
|
from backend_admin.models import Log
|
|
|
|
# Determinar la app llamante a partir del módulo de la vista
|
|
modulo = caller.__class__.__module__
|
|
app_nombre = modulo.split('.')[0]
|
|
|
|
tag_header = request.headers.get('tag')
|
|
|
|
if tag_header:
|
|
usuario_final = tag_header
|
|
elif app_nombre:
|
|
apps_automatizadas = ['automatizados']
|
|
if app_nombre.lower() in apps_automatizadas:
|
|
usuario_final = 'Jenkins'
|
|
else:
|
|
usuario_final = app_nombre
|
|
else:
|
|
usuario_final = user if user else 'orquestador'
|
|
|
|
if log_id is None:
|
|
# --- CREACIÓN: primer registro del ciclo de vida de la petición ---
|
|
path_final = path if path else request.path
|
|
|
|
nuevo_log = Log.objects.create(
|
|
user_id=0,
|
|
user=usuario_final,
|
|
app_id=0,
|
|
remote_address=utils.get_client_ip(request),
|
|
request='',
|
|
response='',
|
|
status_code=status_code if status_code else '0',
|
|
path=path_final,
|
|
method=request.method,
|
|
createdAt=timezone.now(),
|
|
)
|
|
return nuevo_log.pk
|
|
|
|
|
|
else:
|
|
# --- ACTUALIZACIÓN: enriquece el log con datos del procesamiento ---
|
|
datos_a_actualizar = {
|
|
'updatedAt': timezone.now(),
|
|
}
|
|
|
|
if user:
|
|
datos_a_actualizar['user'] = user
|
|
if path:
|
|
datos_a_actualizar['path'] = path
|
|
if body_request is not None:
|
|
# _json_safe convierte dates, Decimals, etc. a tipos JSON válidos
|
|
datos_a_actualizar['request'] = _json_safe(body_request)
|
|
if body_response is not None:
|
|
datos_a_actualizar['response'] = _json_safe(body_response)
|
|
if status_code is not None:
|
|
datos_a_actualizar['status_code'] = str(status_code)
|
|
|
|
Log.objects.filter(pk=log_id).update(**datos_a_actualizar)
|
|
return log_id
|