mirror of
https://github.com/HamzaSha1/zod-backend.git
synced 2025-07-11 07:38:56 +00:00
90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
"""
|
|
web_admin tasks file
|
|
"""
|
|
import datetime
|
|
|
|
# third party imports
|
|
from celery import shared_task
|
|
from templated_email import send_templated_mail
|
|
|
|
# django imports
|
|
from django.conf import settings
|
|
from django.db.models import F, Window
|
|
from django.db.models.functions.window import Rank
|
|
|
|
# local imports
|
|
from base.constants import PENDING, IN_PROGRESS, JUNIOR, GUARDIAN
|
|
from guardian.models import JuniorTask
|
|
from junior.models import JuniorPoints
|
|
from notifications.constants import PENDING_TASK_EXPIRING, IN_PROGRESS_TASK_EXPIRING, NOTIFICATION_DICT, TOP_JUNIOR
|
|
from notifications.models import Notification
|
|
from notifications.utils import send_notification, get_from_user_details, send_notification_multiple_user
|
|
|
|
|
|
@shared_task
|
|
def send_email(recipient_list, template, context: dict = None):
|
|
"""
|
|
used to send otp on email
|
|
:param context:
|
|
:param recipient_list: e-mail list
|
|
:param template: email template
|
|
"""
|
|
if context is None:
|
|
context = {}
|
|
from_email = settings.EMAIL_FROM_ADDRESS
|
|
send_templated_mail(
|
|
template_name=template,
|
|
from_email=from_email,
|
|
recipient_list=recipient_list,
|
|
context=context
|
|
)
|
|
return True
|
|
|
|
|
|
@shared_task()
|
|
def notify_task_expiry():
|
|
"""
|
|
task to send notification for those task which expiring soon
|
|
:return:
|
|
"""
|
|
all_pending_tasks = JuniorTask.objects.filter(
|
|
junior__is_verified=True,
|
|
task_status__in=[PENDING, IN_PROGRESS],
|
|
due_date__range=[datetime.datetime.now().date(),
|
|
(datetime.datetime.now().date() + datetime.timedelta(days=1))])
|
|
if pending_tasks := all_pending_tasks.filter(task_status=PENDING):
|
|
for task in pending_tasks:
|
|
send_notification(PENDING_TASK_EXPIRING, task.guardian.user_id, GUARDIAN, task.junior.auth_id,
|
|
{'task_id': task.id})
|
|
if in_progress_tasks := all_pending_tasks.filter(task_status=IN_PROGRESS):
|
|
for task in in_progress_tasks:
|
|
send_notification(IN_PROGRESS_TASK_EXPIRING, task.junior.auth_id, JUNIOR, task.guardian.user_id,
|
|
{'task_id': task.id})
|
|
return True
|
|
|
|
|
|
@shared_task()
|
|
def notify_top_junior():
|
|
"""
|
|
task to send notification for top leaderboard junior to all junior's
|
|
:return:
|
|
"""
|
|
junior_points_qs = JuniorPoints.objects.filter(
|
|
junior__is_verified=True
|
|
).select_related(
|
|
'junior', 'junior__auth'
|
|
).annotate(rank=Window(
|
|
expression=Rank(),
|
|
order_by=[F('total_points').desc(), 'junior__created_at'])
|
|
).order_by('-total_points', 'junior__created_at')
|
|
|
|
prev_top_position = junior_points_qs.filter(position=1).first()
|
|
new_top_position = junior_points_qs.filter(rank=1).first()
|
|
if prev_top_position != new_top_position:
|
|
send_notification_multiple_user(TOP_JUNIOR, new_top_position.junior.auth_id, JUNIOR,
|
|
{'points': new_top_position.total_points})
|
|
for junior_point in junior_points_qs:
|
|
junior_point.position = junior_point.rank
|
|
junior_point.save()
|
|
return True
|