feat: añadir app general con LogService centralizado
All checks were successful
DEPLOY_MULTI_BRACH/pipeline/pr-dev This commit looks good
DEPLOY_MULTI_BRACH/pipeline/head This commit looks good

- Crear app/general con estructura estándar del proyecto:
  · utilidades/acciones.py → LogService.gestionar_log() (única fuente de logs)
  · utilidades/utils.py → get_client_ip()
  · utilidades/custom_errors.py → ValidationError, ExternalServiceError, NotFoundError
  · exception.py, request.py, serializers.py, validaciones/
- Registrar 'general' en INSTALLED_APPS y añadir general/ a urls.py
- Refactorizar promociones/views.py para usar LogService en lugar de Log directo

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
juanjo
2026-04-16 16:50:53 +02:00
parent 91fc6900eb
commit 2f6564d9a6
19 changed files with 230 additions and 27 deletions

View File

View File

@@ -0,0 +1,81 @@
from backend_admin.models import Log
from django.utils import timezone
from . import utils
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)
"""
def gestionar_log(
self,
request,
log_id=None,
path=None,
user=None,
body_request=None,
body_response=None,
status_code=None,
):
# Determinar la app que llama a partir del módulo de la vista
modulo = self.__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
data_log = {
'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(),
}
nuevo_log = Log.objects.create(**data_log)
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:
datos_a_actualizar['request'] = body_request
if body_response is not None:
datos_a_actualizar['response'] = body_response
if status_code:
datos_a_actualizar['status_code'] = str(status_code)
Log.objects.filter(pk=log_id).update(**datos_a_actualizar)
return log_id

View File

@@ -0,0 +1,33 @@
class ValidationError(Exception):
"""Error de validación de datos de entrada."""
def __init__(self, message, field=None):
self.message = message
self.field = field
super().__init__(message)
def to_response(self):
detail = {'error': self.message}
if self.field:
detail['field'] = self.field
return {'body': {'data': [], **detail}, 'mensaje': self.message}
class ExternalServiceError(Exception):
"""Error en la comunicación con un servicio externo."""
def __init__(self, message, service=None):
self.message = message
self.service = service
super().__init__(message)
def to_response(self):
return {'body': {'data': [], 'error': self.message}, 'mensaje': self.message}
class NotFoundError(Exception):
"""Recurso no encontrado."""
def __init__(self, message='Recurso no encontrado'):
self.message = message
super().__init__(message)
def to_response(self):
return {'body': {'data': [], 'error': self.message}, 'mensaje': self.message}

View File

@@ -0,0 +1,6 @@
def get_client_ip(request):
"""Obtiene la IP real del cliente, considerando proxies."""
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
if x_forwarded_for:
return x_forwarded_for.split(',')[0].strip()
return request.META.get('REMOTE_ADDR')