import uuid from typing import Any, Dict import structlog from django.contrib.auth.models import AbstractUser from django.db import models from django.db.models import JSONField, Max from django.urls import reverse from vbv_lernwelt.core.utils import sanitize_json_data_input logger = structlog.get_logger(__name__) class Organisation(models.Model): organisation_id = models.IntegerField(primary_key=True) name_de = models.CharField(max_length=255) name_fr = models.CharField(max_length=255) name_it = models.CharField(max_length=255) def __str__(self): return f"{self.name_de} ({self.organisation_id})" class Meta: verbose_name = "Organisation" verbose_name_plural = "Organisations" ordering = ["organisation_id"] class Country(models.Model): country_code = models.CharField(max_length=2, primary_key=True) vbv_country_id = models.IntegerField(primary_key=False) name_de = models.CharField(max_length=255) name_fr = models.CharField(max_length=255) name_it = models.CharField(max_length=255) order_id = models.FloatField(default=20) def __str__(self): return f"{self.name_de} ({self.country_code}) ({self.vbv_country_id})" class Meta: verbose_name = "Country" verbose_name_plural = "Countries" ordering = ["order_id", "vbv_country_id"] class User(AbstractUser): """ Default custom user model for VBV Lernwelt. If adding fields that need to be filled at user signup, """ LANGUAGE_CHOICES = ( ("de", "Deutsch"), ("fr", "Français"), ("it", "Italiano"), ) INVOICE_ADDRESS_PRIVATE = "prv" INVOICE_ADDRESS_ORGANISATION = "org" INVOICE_ADDRESS_CHOICES = ( (INVOICE_ADDRESS_PRIVATE, "Private"), (INVOICE_ADDRESS_ORGANISATION, "Organisation"), ) id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) avatar = models.ForeignKey( "media_files.UserImage", null=True, blank=True, on_delete=models.SET_NULL, help_text="Avatar image for the user", ) email = models.EmailField("email address", unique=True) sso_id = models.UUIDField( "SSO subscriber ID", unique=True, null=True, blank=True, default=None ) additional_json_data = JSONField(default=dict, blank=True) language = models.CharField(max_length=2, choices=LANGUAGE_CHOICES, default="de") organisation = models.ForeignKey( Organisation, on_delete=models.SET_NULL, null=True, blank=True ) invoice_address = models.CharField( max_length=3, choices=INVOICE_ADDRESS_CHOICES, default="prv" ) street = models.CharField(max_length=255, blank=True) street_number = models.CharField(max_length=255, blank=True) postal_code = models.CharField(max_length=255, blank=True) city = models.CharField(max_length=255, blank=True) country = models.ForeignKey( Country, related_name="+", on_delete=models.SET_NULL, null=True, blank=True, ) organisation_detail_name = models.CharField(max_length=255, blank=True) organisation_street = models.CharField(max_length=255, blank=True) organisation_street_number = models.CharField(max_length=255, blank=True) organisation_postal_code = models.CharField(max_length=255, blank=True) organisation_city = models.CharField(max_length=255, blank=True) organisation_country = models.ForeignKey( Country, related_name="+", on_delete=models.SET_NULL, null=True, blank=True, ) # fields gathered from cembra pay form birth_date = models.DateField(null=True, blank=True) phone_number = models.CharField(max_length=255, blank=True, default="") # is only set by abacus invoice export code abacus_debitor_number = models.BigIntegerField(unique=True, null=True, blank=True) def set_increment_abacus_debitor_number(self, disable_save=False): if self.abacus_debitor_number: return self # Get the current maximum debitor_number and increment it by 1 current_max = User.objects.aggregate(max_number=Max("abacus_debitor_number"))[ "max_number" ] new_debitor_number = ( current_max if current_max is not None else 60_000_000 ) + 1 self.abacus_debitor_number = new_debitor_number if not disable_save: self.save() return self def create_avatar_url(self, size=400): try: if self.avatar: filter_spec = f"fill-{size}x{size}" self.avatar.get_rendition(filter_spec) url = reverse("user_image", kwargs={"image_id": self.avatar.id}) return f"{url}?filter={filter_spec}" except Exception: logger.warn("could not create avatar url", label="security", exc_info=True) return "/static/avatars/myvbv-default-avatar.png" def update_additional_json_data(self, data: Dict[str, Any]): self.additional_json_data = ( self.additional_json_data | sanitize_json_data_input( { **data, } ) ) @property def avatar_url(self): return self.create_avatar_url() @property def avatar_url_small(self): return self.create_avatar_url(size=96) class SecurityRequestResponseLog(models.Model): created = models.DateTimeField(auto_now_add=True) label = models.CharField(max_length=255, blank=True, default="") type = models.CharField(max_length=255, blank=True, default="") request_trace_id = models.CharField(max_length=255, blank=True, default="") request_method = models.CharField(max_length=255, blank=True, default="") request_full_path = models.CharField(max_length=255, blank=True, default="") request_username = models.CharField(max_length=255, blank=True, default="") request_client_ip = models.CharField(max_length=255, blank=True, default="") request_elapse_time = models.FloatField(default=0) response_status_code = models.CharField(max_length=255, blank=True, default="") category = models.CharField(max_length=255, blank=True, default="") action = models.CharField(max_length=255, blank=True, default="") name = models.CharField(max_length=255, blank=True, default="") ref = models.CharField(max_length=255, blank=True, default="") value = models.IntegerField(default=0) local_url = models.CharField(max_length=255, blank=True, default="") session_key = models.CharField(max_length=255, blank=True, default="") user_agent = models.TextField(blank=True, default="") user_agent_parsed = models.JSONField(blank=True, default=dict) user_agent_os = models.CharField(max_length=255, blank=True, default="") user_agent_browser = models.CharField(max_length=255, blank=True, default="") additional_json_data = models.JSONField(default=dict, blank=True) class Meta: indexes = [ models.Index(fields=["type"]), models.Index(fields=["label"]), models.Index(fields=["category"]), models.Index(fields=["action"]), ] class ExternalApiRequestLog(models.Model): created = models.DateTimeField(auto_now_add=True) api_url = models.TextField(blank=True, default="") api_request_data = models.JSONField(default=dict, blank=True) api_request_verb = models.CharField(max_length=255, blank=True, default="") api_response_status_code = models.IntegerField(default=0) api_response_data = models.TextField(blank=True, default="") request_username = models.CharField(max_length=255, blank=True, default="") request_trace_id = models.CharField(max_length=255, blank=True, default="") elapsed_time = models.FloatField(default=0) additional_json_data = models.JSONField(default=dict, blank=True) def __str__(self): return f"{self.api_request_verb} {self.api_response_status_code} {self.api_url}" class JobLog(models.Model): started = models.DateTimeField(auto_now_add=True) ended = models.DateTimeField(blank=True, null=True) job_name = models.CharField(max_length=255) success = models.BooleanField(default=False) error_message = models.TextField(blank=True, default="") stack_trace = models.TextField(blank=True, default="") json_data = models.JSONField(default=dict) def __str__(self): return "{job_name} {started:%H:%M %d.%m.%Y}".format(**self.__dict__)