# Python imports import logging # Third party imports from celery import shared_task # Django imports from django.core.mail import EmailMultiAlternatives, get_connection from django.template.loader import render_to_string from django.utils.html import strip_tags # Module imports from plane.license.utils.instance_value import get_email_configuration from plane.utils.exception_logger import log_exception @shared_task def send_email_update_magic_code(email, token): try: ( EMAIL_HOST, EMAIL_HOST_USER, EMAIL_HOST_PASSWORD, EMAIL_PORT, EMAIL_USE_TLS, EMAIL_USE_SSL, EMAIL_FROM, ) = get_email_configuration() # Send the mail subject = "Verify your new email address" context = {"code": token, "email": email} html_content = render_to_string("emails/auth/magic_signin.html", context) text_content = strip_tags(html_content) connection = get_connection( host=EMAIL_HOST, port=int(EMAIL_PORT), username=EMAIL_HOST_USER, password=EMAIL_HOST_PASSWORD, use_tls=EMAIL_USE_TLS == "1", use_ssl=EMAIL_USE_SSL == "1", ) msg = EmailMultiAlternatives( subject=subject, body=text_content, from_email=EMAIL_FROM, to=[email], connection=connection, ) msg.attach_alternative(html_content, "text/html") msg.send() logging.getLogger("plane.worker").info("Email sent successfully.") return except Exception as e: log_exception(e) return @shared_task def send_email_update_confirmation(email): """ Send a confirmation email to the user after their email address has been successfully updated. Args: email: The new email address that was successfully updated """ try: ( EMAIL_HOST, EMAIL_HOST_USER, EMAIL_HOST_PASSWORD, EMAIL_PORT, EMAIL_USE_TLS, EMAIL_USE_SSL, EMAIL_FROM, ) = get_email_configuration() # Send the confirmation email subject = "Plane email address successfully updated" context = {"email": email} html_content = render_to_string("emails/user/email_updated.html", context) text_content = strip_tags(html_content) connection = get_connection( host=EMAIL_HOST, port=int(EMAIL_PORT), username=EMAIL_HOST_USER, password=EMAIL_HOST_PASSWORD, use_tls=EMAIL_USE_TLS == "1", use_ssl=EMAIL_USE_SSL == "1", ) msg = EmailMultiAlternatives( subject=subject, body=text_content, from_email=EMAIL_FROM, to=[email], connection=connection, ) msg.attach_alternative(html_content, "text/html") msg.send() logging.getLogger("plane.worker").info(f"Email update confirmation sent successfully to {email}.") return except Exception as e: log_exception(e) return