121 lines
3.9 KiB
Python
121 lines
3.9 KiB
Python
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__": {<field>: <value>, ...}}
|
||
# For deletes: {"__deleted__": {<field>: <value>, ...}}
|
||
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)},
|
||
) |