""" Email Service - Mailgun (primary) with AWS SES fallback """ import os import httpx import secrets 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): # Mailgun settings (primary) 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 if 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 == "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_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.

Verify Email Address

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.

Reset Password

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)