unknown
64 lines
1.8 KiB
Python
64 lines
1.8 KiB
Python
"""
|
|
app/audit.py
|
|
Helper centralizado para registrar eventos de auditoría.
|
|
Se llama desde los routers después de cada operación exitosa.
|
|
"""
|
|
import json
|
|
from datetime import datetime
|
|
from typing import Any, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
import models
|
|
|
|
|
|
def _serialize(obj: Any) -> Optional[str]:
|
|
"""Serializa un objeto a JSON string. Devuelve None si obj es None."""
|
|
if obj is None:
|
|
return None
|
|
if hasattr(obj, '__dict__'):
|
|
# SQLAlchemy model — serializar atributos no privados
|
|
data = {
|
|
k: v for k, v in obj.__dict__.items()
|
|
if not k.startswith('_')
|
|
}
|
|
# Convertir datetime a string para que sea serializable
|
|
for key, value in data.items():
|
|
if isinstance(value, datetime):
|
|
data[key] = value.isoformat()
|
|
return json.dumps(data, default=str)
|
|
if isinstance(obj, dict):
|
|
return json.dumps(obj, default=str)
|
|
return str(obj)
|
|
|
|
|
|
def log_action(
|
|
db: Session,
|
|
entity_type: str,
|
|
entity_id: Any,
|
|
action: str,
|
|
user_id: Optional[Any] = None,
|
|
before: Any = None,
|
|
after: Any = None,
|
|
ip_address: Optional[str] = None,
|
|
) -> None:
|
|
# Convertir "system" (feeder) a -1, y cualquier valor no numérico también
|
|
if user_id == "system" or user_id is None:
|
|
resolved_user_id = -1
|
|
else:
|
|
try:
|
|
resolved_user_id = int(user_id)
|
|
except (ValueError, TypeError):
|
|
resolved_user_id = -1
|
|
|
|
entry = models.AuditLog(
|
|
entity_type=entity_type,
|
|
entity_id=str(entity_id),
|
|
action=action,
|
|
user_id=resolved_user_id,
|
|
before_value=_serialize(before),
|
|
after_value=_serialize(after),
|
|
timestamp=datetime.utcnow(),
|
|
ip_address=ip_address,
|
|
)
|
|
db.add(entry) |