import threading
import logging
from datetime import datetime
import requests
from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from core.utils import _get_language_from_request, _get_default_language
from notification_engine.models import EmailTemplate, EmailTemplateTranslation
SMTP2GO_URL = "https://api.smtp2go.com/v3/email/send"
email_notification_logger = logging.getLogger("email-notification")
class Email:
def __init__(self):
self.subject = ""
self.text_message = ""
self.html_message = ""
self.from_email = ""
self.to_email = ""
def set_subject(self, x): self.subject = x
def set_text_message(self, x): self.text_message = x
def set_html_message(self, x): self.html_message = x
def set_from_email(self, x): self.from_email = x
def set_to_email(self, x): self.to_email = x
def get_subject(self): return self.subject
def get_text_message(self): return self.text_message
def get_html_message(self): return self.html_message
def get_from_email(self): return self.from_email
def get_to_email(self): return self.to_email
def render_email_text(text: str, context: dict) -> str:
text = text or ""
for key, value in context.items():
text = text.replace("{" + key + "}", str(value))
return text
def get_email_translation(template_code: str, request=None, language=None):
template = EmailTemplate.objects.filter(
code=template_code,
is_active=True
).first()
if not template:
raise ImproperlyConfigured(f"Email template '{template_code}' not found")
if language is None and request is not None:
language = _get_language_from_request(request)
if language:
translation = EmailTemplateTranslation.objects.filter(
template=template,
language=language,
is_active=True
).first()
if translation:
return template, translation
default_language = _get_default_language()
if not default_language:
raise ImproperlyConfigured("Default language not found")
translation = EmailTemplateTranslation.objects.filter(
template=template,
language=default_language,
is_active=True
).first()
if not translation:
raise ImproperlyConfigured(
f"Email translation for '{template_code}' not found"
)
return template, translation
def send_otp_via_email(
otp_number: str,
reg_email: str,
template_code: str,
request=None,
language=None,
context: dict = None,
):
template, translation = get_email_translation(
template_code=template_code,
request=request,
language=language,
)
final_context = {
"otp": otp_number,
"app_name": "Wile",
"language_code": translation.language.code,
}
if context:
final_context.update(context)
email = Email()
email.set_subject(render_email_text(translation.subject, final_context))
email.set_text_message(render_email_text(translation.text_message, final_context))
email.set_html_message(render_email_text(translation.html_message, final_context))
email.set_from_email(settings.DEFAULT_FROM_EMAIL)
email.set_to_email(reg_email)
OTPEmailThread(email).start()
def send_email(email: Email):
smtp2go_api_key = settings.SMTP2GO_API_KEY
if not smtp2go_api_key:
raise ImproperlyConfigured("SMTP2GO_API_KEY is not configured")
payload = {
"api_key": smtp2go_api_key,
"to": [email.get_to_email()],
"sender": email.get_from_email(),
"subject": email.get_subject(),
"text_body": email.get_text_message(),
"html_body": email.get_html_message(),
}
response = requests.post(
SMTP2GO_URL,
json=payload,
timeout=20
)
if response.status_code not in [200, 201]:
raise Exception(
f"SMTP2GO failed: {response.status_code} - {response.text}"
)
return response.json()
class OTPEmailThread(threading.Thread):
def __init__(self, email: Email):
self.email = email
threading.Thread.__init__(self)
def run(self):
today = datetime.today()
try:
email_notification_logger.info(
f"{today} : Sending Email to {self.email.get_to_email()}"
)
send_email(self.email)
email_notification_logger.info(
f"{today} : Email sent to {self.email.get_to_email()}"
)
except Exception as exp:
email_notification_logger.error(
f"{today} : Failed Email to {self.email.get_to_email()}"
)
email_notification_logger.error(exp)
# def _reset_base_url() -> str:
# fe = getattr(settings, "FRONTEND_URL", None)
# if not fe or not isinstance(fe, str) or not fe.strip():
# raise ImproperlyConfigured(
# "FRONTEND_URL is not configured. "
# "Set settings.FRONTEND_URL or the FRONTEND_URL environment variable."
# )
# fe = fe.strip()
# if not fe.startswith(("http://", "https://")):
# fe = "https://" + fe
# return urljoin(fe.rstrip("/") + "/", "reset-password")
# def password_reset_email(token: str) -> str:
# """
# Builds the password-reset email body (kept in the same style as your otp_email()).
# """
# base = _reset_base_url()
# sep = "&" if "?" in base else "?"
# reset_link = f"{base}{sep}token={quote(token)}"
# message = f"""\
# Subject: Reset Your Password
# Content-Type: text/html
#
#
# Hi,
# We received a request to reset your password. Click the button below to continue:
#
#
# Reset Password
#
#
# If the button doesn't work, copy and paste this link into your browser:
# {reset_link}
#
# If you didn’t request this, you can safely ignore this email.
# — Wile Security Team
#
#
# """
# return message
# def send_password_reset_email(to_email: str, token: str):
# """
# Usage from your code:
# send_password_reset_email(customer.email, token.token)
# """
# email = Email()
# email.set_subject("Wile — Reset your password")
# email.set_html_message(password_reset_email(token))
# # uses .env if set, otherwise your original default stays unchanged above
# email.set_from_email(os.getenv("DEFAULT_FROM_EMAIL", "no-reply@wile-connect.app"))
# email.set_to_email(to_email)
# OTPEmailThread(email).start()
# # def send_password_reset_email(to_email: str, token: str):
# # """
# # Usage from your code:
# # send_password_reset_email(customer.email, token.token)
# # """
# # email = Email()
# # email.set_subject("Wile — Reset your password")
# # email.get_text_message(password_reset_email(token))
# # # uses .env if set, otherwise your original default stays unchanged above
# # email.set_from_email(os.getenv("DEFAULT_FROM_EMAIL", "no-reply@farmcomp.fi"))
# # email.set_to_email(to_email)
# # send_email(email)
# def send_email(email):
# SMTP2GO_API_KEY = os.getenv("SMTP2GO_API_KEY", None)
# headers = {
# "Content-Type": "application/json",
# "accept": "application/json",
# "X-Smtp2go-Api-Key": SMTP2GO_API_KEY,
# }
# payload = {
# "sender": email.get_from_email(),
# "to": [email.get_to_email()],
# "subject": email.get_subject(),
# "html_body": email.get_html_message(),
# }
# response = requests.post(
# SMTP2GO_URL,
# json=payload,
# headers=headers,
# timeout=10,
# )
# response.raise_for_status() # raises exception on 4xx/5xx
# return response.json()