mpeltriaux
f4541abf20
* refactors geometry field into GeoReferencedMixin, holding more handy methods and used in all models, formerly holding the geometry field * refactors backend admin configuration, so modified, deleted and created are not editable in the backend which also skips loading of all possible choices * fixes typo in sanitize_db command * introduces GeometryConflict model, holding a link between two geometries, where one overlaps the other * adds first (WIP) messages into detail views of ema and intervention for test purposes
269 lines
8.4 KiB
Python
269 lines
8.4 KiB
Python
"""
|
|
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
|
|
from user.models import UserActionLogEntry
|
|
|
|
|
|
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()
|
|
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 ===")
|
|
all_log_entries = UserActionLogEntry.objects.all()
|
|
|
|
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()
|
|
|