228 lines
7 KiB
Python
228 lines
7 KiB
Python
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"{settings.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"{settings.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"{settings.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"{settings.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"{settings.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": settings.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"{settings.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"{settings.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": settings.PROJECT_NAME,
|
|
"contact_email": config.EMAIL_FROM,
|
|
"today": datetime.today(),
|
|
"currency": settings.CURRENCY_CODE,
|
|
},
|
|
),
|
|
to=[promocode.user.email],
|
|
from_email=f"{settings.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)
|