You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
konova/compensation/models/compensation.py

566 lines
17 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 django.urls import reverse
from analysis.settings import LKOMPVZVO_PUBLISH_DATE
from codelist.models import KonovaCode
from compensation.settings import COMPENSATION_IDENTIFIER_TEMPLATE, COMPENSATION_IDENTIFIER_LENGTH, \
COMPENSATION_LANIS_LAYER_NAME_RECORDED, COMPENSATION_LANIS_LAYER_NAME_UNRECORDED, COMPENSATION_LANIS_LAYER_NAME_UNRECORDED_OLD_ENTRY
from konova.sub_settings.django_settings import BASE_URL
from user.models import User, Team
from django.db import models, transaction
from django.db.models import QuerySet, Sum
from django.http import HttpRequest
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, DeadlineType, ResubmitableObjectMixin
from konova.utils.message_templates import DATA_UNSHARED_EXPLANATION, COMPENSATION_REMOVED_TEMPLATE, \
DOCUMENT_REMOVED_TEMPLATE, DEADLINE_REMOVED, DEADLINE_ADDED, \
COMPENSATION_ACTION_REMOVED, COMPENSATION_STATE_REMOVED, INTERVENTION_HAS_REVOCATIONS_TEMPLATE
from user.models import UserActionLogEntry
class AbstractCompensation(BaseObject,
GeoReferencedMixin,
ResubmitableObjectMixin
):
"""
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)
deadline = Deadline.objects.create(
type=form_data["type"],
date=form_data["date"],
comment=form_data["comment"],
created=created_action,
)
self.save()
self.deadlines.add(deadline)
self.mark_as_edited(user, edit_comment=DEADLINE_ADDED)
return deadline
def remove_deadline(self, form):
""" Removes a deadline from the abstract compensation
Args:
form (RemoveDeadlineModalForm): The form holding all relevant data
Returns:
"""
deadline = form.deadline
user = form.user
with transaction.atomic():
deadline.delete()
self.mark_as_edited(user, edit_comment=DEADLINE_REMOVED)
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(
amount=form_data["amount"],
unit=form_data["unit"],
comment=form_data["comment"],
created=user_action,
)
comp_action.action_type.set(form_data.get("action_type", []))
comp_action_details = form_data["action_type_details"]
comp_action.action_type_details.set(comp_action_details)
self.actions.add(comp_action)
return comp_action
def remove_action(self, form):
""" Removes a CompensationAction from the abstract compensation
Args:
form (RemoveCompensationActionModalForm): The form holding all relevant data
Returns:
"""
action = form.action
user = form.user
with transaction.atomic():
action.delete()
self.mark_as_edited(user, edit_comment=COMPENSATION_ACTION_REMOVED)
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():
biotope_type_id = form_data["biotope_type"]
code = KonovaCode.objects.get(id=biotope_type_id)
state = CompensationState.objects.create(
biotope_type=code,
surface=form_data["surface"],
)
state_additional_types = form_data["biotope_extra"]
state.biotope_type_details.set(state_additional_types)
if is_before_state:
self.before_states.add(state)
else:
self.after_states.add(state)
return state
def remove_state(self, form):
""" Removes a CompensationState from the abstract compensation
Args:
form (RemoveCompensationStateModalForm): The form holding all relevant data
Returns:
"""
state = form.state
user = form.user
with transaction.atomic():
state.delete()
self.mark_as_edited(user, edit_comment=COMPENSATION_STATE_REMOVED)
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:
"""
val = qs.aggregate(Sum("surface"))["surface__sum"] or 0
val = float('{:0.2f}'.format(val))
return val
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
def get_finished_deadlines(self):
""" Getter for FINISHED-deadlines
Returns:
queryset (QuerySet): The finished deadlines
"""
return self.deadlines.filter(
type=DeadlineType.FINISHED
)
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 PikMixin(models.Model):
""" Provides PIK flag as Mixin
"""
is_pik = models.BooleanField(
blank=True,
null=True,
default=False,
help_text="Flag if compensation is a 'Produktonsintegrierte Kompensation'"
)
class Meta:
abstract = True
class Compensation(AbstractCompensation, CEFMixin, CoherenceMixin, PikMixin):
"""
Regular compensation, linked to an intervention
"""
intervention = models.ForeignKey(
"intervention.Intervention",
on_delete=models.CASCADE,
null=True,
blank=True,
related_name='compensations'
)
objects = CompensationManager()
identifier_length = COMPENSATION_IDENTIFIER_LENGTH
identifier_template = COMPENSATION_IDENTIFIER_TEMPLATE
def __str__(self):
return "{}".format(self.identifier)
def get_detail_url(self):
return reverse("compensation:detail", args=(self.id,))
def get_detail_url_absolute(self):
return BASE_URL + self.get_detail_url()
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 mark_as_deleted(self, user, send_mail: bool = True):
super().mark_as_deleted(user, send_mail)
if user is not None:
self.intervention.mark_as_edited(user, edit_comment=COMPENSATION_REMOVED_TEMPLATE.format(self.identifier))
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)
def is_only_shared_with(self, user: User):
""" Share check
Checks whether a given user is the only one having shared access to this entry
Args:
user (User): The user to be checked
Returns:
"""
# Compensations inherit their shared state from the interventions
return self.intervention.is_only_shared_with(user)
def share_with_user(self, user: User):
""" Adds user to list of shared access users
Args:
user (User): The user to be added to the object
Returns:
"""
self.intervention.users.add(user)
def share_with_user_list(self, user_list: list):
""" Sets the list of shared access users
Args:
user_list (list): The users to be added to the object
Returns:
"""
self.intervention.users.set(user_list)
def share_with_team(self, team: Team):
""" Adds team to list of shared access teams
Args:
team (Team): The team to be added to the object
Returns:
"""
self.intervention.teams.add(team)
def share_with_team_list(self, team_list: list):
""" Sets the list of shared access teams
Args:
team_list (list): The teams to be added to the object
Returns:
"""
self.intervention.teams.set(team_list)
@property
def shared_users(self) -> QuerySet:
""" Shortcut for fetching the users which have shared access on this object
Returns:
users (QuerySet)
"""
return self.intervention.shared_users
@property
def shared_teams(self) -> QuerySet:
""" Shortcut for fetching the teams which have shared access on this object
Returns:
users (QuerySet)
"""
return self.intervention.shared_teams
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 checked state of the related intervention
Args:
user (User): The performing user
request (HttpRequest): The performing request
edit_comment (str): Additional comment for the log entry
Returns:
"""
self.intervention.set_unchecked()
action = super().mark_as_edited(user, edit_comment=edit_comment)
return action
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()
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 self.intervention.legal.revocations.exists():
messages.error(
request,
INTERVENTION_HAS_REVOCATIONS_TEMPLATE.format(self.intervention.legal.revocations.count()),
extra_tags="danger",
)
super().set_status_messages(request)
return request
@property
def is_recorded(self):
""" Getter for record status as property
Since compensations inherit their record status from their intervention, the intervention's status is being
returned
Returns:
"""
return self.intervention.is_recorded
def get_lanis_layer_name(self):
""" Getter for specific LANIS/WFS object layer
Returns:
"""
retval = None
if self.is_recorded:
retval = COMPENSATION_LANIS_LAYER_NAME_RECORDED
else:
try:
is_old_entry = self.intervention.legal.binding_date < LKOMPVZVO_PUBLISH_DATE
except TypeError:
is_old_entry = False
if is_old_entry:
retval = COMPENSATION_LANIS_LAYER_NAME_UNRECORDED_OLD_ENTRY
else:
retval = COMPENSATION_LANIS_LAYER_NAME_UNRECORDED
return retval
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, user=None, *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
try:
folder_path = self.file.path.split("/")[:-1]
folder_path = "/".join(folder_path)
except ValueError:
folder_path = None
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