112 lines
3.7 KiB
Python
112 lines
3.7 KiB
Python
import logging
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
import emails
|
|
from emails.template import JinjaTemplate
|
|
from jose import jwt
|
|
|
|
from app.core.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def send_email(
|
|
email_to: str,
|
|
subject_template: str = "",
|
|
html_template: str = "",
|
|
environment: Optional[Dict[str, Any]] = None,
|
|
) -> None:
|
|
"""
|
|
Send an email using the SMTP settings in app settings.
|
|
|
|
This is disabled by default in development mode.
|
|
To enable, set EMAILS_ENABLED=True in environment variables.
|
|
"""
|
|
if environment is None:
|
|
environment = {}
|
|
if not settings.EMAILS_ENABLED:
|
|
logger.info(f"Email sending disabled. Would have sent email to {email_to}")
|
|
logger.info(f"Subject: {subject_template}")
|
|
logger.info(f"Content: {html_template}")
|
|
logger.info(f"Data: {environment}")
|
|
return
|
|
|
|
# Prepare message
|
|
message = emails.Message(
|
|
subject=JinjaTemplate(subject_template),
|
|
html=JinjaTemplate(html_template),
|
|
mail_from=(settings.EMAILS_FROM_NAME, settings.EMAILS_FROM_EMAIL),
|
|
)
|
|
|
|
# Send message
|
|
smtp_options = {"host": settings.SMTP_HOST, "port": settings.SMTP_PORT}
|
|
if settings.SMTP_TLS:
|
|
smtp_options["tls"] = True
|
|
if settings.SMTP_USER:
|
|
smtp_options["user"] = settings.SMTP_USER
|
|
if settings.SMTP_PASSWORD:
|
|
smtp_options["password"] = settings.SMTP_PASSWORD
|
|
|
|
try:
|
|
response = message.send(to=email_to, render=environment, smtp=smtp_options)
|
|
logger.info(f"Email sent to {email_to}, status: {response.status_code}")
|
|
except Exception as e:
|
|
logger.error(f"Error sending email to {email_to}: {e}")
|
|
|
|
|
|
def send_verification_email(email_to: str, token: str) -> None:
|
|
"""
|
|
Send an email verification link to a new user.
|
|
"""
|
|
server_host = settings.SERVER_HOST
|
|
frontend_host = settings.FRONTEND_HOST or server_host
|
|
|
|
# Build verification link
|
|
verify_link = f"http://{frontend_host}/{settings.EMAIL_VERIFY_URL}?token={token}"
|
|
|
|
subject = f"{settings.PROJECT_NAME} - Verify Your Email"
|
|
html_template = f"""
|
|
<p>Hello,</p>
|
|
<p>Thank you for registering with {settings.PROJECT_NAME}.</p>
|
|
<p>Please verify your email by clicking on the link below:</p>
|
|
<p><a href="{verify_link}">{verify_link}</a></p>
|
|
<p>This link will expire in {settings.VERIFICATION_TOKEN_EXPIRE_HOURS} hours.</p>
|
|
<p>If you did not register for this service, please ignore this email.</p>
|
|
<p>Best regards,<br/>{settings.PROJECT_NAME} Team</p>
|
|
"""
|
|
|
|
send_email(
|
|
email_to=email_to,
|
|
subject_template=subject,
|
|
html_template=html_template,
|
|
)
|
|
|
|
|
|
def send_password_reset_email(email_to: str, token: str) -> None:
|
|
"""
|
|
Send a password reset link to a user.
|
|
"""
|
|
server_host = settings.SERVER_HOST
|
|
frontend_host = settings.FRONTEND_HOST or server_host
|
|
|
|
# Build password reset link
|
|
reset_link = f"http://{frontend_host}/{settings.PASSWORD_RESET_URL}?token={token}"
|
|
|
|
subject = f"{settings.PROJECT_NAME} - Password Reset"
|
|
html_template = f"""
|
|
<p>Hello,</p>
|
|
<p>You have requested to reset your password for {settings.PROJECT_NAME}.</p>
|
|
<p>Please click on the link below to set a new password:</p>
|
|
<p><a href="{reset_link}">{reset_link}</a></p>
|
|
<p>This link will expire in {settings.PASSWORD_RESET_TOKEN_EXPIRE_HOURS} hours.</p>
|
|
<p>If you did not request a password reset, please ignore this email.</p>
|
|
<p>Best regards,<br/>{settings.PROJECT_NAME} Team</p>
|
|
"""
|
|
|
|
send_email(
|
|
email_to=email_to,
|
|
subject_template=subject,
|
|
html_template=html_template,
|
|
) |