shiftregister/shiftregister/app/models.py

import secrets
from datetime import timedelta

from django.db import models
from django.db.models import Case, Count, ExpressionWrapper, F, Q, When
from django.shortcuts import reverse
from django.template import Context, Template
from django.utils import timezone
from dynamic_preferences.registries import global_preferences_registry
from phonenumber_field.modelfields import PhoneNumberField

from shiftregister.messaging import message

global_preferences = global_preferences_registry.manager()


class Room(models.Model):
    name = models.CharField(max_length=200, primary_key=True)
    required_helpers = models.IntegerField()
    meeting_location = models.TextField(default="Infopoint")
    description = models.TextField(blank=True, default="")
    send_reminders = models.BooleanField(default=True)

    def __str__(self):
        return self.name

    def valid_shifts(self):
        return self.shift_set.filter(deleted=False)


class Shift(models.Model):
    room = models.ForeignKey(Room, on_delete=models.RESTRICT)
    start_at = models.DateTimeField(db_index=True)
    duration = models.DurationField()
    required_helpers = models.IntegerField(
        default=0, help_text="When this is set to zero, the room value is used instead."
    )
    description = models.TextField(blank=True, default="")
    deleted = models.BooleanField(default=False, db_index=True)

    def with_reg_count():
        return Shift.objects.annotate(
            reg_count=Count(
                "shiftregistration",
                distinct=True,
                filter=Q(
                    shiftregistration__state__in=(
                        ShiftRegistration.RegState.REGISTERED,
                        ShiftRegistration.RegState.CHECKED_IN,
                    )
                ),
            )
        ).select_related("room")

    def __str__(self):
        return f"{self.room.name}: {self.start_at}"

    def has_ended(self):
        return (self.start_at + self.duration) < timezone.now()

    def is_running(self):
        return (self.start_at <= timezone.now()) and (not self.has_ended())

    def registration_count(self):
        return (
            self.reg_count
            if hasattr(self, "reg_count")
            else self.shiftregistration_set.filter(
                state__in=[
                    ShiftRegistration.RegState.REGISTERED,
                    ShiftRegistration.RegState.CHECKED_IN,
                ]
            ).count()
        )

    def valid_registrations(self):
        return self.shiftregistration_set.filter(
            state__in=[
                ShiftRegistration.RegState.REGISTERED,
                ShiftRegistration.RegState.CHECKED_IN,
            ],
        )


class Helper(models.Model):
    phone = PhoneNumberField(unique=True, editable=False)
    name = models.CharField(max_length=200)
    # change this to a generic state variable to allow for number blocking/account deactivation?
    number_validated = models.BooleanField(default=False)
    asta_confirmed = models.BooleanField(default=False)

    def __str__(self):
        return self.name

    def send_confirmation(self):
        (token, created) = LoginToken.objects.get_or_create(helper=self)
        token.send()
        return token

    # current or next shift
    def important_shift(self):
        ret = (
            ShiftRegistration.objects.annotate(
                shift_end=ExpressionWrapper(
                    F("shift__start_at") + F("shift__duration"),
                    output_field=models.DateTimeField(),
                )
            )
            .filter(
                helper=self,
                shift_end__gte=timezone.now(),
                shift__deleted=False,
                state__in=[
                    ShiftRegistration.RegState.REGISTERED,
                    ShiftRegistration.RegState.CHECKED_IN,
                ],
            )
            .order_by("shift__start_at")
            .first()
        )
        if ret:
            return ret.shift


class ShiftRegistration(models.Model):
    class Meta:
        unique_together = (("shift", "helper"),)

    # use restrict for now as Model.delete is not called
    shift = models.ForeignKey(Shift, on_delete=models.RESTRICT)
    helper = models.ForeignKey(Helper, on_delete=models.CASCADE)
    reminder_sent = models.BooleanField(default=False)

    class RegState(models.TextChoices):
        # default is registered
        REGISTERED = "REG", "registriert"
        CHECKED_IN = "CHECK", "checkin"
        # cancel via infopoint
        CANCELED = "CANCEL", "abgemeldet"
        # did not attend shift
        FAILED = "FAIL", "nicht angetreten"

    state = models.CharField(
        max_length=7,
        choices=RegState.choices,
        default=RegState.REGISTERED,
    )

    def can_cancel(self):
        if self.state != self.RegState.REGISTERED:
            return False
        return self.shift.start_at > (
            timezone.now()
            + global_preferences_registry.manager()["helper__min_cancel_time"]
        )

    def send_reminder(self):
        url = reverse("shift", kwargs={"shiftid": self.shift.pk})
        template = Template(
            'Deine Schicht beginnt um {{ start_at|date:"H:i" }}, bitte komm 15 Minuten vorher an den Infopoint https://helfen.kntkt.de{{ url }} Du kommst ab jetzt ohne Anstehen aufs Gelände'
        )
        text = template.render(Context({"start_at": self.shift.start_at, "url": url}))
        msg = Message(to=self.helper, text=text)
        msg.save()
        self.reminder_sent = True
        self.save()

    def __str__(self):
        return f"{self.helper.name}: {self.shift}"

    def is_pending(self):
        return self.state == self.RegState.REGISTERED

    def is_checked_in(self):
        return self.state == self.RegState.CHECKED_IN

    def is_withdrawn(self):
        return self.state in (self.RegState.CANCELED, self.RegState.FAILED)


class Message(models.Model):
    # remove limit and send long messages in multiple messages?
    text = models.CharField(max_length=160)
    to = models.ForeignKey(Helper, on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)
    sent_at = models.DateTimeField(blank=True, null=True)

    def __str__(self):
        return f"{self.to.name}({self.created_at}): {self.text}"

    def as_outbound(self):
        return message.Message(
            self.pk,
            recipient=self.to.phone,
            text=self.text,
            type=message.MessageType.OUTBOUND,
            created_at=self.created_at,
        )


def gen_token():
    return secrets.token_urlsafe(
        15
    )  # returns 15 bytes Base64-encoded (times 1.333...) = 20 characters


class LoginToken(models.Model):
    id = models.CharField(
        max_length=20, primary_key=True, default=gen_token, editable=False
    )
    helper = models.ForeignKey(Helper, on_delete=models.CASCADE)
    sent_at = models.DateTimeField(auto_now_add=True)
    send_count = models.IntegerField(default=0)

    def send(self):
        text = f"Dein Registrierungslink zum Helfer*innensystem: https://helfen.kntkt.de{self.get_absolute_url()}\nWenn du dich nicht registriert hast, ignoriere diese SMS."
        msg = Message(to=self.helper, text=text)
        msg.save()
        self.sent_at = timezone.now()
        self.send_count += 1
        self.save()
        # import here to break import cycle
        from .tasks import send_message

        send_message.delay(msg.pk)

    def get_absolute_url(self):
        return reverse("token_login", kwargs={"token": self.id})