diff --git a/konova/management/commands/generate_report.py b/konova/management/commands/generate_report.py new file mode 100644 index 00000000..be0531ff --- /dev/null +++ b/konova/management/commands/generate_report.py @@ -0,0 +1,134 @@ +""" +Author: Michel Peltriaux +Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany +Contact: michel.peltriaux@sgdnord.rlp.de +Created on: 26.10.22 + +""" +import zipfile +from io import BytesIO + +from django.core.mail import EmailMessage +from django.utils import timezone +from django.utils.datetime_safe import datetime + +from analysis.utils.excel.excel import TempExcelFile +from analysis.utils.report import TimespanReport +from codelist.models import KonovaCode +from codelist.settings import CODELIST_CONSERVATION_OFFICE_ID +from konova.management.commands.setup import BaseKonovaCommand +from konova.sub_settings.django_settings import DEFAULT_FROM_EMAIL, ADMINS + + +class Command(BaseKonovaCommand): + help = "Generates reports for conservation offices" + + __from_key = "from" + __to_key = "to" + __for_key = "for" + + from_date = None + to_date = None + offices = None + + def add_arguments(self, parser): + try: + parser.add_argument(f"--{self.__from_key}", type=str, nargs="+") + parser.add_argument(f"--{self.__to_key}", type=str, nargs="+") + parser.add_argument(f"--{self.__for_key}", type=str, nargs="*") + except ValueError as e: + self._write_error(f"Argument error: {e}") + exit(-1) + + def handle(self, *args, **options): + self.__check_arguments(options) + in_memory_zipped_reports = self.generate_reports_in_memory_zipped() + self.send_mail_to_admins(in_memory_zipped_reports) + + def generate_reports_in_memory_zipped(self): + """ + Generates in-memory reports and zips into in-memory zip file. + + Returns: + + """ + memory_zip = BytesIO() + with zipfile.ZipFile(memory_zip, "w", compression=zipfile.ZIP_DEFLATED) as zf: + for office in self.offices: + self._write_warning(f" Process report for {office.long_name}...") + report = TimespanReport(office.id, self.from_date, self.to_date) + excel_file = TempExcelFile(report.excel_template_path, report.excel_map) + zf.writestr(zinfo_or_arcname=f"{office.long_name}_{self.from_date.date()}_{self.to_date.date()}.xlsx", data=excel_file.stream) + self._write_success(f"Reports generated for {self.offices.count()} offices and zipped.") + return memory_zip.getvalue() + + def __check_arguments(self, options): + """ + Checks given parameters for validity + + Args: + options (): + + Returns: + + """ + _from_value = options.get(self.__from_key, [None])[0] + _to_value = options.get(self.__to_key, [None])[0] + _for_value = options.get(self.__for_key, []) + + # Check ISO dates + try: + _from_date = timezone.make_aware(datetime.fromisoformat(_from_value)) + _to_date = timezone.make_aware(datetime.fromisoformat(_to_value)) + except Exception as e: + self._write_warning(f"One of the dates is not in ISO format (YYYY-MM-DD). {e}") + exit(-1) + + # Check conservation office IDs + _filter = { + "is_archived": False, + "is_leaf": True, + "code_lists__in": [CODELIST_CONSERVATION_OFFICE_ID], + } + offices = KonovaCode.objects.filter(**_filter) + if _for_value is not None and len(_for_value) != 0: + # Specifc offices requested + offices = offices.filter(short_name__in=_for_value) + all_requested_offices_exist = offices.count() == len(_for_value) + if not all_requested_offices_exist: + offices_short_name = set(offices.values_list("short_name", flat=True)) + missing_ids = [] + for val in _for_value: + if val not in offices_short_name: + missing_ids.append(val) + self._write_warning( + f"Unknown offices: {missing_ids}" + ) + exit(-1) + + self.offices = offices + self.from_date = _from_date + self.to_date = _to_date + + def send_mail_to_admins(self, attachment): + office_names = [office.long_name for office in self.offices] + admin_mails = [admin[1] for admin in ADMINS] + + date_from_date = self.from_date.date() + date_to_date = self.to_date.date() + + mail = EmailMessage( + subject="Konova reports generated", + body=f"Zipped reports attached from {date_from_date} to {date_to_date} for {office_names}.", + from_email=DEFAULT_FROM_EMAIL, + to=admin_mails, + attachments=[ + ( + f"reports_{date_from_date}_{date_to_date}.zip", + attachment, + "application/zip" + ) + ] + ) + success = mail.send() + self._write_success(f"Mails with zip as attachment sent to {admin_mails}.")