"""
Email Service - Postal SMTP (primary), Mailgun, or AWS SES fallback
"""
import os
import httpx
import secrets
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime, timedelta
from typing import Optional
# Optional: Keep boto3 for legacy/fallback
try:
import boto3
from botocore.exceptions import ClientError
BOTO3_AVAILABLE = True
except ImportError:
BOTO3_AVAILABLE = False
class EmailService:
def __init__(self):
# Postal SMTP settings (primary - self-hosted)
self.smtp_host = os.getenv("SMTP_HOST")
self.smtp_port = int(os.getenv("SMTP_PORT", "25"))
self.smtp_username = os.getenv("SMTP_USERNAME")
self.smtp_password = os.getenv("SMTP_PASSWORD")
self.smtp_use_tls = os.getenv("SMTP_USE_TLS", "true").lower() == "true"
# Mailgun settings (alternative)
self.mailgun_api_key = os.getenv("MAILGUN_API_KEY")
self.mailgun_domain = os.getenv("MAILGUN_DOMAIN")
self.mailgun_api_base = os.getenv("MAILGUN_API_BASE", "https://api.mailgun.net/v3")
# AWS SES settings (fallback)
self.aws_region = os.getenv("AWS_SES_REGION", "us-east-1")
self.aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID")
self.aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY")
# Common settings
self.email_from = os.getenv("EMAIL_FROM", "noreply@elmeg.xyz")
self.frontend_url = os.getenv("FRONTEND_URL", "https://elmeg.xyz")
# Determine which provider to use (priority: SMTP -> Mailgun -> SES -> Dummy)
if self.smtp_host and self.smtp_username and self.smtp_password:
self.provider = "smtp"
print(f"Email service: Using SMTP ({self.smtp_host}:{self.smtp_port})")
elif self.mailgun_api_key and self.mailgun_domain:
self.provider = "mailgun"
print(f"Email service: Using Mailgun ({self.mailgun_domain})")
elif BOTO3_AVAILABLE and self.aws_access_key_id and self.aws_secret_access_key:
self.provider = "ses"
self.ses_client = boto3.client(
"ses",
region_name=self.aws_region,
aws_access_key_id=self.aws_access_key_id,
aws_secret_access_key=self.aws_secret_access_key,
)
print("Email service: Using AWS SES")
else:
self.provider = "dummy"
print("WARNING: No email credentials found. Running in dummy mode.")
def send_email(self, to_email: str, subject: str, html_content: str, text_content: str):
"""Send an email using configured provider"""
if self.provider == "smtp":
return self._send_smtp(to_email, subject, html_content, text_content)
elif self.provider == "mailgun":
return self._send_mailgun(to_email, subject, html_content, text_content)
elif self.provider == "ses":
return self._send_ses(to_email, subject, html_content, text_content)
else:
return self._send_dummy(to_email, subject, text_content)
def _send_smtp(self, to_email: str, subject: str, html_content: str, text_content: str):
"""Send email via SMTP (Postal or any SMTP server)"""
try:
# Create message
msg = MIMEMultipart("alternative")
msg["Subject"] = subject
msg["From"] = f"Elmeg <{self.email_from}>"
msg["To"] = to_email
# Attach text and HTML parts
msg.attach(MIMEText(text_content, "plain"))
msg.attach(MIMEText(html_content, "html"))
# Connect and send
if self.smtp_use_tls:
server = smtplib.SMTP(self.smtp_host, self.smtp_port)
server.starttls()
else:
server = smtplib.SMTP(self.smtp_host, self.smtp_port)
server.login(self.smtp_username, self.smtp_password)
server.sendmail(self.email_from, to_email, msg.as_string())
server.quit()
print(f"Email sent via SMTP to {to_email}")
return True
except Exception as e:
print(f"Error sending email via SMTP: {e}")
return False
def _send_mailgun(self, to_email: str, subject: str, html_content: str, text_content: str):
"""Send email via Mailgun API"""
try:
with httpx.Client() as client:
response = client.post(
f"{self.mailgun_api_base}/{self.mailgun_domain}/messages",
auth=("api", self.mailgun_api_key),
data={
"from": f"Elmeg <{self.email_from}>",
"to": to_email,
"subject": subject,
"text": text_content,
"html": html_content,
},
timeout=30.0,
)
if response.status_code == 200:
print(f"Email sent via Mailgun to {to_email}")
return response.json()
else:
print(f"Mailgun error: {response.status_code} - {response.text}")
return False
except Exception as e:
print(f"Error sending email via Mailgun: {e}")
return False
def _send_ses(self, to_email: str, subject: str, html_content: str, text_content: str):
"""Send email via AWS SES (fallback)"""
try:
response = self.ses_client.send_email(
Source=self.email_from,
Destination={"ToAddresses": [to_email]},
Message={
"Subject": {"Data": subject, "Charset": "UTF-8"},
"Body": {
"Html": {"Data": html_content, "Charset": "UTF-8"},
"Text": {"Data": text_content, "Charset": "UTF-8"},
},
},
)
print(f"Email sent via SES to {to_email}")
return response
except ClientError as e:
print(f"SES error: {e.response['Error']['Message']}")
return False
def _send_dummy(self, to_email: str, subject: str, text_content: str):
"""Dummy mode - print to console"""
print(f"\n{'='*50}")
print(f"DUMMY EMAIL to: {to_email}")
print(f"Subject: {subject}")
print("-" * 50)
print(text_content)
print(f"{'='*50}\n")
return True
# Global instance
email_service = EmailService()
# --- Helper Functions ---
def generate_token() -> str:
"""Generate a secure random token"""
return secrets.token_urlsafe(32)
def get_verification_expiry() -> datetime:
"""Get expiration time for verification token (48 hours)"""
return datetime.utcnow() + timedelta(hours=48)
def get_reset_expiry() -> datetime:
"""Get expiration time for reset token (1 hour)"""
return datetime.utcnow() + timedelta(hours=1)
def send_verification_email(to_email: str, token: str):
"""Send account verification email"""
verify_url = f"{email_service.frontend_url}/verify-email?token={token}"
subject = "Verify your Elmeg account"
html_content = f"""
Welcome to Elmeg! 🎵
Thanks for joining the community. Please verify your email address to get started.
Or copy this link to your browser:
{verify_url}
If you didn't create an account, you can safely ignore this email.
"""
text_content = f"""
Welcome to Elmeg!
Please verify your email address by visiting this link:
{verify_url}
If you didn't create an account, safely ignore this email.
"""
return email_service.send_email(to_email, subject, html_content, text_content)
def send_password_reset_email(to_email: str, token: str):
"""Send password reset email"""
reset_url = f"{email_service.frontend_url}/reset-password?token={token}"
subject = "Reset your Elmeg password"
html_content = f"""
Password Reset Request
We received a request to reset your password. Click the button below to choose a new one.
Or copy this link to your browser:
{reset_url}
⏰ This link expires in 1 hour.
If you didn't request a password reset, you can safely ignore this email.
"""
text_content = f"""
Reset your Elmeg password
Click the link below to choose a new password:
{reset_url}
This link expires in 1 hour.
If you didn't request a password reset, safely ignore this email.
"""
return email_service.send_email(to_email, subject, html_content, text_content)
def send_reply_notification_email(to_email: str, replier_name: str, original_content: str, link: str):
"""Send notification when someone replies to user's comment/review"""
subject = f"{replier_name} replied to your comment on Elmeg"
html_content = f"""
New Reply
{replier_name} replied to your comment:
"{original_content[:100]}{'...' if len(original_content) > 100 else ''}"
You can disable these notifications in your settings.
"""
text_content = f"""
{replier_name} replied to your comment on Elmeg
Your comment: "{original_content[:100]}{'...' if len(original_content) > 100 else ''}"
View the reply: {email_service.frontend_url}{link}
To disable these notifications, visit {email_service.frontend_url}/settings
"""
return email_service.send_email(to_email, subject, html_content, text_content)
def send_mention_notification_email(to_email: str, mentioner_name: str, context: str, link: str):
"""Send notification when someone mentions user in a comment"""
subject = f"{mentioner_name} mentioned you on Elmeg"
html_content = f"""
You were mentioned!
{mentioner_name} mentioned you in a comment:
"{context[:150]}{'...' if len(context) > 150 else ''}"
You can disable these notifications in your settings.
"""
text_content = f"""
{mentioner_name} mentioned you in a comment on Elmeg
"{context[:150]}{'...' if len(context) > 150 else ''}"
View the comment: {email_service.frontend_url}{link}
To disable these notifications, visit {email_service.frontend_url}/settings
"""
return email_service.send_email(to_email, subject, html_content, text_content)
def send_chase_notification_email(to_email: str, song_title: str, show_date: str, venue_name: str, link: str):
"""Send notification when a song the user is chasing gets played"""
subject = f"Chase Alert: {song_title} was played!"
html_content = f"""
Chase Alert!
A song you're chasing was just played!
{song_title}
{show_date} - {venue_name}
You can disable chase alerts in your settings.
"""
text_content = f"""
Chase Alert: {song_title} was played!
{show_date} at {venue_name}
View the performance: {email_service.frontend_url}{link}
To disable chase alerts, visit {email_service.frontend_url}/settings
"""
return email_service.send_email(to_email, subject, html_content, text_content)