import os import boto3 import secrets from datetime import datetime, timedelta from botocore.exceptions import ClientError from typing import Optional class EmailService: def __init__(self): self.region_name = 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") self.email_from = os.getenv("EMAIL_FROM", "noreply@elmeg.xyz") self.frontend_url = os.getenv("FRONTEND_URL", "https://elmeg.xyz") # Initialize SES client if credentials exist if self.aws_access_key_id and self.aws_secret_access_key: self.client = boto3.client( "ses", region_name=self.region_name, aws_access_key_id=self.aws_access_key_id, aws_secret_access_key=self.aws_secret_access_key, ) else: self.client = None print("WARNING: AWS credentials not found. Email service running in dummy mode.") def send_email(self, to_email: str, subject: str, html_content: str, text_content: str): """Send an email using AWS SES""" if not self.client: print(f"DUMMY EMAIL to {to_email}: {subject}") print(text_content) return True try: response = self.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", }, }, }, ) return response except ClientError as e: print(f"Error sending email: {e.response['Error']['Message']}") return False # Global instance email_service = EmailService() # --- Helper Functions (used by auth router) --- 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 signing up. 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)