""" Author: Michel Peltriaux Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany Contact: michel.peltriaux@sgdnord.rlp.de Created on: 15.11.21 """ from django.contrib.auth.models import User from django.db import models, transaction from django.db.models import QuerySet from django.utils.translation import gettext_lazy as _ from intervention.managers import InterventionManager from intervention.models.legal import Legal from intervention.models.responsibility import Responsibility from intervention.models.revocation import RevocationDocument from intervention.utils.quality import InterventionQualityChecker from konova.models import generate_document_file_upload_path, AbstractDocument, Geometry, BaseObject, ShareableObject, \ RecordableObject, CheckableObject from konova.settings import LANIS_LINK_TEMPLATE, LANIS_ZOOM_LUT, DEFAULT_SRID_RLP from user.models import UserAction, UserActionLogEntry class Intervention(BaseObject, ShareableObject, RecordableObject, CheckableObject): """ Interventions are e.g. construction sites where nature used to be. """ responsible = models.OneToOneField( Responsibility, on_delete=models.SET_NULL, null=True, blank=True, help_text="Holds data on responsible organizations ('Zulassungsbehörde', 'Eintragungsstelle')" ) legal = models.OneToOneField( Legal, on_delete=models.SET_NULL, null=True, blank=True, help_text="Holds data on legal dates or law" ) geometry = models.ForeignKey(Geometry, null=True, blank=True, on_delete=models.SET_NULL) objects = InterventionManager() def __str__(self): return "{} ({})".format(self.identifier, self.title) def save(self, *args, **kwargs): """ Custom save functionality Performs some pre-save checks: 1. Checking for existing identifiers Args: *args (): **kwargs (): Returns: """ if self.identifier is None or len(self.identifier) == 0: # No identifier given by the user self.identifier = self.generate_new_identifier() # Before saving, make sure the given identifier is not used in the meanwhile while Intervention.objects.filter(identifier=self.identifier).exclude(id=self.id).exists(): self.identifier = self.generate_new_identifier() super().save(*args, **kwargs) def delete(self, using=None, keep_parents=False): to_delete = [ self.legal, self.responsible, self.geometry, self.log.all() ] for entry in to_delete: try: entry.delete() except AttributeError: pass super().delete(using, keep_parents) def quality_check(self) -> InterventionQualityChecker: """ Quality check Returns: ret_msgs (list): Holds error messages """ checker = InterventionQualityChecker(obj=self) checker.run_check() return checker def get_LANIS_link(self) -> str: """ Generates a link for LANIS depending on the geometry Returns: """ try: geom = self.geometry.geom.transform(DEFAULT_SRID_RLP, clone=True) x = geom.centroid.x y = geom.centroid.y area = int(geom.envelope.area) z_l = 16 for k_area, v_zoom in LANIS_ZOOM_LUT.items(): if k_area < area: z_l = v_zoom break zoom_lvl = z_l except AttributeError: # If no geometry has been added, yet. x = 1 y = 1 zoom_lvl = 6 return LANIS_LINK_TEMPLATE.format( zoom_lvl, x, y, ) def get_documents(self) -> (QuerySet, QuerySet): """ Getter for all documents of an intervention Returns: revoc_docs (QuerySet): The queryset of a revocation document regular_docs (QuerySet): The queryset of regular other documents """ revoc_docs = RevocationDocument.objects.filter( instance__in=self.legal.revocations.all() ) regular_docs = InterventionDocument.objects.filter( instance=self ) return revoc_docs, regular_docs def toggle_recorded(self, user: User): """ Toggle the recorded state For interventions the recorded action needs to be added to their compensation objects as well Args: user (User): The performing user Returns: """ log_entry = super().toggle_recorded(user) # Add this action to the linked compensation logs as well comps = self.compensations.all() for comp in comps: comp.log.add(log_entry) def toggle_checked(self, user: User) -> UserActionLogEntry: """ Toggle the checked state For interventions the checked action needs to be added to their compensation objects as well Args: user (User): The performing user Returns: """ log_entry = super().toggle_checked(user) # Leave if the log_entry is None (means "unchecked") if log_entry is None: return # Add this action to the linked compensation logs as well comps = self.compensations.all() for comp in comps: comp.log.add(log_entry) def add_payment(self, form): """ Adds a new payment to the intervention Args: form (NewPaymentForm): The form holding the data Returns: """ from compensation.models import Payment form_data = form.cleaned_data user = form.user with transaction.atomic(): created_action = UserActionLogEntry.objects.create( user=user, action=UserAction.CREATED, ) edited_action = UserActionLogEntry.objects.create( user=user, action=UserAction.EDITED, comment=_("Added payment"), ) pay = Payment.objects.create( created=created_action, amount=form_data.get("amount", -1), due_on=form_data.get("due", None), comment=form_data.get("comment", None), intervention=self, ) self.log.add(edited_action) self.modified = edited_action self.save() return pay class InterventionDocument(AbstractDocument): """ Specializes document upload for an intervention with certain path """ instance = models.ForeignKey( Intervention, on_delete=models.CASCADE, related_name="documents", ) file = models.FileField( upload_to=generate_document_file_upload_path, max_length=1000, ) def delete(self, *args, **kwargs): """ Custom delete functionality for InterventionDocuments. Removes the folder from the file system if there are no further documents for this entry. Args: *args (): **kwargs (): Returns: """ revoc_docs, other_intervention_docs = self.instance.get_documents() folder_path = None if revoc_docs.count() == 0 and other_intervention_docs.count() == 1: # The only file left for this intervention is the one which is currently processed and will be deleted # Make sure that the intervention folder itself is deleted as well, not only the file # Therefore take the folder path from the file path folder_path = self.file.path.split("/")[:-1] folder_path = "/".join(folder_path) # Remove the file itself super().delete(*args, **kwargs) # If a folder path has been set, we need to delete the whole folder! if folder_path is not None: try: shutil.rmtree(folder_path) pass except FileNotFoundError: # Folder seems to be missing already... pass