Refactoring
* splits compensation/models.py into subpackage * renames base objects by adding suffix Mixin
This commit is contained in:
336
compensation/models/compensation.py
Normal file
336
compensation/models/compensation.py
Normal file
@@ -0,0 +1,336 @@
|
||||
"""
|
||||
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.auth.models import User
|
||||
from django.db import models, transaction
|
||||
from django.db.models import QuerySet, Sum
|
||||
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 intervention.models import Responsibility, Intervention
|
||||
from konova.models import BaseObject, AbstractDocument, Geometry, Deadline, generate_document_file_upload_path
|
||||
from konova.settings import DEFAULT_SRID_RLP, LANIS_LINK_TEMPLATE
|
||||
from user.models import UserActionLogEntry, UserAction
|
||||
|
||||
|
||||
class AbstractCompensation(BaseObject):
|
||||
"""
|
||||
Abstract compensation model which holds basic attributes, shared by subclasses like the regular Compensation,
|
||||
EMA or EcoAccount.
|
||||
|
||||
"""
|
||||
responsible = models.OneToOneField(
|
||||
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="+")
|
||||
|
||||
geometry = models.ForeignKey(Geometry, null=True, blank=True, on_delete=models.SET_NULL)
|
||||
|
||||
class Meta:
|
||||
abstract = True
|
||||
|
||||
def add_new_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.objects.create(
|
||||
user=user,
|
||||
action=UserAction.CREATED
|
||||
)
|
||||
deadline = Deadline.objects.create(
|
||||
type=form_data["type"],
|
||||
date=form_data["date"],
|
||||
comment=form_data["comment"],
|
||||
created=created_action,
|
||||
)
|
||||
edited_action = UserActionLogEntry.objects.create(
|
||||
user=user,
|
||||
action=UserAction.EDITED,
|
||||
comment=_("Added deadline")
|
||||
)
|
||||
self.modified = edited_action
|
||||
self.save()
|
||||
self.log.add(edited_action)
|
||||
self.deadlines.add(deadline)
|
||||
return deadline
|
||||
|
||||
def add_new_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.objects.create(
|
||||
user=user,
|
||||
action=UserAction.CREATED,
|
||||
)
|
||||
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,
|
||||
)
|
||||
edited_action = UserActionLogEntry.objects.create(
|
||||
user=user,
|
||||
action=UserAction.EDITED,
|
||||
comment=_("Added action"),
|
||||
)
|
||||
self.modified = edited_action
|
||||
self.save()
|
||||
self.log.add(edited_action)
|
||||
self.actions.add(comp_action)
|
||||
return comp_action
|
||||
|
||||
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
|
||||
|
||||
|
||||
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,
|
||||
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)
|
||||
|
||||
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 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
|
||||
user = form.user
|
||||
with transaction.atomic():
|
||||
user_action = UserActionLogEntry.objects.create(
|
||||
user=user,
|
||||
action=UserAction.EDITED,
|
||||
comment=_("Added state")
|
||||
)
|
||||
self.log.add(user_action)
|
||||
self.modified = user_action
|
||||
self.save()
|
||||
|
||||
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
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user