212 lines
7.6 KiB
Python
212 lines
7.6 KiB
Python
"""
|
|
Author: Michel Peltriaux
|
|
Created on: 16.07.26
|
|
|
|
"""
|
|
from django.db.models import QuerySet
|
|
|
|
from compensation.models import Compensation, EcoAccount
|
|
from ema.models import Ema
|
|
from intervention.models import Intervention
|
|
from konova.management.commands.setup import BaseKonovaCommand
|
|
from user.models import User
|
|
|
|
|
|
class Command(BaseKonovaCommand):
|
|
help = "Mass-deletes entries according to user input"
|
|
|
|
_DELETE_CLS = None
|
|
_DELETE_COMMENT = None
|
|
_DELETE_USER = None
|
|
_DELETE_FROM_DB = False
|
|
|
|
def handle(self, *args, **options):
|
|
self._get_delete_user()
|
|
self._get_object_type()
|
|
filter = self._get_filter()
|
|
self._get_delete_comment()
|
|
self._show_config(filter)
|
|
entries = self._collect_entries(filter)
|
|
self._process_delete(entries)
|
|
|
|
def _get_delete_user(self):
|
|
self._DELETE_USER = User.objects.filter(
|
|
is_superuser=True,
|
|
is_staff=True
|
|
).order_by(
|
|
"id"
|
|
).first()
|
|
|
|
if not self._DELETE_USER:
|
|
self._write_error("No admin-staff user could be found. Please check the admin users!")
|
|
|
|
def _get_object_type(self):
|
|
object_types = {
|
|
"eiv": Intervention,
|
|
"kom": Compensation,
|
|
"oek": EcoAccount,
|
|
"ema": Ema,
|
|
}
|
|
object_types_str = "|".join(object_types.keys())
|
|
object_type_select = input(f"Which object type shall be processed? ({object_types_str}): ").lower()
|
|
try:
|
|
self._DELETE_CLS = object_types[object_type_select]
|
|
if self._DELETE_CLS == EcoAccount:
|
|
oek_deduction_remove_understood = input(f"Please be aware that deleting OEKs will result in TOTAL REMOVING of their deductions without any chance to restore them. Do you want to continue? (y|n): ")
|
|
oek_deduction_remove_understood = oek_deduction_remove_understood.lower() == "y"
|
|
if not oek_deduction_remove_understood:
|
|
self._write_error("Removing of OEK canceled. Abort.")
|
|
exit(-1)
|
|
|
|
except KeyError:
|
|
self._write_error(f"'{object_type_select}' not a valid option of {object_types_str}. Abort.")
|
|
exit(-1)
|
|
|
|
def _get_filter(self):
|
|
filter_attribute = input("What do you want to filter entries by? (identifier|created_by_user): ").lower()
|
|
filter = {}
|
|
try:
|
|
assert filter_attribute in ["identifier", "created_by_user"]
|
|
if filter_attribute == "identifier":
|
|
filter_attribute = "identifier__contains"
|
|
elif filter_attribute == "created_by_user":
|
|
filter_attribute = "created__user__username"
|
|
except AssertionError:
|
|
self._write_error(f"'{filter_attribute}' unknown. Abort")
|
|
exit(-1)
|
|
|
|
filter_val = input("Use this value to search for: ")
|
|
try:
|
|
assert len(filter_val) > 0
|
|
except AssertionError:
|
|
self._write_error(f"'{filter_val}' is empty. Abort.")
|
|
exit(-1)
|
|
|
|
filter[filter_attribute] = filter_val
|
|
return filter
|
|
|
|
def _get_delete_comment(self):
|
|
self._DELETE_COMMENT = input("Comment to store in delete action: ")
|
|
|
|
def _show_config(self, filter: dict):
|
|
assert self._DELETE_CLS is not None
|
|
assert self._DELETE_COMMENT is not None
|
|
|
|
self._write_warning("You are about to delete entries with:")
|
|
self._write_warning(f" Object type: {self._DELETE_CLS.__name__}")
|
|
self._write_warning(f" Filter by: '{filter}'")
|
|
self._write_warning(f" Delete performing user: '{self._DELETE_USER}'")
|
|
|
|
def _collect_entries(self, filter: dict):
|
|
return self._DELETE_CLS.objects.filter(
|
|
deleted__isnull=True
|
|
).filter(
|
|
**filter
|
|
)
|
|
|
|
def _process_delete(self, entries: QuerySet):
|
|
""" Wrapping delete method. Forces user to decide whether entries will be marked as deleted or shall
|
|
be removed completely from database
|
|
|
|
Args:
|
|
entries (QuerySet): Entries to be processed
|
|
|
|
Returns:
|
|
|
|
"""
|
|
delete_type = input("Do you want 1) mark entries as deleted or 2) remove them from the database? (1|2): ")
|
|
try:
|
|
delete_type = int(delete_type)
|
|
assert delete_type in (1, 2)
|
|
self._DELETE_FROM_DB = delete_type == 2
|
|
except (ValueError, AssertionError) as e:
|
|
self._write_error(f"Entered unknown '{delete_type}'. Abort.")
|
|
exit(-1)
|
|
if self._DELETE_FROM_DB:
|
|
self._process_delete_from_db(entries)
|
|
else:
|
|
self._process_mark_as_deleted(entries)
|
|
|
|
|
|
def _process_mark_as_deleted(self, entries: QuerySet):
|
|
""" Method do mark chosen entries as deleted
|
|
|
|
Args:
|
|
entries (QuerySet): Entries to be processed
|
|
|
|
Returns:
|
|
|
|
"""
|
|
self._write_warning(f" Comment for delete action: '{self._DELETE_COMMENT}'")
|
|
|
|
self._check_entries_sample(entries)
|
|
|
|
entries_count = entries.count()
|
|
i = 0
|
|
for element in entries:
|
|
if i % 10 == 0:
|
|
self._write_warning(f" Deleted {i}/{entries_count}")
|
|
element.mark_as_deleted(user=self._DELETE_USER, send_mail=False, comment=self._DELETE_COMMENT)
|
|
i += 1
|
|
self._write_warning(f" Deleted {entries_count}/{entries_count}")
|
|
self._write_success("Entries deleted. Bye.")
|
|
|
|
def _process_delete_from_db(self, entries: QuerySet):
|
|
""" Method do REMOVE entries from the database
|
|
|
|
Args:
|
|
entries (QuerySet): Entries to be processed
|
|
|
|
Returns:
|
|
|
|
"""
|
|
self._check_entries_sample(entries)
|
|
|
|
entries_count = entries.count()
|
|
input_num_entries = input(f"THIS IS THE FINAL WARNING! {entries_count} ENTRIES WILL BE REMOVED FROM THE DATABASE! ENTER THE NUMBER OF ENTRIES TO BE DELETED TO CONTINUE: ")
|
|
try:
|
|
input_num_entries = int(input_num_entries)
|
|
except ValueError:
|
|
self._write_error(f"'{input_num_entries}' is not an integer. Abort.")
|
|
exit(-1)
|
|
if entries_count == input_num_entries:
|
|
i = 0
|
|
for element in entries:
|
|
if i % 10 == 0:
|
|
self._write_warning(f" Deleted {i}/{entries_count}")
|
|
element.delete() # call individual delete routine to enforce custom delete logic (not called on bulk deleting!)
|
|
i += 1
|
|
|
|
self._write_warning(f" Deleted {entries_count}/{entries_count}")
|
|
self._write_success("Entries deleted. Bye.")
|
|
else:
|
|
self._write_error(f"'{input_num_entries}' was not correct. Abort.")
|
|
exit(-1)
|
|
|
|
def _check_entries_sample(self, entries):
|
|
""" Shows a sample from the entries on the terminal, so the user can decide whether the results are as expected
|
|
|
|
Args:
|
|
entries (QuerySet): Entries to be processed
|
|
|
|
Returns:
|
|
|
|
"""
|
|
|
|
entries_count = entries.count()
|
|
if entries_count == 0:
|
|
self._write_success(f"Found {entries_count} matches to delete. Nothing to do here. Bye.")
|
|
exit(0)
|
|
|
|
self._write_warning(
|
|
f"Found {entries_count} matches to delete. Please double-check whether these samples from the results are what you expected: ")
|
|
first_three_elements = entries[:3]
|
|
for element in first_three_elements:
|
|
self._write_warning(f" {element.identifier} ({element.id})")
|
|
|
|
delete_input = input("Are you ready to delete them now? (y|n): ").lower()
|
|
delete_now = delete_input == "y"
|
|
if not delete_now:
|
|
self._write_error(f"Entered '{delete_input}'. Abort.")
|
|
exit(-1)
|