from datetime import datetime from celery.app import shared_task from celery.utils.log import get_task_logger from constance import config from django.conf import settings from django.core.mail import EmailMessage from django.template.loader import render_to_string from django.utils.translation import activate from django.utils.translation import gettext_lazy as _ from core.models import Order, OrderProduct, PromoCode from core.utils import get_dynamic_email_connection logger = get_task_logger(__name__) @shared_task(queue="default") def contact_us_email(contact_info) -> tuple[bool, str]: logger.debug( "Contact us email sent to %s with subject %s", contact_info.get("email"), contact_info.get("subject", "Without subject"), ) email = EmailMessage( _(f"{config.PROJECT_NAME} | contact us initiated"), render_to_string( "contact_us_email.html", { "email": contact_info.get("email"), "name": contact_info.get("name"), "subject": contact_info.get("subject", "Without subject"), "phone_number": contact_info.get("phone_number", "Without phone number"), "message": contact_info.get("message"), "config": config, }, ), to=[config.EMAIL_FROM], from_email=f"{config.PROJECT_NAME} <{config.EMAIL_FROM}>", connection=get_dynamic_email_connection(), ) email.content_subtype = "html" email.send() return True, str(contact_info.get("email")) @shared_task(queue="default") def send_order_created_email(order_pk: str) -> tuple[bool, str]: try: order = Order.objects.get(pk=order_pk) except Order.DoesNotExist: return False, f"Order not found with the given pk: {order_pk}" if type(order.attributes) is not dict: order.attributes = {} if not any([order.user, order.attributes.get("email", None), order.attributes.get("customer_email", None)]): return False, f"Order's user not found with the given pk: {order_pk}" language = settings.LANGUAGE_CODE recipient = order.attributes.get("customer_email", "") if order.user: recipient = order.user.email language = order.user.language activate(language) if not order.is_whole_digital: email = EmailMessage( _(f"{config.PROJECT_NAME} | order confirmation"), render_to_string( "digital_order_created_email.html" if order.is_whole_digital else "shipped_order_created_email.html", { "order": order, "today": datetime.today(), "config": config, "total_price": order.total_price, }, ), to=[recipient], from_email=f"{config.PROJECT_NAME} <{config.EMAIL_FROM}>", connection=get_dynamic_email_connection(), ) email.content_subtype = "html" email.send() return True, str(order.uuid) @shared_task(queue="default") def send_order_finished_email(order_pk: str) -> tuple[bool, str]: def send_digital_assets_email(ops: list[OrderProduct]) -> None: if len(ops) <= 0: return if not order.user: return activate(order.user.language) email = EmailMessage( _(f"{config.PROJECT_NAME} | order delivered"), render_to_string( template_name="digital_order_delivered_email.html", context={ "order_uuid": order.human_readable_id, "user_first_name": "" or order.user.first_name, "order_products": ops, "project_name": config.PROJECT_NAME, "contact_email": config.EMAIL_FROM, "total_price": round(sum(0.0 or op.buy_price for op in ops), 2), # type: ignore [misc] "display_system_attributes": order.user.has_perm("core.view_order"), "today": datetime.today(), }, ), to=[order.user.email], from_email=f"{config.PROJECT_NAME} <{config.EMAIL_FROM}>", connection=get_dynamic_email_connection(), ) email.content_subtype = "html" result = email.send() logger.debug("Order %s: Tried to send email to %s, resulted with %s", order.pk, order.user.email, result) def send_thank_you_email(ops: list[OrderProduct]) -> None: if ops: pass if not order.user: return activate(order.user.language) pass try: order = Order.objects.get(pk=order_pk) except Order.DoesNotExist: return False, f"Order not found with the given pk: {order_pk}" if not order.user: return False, f"Order's user not found with the given pk: {order_pk}" digital_ops = [] for digital_op in order.order_products.filter( product__is_digital=True, status__in=[ "FINISHED", "DELIVERED", "ACCEPTED", ], ): digital_ops.append(digital_op) send_digital_assets_email(digital_ops) shipped_ops = [] for shipped_op in order.order_products.filter( product__is_digital=False, status__in=[ "FINISHED", "DELIVERED", "ACCEPTED", ], ): shipped_ops.append(shipped_op) send_thank_you_email(shipped_ops) return True, str(order.uuid) @shared_task(queue="default") def send_promocode_created_email(promocode_pk: str) -> tuple[bool, str]: try: promocode = PromoCode.objects.get(pk=promocode_pk) except PromoCode.DoesNotExist: return False, f"Promocode not found with the given pk: {promocode_pk}" if not promocode.user: return True, "The promocode has no user" activate(promocode.user.language) email = EmailMessage( _(f"{config.PROJECT_NAME} | promocode granted"), render_to_string( template_name="promocode_granted_email.html", context={ "promocode": promocode, "user_first_name": "" or promocode.user.first_name, "project_name": config.PROJECT_NAME, "contact_email": config.EMAIL_FROM, "today": datetime.today(), "currency": settings.CURRENCY_CODE, }, ), to=[promocode.user.email], from_email=f"{config.PROJECT_NAME} <{config.EMAIL_FROM}>", connection=get_dynamic_email_connection(), ) email.content_subtype = "html" result = email.send() logger.debug( "Promocode %s: Tried to send email to %s, resulted with %s", promocode.pk, promocode.user.email, result ) return True, str(promocode.uuid)