from django.core.validators import RegexValidator 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 _ # action_code values are dotted, e.g. "billboard.approved", "escrow.timeout" # (see docs/frontend-notification-click-actions.md) — a plain SlugField # rejects the dot, so every real call using the documented catalog gets a # 400 from the serializer. This mirrors slug's allowed charset but adds ".". validate_action_code = RegexValidator( regex=r'^[-a-zA-Z0-9_.]+\Z', message="Enter a valid action code: letters, numbers, underscores, hyphens, or dots.", ) 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 NotificationLink(BaseModel): """ Admin-managed action_code -> URL mapping for notification click actions. Producer services send an `action_code` (documented per notification type) instead of a raw URL, so where a tap should navigate is a config change here, not a code deploy across every producer service. """ action_code = models.CharField(max_length=100, unique=True, db_index=True, validators=[validate_action_code]) description = models.CharField(max_length=255, blank=True) url_template = models.CharField( max_length=1000, help_text="Use {object_id} as a placeholder, e.g. https://app.gooyal.ir/ads/{object_id}", ) is_active = models.BooleanField(default=True) def __str__(self): return self.action_code def resolve(self, object_id=None): if object_id and '{object_id}' in self.url_template: return self.url_template.format(object_id=object_id) return self.url_template 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) # Alternative to a raw click_url: a documented per-notification-type # action_code, resolved against NotificationLink at send time (see # get_gotify_extras). click_object_id_value is substituted into that # action_code's {object_id} placeholder. action_code = models.CharField(max_length=100, null=True, blank=True, db_index=True, validators=[validate_action_code]) click_object_id_value = models.CharField(max_length=255, 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 resolve_click_url(self): """An explicit click_url always wins; otherwise resolve via `action_code` against NotificationLink. Returns None (not an error) if `action_code` isn't set, or isn't configured/active yet — not every notification is clickable, and a not-yet-configured action_code shouldn't block delivery.""" if self.click_url: return self.click_url if not self.action_code: return None link = NotificationLink.objects.filter(action_code=self.action_code, is_active=True).first() if not link: return None return link.resolve(self.click_object_id_value) def get_gotify_extras(self): """Merge the resolved click url into extras using Gotify's click-action convention, without disturbing any other extras keys the caller already set. Also passes action_code through as-is, when set — even if it didn't resolve to a click_url yet (e.g. no NotificationLink row for it) — so the frontend can see which action_code a notification carries, not just the URL it happened to resolve to.""" extras = dict(self.extras or {}) click_url = self.resolve_click_url() if click_url or self.action_code: notification_extra = dict(extras.get(self.GOTIFY_CLICK_EXTRA_KEY) or {}) if click_url: notification_extra["click"] = {"url": click_url} if self.action_code: notification_extra["action_code"] = self.action_code 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, action_code=data[user_uuid].get("action_code") or None, click_object_id_value=data[user_uuid].get("click_object_id_value") 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 "" action_code = str(sh.cell(rowx=row, colx=6).value).strip() if sh.ncols > 6 else "" click_object_id_value = str(sh.cell(rowx=row, colx=7).value).strip() if sh.ncols > 7 else "" push_dict[user_uuid] = { "title" : title, "message" : message, "priority" : priority, "extras" : extras, "click_url": click_url or None, "action_code": action_code or None, "click_object_id_value": click_object_id_value or None, } return push_dict