Amazon SES 电商邮件运营：从零到一搭建发送平台

import smtplib
from email.mime.text import MIMEText

# SMTP 配置
smtp_host = "email-smtp.us-east-1.amazonaws.com"
smtp_port = 587
smtp_username = "YOUR_SMTP_USERNAME"
smtp_password = "YOUR_SMTP_PASSWORD"

# 发送邮件
msg = MIMEText("您的订单已发货！", "plain", "utf-8")
msg["Subject"] = "订单发货通知"
msg["From"] = "noreply@example.com"
msg["To"] = "customer@example.com"

with smtplib.SMTP(smtp_host, smtp_port) as server:
    server.starttls()
    server.login(smtp_username, smtp_password)
    server.send_message(msg)

import boto3

ses_client = boto3.client('ses', region_name='us-east-1')

response = ses_client.send_email(
    Source='noreply@example.com',
    Destination={'ToAddresses': ['customer@example.com']},
    Message={
        'Subject': {'Data': '订单发货通知', 'Charset': 'UTF-8'},
        'Body': {
            'Text': {'Data': '您的订单已发货！', 'Charset': 'UTF-8'}
        }
    }
)

# 查询预热进度
aws sesv2 get-dedicated-ip --ip 192.0.2.1

# 输出示例
{
  "DedicatedIp": {
    "Ip": "192.0.2.1",
    "WarmupStatus": "IN_PROGRESS",
    "WarmupPercentage": 45,  # 当前预热进度 45%
    "PoolName": "my-pool"
  }
}

# 手动调整预热进度（高级用户）
aws sesv2 put-dedicated-ip-warmup-attributes \
  --ip 192.0.2.1 \
  --warmup-percentage 50

# 创建托管 IP 池
aws sesv2 create-dedicated-ip-pool \
  --pool-name managed-pool \
  --scaling-mode MANAGED

# AWS 会自动：
# 1. 根据发送量决定需要几个 IP
# 2. 自动申请和预热新 IP
# 3. 自动释放不需要的 IP

# 创建 Configuration Set
aws sesv2 create-configuration-set \
  --configuration-set-name ecommerce-emails

# 添加事件目标（发送到 CloudWatch）
aws sesv2 create-configuration-set-event-destination \
  --configuration-set-name ecommerce-emails \
  --event-destination-name cloudwatch-destination \
  --event-destination '{
    "CloudWatchDestination": {
      "DimensionConfigurations": [{
        "DimensionName": "campaign",
        "DimensionValueSource": "MESSAGE_TAG",
        "DefaultDimensionValue": "default"
      }]
    }
  }'

# 测试成功送达
test_addresses = {
    'success': 'success@simulator.amazonses.com',
    'bounce': 'bounce@simulator.amazonses.com',
    'complaint': 'complaint@simulator.amazonses.com',
    'suppressed': 'suppressed@simulator.amazonses.com'
}

# 示例：清理退信邮箱
def clean_bounce_list(email_list, bounce_records):
    """移除硬退信邮箱"""
    hard_bounces = {r['email'] for r in bounce_records if r['type'] == 'Permanent'}
    return [email for email in email_list if email not in hard_bounces]

# 示例：用户分群
def segment_users(users):
    """根据活跃度分群"""
    return {
        'active': [u for u in users if u['last_open_days'] < 30],
        'inactive': [u for u in users if 30 <= u['last_open_days'] < 90],
        'dormant': [u for u in users if u['last_open_days'] >= 90]
    }

# 计算关键指标
def calculate_metrics(campaign_data):
    sent = campaign_data['sent']
    delivered = campaign_data['delivered']
    opened = campaign_data['opened']
    clicked = campaign_data['clicked']
    
    return {
        'delivery_rate': delivered / sent * 100,
        'open_rate': opened / delivered * 100,
        'click_rate': clicked / delivered * 100,
        'ctr': clicked / opened * 100 if opened > 0 else 0
    }

