mpeltriaux
02ce78551c
* introduces new command for mass deleting entries based on specific pattern in identifier
* command guides user through parameter collecting process
113 lines
4.4 KiB
Python
113 lines
4.4 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_IDENTIFIER_ILIKE_PARAM = None
|
|
_DELETE_USER = None
|
|
|
|
def handle(self, *args, **options):
|
|
self._get_delete_user()
|
|
self._get_object_type()
|
|
self._get_identifier_ilike_param()
|
|
self._get_delete_comment()
|
|
self._show_config()
|
|
entries = self._collect_entries()
|
|
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_identifier_ilike_param(self):
|
|
self._DELETE_IDENTIFIER_ILIKE_PARAM = input("Use this substring to search in identifiers: ")
|
|
|
|
def _get_delete_comment(self):
|
|
self._DELETE_COMMENT = input("Comment to store in delete action: ")
|
|
|
|
def _show_config(self):
|
|
assert self._DELETE_CLS is not None
|
|
assert self._DELETE_IDENTIFIER_ILIKE_PARAM 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" Identifier pattern: '*{self._DELETE_IDENTIFIER_ILIKE_PARAM}*'")
|
|
self._write_warning(f" Comment for delete action: '{self._DELETE_COMMENT}'")
|
|
self._write_warning(f" Delete performing user: '{self._DELETE_USER}'")
|
|
|
|
def _collect_entries(self):
|
|
return self._DELETE_CLS.objects.filter(
|
|
deleted__isnull=True,
|
|
identifier__contains=self._DELETE_IDENTIFIER_ILIKE_PARAM
|
|
)
|
|
|
|
def _process_delete(self, entries: QuerySet):
|
|
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 unknown '{delete_input}'. Abort.")
|
|
exit(-1)
|
|
else:
|
|
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.") |