"""Webapp app models.""" from django.db import models from django.template.loader import render_to_string from django.utils import timezone from django.db.models.signals import post_save from django.dispatch import receiver # from datetime import timedelta import requests from uuid import uuid4 import logging from django.utils.translation import ugettext_lazy as _ from django.db.models import JSONField from auditlog.registry import auditlog from polymorphic.models import PolymorphicModel from webapp.utils import month_from_now, send_signup_email from sikweb.settings import FRONTEND_URL VERBOSE_NAME = _("Webapp") EMAIL_REGEX = r"(^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$)" class Tag(models.Model): """Model for tag.""" id = models.AutoField(primary_key=True) slug = models.SlugField(unique=True) name = models.CharField(max_length=127) icon = models.ImageField() class Meta: verbose_name = _("Tag") verbose_name_plural = _("Tags") def __str__(self): return _("Tag: {}").format(self.slug) class BaseFeed(models.Model): """Model containing something showing on some info feed.""" id = models.AutoField(primary_key=True) deleted = models.BooleanField(default=False) title = models.CharField(max_length=255) description = models.CharField(max_length=255) content = models.TextField() image = models.ImageField(blank=True, null=True) tags = models.ManyToManyField(Tag, related_name="feeds", blank=True) # Require explicit publishing from creator isPublished = models.BooleanField(default=False) # Automatically publish after this time, unless still in draft (!isPublished) publishAt = models.DateTimeField(default=timezone.now) autoUnpublish = models.BooleanField(default=False) # Automatically unpublish after this if auto_unpublish==True unpublishAt = models.DateTimeField(default=month_from_now) webhookUrl = "" hookType = "" wasPublishedBefore = False def __init__(self, *args, **kwargs): super(BaseFeed, self).__init__(*args, **kwargs) self.wasPublishedBefore = self.isPublished def __str__(self): delete_str = _("Deleted: ") if self.deleted else "" return _("{}{}: {}").format(delete_str, self._meta.verbose_name, self.title) def save(self, force_insert=False, force_update=False, *args, **kwargs): created = self.pk is None super(BaseFeed, self).save(force_insert, force_update, *args, **kwargs) if self.isPublished and (created or not self.wasPublishedBefore): self.refresh_from_db() # Fetch so we can use primary key url = f"{self.webhookUrl}/{self.pk}" processHooks( message=generateMessage( f"Uusi {self._meta.verbose_name}", self.title, self.description, url ), eventType=self.hookType, ) self.wasPublishedBefore = self.isPublished class Feed(BaseFeed): """Model representing feed.""" webhookUrl = f"https://{FRONTEND_URL}/feed" hookType = "feed" class Meta: verbose_name = _("Feed") verbose_name_plural = _("Feeds") class Event(BaseFeed): """Model for event in guild calendar""" start_time = models.DateTimeField(default=timezone.now) end_time = models.DateTimeField(default=timezone.now) location = models.CharField(max_length=255, blank=True) signupForm = models.ManyToManyField("SignupForm", blank=True, related_name="event") webhookUrl = f"https://{FRONTEND_URL}/events" hookType = "event" class Meta: verbose_name = _("Event") verbose_name_plural = _("Events") class JobAd(BaseFeed): """Job advertisements shown on Corporate relations page""" webhookUrl = f"https://{FRONTEND_URL}/jobads" hookType = "jobad" class Meta: verbose_name = _("JobAd") verbose_name_plural = _("JobAds") class TemplateQuestion(models.Model): """ Stores template questions for signup forms as JSON format. Used in signup form creation. """ id = models.AutoField(primary_key=True) name = models.CharField(max_length=255) questions = JSONField() deleted = models.BooleanField(default=False) class Meta: verbose_name = _("Template question") verbose_name_plural = _("Template questions") def __str__(self): return _("Template questions: {}").format(self.name) class SignupForm(models.Model): """Model for event signup form. Stores questions in JSON format.""" id = models.AutoField(primary_key=True) title = models.CharField(max_length=255) deleted = models.BooleanField(default=False) visible = models.BooleanField(default=True) start_time = models.DateTimeField(default=timezone.now) end_time = models.DateTimeField(default=timezone.now) questions = JSONField() schema = JSONField() quota = models.PositiveIntegerField(blank=True, null=True) email_content = models.TextField(blank=True) class Meta: verbose_name = _("Signup form") verbose_name_plural = _("Signup forms") def __str__(self): delete_str = _("Deleted: ") if self.deleted else "" return _("#{} {}{}").format(self.id, delete_str, self.title) @property def signups(self): return Signup.objects.filter(signupForm=self, deleted=False).order_by("pk") @property def isOpen(self): now = timezone.now() return self.start_time <= now and now < self.end_time class Signup(models.Model): """ Actual signup into any SignupForm. Deletes are soft. """ id = models.AutoField(primary_key=True) signupForm = models.ForeignKey("SignupForm", on_delete=models.CASCADE) deleted = models.BooleanField(default=False) time = models.DateTimeField(default=timezone.now) answer = JSONField() # Answer we use in signupForm signups field. Frontend uses first questions answer as this value. list_name = models.CharField(_("Name"), max_length=255) # If there is email in questions, we save it as own field email = models.EmailField(blank=True, null=True) # Random unique identifier. Used for signup editing by the user. uuid = models.UUIDField(default=uuid4, editable=False) signupForm = models.ForeignKey("SignupForm", on_delete=models.CASCADE) class Meta: verbose_name = _("Sign-up") verbose_name_plural = _("Sign-ups") def __str__(self): delete_str = _("Deleted: ") if self.deleted else "" return f"{self.signupForm}: {delete_str}{self.list_name} ({self.pk})" @receiver(post_save, sender=Signup) def email_on_signup(sender, instance, created, **kwargs): if created and instance.email: # TODO: Possible bug due to many-to-many relationship with events and forms. # TODO: Subject field crashes with lazy loaded translations. try: # subject = _(f"Olet ilmoittautunut tapahtumaan {instance.signupForm.event.first().title}") subject = f"Olet ilmoittautunut tapahtumaan {instance.signupForm.event.first().title}" except AttributeError: # subject = _(f"Olet ilmoittautunut ilmoon {instance.signupForm.title}") subject = f"Olet ilmoittautunut ilmoon {instance.signupForm.title}" send_signup_email( instance.email, subject, instance.id, instance.uuid, instance.signupForm.email_content, ) def generateMessage(heading: str, title: str, description: str, url: str): return render_to_string( "webapp/tg_message.tpl", {"heading": heading, "title": title, "description": description, "url": url}, ) def processHooks(message: str, eventType: str): allHooks = BaseWebhook.objects.all() for hook in list(allHooks): if hook.plugs[eventType] is True: hook.broadcast(message) class BaseWebhook(PolymorphicModel): """Webhook base class instance""" id = models.AutoField(primary_key=True) name = models.CharField(max_length=255) url = ( models.URLField() ) # URL where webhook message is broadcast. For example Telegram webhook API # "Plugs"; which notifications are sent to this specific webhook instance kaehmy_submit = models.BooleanField(_("Hook Kaehmys"), default=False) ohlhafv_submit = models.BooleanField(_("Hook Ohlhafv challenges"), default=False) feed_published = models.BooleanField(_("Hook published news"), default=False) jobad_published = models.BooleanField(_("Hook published Job Ads"), default=False) event_published = models.BooleanField(_("Hook published events"), default=False) signup_opened = models.BooleanField(_("Hook opened signups"), default=False) @property def plugs(self): return { "kaehmy": self.kaehmy_submit, "ohlhafv": self.ohlhafv_submit, "feed": self.feed_published, "jobad": self.jobad_published, "event": self.event_published, "signup": self.signup_opened, } def parseData(self): pass def broadcast(self, message): resp = requests.post(self.url, json=self.parseData(message)) logging.debug(f"Webhook API response: HTTP{resp.status_code}") logging.debug(resp.content) class GenericWebhook(BaseWebhook): class Meta: proxy = True verbose_name = _("Webhook") verbose_name_plural = _("Webhooks") def __str__(self): return 'Webhook "{}"'.format(self.name) def parseData(self, message): return {"text": message} class TelegramHook(BaseWebhook): """Model containing the channel id of a Telegram chat""" class Meta: verbose_name = _("Telegram channel") verbose_name_plural = _("Telegram channels") channel_id = models.CharField(max_length=255, unique=True) def __str__(self): return 'Telegram channel: "{}"'.format(self.name) def parseData(self, message): return {"text": message, "chat_id": self.channel_id, "parse_mode": "Markdown"} auditlog.register(Tag) auditlog.register(Feed) auditlog.register(Event) auditlog.register(JobAd) auditlog.register(TemplateQuestion) auditlog.register(SignupForm) auditlog.register(Signup) auditlog.register(GenericWebhook) auditlog.register(TelegramHook)