notifications/apps/push_notifications/models.py

from django.db import models
from oauth2_provider.settings import oauth2_settings

from apps.users.models import User
from utils.models import BaseModel
from utils.clients import gotify
from django.utils.translation import gettext_lazy as _


class PushUserQuerySet(models.QuerySet):
    def submit(self, user):
        push_user, created = self.get_or_create(user=user)
        push_user: PushUser = push_user
        push_user.submit_user()
        push_user.submit_client()
        push_user.submit_application()
        return push_user


class PushUser(BaseModel):
    user = models.ForeignKey(User, on_delete=models.PROTECT)
    client_token = models.CharField(max_length=32, null=True, blank=True)
    application_token = models.CharField(max_length=32, null=True, blank=True)
    app_id = None

    objects = PushUserQuerySet.as_manager()

    def submit_user(self):
        response = gotify.submit_user(str(self.user.uuid))

    def submit_client(self):
        if not self.client_token:
            response = gotify.submit_client(str(self.user.uuid))
            self.client_token = response.token
            self.save()

    def submit_application(self):
        if not self.application_token:
            response = gotify.submit_application(self.client_token, str(self.user.uuid))
            self.application_token = response.token
            self.save()

    def send_push(self):
        return


class PushMessage(BaseModel):
    # Gotify's reserved extras namespace for client-side actions. Official
    # clients (Android/iOS/web) open this URL when the notification is tapped:
    # https://gotify.net/docs/pushmsg#extras
    GOTIFY_CLICK_EXTRA_KEY = "client::notification"

    class StateChoices(models.IntegerChoices):
        INIT = 0, _('init')
        START = 1, _('Start')
        DONE = 2, _('done')
        TIMEOUT = 3, _('timeout')
        FAILED = 4, _('failed')

    push_user = models.ForeignKey(PushUser, on_delete=models.PROTECT)
    application = models.ForeignKey(oauth2_settings.APPLICATION_MODEL, on_delete=models.PROTECT,
                                    related_name='+', null=True, blank=True)
    title = models.CharField(max_length=128)
    message = models.TextField()
    priority = models.IntegerField(default=5)
    extras = models.JSONField(default=dict)
    click_url = models.URLField(max_length=1000, null=True, blank=True)
    stats = models.IntegerField(default=StateChoices.INIT, choices=StateChoices.choices)

    bulk = models.ForeignKey("BulkPushMessage", null=True, blank=True, on_delete=models.PROTECT)

    def get_gotify_extras(self):
        """Merge click_url into extras using Gotify's click-action convention,
        without disturbing any other extras keys the caller already set."""
        extras = dict(self.extras or {})
        if self.click_url:
            notification_extra = dict(extras.get(self.GOTIFY_CLICK_EXTRA_KEY) or {})
            notification_extra["click"] = {"url": self.click_url}
            extras[self.GOTIFY_CLICK_EXTRA_KEY] = notification_extra
        return extras

    def send_push(self):
        from apps.push_notifications.tasks import send_push_notification
        send_push_notification.delay(str(self.uuid))


class BulkPushMessage(BaseModel):
    class StateChoices(models.IntegerChoices):
        INIT = 0, _('init')
        START = 1, _('Start')
        DONE = 2, _('done')
        TIMEOUT = 3, _('timeout')
        FAILED = 4, _('failed')

    file = models.FileField(null=True, blank=True, upload_to="bulk_push")
    success_count = models.PositiveIntegerField(default=0)
    failure_count = models.PositiveIntegerField(default=0)
    state = models.IntegerField(choices=StateChoices.choices, default=StateChoices.INIT)

    def bulk_push_task(self):
        from .tasks import bulk_push
        bulk_push.delay(str(self.pk))

    def push_to_all(self, data: dict):
        to_start = BulkPushMessage.objects.filter(pk=self.pk, state=self.StateChoices.INIT).update(state=self.StateChoices.START)

        if not to_start:
            raise Exception('not in ini state')

        for user_uuid in data:
            push_user = PushUser.objects.filter(user_id=user_uuid).first()
            if push_user:
                push_message = PushMessage.objects.create(push_user=push_user,
                                                          title=data[user_uuid].get('title'),
                                                          message=data[user_uuid].get('message'),
                                                          priority=data[user_uuid].get('priority', 5),
                                                          extras=data[user_uuid].get("extras", {}),
                                                          click_url=data[user_uuid].get("click_url") or None)
                push_message.send_push()

                self.success_count += 1

            else:
                self.failure_count += 1

            self.save()

        to_done = BulkPushMessage.objects.filter(pk=self.pk, state=self.StateChoices.START).update(
            state=self.StateChoices.DONE)

    def extract_data(self):
        push_dict = {}
        import xlrd
        import json
        book = xlrd.open_workbook(self.file.path)
        sh = book.sheet_by_index(0)

        for row in range(1, sh.nrows):
            user_uuid = sh.cell(rowx=row, colx=0).value.strip()
            title = sh.cell(rowx=row, colx=1).value.strip()
            message = sh.cell(rowx=row, colx=2).value.strip()
            priority = int(float(str(sh.cell(rowx=row, colx=3).value).strip()))
            extras_str = sh.cell(rowx=row, colx=4).value.strip()
            if extras_str:
                extras = json.loads(extras_str)
            else:
                extras = {}
            click_url = str(sh.cell(rowx=row, colx=5).value).strip() if sh.ncols > 5 else ""

            push_dict[user_uuid] = {
                "title" : title,
                "message" : message,
                "priority" : priority,
                "extras" : extras,
                "click_url": click_url or None,
            }
        return push_dict