mpeltriaux
fa01733815
* adds support for POST of new compensations * adds shared_users property to BaseObject and Compensation to simplify fetching of shared users (Compensation inherits from intervention) * extends compensation admin index * modifies compensation manager which led to invisibility of deleted entries in the admin backend * fixes bug in sanitize_db.py where CREATED useractions would be removed if they are not found on any log but still are used on the .created attribute of the objects
359 lines
11 KiB
Python
359 lines
11 KiB
Python
"""
|
|
Author: Michel Peltriaux
|
|
Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany
|
|
Contact: michel.peltriaux@sgdnord.rlp.de
|
|
Created on: 16.11.21
|
|
|
|
"""
|
|
import shutil
|
|
|
|
from django.contrib import messages
|
|
from user.models import User
|
|
from django.db import models, transaction
|
|
from django.db.models import QuerySet, Sum
|
|
from django.http import HttpRequest
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from compensation.managers import CompensationManager
|
|
from compensation.models import CompensationState, CompensationAction
|
|
from compensation.utils.quality import CompensationQualityChecker
|
|
from konova.models import BaseObject, AbstractDocument, Deadline, generate_document_file_upload_path, \
|
|
GeoReferencedMixin
|
|
from konova.settings import DEFAULT_SRID_RLP, LANIS_LINK_TEMPLATE
|
|
from konova.utils.message_templates import DATA_UNSHARED_EXPLANATION
|
|
from user.models import UserActionLogEntry
|
|
|
|
|
|
class AbstractCompensation(BaseObject, GeoReferencedMixin):
|
|
"""
|
|
Abstract compensation model which holds basic attributes, shared by subclasses like the regular Compensation,
|
|
EMA or EcoAccount.
|
|
|
|
"""
|
|
responsible = models.OneToOneField(
|
|
"intervention.Responsibility",
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
blank=True,
|
|
help_text="Holds data on responsible organizations ('Zulassungsbehörde', 'Eintragungsstelle') and handler",
|
|
)
|
|
|
|
before_states = models.ManyToManyField(CompensationState, blank=True, related_name='+', help_text="Refers to 'Ausgangszustand Biotop'")
|
|
after_states = models.ManyToManyField(CompensationState, blank=True, related_name='+', help_text="Refers to 'Zielzustand Biotop'")
|
|
actions = models.ManyToManyField(CompensationAction, blank=True, help_text="Refers to 'Maßnahmen'")
|
|
|
|
deadlines = models.ManyToManyField("konova.Deadline", blank=True, related_name="+")
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def add_deadline(self, form) -> Deadline:
|
|
""" Adds a new deadline to the abstract compensation
|
|
|
|
Args:
|
|
form (NewDeadlineModalForm): The form holding all relevant data
|
|
|
|
Returns:
|
|
|
|
"""
|
|
form_data = form.cleaned_data
|
|
user = form.user
|
|
with transaction.atomic():
|
|
created_action = UserActionLogEntry.get_created_action(user)
|
|
edited_action = UserActionLogEntry.get_edited_action(user, _("Added deadline"))
|
|
|
|
deadline = Deadline.objects.create(
|
|
type=form_data["type"],
|
|
date=form_data["date"],
|
|
comment=form_data["comment"],
|
|
created=created_action,
|
|
)
|
|
|
|
self.modified = edited_action
|
|
self.save()
|
|
self.log.add(edited_action)
|
|
self.deadlines.add(deadline)
|
|
return deadline
|
|
|
|
def add_action(self, form) -> CompensationAction:
|
|
""" Adds a new action to the compensation
|
|
|
|
Args:
|
|
form (NewActionModalForm): The form holding all relevant data
|
|
|
|
Returns:
|
|
|
|
"""
|
|
form_data = form.cleaned_data
|
|
user = form.user
|
|
with transaction.atomic():
|
|
user_action = UserActionLogEntry.get_created_action(user)
|
|
comp_action = CompensationAction.objects.create(
|
|
action_type=form_data["action_type"],
|
|
amount=form_data["amount"],
|
|
unit=form_data["unit"],
|
|
comment=form_data["comment"],
|
|
created=user_action,
|
|
)
|
|
self.actions.add(comp_action)
|
|
return comp_action
|
|
|
|
def add_state(self, form, is_before_state: bool) -> CompensationState:
|
|
""" Adds a new compensation state to the compensation
|
|
|
|
Args:
|
|
form (NewStateModalForm): The form, holding all relevant data
|
|
is_before_state (bool): Whether this is a new before_state or after_state
|
|
|
|
Returns:
|
|
|
|
"""
|
|
form_data = form.cleaned_data
|
|
with transaction.atomic():
|
|
state = CompensationState.objects.create(
|
|
biotope_type=form_data["biotope_type"],
|
|
surface=form_data["surface"],
|
|
)
|
|
if is_before_state:
|
|
self.before_states.add(state)
|
|
else:
|
|
self.after_states.add(state)
|
|
return state
|
|
|
|
def get_surface_after_states(self) -> float:
|
|
""" Calculates the compensation's/account's surface
|
|
|
|
Returns:
|
|
sum_surface (float)
|
|
"""
|
|
return self._calc_surface(self.after_states.all())
|
|
|
|
def get_surface_before_states(self) -> float:
|
|
""" Calculates the compensation's/account's surface
|
|
|
|
Returns:
|
|
sum_surface (float)
|
|
"""
|
|
return self._calc_surface(self.before_states.all())
|
|
|
|
def _calc_surface(self, qs: QuerySet):
|
|
""" Calculates the surface sum of a given queryset
|
|
|
|
Args:
|
|
qs (QuerySet): The queryset containing CompensationState entries
|
|
|
|
Returns:
|
|
|
|
"""
|
|
return qs.aggregate(Sum("surface"))["surface__sum"] or 0
|
|
|
|
def quality_check(self) -> CompensationQualityChecker:
|
|
""" Performs data quality check
|
|
|
|
Returns:
|
|
checker (CompensationQualityChecker): Holds validity data and error messages
|
|
"""
|
|
checker = CompensationQualityChecker(self)
|
|
checker.run_check()
|
|
return checker
|
|
|
|
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
|
|
"""
|
|
if not self.is_shared_with(request.user):
|
|
messages.info(request, DATA_UNSHARED_EXPLANATION)
|
|
request = self.set_geometry_conflict_message(request)
|
|
return request
|
|
|
|
|
|
class CEFMixin(models.Model):
|
|
""" Provides CEF flag as Mixin
|
|
|
|
"""
|
|
is_cef = models.BooleanField(
|
|
blank=True,
|
|
null=True,
|
|
default=False,
|
|
help_text="Flag if compensation is a 'CEF-Maßnahme'"
|
|
)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
|
|
class CoherenceMixin(models.Model):
|
|
""" Provides coherence keeping flag as Mixin
|
|
|
|
"""
|
|
is_coherence_keeping = models.BooleanField(
|
|
blank=True,
|
|
null=True,
|
|
default=False,
|
|
help_text="Flag if compensation is a 'Kohärenzsicherung'"
|
|
)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
|
|
class Compensation(AbstractCompensation, CEFMixin, CoherenceMixin):
|
|
"""
|
|
Regular compensation, linked to an intervention
|
|
"""
|
|
intervention = models.ForeignKey(
|
|
"intervention.Intervention",
|
|
on_delete=models.CASCADE,
|
|
null=True,
|
|
blank=True,
|
|
related_name='compensations'
|
|
)
|
|
|
|
objects = CompensationManager()
|
|
|
|
def __str__(self):
|
|
return "{}".format(self.identifier)
|
|
|
|
def save(self, *args, **kwargs):
|
|
if self.identifier is None or len(self.identifier) == 0:
|
|
# Create new identifier is none was given
|
|
self.identifier = self.generate_new_identifier()
|
|
|
|
# Before saving, make sure a given identifier has not been taken already in the meanwhile
|
|
while Compensation.objects.filter(identifier=self.identifier).exclude(id=self.id).exists():
|
|
self.identifier = self.generate_new_identifier()
|
|
super().save(*args, **kwargs)
|
|
|
|
def is_shared_with(self, user: User):
|
|
""" Access check
|
|
|
|
Checks whether a given user has access to this object
|
|
|
|
Args:
|
|
user (User): The user to be checked
|
|
|
|
Returns:
|
|
|
|
"""
|
|
# Compensations inherit their shared state from the interventions
|
|
return self.intervention.is_shared_with(user)
|
|
|
|
@property
|
|
def shared_users(self) -> QuerySet:
|
|
""" Shortcut for fetching the users which have shared access on this object
|
|
|
|
Returns:
|
|
users (QuerySet)
|
|
"""
|
|
return self.intervention.users.all()
|
|
|
|
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
|
|
zoom_lvl = 16
|
|
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:
|
|
""" Getter for all documents of a compensation
|
|
|
|
Returns:
|
|
docs (QuerySet): The queryset of all documents
|
|
"""
|
|
docs = CompensationDocument.objects.filter(
|
|
instance=self
|
|
)
|
|
return docs
|
|
|
|
def mark_as_edited(self, user: User, request: HttpRequest = None, edit_comment: str = None):
|
|
""" Performs internal logic for setting the recordedd/checked state of the related intervention
|
|
|
|
Args:
|
|
user (User): The performing user
|
|
request (HttpRequest): The performing request
|
|
|
|
Returns:
|
|
|
|
"""
|
|
self.intervention.mark_as_edited(user, request, edit_comment)
|
|
|
|
def is_ready_for_publish(self) -> bool:
|
|
""" Not inherited by RecordableObjectMixin
|
|
|
|
Simplifies same usage for compensations as for other datatypes
|
|
|
|
Returns:
|
|
is_ready (bool): True|False
|
|
"""
|
|
return self.intervention.is_ready_for_publish()
|
|
|
|
|
|
class CompensationDocument(AbstractDocument):
|
|
"""
|
|
Specializes document upload for revocations with certain path
|
|
"""
|
|
instance = models.ForeignKey(
|
|
Compensation,
|
|
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 CompensationDocuments.
|
|
Removes the folder from the file system if there are no further documents for this entry.
|
|
|
|
Args:
|
|
*args ():
|
|
**kwargs ():
|
|
|
|
Returns:
|
|
|
|
"""
|
|
comp_docs = self.instance.get_documents()
|
|
|
|
folder_path = None
|
|
if comp_docs.count() == 1:
|
|
# The only file left for this compensation is the one which is currently processed and will be deleted
|
|
# Make sure that the compensation 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)
|
|
except FileNotFoundError:
|
|
# Folder seems to be missing already...
|
|
pass
|