72 lines
2.8 KiB
Python
72 lines
2.8 KiB
Python
from django.utils import timezone
|
|
|
|
from apps.account.models import User
|
|
|
|
from .models import Notification, NotificationRead
|
|
|
|
# آستانه‌ی هشدار پایان دوره‌ی آزمایشی — هر روزی که ≤ این عدد مانده باشد نشان داده می‌شود.
|
|
TRIAL_WARNING_DAYS = 3
|
|
|
|
|
|
class NotificationService:
|
|
@staticmethod
|
|
def notify(user: User, title: str, body: str = "") -> Notification:
|
|
"""اعلان شخصی برای یک کاربر مشخص می‌سازد."""
|
|
return Notification.objects.create(user=user, title=title, body=body)
|
|
|
|
@staticmethod
|
|
def trial_warning(user: User) -> dict | None:
|
|
if not user.is_in_trial or not user.trial_expires_at:
|
|
return None
|
|
days_left = (user.trial_expires_at - timezone.now()).days
|
|
if days_left > TRIAL_WARNING_DAYS:
|
|
return None
|
|
return {
|
|
"days_left": max(days_left, 0),
|
|
"title": "دوره‌ی رایگانت داره تموم میشه!",
|
|
"body": (
|
|
f"فقط {max(days_left, 0)} روز از دوره‌ی آزمایشی‌ات مونده — برای این‌که کارت زمین نمونه، "
|
|
"از فروشگاه یکی از ابزارها رو با جم فعال کن."
|
|
),
|
|
}
|
|
|
|
@staticmethod
|
|
def list_for_user(user: User) -> dict:
|
|
notifications = Notification.objects.filter(user=user) | Notification.objects.filter(user__isnull=True)
|
|
notifications = notifications.order_by("-created_at")
|
|
read_ids = set(NotificationRead.objects.filter(user=user).values_list("notification_id", flat=True))
|
|
|
|
items = [
|
|
{
|
|
"id": n.id,
|
|
"title": n.title,
|
|
"body": n.body,
|
|
"created_at": n.created_at,
|
|
"is_read": n.id in read_ids,
|
|
}
|
|
for n in notifications
|
|
]
|
|
unread_count = sum(1 for i in items if not i["is_read"])
|
|
return {
|
|
"unread_count": unread_count,
|
|
"trial_warning": NotificationService.trial_warning(user),
|
|
"items": items,
|
|
}
|
|
|
|
@staticmethod
|
|
def mark_read(user: User, notification_id: int) -> None:
|
|
NotificationRead.objects.get_or_create(user=user, notification_id=notification_id)
|
|
|
|
@staticmethod
|
|
def mark_all_read(user: User) -> None:
|
|
notifications = Notification.objects.filter(user=user) | Notification.objects.filter(user__isnull=True)
|
|
already_read = set(NotificationRead.objects.filter(user=user).values_list("notification_id", flat=True))
|
|
NotificationRead.objects.bulk_create(
|
|
[
|
|
NotificationRead(user=user, notification_id=n.id)
|
|
for n in notifications
|
|
if n.id not in already_read
|
|
],
|
|
ignore_conflicts=True,
|
|
)
|