Ali Asadi
707407bb52
The click_url a producer sends had to be built in that producer's own
Python code (utils/deep_links.py in advertising), so changing the URL
scheme meant a code deploy. Adds NotificationLink: an admin-editable
(code, url_template, is_active) row per notification type, with
url_template supporting a {object_id} placeholder.
PushMessage gains `code` and `click_object_id` — a producer can send either
a raw click_url (unchanged, still wins if both are given) or a code + the
referenced object's id, and resolve_click_url() looks up the template at
send time. An unset or not-yet-configured/inactive code resolves to no
click action rather than an error, consistent with "not every notification
is clickable." Wired through both the single-push API (serializer) and the
bulk Excel path (two new optional columns).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
208 lines
8.5 KiB
Python
208 lines
8.5 KiB
Python
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 NotificationLink(BaseModel):
|
|
"""
|
|
Admin-managed code -> URL mapping for notification click actions. Producer
|
|
services send a `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.
|
|
"""
|
|
code = models.SlugField(max_length=100, unique=True, db_index=True)
|
|
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.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 code,
|
|
# resolved against NotificationLink at send time (see get_gotify_extras).
|
|
# click_object_id is substituted into that code's {object_id} placeholder.
|
|
code = models.SlugField(max_length=100, null=True, blank=True, db_index=True)
|
|
click_object_id = 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 `code`
|
|
against NotificationLink. Returns None (not an error) if `code` isn't
|
|
set, or isn't configured/active yet — not every notification is
|
|
clickable, and a not-yet-configured code shouldn't block delivery."""
|
|
if self.click_url:
|
|
return self.click_url
|
|
if not self.code:
|
|
return None
|
|
link = NotificationLink.objects.filter(code=self.code, is_active=True).first()
|
|
if not link:
|
|
return None
|
|
return link.resolve(self.click_object_id)
|
|
|
|
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."""
|
|
extras = dict(self.extras or {})
|
|
click_url = self.resolve_click_url()
|
|
if click_url:
|
|
notification_extra = dict(extras.get(self.GOTIFY_CLICK_EXTRA_KEY) or {})
|
|
notification_extra["click"] = {"url": 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,
|
|
code=data[user_uuid].get("code") or None,
|
|
click_object_id=data[user_uuid].get("click_object_id") 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 ""
|
|
code = str(sh.cell(rowx=row, colx=6).value).strip() if sh.ncols > 6 else ""
|
|
click_object_id = 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,
|
|
"code": code or None,
|
|
"click_object_id": click_object_id or None,
|
|
}
|
|
return push_dict
|