Refactoring
* splits intervention/models.py into subpackage
This commit is contained in:
257
intervention/models/intervention.py
Normal file
257
intervention/models/intervention.py
Normal file
@@ -0,0 +1,257 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user