@ -0,0 +1,137 @@
|
||||
"""
|
||||
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()
|
||||
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}'")
|
||||
Loading…
Reference in New Issue