schon/engine/core/utils/emailing.py

import logging
from datetime import datetime

from celery.app import shared_task
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 engine.core.models import Order, OrderProduct, PromoCode
from engine.core.utils import get_dynamic_email_connection

logger = logging.getLogger(__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(
            "../templates/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="../templates/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="../templates/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)