From 7b760362173f5e0cc798e0d48996ab3f8db6d7be Mon Sep 17 00:00:00 2001 From: Joshua Laymon Date: Sun, 31 Aug 2025 13:01:02 +0000 Subject: [PATCH] Add web/core/models_audit.py --- web/core/models_audit.py | 121 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 121 insertions(+) create mode 100644 web/core/models_audit.py diff --git a/web/core/models_audit.py b/web/core/models_audit.py new file mode 100644 index 0000000..43b3ba1 --- /dev/null +++ b/web/core/models_audit.py @@ -0,0 +1,121 @@ +from django.db import models +from django.utils import timezone +from django.dispatch import receiver +from django.db.models.signals import pre_save, post_save, post_delete +from django.forms.models import model_to_dict +from django.apps import apps + +class AuditLog(models.Model): + ACTION_CREATE = "create" + ACTION_UPDATE = "update" + ACTION_DELETE = "delete" + ACTIONS = [ + (ACTION_CREATE, "Created"), + (ACTION_UPDATE, "Updated"), + (ACTION_DELETE, "Deleted"), + ] + + entry_id = models.IntegerField(db_index=True) + action = models.CharField(max_length=10, choices=ACTIONS, db_index=True) + username = models.CharField(max_length=255, blank=True, default="") + timestamp = models.DateTimeField(default=timezone.now, db_index=True) + # For updates: {"field": ["old","new"], ...} + # For creates: {"__created__": {: , ...}} + # For deletes: {"__deleted__": {: , ...}} + changes = models.JSONField(null=True, blank=True) + + class Meta: + ordering = ["-timestamp"] + indexes = [ + models.Index(fields=["timestamp"]), + models.Index(fields=["action"]), + models.Index(fields=["entry_id"]), + ] + + def __str__(self): + 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 + +def _tracked_field_names(): + names = [] + for f in Entry._meta.get_fields(): + if getattr(f, "many_to_many", False): + continue + if not getattr(f, "concrete", False): + continue + if getattr(f, "primary_key", False): + continue + if getattr(f, "editable", True) is False: + continue + names.append(f.name) + return names + +_TRACKED = _tracked_field_names() + +def _serialize_entry(obj): + d = model_to_dict(obj, fields=_TRACKED) + for k, v in d.items(): + d[k] = "" if v is None else str(v) + return d + +# ---- get current username from middleware ---- +from .middleware import get_current_username # you’ll add this in step 2 + +# ---- signals ---- + +@receiver(pre_save, sender=Entry) +def _audit_entry_pre_save(sender, instance, **kwargs): + # For updates, stash old snapshot on the instance + if instance.pk: + try: + existing = Entry.objects.get(pk=instance.pk) + instance.__audit_old__ = _serialize_entry(existing) + except Entry.DoesNotExist: + instance.__audit_old__ = None + else: + instance.__audit_old__ = None + +@receiver(post_save, sender=Entry) +def _audit_entry_save(sender, instance, created, **kwargs): + from .models_audit import AuditLog + if created: + AuditLog.objects.create( + entry_id=instance.id, + action=AuditLog.ACTION_CREATE, + username=get_current_username(), + changes={"__created__": _serialize_entry(instance)}, + ) + else: + old = getattr(instance, "__audit_old__", None) + new = _serialize_entry(instance) + if old is None: + changes = {"__created__": new} + action = AuditLog.ACTION_CREATE + else: + diffs = {} + for field in _TRACKED: + ov = old.get(field, "") + nv = new.get(field, "") + if ov != nv: + diffs[field] = [ov, nv] + changes = diffs or None + action = AuditLog.ACTION_UPDATE + AuditLog.objects.create( + entry_id=instance.id, + action=action, + username=get_current_username(), + changes=changes, + ) + +@receiver(post_delete, sender=Entry) +def _audit_entry_delete(sender, instance, **kwargs): + from .models_audit import AuditLog + AuditLog.objects.create( + entry_id=instance.id, + action=AuditLog.ACTION_DELETE, + username=get_current_username(), + changes={"__deleted__": _serialize_entry(instance)}, + ) \ No newline at end of file