mpeltriaux
eb975cd3c5
* changes trigger for sending data to EGON: on each new payment, edited payment or deleted payment action, the data will be sent to EGON instead only once on "recording"
403 lines
13 KiB
Python
403 lines
13 KiB
Python
"""
|
|
Author: Michel Peltriaux
|
|
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
|
|
Contact: michel.peltriaux@sgdnord.rlp.de
|
|
Created on: 15.11.21
|
|
|
|
"""
|
|
import shutil
|
|
|
|
from django.contrib import messages
|
|
from django.core.exceptions import ObjectDoesNotExist
|
|
from django.db.models.fields.files import FieldFile
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
|
|
from intervention.tasks import celery_export_to_egon
|
|
from user.models import User
|
|
from django.db import models, transaction
|
|
from django.db.models import QuerySet
|
|
from django.http import HttpRequest
|
|
|
|
from intervention.managers import InterventionManager
|
|
from intervention.models.legal import Legal
|
|
from intervention.models.responsibility import Responsibility
|
|
from intervention.models.revocation import RevocationDocument, Revocation
|
|
from intervention.utils.quality import InterventionQualityChecker
|
|
from konova.models import generate_document_file_upload_path, AbstractDocument, BaseObject, \
|
|
ShareableObjectMixin, \
|
|
RecordableObjectMixin, CheckableObjectMixin, GeoReferencedMixin
|
|
from konova.settings import LANIS_LINK_TEMPLATE, LANIS_ZOOM_LUT, DEFAULT_SRID_RLP
|
|
from konova.utils.message_templates import DATA_UNSHARED_EXPLANATION, DOCUMENT_REMOVED_TEMPLATE, \
|
|
PAYMENT_REMOVED, PAYMENT_ADDED, REVOCATION_REMOVED, INTERVENTION_HAS_REVOCATIONS_TEMPLATE
|
|
from user.models import UserActionLogEntry
|
|
|
|
|
|
class Intervention(BaseObject, ShareableObjectMixin, RecordableObjectMixin, CheckableObjectMixin, GeoReferencedMixin):
|
|
"""
|
|
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"
|
|
)
|
|
|
|
objects = InterventionManager()
|
|
|
|
def __str__(self):
|
|
return f"{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_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 set_unchecked(self):
|
|
super().set_unchecked()
|
|
|
|
def set_checked(self, user: User) -> UserActionLogEntry:
|
|
log_entry = super().set_checked(user)
|
|
if log_entry is not None:
|
|
self.add_log_entry_to_compensations(log_entry)
|
|
return log_entry
|
|
|
|
def set_unrecorded(self, user: User):
|
|
log_entry = super().set_unrecorded(user)
|
|
self.add_log_entry_to_compensations(log_entry)
|
|
return log_entry
|
|
|
|
def send_data_to_egon(self):
|
|
""" Performs the export to rabbitmq of this intervention's data
|
|
|
|
FOLLOWING BACKWARDS COMPATIBILITY LOGIC
|
|
|
|
Returns:
|
|
|
|
"""
|
|
celery_export_to_egon.delay(self.id)
|
|
|
|
def set_recorded(self, user: User) -> UserActionLogEntry:
|
|
log_entry = super().set_recorded(user)
|
|
self.add_log_entry_to_compensations(log_entry)
|
|
return log_entry
|
|
|
|
def add_log_entry_to_compensations(self, log_entry: UserActionLogEntry):
|
|
""" Adds the log entry to related compensations
|
|
|
|
Args:
|
|
log_entry (UserActionLogEntry): The log entry
|
|
|
|
Returns:
|
|
|
|
"""
|
|
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.get_created_action(user)
|
|
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.mark_as_edited(user, form.request, edit_comment=PAYMENT_ADDED)
|
|
|
|
self.send_data_to_egon()
|
|
return pay
|
|
|
|
def add_revocation(self, form):
|
|
""" Adds a new revocation to the intervention
|
|
|
|
Args:
|
|
form (NewRevocationModalForm): The form holding the data
|
|
|
|
Returns:
|
|
|
|
"""
|
|
form_data = form.cleaned_data
|
|
user = form.user
|
|
with transaction.atomic():
|
|
created_action = UserActionLogEntry.get_created_action(user)
|
|
|
|
revocation = Revocation.objects.create(
|
|
date=form_data["date"],
|
|
legal=self.legal,
|
|
comment=form_data["comment"],
|
|
created=created_action,
|
|
)
|
|
|
|
if form_data["file"]:
|
|
RevocationDocument.objects.create(
|
|
title="revocation_of_{}".format(self.identifier),
|
|
date_of_creation=form_data["date"],
|
|
comment=form_data["comment"],
|
|
file=form_data["file"],
|
|
instance=revocation
|
|
)
|
|
return revocation
|
|
|
|
def edit_revocation(self, form):
|
|
""" Updates a revocation of the intervention
|
|
|
|
Args:
|
|
form (EditRevocationModalForm): The form holding the data
|
|
|
|
Returns:
|
|
|
|
"""
|
|
form_data = form.cleaned_data
|
|
file = form_data.get("file", None)
|
|
|
|
revocation = form.revocation
|
|
revocation.date = form_data.get("date", None)
|
|
revocation.comment = form_data.get("comment", None)
|
|
|
|
with transaction.atomic():
|
|
try:
|
|
revocation.document.date_of_creation = revocation.date
|
|
revocation.document.comment = revocation.comment
|
|
if not isinstance(file, FieldFile):
|
|
revocation.document.replace_file(file)
|
|
revocation.document.save()
|
|
except ObjectDoesNotExist:
|
|
revocation.document = RevocationDocument.objects.create(
|
|
title="revocation_of_{}".format(self.identifier),
|
|
date_of_creation=revocation.date,
|
|
comment=revocation.comment,
|
|
file=file,
|
|
instance=revocation
|
|
)
|
|
revocation.save()
|
|
|
|
return revocation
|
|
|
|
def remove_revocation(self, form):
|
|
""" Removes a revocation from the intervention
|
|
|
|
Args:
|
|
form (RemoveRevocationModalForm): The form holding all relevant data
|
|
|
|
Returns:
|
|
|
|
"""
|
|
revocation = form.revocation
|
|
user = form.user
|
|
with transaction.atomic():
|
|
revocation.delete()
|
|
self.mark_as_edited(user, request=form.request, edit_comment=REVOCATION_REMOVED)
|
|
|
|
def mark_as_edited(self, performing_user: User, request: HttpRequest = None, edit_comment: str = None, reset_recorded: bool = True):
|
|
""" In case the object or a related object changed, internal processes need to be started, such as
|
|
unrecord and uncheck
|
|
|
|
Args:
|
|
performing_user (User): The user which performed the editing action
|
|
request (HttpRequest): The used request for this action
|
|
edit_comment (str): Additional comment for the log entry
|
|
reset_recorded (bool): Whether the record-state of the object should be reset
|
|
|
|
Returns:
|
|
|
|
"""
|
|
action = super().mark_as_edited(performing_user, edit_comment=edit_comment)
|
|
if reset_recorded:
|
|
self.unrecord(performing_user, request)
|
|
if self.checked:
|
|
self.set_unchecked()
|
|
return action
|
|
|
|
def set_status_messages(self, request: HttpRequest):
|
|
""" Setter for different information that need to be rendered
|
|
|
|
Adds messages to the given HttpRequest
|
|
|
|
Args:
|
|
request (HttpRequest): The incoming request
|
|
|
|
Returns:
|
|
request (HttpRequest): The modified request
|
|
"""
|
|
# Inform user about revocation
|
|
if self.legal.revocations.exists():
|
|
messages.error(
|
|
request,
|
|
INTERVENTION_HAS_REVOCATIONS_TEMPLATE.format(self.legal.revocations.count()),
|
|
extra_tags="danger",
|
|
)
|
|
if not self.is_shared_with(request.user):
|
|
messages.info(request, DATA_UNSHARED_EXPLANATION)
|
|
request = self.set_geometry_conflict_message(request)
|
|
return request
|
|
|
|
def is_ready_for_publish(self) -> bool:
|
|
""" Checks whether the data passes all constraints for being publishable
|
|
|
|
Returns:
|
|
is_ready (bool) : True|False
|
|
"""
|
|
now_date = timezone.now().date()
|
|
binding_date = self.legal.binding_date
|
|
is_binding_date_ready = binding_date is not None and binding_date <= now_date
|
|
is_recorded = self.recorded is not None
|
|
is_free_of_revocations = not self.legal.revocations.exists()
|
|
is_ready = is_binding_date_ready \
|
|
and is_recorded \
|
|
and is_free_of_revocations
|
|
return is_ready
|
|
|
|
def get_share_link(self):
|
|
""" Returns the share url for the object
|
|
|
|
Returns:
|
|
|
|
"""
|
|
return reverse("intervention:share", args=(self.id, self.access_token))
|
|
|
|
def remove_payment(self, form):
|
|
""" Removes a Payment from the intervention
|
|
|
|
Args:
|
|
form (RemovePaymentModalForm): The form holding all relevant data
|
|
|
|
Returns:
|
|
|
|
"""
|
|
payment = form.payment
|
|
user = form.user
|
|
with transaction.atomic():
|
|
payment.delete()
|
|
self.mark_as_edited(user, request=form.request, edit_comment=PAYMENT_REMOVED)
|
|
self.send_data_to_egon()
|
|
|
|
|
|
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, user=None, *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)
|
|
|
|
if user:
|
|
self.instance.mark_as_edited(user, edit_comment=DOCUMENT_REMOVED_TEMPLATE.format(self.title))
|
|
|
|
# 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)
|
|
except FileNotFoundError:
|
|
# Folder seems to be missing already...
|
|
pass
|