1""" 2Base model definitions for audit logging. These may be subclassed to accommodate specific models 3such as Page, but the definitions here should remain generic and not depend on the base 4wagtail.core.models module or specific models such as Page. 5""" 6 7import json 8 9from django.conf import settings 10from django.contrib.auth import get_user_model 11from django.contrib.contenttypes.models import ContentType 12from django.core.exceptions import ValidationError 13from django.db import models 14from django.utils import timezone 15from django.utils.functional import cached_property 16from django.utils.translation import gettext_lazy as _ 17 18 19class LogEntryQuerySet(models.QuerySet): 20 def get_users(self): 21 """ 22 Returns a QuerySet of Users who have created at least one log entry in this QuerySet. 23 24 The returned queryset is ordered by the username. 25 """ 26 User = get_user_model() 27 return User.objects.filter( 28 pk__in=set(self.values_list('user__pk', flat=True)) 29 ).order_by(User.USERNAME_FIELD) 30 31 32class BaseLogEntryManager(models.Manager): 33 def get_queryset(self): 34 return LogEntryQuerySet(self.model, using=self._db) 35 36 def get_instance_title(self, instance): 37 return str(instance) 38 39 def log_action(self, instance, action, **kwargs): 40 """ 41 :param instance: The model instance we are logging an action for 42 :param action: The action. Should be namespaced to app (e.g. wagtail.create, wagtail.workflow.start) 43 :param kwargs: Addition fields to for the model deriving from BaseLogEntry 44 - user: The user performing the action 45 - title: the instance title 46 - data: any additional metadata 47 - content_changed, deleted - Boolean flags 48 :return: The new log entry 49 """ 50 data = kwargs.pop('data', '') 51 title = kwargs.pop('title', None) 52 if not title: 53 title = self.get_instance_title(instance) 54 55 timestamp = kwargs.pop('timestamp', timezone.now()) 56 return self.model.objects.create( 57 content_type=ContentType.objects.get_for_model(instance, for_concrete_model=False), 58 label=title, 59 action=action, 60 timestamp=timestamp, 61 data_json=json.dumps(data), 62 **kwargs, 63 ) 64 65 def get_for_model(self, model): 66 # Return empty queryset if the given object is not valid. 67 if not issubclass(model, models.Model): 68 return self.none() 69 70 ct = ContentType.objects.get_for_model(model) 71 72 return self.filter(content_type=ct) 73 74 def get_for_user(self, user_id): 75 return self.filter(user=user_id) 76 77 78class BaseLogEntry(models.Model): 79 content_type = models.ForeignKey( 80 ContentType, 81 models.SET_NULL, 82 verbose_name=_('content type'), 83 blank=True, null=True, 84 related_name='+', 85 ) 86 label = models.TextField() 87 88 action = models.CharField(max_length=255, blank=True, db_index=True) 89 data_json = models.TextField(blank=True) 90 timestamp = models.DateTimeField(verbose_name=_('timestamp (UTC)')) 91 92 user = models.ForeignKey( 93 settings.AUTH_USER_MODEL, 94 null=True, # Null if actioned by system 95 blank=True, 96 on_delete=models.DO_NOTHING, 97 db_constraint=False, 98 related_name='+', 99 ) 100 101 # Flags for additional context to the 'action' made by the user (or system). 102 content_changed = models.BooleanField(default=False, db_index=True) 103 deleted = models.BooleanField(default=False) 104 105 objects = BaseLogEntryManager() 106 107 action_registry = None 108 109 class Meta: 110 abstract = True 111 verbose_name = _('log entry') 112 verbose_name_plural = _('log entries') 113 ordering = ['-timestamp'] 114 115 def save(self, *args, **kwargs): 116 self.full_clean() 117 return super().save(*args, **kwargs) 118 119 def clean(self): 120 self.action_registry.scan_for_actions() 121 122 if self.action not in self.action_registry.actions: 123 raise ValidationError({'action': _("The log action '{}' has not been registered.").format(self.action)}) 124 125 def __str__(self): 126 return "LogEntry %d: '%s' on '%s'" % ( 127 self.pk, self.action, self.object_verbose_name() 128 ) 129 130 @cached_property 131 def user_display_name(self): 132 """ 133 Returns the display name of the associated user; 134 get_full_name if available and non-empty, otherwise get_username. 135 Defaults to 'system' when none is provided 136 """ 137 if self.user_id: 138 user = self.user 139 if user is None: 140 # User has been deleted. Using a string placeholder as the user id could be non-numeric 141 return _('user %(id)s (deleted)') % {'id': self.user_id} 142 143 try: 144 full_name = user.get_full_name().strip() 145 except AttributeError: 146 full_name = '' 147 return full_name or user.get_username() 148 149 else: 150 return _('system') 151 152 @cached_property 153 def data(self): 154 """ 155 Provides deserialized data 156 """ 157 if self.data_json: 158 return json.loads(self.data_json) 159 else: 160 return {} 161 162 @cached_property 163 def object_verbose_name(self): 164 model_class = self.content_type.model_class() 165 if model_class is None: 166 return self.content_type_id 167 168 return model_class._meta.verbose_name.title 169 170 def object_id(self): 171 raise NotImplementedError 172 173 def format_message(self): 174 return self.action_registry.format_message(self) 175 176 def format_comment(self): 177 return self.action_registry.format_comment(self) 178 179 @property 180 def comment(self): 181 return self.format_comment() 182