""" Author: Michel Peltriaux Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany Contact: michel.peltriaux@sgdnord.rlp.de Created on: 16.11.21 """ from compensation.models import CompensationState, Compensation, EcoAccount, CompensationAction from ema.models import Ema from intervention.models import Intervention from konova.management.commands.setup import BaseKonovaCommand from konova.models import Deadline, Geometry, Parcel, District from user.models import UserActionLogEntry, UserAction class Command(BaseKonovaCommand): help = "Checks the database' sanity and removes unused entries" def handle(self, *args, **options): try: self.sanitize_log_entries() self.sanitize_compensation_states() self.sanitize_actions() self.sanitize_deadlines() self.sanitize_geometries() self.sanitize_parcels_and_districts() except KeyboardInterrupt: self._break_line() exit(-1) def get_all_log_entries_ids(self, cls): """ Getter for all log entry ids of a model Args: cls (Intervention|Compensation|Ema|EcoAccount): The model class Returns: """ all_objects = cls.objects.all().prefetch_related( "log", ) all_ids = all_objects.exclude( log__isnull=True, ).values_list( "log__id", flat=True, ) return all_ids def sanitize_log_entries(self): """ Removes log entries which are not attached to any logs Returns: """ self._write_warning("=== Sanitize log entries ===") # Exclude created log entries from being cleaned, since they can be part of objects which do not have logs # Being in a log (or not) is essential for this cleanup all_log_entries = UserActionLogEntry.objects.all().exclude( action=UserAction.CREATED ) intervention_log_entries_ids = self.get_all_log_entries_ids(Intervention) attached_log_entries_id = intervention_log_entries_ids.union( self.get_all_log_entries_ids(Compensation), self.get_all_log_entries_ids(EcoAccount), self.get_all_log_entries_ids(Ema), ) unattached_log_entries = all_log_entries.exclude(id__in=attached_log_entries_id) num_entries = unattached_log_entries.count() if num_entries > 0: self._write_error(f"Found {num_entries} log entries not attached to anything. Delete now...") unattached_log_entries.delete() self._write_success("Log entries deleted.") else: self._write_success("No unattached log entries found.") self._break_line() def get_all_action_ids(self, cls): """ Getter for all action ids of a model Args: cls (Compensation|Ema|EcoAccount): The model class Returns: """ all_objects = cls.objects.all().prefetch_related( "actions", ) all_ids = all_objects.exclude( actions__isnull=True, ).values_list( "actions__id", flat=True, ) return all_ids def sanitize_actions(self): """ Removes actions which are not attached to any entries Returns: """ self._write_warning("=== Sanitize compensation actions ===") all_actions = CompensationAction.objects.all() compensation_action_ids = self.get_all_action_ids(Compensation) attached_action_ids = compensation_action_ids.union( self.get_all_action_ids(EcoAccount), self.get_all_action_ids(Ema), ) unattached_actions = all_actions.exclude(id__in=attached_action_ids) num_entries = unattached_actions.count() if num_entries > 0: self._write_error(f"Found {num_entries} actions not attached to anything. Delete now...") unattached_actions.delete() self._write_success("Actions deleted.") else: self._write_success("No unattached actions found.") self._break_line() def get_all_deadline_ids(self, cls): """ Getter for all deadline ids of a model Args: cls (Compensation|Ema|EcoAccount): The model class Returns: """ all_objects = cls.objects.all().prefetch_related( "deadlines", ) all_ids = all_objects.exclude( deadlines__isnull=True, ).values_list( "deadlines__id", flat=True, ) return all_ids def sanitize_deadlines(self): """ Removes deadlines which are not attached to any entries Returns: """ self._write_warning("=== Sanitize deadlines ===") all_deadlines = Deadline.objects.all() compensation_deadline_ids = self.get_all_deadline_ids(Compensation) attached_deadline_ids = compensation_deadline_ids.union( self.get_all_deadline_ids(EcoAccount), self.get_all_deadline_ids(Ema), ) unattached_deadlines = all_deadlines.exclude(id__in=attached_deadline_ids) num_entries = unattached_deadlines.count() if num_entries > 0: self._write_error(f"Found {num_entries} deadlines not attached to anything. Delete now...") unattached_deadlines.delete() self._write_success("Deadlines deleted.") else: self._write_success("No unattached deadlines found.") self._break_line() def get_all_geometry_ids(self, cls): """ Getter for all geometry ids of a model Args: cls (Intervention|Compensation|Ema|EcoAccount): The model class Returns: """ all_objects = cls.objects.all().prefetch_related( "geometry", ) all_ids = all_objects.exclude( geometry__isnull=True, ).values_list( "geometry__id", flat=True, ) return all_ids def sanitize_geometries(self): """ Removes geometries which are not attached to any entries Returns: """ self._write_warning("=== Sanitize geometries ===") all_geometries = Geometry.objects.all() compensation_geometry_ids = self.get_all_geometry_ids(Compensation) attached_geometry_ids = compensation_geometry_ids.union( self.get_all_geometry_ids(Intervention), self.get_all_geometry_ids(EcoAccount), self.get_all_geometry_ids(Ema), ) unattached_geometries = all_geometries.exclude(id__in=attached_geometry_ids) num_entries = unattached_geometries.count() if num_entries > 0: self._write_error(f"Found {num_entries} geometries not attached to anything. Delete now...") unattached_geometries.delete() self._write_success("Geometries deleted.") else: self._write_success("No unattached geometries found.") self._break_line() def get_all_state_ids(self, cls): """ Getter for all states (before and after) of a class Args: cls (): Returns: """ all_objects = cls.objects.all().prefetch_related( "before_states", "after_states", ) before_state_ids = all_objects.exclude( before_states__isnull=True, ).values_list( "before_states__id", flat=True, ) after_state_ids = all_objects.exclude( after_states__isnull=True, ).values_list( "after_states__id", flat=True, ) all_ids = after_state_ids.union(before_state_ids) return all_ids def sanitize_compensation_states(self): """ Removes unattached compensation states Returns: """ self._write_warning("=== Sanitize compensation states ===") all_states = CompensationState.objects.all() compensation_state_ids = self.get_all_state_ids(Compensation) account_state_ids = self.get_all_state_ids(EcoAccount) ema_state_ids = self.get_all_state_ids(Ema) attached_state_ids = compensation_state_ids.union(account_state_ids, ema_state_ids) unattached_states = all_states.exclude( id__in=attached_state_ids ) num_unattached_states = unattached_states.count() if num_unattached_states > 0: self._write_error(f"Found {num_unattached_states} unused compensation states. Delete now...") unattached_states.delete() self._write_success("Unused states deleted.") else: self._write_success("No unused states found.") self._break_line() def sanitize_parcels_and_districts(self): """ Removes unattached parcels and districts Returns: """ self._write_warning("=== Sanitize parcels and districts ===") unrelated_parcels = Parcel.objects.filter( geometries=None, ) num_unrelated_parcels = unrelated_parcels.count() if num_unrelated_parcels > 0: self._write_error(f"Found {num_unrelated_parcels} unrelated parcel entries. Delete now...") unrelated_parcels.delete() self._write_success("Unrelated parcels deleted.") else: self._write_success("No unrelated parcels found.") unrelated_districts = District.objects.filter( parcels=None, ) num_unrelated_districts = unrelated_districts.count() if num_unrelated_districts > 0: self._write_error(f"Found {num_unrelated_districts} unrelated district entries. Delete now...") unrelated_districts.delete() self._write_success("Unrelated districts deleted.") else: self._write_success("No unrelated districts found.") self._break_line()