Update web/core/models_audit.py

This commit is contained in:
Joshua Laymon 2025-08-31 13:07:58 +00:00
parent b9a8efa116
commit a107c677ef

View File

@ -3,7 +3,6 @@ from django.utils import timezone
from django.dispatch import receiver from django.dispatch import receiver
from django.db.models.signals import pre_save, post_save, post_delete from django.db.models.signals import pre_save, post_save, post_delete
from django.forms.models import model_to_dict from django.forms.models import model_to_dict
from django.apps import apps
class AuditLog(models.Model): class AuditLog(models.Model):
ACTION_CREATE = "create" ACTION_CREATE = "create"
@ -35,13 +34,13 @@ class AuditLog(models.Model):
def __str__(self): def __str__(self):
return f"{self.timestamp:%Y-%m-%d %H:%M:%S} {self.action} #{self.entry_id} by {self.username or '-'}" return f"{self.timestamp:%Y-%m-%d %H:%M:%S} {self.action} #{self.entry_id} by {self.username or '-'}"
# ---- helpers for diffs ----
Entry = apps.get_model("core", "Entry") # avoids circular import # ---- helpers for diffs (lazy: derive from sender) ----
def _tracked_field_names(): def _tracked_field_names(sender):
"""Return concrete, editable, non-PK field names for the given model."""
names = [] names = []
for f in Entry._meta.get_fields(): for f in sender._meta.get_fields():
if getattr(f, "many_to_many", False): if getattr(f, "many_to_many", False):
continue continue
if not getattr(f, "concrete", False): if not getattr(f, "concrete", False):
@ -53,50 +52,61 @@ def _tracked_field_names():
names.append(f.name) names.append(f.name)
return names return names
_TRACKED = _tracked_field_names() def _serialize_instance(sender, obj):
"""Serialize only tracked fields; stringify values for safety."""
def _serialize_entry(obj): fields = _tracked_field_names(sender)
d = model_to_dict(obj, fields=_TRACKED) d = model_to_dict(obj, fields=fields)
for k, v in d.items(): for k, v in d.items():
d[k] = "" if v is None else str(v) d[k] = "" if v is None else str(v)
return d return d
# ---- get current username from middleware ---- def _is_core_entry(sender):
from .middleware import get_current_username # youll add this in step 2 """True iff the signal sender is core.Entry (no import-time lookups)."""
return getattr(sender, "_meta", None) and sender._meta.label_lower == "core.entry"
# ---- signals ----
@receiver(pre_save, sender=Entry) # ---- current user from middleware ----
from .middleware import get_current_username
# ---- signals (note: no sender=... at decorator time) ----
@receiver(pre_save)
def _audit_entry_pre_save(sender, instance, **kwargs): def _audit_entry_pre_save(sender, instance, **kwargs):
if not _is_core_entry(sender):
return
# For updates, stash old snapshot on the instance # For updates, stash old snapshot on the instance
if instance.pk: if instance.pk:
try: try:
existing = Entry.objects.get(pk=instance.pk) existing = sender.objects.get(pk=instance.pk)
instance.__audit_old__ = _serialize_entry(existing) instance.__audit_old__ = _serialize_instance(sender, existing)
except Entry.DoesNotExist: except sender.DoesNotExist:
instance.__audit_old__ = None instance.__audit_old__ = None
else: else:
instance.__audit_old__ = None instance.__audit_old__ = None
@receiver(post_save, sender=Entry)
@receiver(post_save)
def _audit_entry_save(sender, instance, created, **kwargs): def _audit_entry_save(sender, instance, created, **kwargs):
from .models_audit import AuditLog if not _is_core_entry(sender):
return
from .models_audit import AuditLog # local import to avoid circulars
if created: if created:
AuditLog.objects.create( AuditLog.objects.create(
entry_id=instance.id, entry_id=instance.id,
action=AuditLog.ACTION_CREATE, action=AuditLog.ACTION_CREATE,
username=get_current_username(), username=get_current_username(),
changes={"__created__": _serialize_entry(instance)}, changes={"__created__": _serialize_instance(sender, instance)},
) )
else: else:
old = getattr(instance, "__audit_old__", None) old = getattr(instance, "__audit_old__", None)
new = _serialize_entry(instance) new = _serialize_instance(sender, instance)
if old is None: if old is None:
changes = {"__created__": new} changes = {"__created__": new}
action = AuditLog.ACTION_CREATE action = AuditLog.ACTION_CREATE
else: else:
diffs = {} diffs = {}
for field in _TRACKED: for field in _tracked_field_names(sender):
ov = old.get(field, "") ov = old.get(field, "")
nv = new.get(field, "") nv = new.get(field, "")
if ov != nv: if ov != nv:
@ -110,12 +120,15 @@ def _audit_entry_save(sender, instance, created, **kwargs):
changes=changes, changes=changes,
) )
@receiver(post_delete, sender=Entry)
@receiver(post_delete)
def _audit_entry_delete(sender, instance, **kwargs): def _audit_entry_delete(sender, instance, **kwargs):
if not _is_core_entry(sender):
return
from .models_audit import AuditLog from .models_audit import AuditLog
AuditLog.objects.create( AuditLog.objects.create(
entry_id=instance.id, entry_id=instance.id,
action=AuditLog.ACTION_DELETE, action=AuditLog.ACTION_DELETE,
username=get_current_username(), username=get_current_username(),
changes={"__deleted__": _serialize_entry(instance)}, changes={"__deleted__": _serialize_instance(sender, instance)},
) )