Aarni Halinen
359 lines
12 KiB
Python
359 lines
12 KiB
Python
"""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.contrib.postgres.fields 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."""
|
|
|
|
class Meta:
|
|
verbose_name = _("Tag")
|
|
verbose_name_plural = _("Tags")
|
|
|
|
slug = models.SlugField(unique=True)
|
|
name = models.CharField(max_length=127)
|
|
icon = models.ImageField()
|
|
|
|
def __str__(self):
|
|
return _("Tag: {}").format(self.slug)
|
|
|
|
|
|
class BaseFeed(models.Model):
|
|
"""Model containing something showing on some info feed."""
|
|
|
|
tags = models.ManyToManyField(Tag, related_name="feeds", blank=True)
|
|
visible = models.BooleanField(default=True)
|
|
title = models.CharField(max_length=255)
|
|
description = models.CharField(max_length=255)
|
|
content = models.TextField()
|
|
image = models.ImageField(blank=True, null=True)
|
|
|
|
|
|
class Feed(BaseFeed):
|
|
"""Model representing feed."""
|
|
|
|
class Meta:
|
|
verbose_name = _("Feed")
|
|
verbose_name_plural = _("Feeds")
|
|
|
|
publish_time = models.DateTimeField(default=timezone.now)
|
|
autohide = models.DateTimeField(default=month_from_now)
|
|
autohide_enabled = models.BooleanField(default=False)
|
|
deleted = models.BooleanField(default=False)
|
|
|
|
def __str__(self):
|
|
delete_str = _("Deleted: ") if self.deleted else ""
|
|
return _("{}Feed: {}").format(delete_str, self.title)
|
|
|
|
__previousVisible = False
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super(Feed, self).__init__(*args, **kwargs)
|
|
self.__previousVisible = self.visible
|
|
|
|
def save(self, force_insert=False, force_update=False, *args, **kwargs):
|
|
created = self.pk is None
|
|
super(Feed, self).save(force_insert, force_update, *args, **kwargs)
|
|
|
|
if self.visible and (created or not self.__previousVisible):
|
|
self.refresh_from_db() # Fetch so we can use primary key
|
|
url = f"https://{FRONTEND_URL}/feed/{self.pk}"
|
|
processHooks(
|
|
message=generateMessage(
|
|
"Uusi uutinen", self.title, self.description, url
|
|
),
|
|
eventType="feed",
|
|
)
|
|
self.__previousVisible = self.visible
|
|
|
|
|
|
class Event(BaseFeed):
|
|
"""Model for event in guild calendar"""
|
|
|
|
class Meta:
|
|
verbose_name = _("Event")
|
|
verbose_name_plural = _("Events")
|
|
|
|
start_time = models.DateTimeField(default=timezone.now)
|
|
end_time = models.DateTimeField(default=timezone.now)
|
|
signupForm = models.ManyToManyField("SignupForm", blank=True, related_name="event")
|
|
location = models.CharField(max_length=255, blank=True)
|
|
deleted = models.BooleanField(default=False)
|
|
|
|
def __str__(self):
|
|
delete_str = _("Deleted: ") if self.deleted else ""
|
|
return _("{}Event: {}").format(delete_str, self.title)
|
|
|
|
__previousVisible = False
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super(Event, self).__init__(*args, **kwargs)
|
|
self.__previousVisible = self.visible
|
|
|
|
def save(self, force_insert=False, force_update=False, *args, **kwargs):
|
|
created = self.pk is None
|
|
super(Event, self).save(force_insert, force_update, *args, **kwargs)
|
|
|
|
if self.visible and (created or not self.__previousVisible):
|
|
self.refresh_from_db() # Fetch so we can use primary key
|
|
url = f"https://{FRONTEND_URL}/events/{self.pk}"
|
|
processHooks(
|
|
message=generateMessage(
|
|
"Uusi tapahtuma", self.title, self.description, url
|
|
),
|
|
eventType="event",
|
|
)
|
|
self.__previousVisible = self.visible
|
|
|
|
|
|
class TemplateQuestion(models.Model):
|
|
"""
|
|
Stores template questions for signup forms as JSON format. Used in signup form creation.
|
|
"""
|
|
|
|
class Meta:
|
|
verbose_name = _("Template question")
|
|
verbose_name_plural = _("Template questions")
|
|
|
|
name = models.CharField(max_length=255)
|
|
questions = JSONField()
|
|
deleted = models.BooleanField(default=False)
|
|
|
|
def __str__(self):
|
|
return _("Template questions: {}").format(self.name)
|
|
|
|
|
|
class SignupForm(models.Model):
|
|
"""Model for event signup form. Stores questions in JSON format."""
|
|
|
|
class Meta:
|
|
verbose_name = _("Signup form")
|
|
verbose_name_plural = _("Signup forms")
|
|
|
|
title = models.CharField(max_length=255)
|
|
start_time = models.DateTimeField(default=timezone.now)
|
|
end_time = models.DateTimeField(default=timezone.now)
|
|
questions = JSONField()
|
|
schema = JSONField()
|
|
visible = models.BooleanField(default=True)
|
|
quota = models.PositiveIntegerField(blank=True, null=True)
|
|
email_content = models.TextField(blank=True)
|
|
deleted = models.BooleanField(default=False)
|
|
|
|
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.
|
|
"""
|
|
|
|
class Meta:
|
|
verbose_name = _("Sign-up")
|
|
verbose_name_plural = _("Sign-ups")
|
|
|
|
signupForm = models.ForeignKey("SignupForm", on_delete=models.CASCADE)
|
|
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)
|
|
deleted = models.BooleanField(default=False)
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
class BaseRole(models.Model):
|
|
"""Base model for occupations/roles."""
|
|
|
|
name = models.CharField(_("Name"), max_length=255)
|
|
is_board = models.BooleanField(_("Board member"))
|
|
|
|
def __str__(self):
|
|
n = self.name.capitalize()
|
|
return "{} ({})".format(n, _("board member")) if self.is_board else n
|
|
|
|
|
|
class JobAd(models.Model):
|
|
"""Job advertisements shown on Corporate relations page"""
|
|
|
|
class Meta:
|
|
verbose_name = _("JobAd")
|
|
verbose_name_plural = _("JobAds")
|
|
|
|
title = models.CharField(max_length=255)
|
|
description = models.CharField(max_length=255)
|
|
content = models.TextField()
|
|
visible = models.BooleanField(default=True)
|
|
created_at = models.DateTimeField(default=timezone.now)
|
|
autohide_at = models.DateTimeField(default=month_from_now)
|
|
autohide_enabled = models.BooleanField(default=False)
|
|
deleted = models.BooleanField(default=False)
|
|
|
|
def __str__(self):
|
|
delete_str = _("Deleted: ") if self.deleted else ""
|
|
return f"{delete_str}{self.title}"
|
|
|
|
__previousVisible = False
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super(JobAd, self).__init__(*args, **kwargs)
|
|
self.__previousVisible = self.visible
|
|
|
|
def save(self, force_insert=False, force_update=False, *args, **kwargs):
|
|
created = self.pk is None
|
|
super(JobAd, self).save(force_insert, force_update, *args, **kwargs)
|
|
|
|
if self.visible and (created or not self.__previousVisible):
|
|
self.refresh_from_db() # Fetch so we can use primary key
|
|
url = f"https://{FRONTEND_URL}/jobads/{self.pk}"
|
|
processHooks(
|
|
message=generateMessage(
|
|
"Uusi työpaikkailmoitus", self.title, self.description, url
|
|
),
|
|
eventType="jobad",
|
|
)
|
|
self.__previousVisible = self.visible
|
|
|
|
|
|
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"""
|
|
|
|
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(SignupForm)
|
|
auditlog.register(Signup)
|
|
auditlog.register(JobAd)
|
|
auditlog.register(GenericWebhook)
|
|
auditlog.register(TelegramHook)
|