# 1. 启用 VDM（通过控制台或 CLI）
aws sesv2 put-account-vdm-attributes \
  --vdm-attributes '{
    "VdmEnabled": "ENABLED",
    "DashboardAttributes": {"EngagementMetrics": "ENABLED"},
    "GuardianAttributes": {"OptimizedSharedDelivery": "ENABLED"}
  }'

# 2. 启用参与度追踪（打开和点击）
aws sesv2 put-account-vdm-attributes \
  --vdm-attributes '{
    "DashboardAttributes": {
      "EngagementMetrics": "ENABLED"
    }
  }'

 	# 通过 CLI 导出 VDM 数据
aws sesv2 create-export-job \
  --export-data-source '{
    "MetricsDataSource": {
      "Namespace": "VDM",
      "Metrics": [
        {"Name": "SEND"},
        {"Name": "DELIVERY"},
        {"Name": "OPEN"},
        {"Name": "CLICK"}
      ],
      "StartDate": "2024-12-01T00:00:00Z",
      "EndDate": "2024-12-07T23:59:59Z"
    }
  }' \
  --export-destination '{
    "DataFormat": "CSV",
    "S3Url": "s3://my-bucket/vdm-exports/"
  }'

 	import pandas as pd

# 读取导出的 CSV 数据
df = pd.read_csv('vdm-export.csv')

# 计算关键指标
total_sent = df['SEND'].sum()
total_delivered = df['DELIVERY'].sum()
total_opens = df['OPEN'].sum()
total_clicks = df['CLICK'].sum()

print(f"送达率: {total_delivered/total_sent:.2%}")
print(f"打开率 (UV): {total_opens/total_delivered:.2%}")
print(f"点击率 (UV): {total_clicks/total_delivered:.2%}")

# 创建退信率告警
aws cloudwatch put-metric-alarm \
  --alarm-name ses-high-bounce-rate \
  --alarm-description "SES bounce rate > 5%" \
  --metric-name Reputation.BounceRate \
  --namespace AWS/SES \
  --statistic Average \
  --period 300 \
  --threshold 0.05 \
  --comparison-operator GreaterThanThreshold \
  --evaluation-periods 1 \
  --alarm-actions arn:aws:sns:us-east-1:123456789012:ses-alerts

# 简单的 A/B 测试实现
def ab_test_send(user_list, subject_a, subject_b, split_ratio=0.5):
    """对比两个主题行的效果"""
    import random
    
    group_a = random.sample(user_list, int(len(user_list) * split_ratio))
    group_b = [u for u in user_list if u not in group_a]
    
    # 发送 A 组
    send_campaign(group_a, subject=subject_a, campaign_id='test_a')
    
    # 发送 B 组
    send_campaign(group_b, subject=subject_b, campaign_id='test_b')
    
    return {'group_a': len(group_a), 'group_b': len(group_b)}

# 使用 SES 模板实现个性化
template = {
    "TemplateName": "order-confirmation",
    "SubjectPart": "{{customer_name}}，您的订单 {{order_id}} 已确认",
    "HtmlPart": """
    <h1>感谢您的购买，{{customer_name}}！</h1>
    <p>订单号：{{order_id}}</p>
    <p>预计送达：{{delivery_date}}</p>
    """
}

# 发送个性化邮件
ses_client.send_templated_email(
    Source='noreply@example.com',
    Destination={'ToAddresses': ['customer@example.com']},
    Template='order-confirmation',
    TemplateData=json.dumps({
        'customer_name': '张三',
        'order_id': 'ORD-12345',
        'delivery_date': '2024-12-10'
    })
)

# 使用 SNS + Lambda 自动处理
def lambda_handler(event, context):
    """处理 SES 通知"""
    message = json.loads(event['Records'][0]['Sns']['Message'])
    
    if message['notificationType'] == 'Bounce':
        if message['bounce']['bounceType'] == 'Permanent':
            # 从数据库移除硬退信邮箱
            remove_from_list(message['bounce']['bouncedRecipients'])
    
    elif message['notificationType'] == 'Complaint':
        # 立即停止向投诉用户发送
        add_to_suppression_list(message['complaint']['complainedRecipients'])