"""
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.urls import reverse
from compensation.settings import ECO_ACCOUNT_IDENTIFIER_TEMPLATE, ECO_ACCOUNT_IDENTIFIER_LENGTH, \
ECO_ACCOUNT_LANIS_LAYER_NAME_RECORDED, ECO_ACCOUNT_LANIS_LAYER_NAME_UNRECORDED
from konova.sub_settings.django_settings import BASE_URL
from konova.utils.message_templates import DEDUCTION_REMOVED, DOCUMENT_REMOVED_TEMPLATE
from django.core.validators import MinValueValidator
from django.db import models
from django.db.models import Sum, QuerySet
from compensation.managers import EcoAccountManager, EcoAccountDeductionManager
from compensation.models.compensation import AbstractCompensation, PikMixin
from compensation.utils.quality import EcoAccountQualityChecker
from konova.models import ShareableObjectMixin, RecordableObjectMixin, AbstractDocument, BaseResource, \
generate_document_file_upload_path
from konova.tasks import celery_send_mail_deduction_changed, celery_send_mail_deduction_changed_team
class EcoAccount(AbstractCompensation, ShareableObjectMixin, RecordableObjectMixin, PikMixin):
"""
An eco account is a kind of 'prepaid' compensation. It can be compared to an account that already has been filled
with some kind of currency. From this account one is able to deduct currency for current projects.
"""
deductable_surface = models.FloatField(
blank=True,
null=True,
help_text="Amount of deductable surface - can be lower than the total surface due to deduction limitations",
default=0,
)
deductable_rest = models.FloatField(
blank=True,
null=True,
help_text="Amount of deductable rest",
default=0,
)
legal = models.OneToOneField(
"intervention.Legal",
on_delete=models.SET_NULL,
null=True,
blank=True,
help_text="Holds data on legal dates or law"
)
objects = EcoAccountManager()
identifier_length = ECO_ACCOUNT_IDENTIFIER_LENGTH
identifier_template = ECO_ACCOUNT_IDENTIFIER_TEMPLATE
def __str__(self):
return f"{self.identifier} ({self.title})"
def get_detail_url(self):
return reverse("compensation:acc: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 if none was given
self.identifier = self.generate_new_identifier()
# Before saving, make sure the given identifier is not used, yet
while EcoAccount.objects.filter(identifier=self.identifier).exclude(id=self.id).exists():
self.identifier = self.generate_new_identifier()
super().save(*args, **kwargs)
@property
def deductions_surface_sum(self) -> float:
""" Shortcut for get_deductions_surface.
Can be used in templates
Returns:
sum_surface (float)
"""
return self.get_deductions_surface()
def get_deductions_surface(self) -> float:
""" Calculates the account's deductions surface sum
Returns:
sum_surface (float)
"""
val = self.deductions.all().aggregate(Sum("surface"))["surface__sum"] or 0
val = float('{:0.2f}'.format(val))
return val
def __calculate_deductable_rest(self):
""" Calculates available rest surface of the eco account
Args:
Returns:
ret_val_total (float): Total amount
"""
deductions_surfaces = self.get_deductions_surface()
available_surface = self.deductable_surface
if available_surface is None:
# Fallback!
available_surface = deductions_surfaces
else:
available_surface = float(available_surface)
ret_val = available_surface - deductions_surfaces
return ret_val
def quality_check(self) -> EcoAccountQualityChecker:
""" Quality check
Returns:
ret_msgs (EcoAccountQualityChecker): Holds validity and error messages
"""
checker = EcoAccountQualityChecker(self)
checker.run_check()
return checker
def get_documents(self) -> QuerySet:
""" Getter for all documents of an EcoAccount
Returns:
docs (QuerySet): The queryset of all documents
"""
docs = EcoAccountDocument.objects.filter(
instance=self
)
return docs
def is_ready_for_publish(self) -> bool:
""" Checks whether the data passes all constraints for being publishable
Returns:
is_ready (bool) : True|False
"""
is_recorded = self.recorded is not None
is_ready = is_recorded
return is_ready
def get_share_link(self):
""" Returns the share url for the object
Returns:
"""
return reverse("compensation:acc:share-token", args=(self.id, self.access_token))
def send_notification_mail_on_deduction_change(self, data_change: dict):
""" Sends notification mails for changes on the deduction
Args:
data_change ():
Returns:
"""
# Send mail
shared_users = self.shared_users.values_list("id", flat=True)
for user_id in shared_users:
celery_send_mail_deduction_changed.delay(self.id, self.get_app_object_tuple(), user_id, data_change)
# Send mail
shared_teams = self.shared_teams.values_list("id", flat=True)
for team_id in shared_teams:
celery_send_mail_deduction_changed_team.delay(self.id, self.get_app_object_tuple(), team_id, data_change)
def update_deductable_rest(self):
"""
Updates deductable_rest, which holds the amount of rest surface for this account.
Returns:
"""
self.deductable_rest = self.__calculate_deductable_rest()
self.save()
def get_deductable_rest_relative(self):
"""
Returns deductable_rest relative to deductable_surface mapped to [0,100]
Returns:
"""
try:
ret_val = int((self.deductable_rest / (self.deductable_surface or 0)) * 100)
except ZeroDivisionError:
ret_val = 0
return ret_val
def get_lanis_layer_name(self):
""" Getter for specific LANIS/WFS object layer
Returns:
"""
retval = None
if self.is_recorded:
retval = ECO_ACCOUNT_LANIS_LAYER_NAME_RECORDED
else:
retval = ECO_ACCOUNT_LANIS_LAYER_NAME_UNRECORDED
return retval
class EcoAccountDocument(AbstractDocument):
"""
Specializes document upload for revocations with certain path
"""
instance = models.ForeignKey(
EcoAccount,
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 EcoAccountDocuments.
Removes the folder from the file system if there are no further documents for this entry.
Args:
*args ():
**kwargs ():
Returns:
"""
acc_docs = self.instance.get_documents()
folder_path = None
if acc_docs.count() == 1:
# The only file left for this eco account 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
class EcoAccountDeduction(BaseResource):
"""
A deduction object for eco accounts
"""
account = models.ForeignKey(
EcoAccount,
on_delete=models.SET_NULL,
null=True,
blank=True,
help_text="Deducted from",
related_name="deductions",
)
surface = models.FloatField(
null=True,
blank=True,
help_text="Amount deducted (m²)",
validators=[
MinValueValidator(limit_value=0.00),
]
)
intervention = models.ForeignKey(
"intervention.Intervention",
on_delete=models.CASCADE,
null=True,
blank=True,
help_text="Deducted for",
related_name="deductions",
)
objects = EcoAccountDeductionManager()
def __str__(self):
return "{} of {}".format(self.surface, self.account)
def delete(self, user=None, *args, **kwargs):
if user is not None:
self.intervention.mark_as_edited(user, edit_comment=DEDUCTION_REMOVED)
self.account.mark_as_edited(user, edit_comment=DEDUCTION_REMOVED)
super().delete(*args, **kwargs)
self.account.update_deductable_rest()
def save(self, *args, **kwargs):
super().save(*args, **kwargs)
self.account.update_deductable_rest()