Command

* adds new 'generate_report' command
   * generates TimeSpanReports for given conservation offices
   * zips reports into archive
   * sents archive to ADMINS mails

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}.")