""" Author: Michel Peltriaux Organization: Struktur- und Genehmigungsdirektion Nord, Rhineland-Palatinate, Germany Contact: michel.peltriaux@sgdnord.rlp.de Created on: 17.11.20 """ from django.contrib.auth.models import User from django.contrib.gis.db import models from django.db import transaction from django.utils import timezone from django.utils.timezone import now from intervention.settings import INTERVENTION_IDENTIFIER_LENGTH, INTERVENTION_IDENTIFIER_TEMPLATE from konova.models import BaseObject, Geometry, UuidModel from konova.utils import generators from konova.utils.generators import generate_random_string from organisation.models import Organisation from user.models import UserActionLogEntry, UserAction class ResponsibilityData(UuidModel): """ Holds intervention data about responsible organizations and their file numbers for this case """ registration_office = models.ForeignKey(Organisation, on_delete=models.SET_NULL, null=True, related_name="+") registration_file_number = models.CharField(max_length=1000, blank=True, null=True) conservation_office = models.ForeignKey(Organisation, on_delete=models.SET_NULL, null=True, related_name="+") conservation_file_number = models.CharField(max_length=1000, blank=True, null=True) handler = models.CharField(max_length=500, null=True, blank=True, help_text="Refers to 'Eingriffsverursacher'") def __str__(self): return "ZB: {} | ETS: {} | Handler: {}".format( self.registration_office, self.conservation_office, self.handler ) class LegalData(UuidModel): """ Holds intervention legal data such as important dates, laws or responsible handler """ # Refers to "zugelassen am" registration_date = models.DateField(null=True, blank=True, help_text="Refers to 'Zugelassen am'") # Refers to "Bestandskraft am" binding_date = models.DateField(null=True, blank=True, help_text="Refers to 'Bestandskraft am'") process_type = models.CharField(max_length=500, null=True, blank=True) law = models.CharField(max_length=500, null=True, blank=True) def __str__(self): return "{} | {} | {}".format( self.process_type, self.law, self.id ) class Intervention(BaseObject): """ Interventions are e.g. construction sites where nature used to be. """ responsible = models.OneToOneField( ResponsibilityData, on_delete=models.SET_NULL, null=True, blank=True, related_name='+', help_text="Holds data on responsible organizations ('Zulassungsbehörde', 'Eintragungsstelle')" ) legal = models.OneToOneField( LegalData, on_delete=models.SET_NULL, null=True, blank=True, related_name='+', help_text="Holds data on legal dates or law" ) geometry = models.ForeignKey(Geometry, null=True, blank=True, on_delete=models.SET_NULL) documents = models.ManyToManyField("konova.Document", blank=True) # Checks - Refers to "Genehmigen" but optional checked = models.OneToOneField( UserActionLogEntry, on_delete=models.SET_NULL, null=True, blank=True, help_text="Holds data on user and timestamp of this action", related_name="+" ) # Refers to "verzeichnen" recorded = models.OneToOneField( UserActionLogEntry, on_delete=models.SET_NULL, null=True, blank=True, help_text="Holds data on user and timestamp of this action", related_name="+" ) # Holds which intervention is simply a newer version of this dataset next_version = models.ForeignKey("Intervention", null=True, blank=True, on_delete=models.DO_NOTHING) # Users having access on this object users = models.ManyToManyField(User) access_token = models.CharField( max_length=255, null=True, blank=True, help_text="Used for sharing access", ) def __str__(self): return "{} ({})".format(self.identifier, self.title) def delete(self, *args, **kwargs): """ Custom delete functionality Does not delete from database but sets a timestamp for being deleted on and which user deleted the object Args: *args (): **kwargs (): Returns: """ _now = timezone.now() _user = kwargs.get("user", None) with transaction.atomic(): # "Delete" related compensations as well coms = self.compensations.all() action = UserActionLogEntry.objects.create( user=_user, timestamp=_now, action=UserAction.DELETED ) for com in coms: com.deleted = action com.save() self.deleted = action self.save() @staticmethod def _generate_new_identifier() -> str: """ Generates a new identifier for the intervention object Returns: str """ curr_month = str(now().month) curr_year = str(now().year) rand_str = generate_random_string( length=INTERVENTION_IDENTIFIER_LENGTH, only_numbers=True, ) _str = "{}{}{}".format(curr_month, curr_year, rand_str) return INTERVENTION_IDENTIFIER_TEMPLATE.format(_str) def generate_access_token(self, make_unique: bool = False, rec_depth: int = 5): """ Creates a new access token for the intervention Tokens are not used for identification of a table row. The share logic checks the intervention id as well as the given token. Therefore two different interventions can hold the same access_token without problems. For (possible) future changes to the share logic, the make_unique parameter may be used for checking whether the access_token is already used in any intervention. If so, tokens will be generated as long as a free token can be found. Args: make_unique (bool): Perform check on uniqueness over all intervention entries rec_depth (int): How many tries for generating a free random token (only if make_unique) Returns: """ # Make sure we won't end up in an infinite loop of trying to generate access_tokens rec_depth = rec_depth - 1 if rec_depth < 0 and make_unique: raise RuntimeError( "Access token generating for {} does not seem to find a free random token! Aborted!".format(self.id) ) # Create random token token = generators.generate_random_string(15) token_used_in = Intervention.objects.filter(access_token=token) # Make sure the token is not used anywhere as access_token, yet. # Make use of QuerySet lazy method for checking if it exists or not. if token_used_in and make_unique: self.generate_access_token(make_unique, rec_depth) else: self.access_token = token self.save() def save(self, *args, **kwargs): if self.identifier is None or len(self.identifier) == 0: # Create new identifier new_id = self._generate_new_identifier() while Intervention.objects.filter(identifier=new_id).exists(): new_id = self._generate_new_identifier() self.identifier = new_id super().save(*args, **kwargs) def is_shared_with(self, user: User): """ Access check Checks whether a given user has access to this intervention Args: user (): Returns: """ return self.users.filter(username=user.username).exists()