mpeltriaux
be885306c5
* adds model and form mixin for PIK * integrates mixins for compensation, ema and ecoaccount * adds migration files * extends API * extends API test data * adds is_xy fields to compensation, ema and ecoaccount reports * adds is_pik information to detail views * adds/updates translations
502 lines
15 KiB
Python
502 lines
15 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 codelist.models import KonovaCode
|
|
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 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, COMPENSATION_REMOVED_TEMPLATE, \
|
|
DOCUMENT_REMOVED_TEMPLATE, COMPENSATION_EDITED_TEMPLATE, DEADLINE_REMOVED, ADDED_DEADLINE, \
|
|
COMPENSATION_ACTION_REMOVED, COMPENSATION_STATE_REMOVED, INTERVENTION_HAS_REVOCATIONS_TEMPLATE
|
|
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)
|
|
|
|
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=ADDED_DEADLINE)
|
|
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:
|
|
|
|
"""
|
|
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 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()
|
|
|
|
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 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 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.users.all()
|
|
|
|
@property
|
|
def shared_teams(self) -> QuerySet:
|
|
""" Shortcut for fetching the teams which have shared access on this object
|
|
|
|
Returns:
|
|
users (QuerySet)
|
|
"""
|
|
return self.intervention.teams.all()
|
|
|
|
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, reset_recorded: bool = True):
|
|
""" Performs internal logic for setting the recordedd/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
|
|
reset_recorded (bool): Whether the record-state of the object should be reset
|
|
|
|
Returns:
|
|
|
|
"""
|
|
self.intervention.unrecord(user, request)
|
|
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
|
|
|
|
|
|
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
|
|
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
|