ses_email_service.py•5.81 kB
"""
SES-based email service for Lambda functions.
Replaces SMTP-based email notifications with AWS SES.
"""
import json
from datetime import datetime
from typing import Dict, Any, List
import boto3
import pystache
class SESEmailService:
"""
Email service using AWS SES instead of SMTP.
Supports HTML templates stored in S3.
"""
def __init__(self, ses_client, s3_client, templates_bucket: str):
"""
Initialize SES email service.
Args:
ses_client: Boto3 SES client
s3_client: Boto3 S3 client
templates_bucket: S3 bucket containing email templates
"""
self.ses_client = ses_client
self.s3_client = s3_client
self.templates_bucket = templates_bucket
def send_notification(self, task_config: Dict[str, Any], new_items: List[Dict[str, Any]]) -> bool:
"""
Send email notification about new items found.
Args:
task_config: Task configuration dictionary
new_items: List of new items to include in email
Returns:
True if email was sent successfully, False otherwise
"""
try:
# Get email template from S3
template_html = self._get_email_template()
# Prepare template data
template_data = {
'task_description': task_config['task_description'],
'timestamp': datetime.now().strftime('%Y-%m-%d %H:%M:%S UTC'),
'new_items_count': len(new_items),
'new_items': new_items
}
# Render HTML email
html_body = pystache.render(template_html, template_data)
# Create plain text version
text_body = self._create_text_version(task_config, new_items)
# Send email via SES
response = self.ses_client.send_email(
Source=task_config['sender_email'],
Destination={
'ToAddresses': [task_config['recipient_email']]
},
Message={
'Subject': {
'Data': f"New items found: {task_config['task_description'][:50]}...",
'Charset': 'UTF-8'
},
'Body': {
'Text': {
'Data': text_body,
'Charset': 'UTF-8'
},
'Html': {
'Data': html_body,
'Charset': 'UTF-8'
}
}
}
)
print(f"Email sent successfully. MessageId: {response['MessageId']}")
return True
except Exception as e:
print(f"Error sending email notification: {e}")
return False
def _get_email_template(self) -> str:
"""
Get email template from S3.
Returns:
HTML template string
"""
try:
response = self.s3_client.get_object(
Bucket=self.templates_bucket,
Key='notification_template.html'
)
return response['Body'].read().decode('utf-8')
except Exception as e:
print(f"Error getting email template from S3: {e}")
# Return a simple fallback template
return self._get_fallback_template()
def _get_fallback_template(self) -> str:
"""Fallback HTML template if S3 template is not available."""
return """
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<title>TimeLooker Notification</title>
</head>
<body style="font-family: Arial, sans-serif; line-height: 1.6; color: #333;">
<h1>🔍 TimeLooker Notification</h1>
<p><strong>Task:</strong> {{task_description}}</p>
<p><strong>Time:</strong> {{timestamp}}</p>
<p><strong>New items found:</strong> {{new_items_count}}</p>
<h2>New Items:</h2>
{{#new_items}}
<div style="border: 1px solid #ddd; padding: 10px; margin: 10px 0;">
<h3>{{name}}</h3>
<p><strong>Source:</strong> {{source}}</p>
<p><strong>Location:</strong> {{location}}</p>
{{#description}}<p>{{description}}</p>{{/description}}
{{#url}}<p><a href="{{url}}">{{url}}</a></p>{{/url}}
</div>
{{/new_items}}
<p><em>This is an automated message from TimeLooker.</em></p>
</body>
</html>
"""
def _create_text_version(self, task_config: Dict[str, Any], new_items: List[Dict[str, Any]]) -> str:
"""
Create plain text version of the email.
Args:
task_config: Task configuration
new_items: List of new items
Returns:
Plain text email body
"""
text = f"""TimeLooker Notification
Task: {task_config['task_description']}
Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S UTC')}
New items found: {len(new_items)}
New Items:
{'='*50}
"""
for i, item in enumerate(new_items, 1):
text += f"{i}. {item.get('name', 'Unknown Item')}\n"
text += f" Source: {item.get('source', 'Unknown')}\n"
text += f" Location: {item.get('location', 'Not specified')}\n"
if item.get('description'):
text += f" Description: {item.get('description')}\n"
if item.get('url'):
text += f" URL: {item.get('url')}\n"
text += "\n"
text += f"""
{'='*50}
This is an automated message from TimeLooker.
"""
return text