66 lines
2.5 KiB
Python
66 lines
2.5 KiB
Python
"""Account app celery tasks."""
|
|
import datetime
|
|
import inspect
|
|
import logging
|
|
|
|
from celery import shared_task
|
|
from django.utils.translation import gettext_lazy as _
|
|
from celery.schedules import crontab
|
|
from celery.task import periodic_task
|
|
|
|
from account.models import User, UserRole, Role
|
|
|
|
logging.basicConfig(format='[%(levelname)s] %(message)s', level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def send_email(user_id: int, subject: str, message_prop: str, country_code: str, emails=None):
|
|
try:
|
|
user = User.objects.get(id=user_id)
|
|
user.send_email(subject=_(subject),
|
|
message=getattr(user, message_prop, lambda _: '')(country_code, user.username, _(subject)),
|
|
emails=emails)
|
|
except:
|
|
cur_frame = inspect.currentframe()
|
|
cal_frame = inspect.getouterframes(cur_frame, 2)
|
|
logger.error(f'METHOD_NAME: {cal_frame[1][3]}\n'
|
|
f'DETAIL: Exception occurred for user: {user_id}')
|
|
|
|
|
|
@shared_task
|
|
def send_reset_password_email(user_id, country_code):
|
|
"""Send email to user for reset password."""
|
|
send_email(user_id, 'Password_resetting', 'reset_password_template', country_code)
|
|
|
|
|
|
@shared_task
|
|
def confirm_new_email_address(user_id, country_code):
|
|
"""Send email to user new email."""
|
|
send_email(user_id, 'Confirm new email address', 'confirm_email_template', country_code)
|
|
|
|
|
|
@shared_task
|
|
def change_email_address(user_id, country_code, emails=None):
|
|
"""Send email to user new email."""
|
|
send_email(user_id, 'Validate new email address', 'change_email_template', country_code, emails)
|
|
|
|
|
|
@shared_task
|
|
def send_password_changed_email(user_id, country_code):
|
|
"""Send email which notifies user that his password had changed"""
|
|
send_email(user_id, 'Notify password changed', 'notify_password_changed_template', country_code)
|
|
|
|
|
|
@periodic_task(run_every=crontab(hour='*/3'))
|
|
def clean_unaccepted_establishment_team_invites():
|
|
"""This task cleans unaccepted (for 7 days) establishment team membership invites."""
|
|
week_ago = datetime.datetime.now() - datetime.timedelta(days=7)
|
|
roles_to_remove = UserRole.objects.filter(role__role=Role.ESTABLISHMENT_ADMINISTRATOR,
|
|
for_team=True, created__lte=week_ago)\
|
|
.exclude(state=UserRole.VALIDATED).prefetch_related('user', 'role')
|
|
for user_role in roles_to_remove:
|
|
user = user_role.user
|
|
user_role.delete()
|
|
if user.last_login is None:
|
|
user.delete()
|