mpeltriaux
62e1b046c3
* refactors command update_all_parcels into recalculate_parcels * fixes bug in command generate_report
138 lines
5.0 KiB
Python
138 lines
5.0 KiB
Python
"""
|
|
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 datetime import datetime
|
|
from io import BytesIO
|
|
|
|
from django.core.mail import EmailMessage
|
|
from django.utils import timezone
|
|
|
|
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()
|
|
if success == 1:
|
|
self._write_success(f"Mails with zip as attachment sent to {admin_mails}.")
|
|
else:
|
|
self._write_error(f"Something went wrong during mail sending. Returned sending code was '{success}'")
|