如何通过AWS SES获取收件人邮箱地址及邮件投递状态？

核心思路 #

AWS SES配置集默认不会在事件日志中直接关联收件人邮箱，需在发送邮件时主动添加自定义标识，再结合日志/事件通知完成关联。

具体步骤 #

1. 发送邮件时嵌入自定义关联标识 #

发送邮件（API/SMTP方式）时，在邮件头中添加两个自定义字段：

• X-Customer-Email: 直接填入收件人邮箱
• X-Message-ID: 生成唯一标识（可使用时间戳+表单提交ID组合，确保全局唯一）

同时，将X-Message-ID、收件人邮箱、完整邮件内容存入自有数据库（如DynamoDB、MySQL），作为原始发送记录。

boto3发送示例：

import boto3
from uuid import uuid4
from datetime import datetime

ses = boto3.client('ses', region_name='us-east-1')
recipient = 'customer@example.com'
msg_id = f"web-form-{uuid4()}"  # 生成唯一ID

# 保存原始记录到数据库（以DynamoDB为例）
dynamodb = boto3.resource('dynamodb')
db_table = dynamodb.Table('ses_outbound_logs')
db_table.put_item(
    Item={
        'msg_id': msg_id,
        'recipient': recipient,
        'subject': '表单提交确认',
        'body': '感谢您提交表单，我们将尽快处理',
        'send_time': datetime.utcnow().isoformat(),
        'delivery_status': 'Pending'
    }
)

# 发送邮件并附加自定义头
ses.send_email(
    Source='your-verified-sender@example.com',
    Destination={'ToAddresses': [recipient]},
    Message={
        'Subject': {'Data': '表单提交确认'},
        'Body': {'Text': {'Data': '感谢您提交表单，我们将尽快处理'}}
    },
    ConfigurationSetName='your-config-set',
    Headers={
        'X-Customer-Email': recipient,
        'X-Message-ID': msg_id
    }
)

2. 配置集日志/事件通知 #

确保配置集已开启Send、Delivery、Bounce、Complaint四类事件的记录，可选择输出到：

• CloudWatch Logs: 适合实时查询和分析
• S3: 适合批量归档和离线处理
• Lambda: 适合实时触发状态更新

3. 关联发送记录与投递状态 #

方式一：CloudWatch Logs查询关联

使用CloudWatch Insights编写查询语句，直接提取并关联收件人邮箱与状态：

fields @timestamp, 
       eventType as delivery_status,
       mail.headers['X-Customer-Email'] as recipient,
       mail.headers['X-Message-ID'] as msg_id
filter eventType in ['Send', 'Delivery', 'Bounce', 'Complaint']
sort @timestamp desc

可将查询结果导出，或通过API拉取后更新自有数据库。

方式二：Lambda实时更新

给配置集添加Lambda事件目标，当SES产生事件时自动触发Lambda，解析自定义头并更新数据库状态：

import boto3
import json
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
db_table = dynamodb.Table('ses_outbound_logs')

def lambda_handler(event, context):
    for record in event['Records']:
        ses_event = json.loads(record['body'])
        event_type = ses_event['eventType']
        msg_id = next(h['value'] for h in ses_event['mail']['headers'] if h['name'] == 'X-Message-ID')
        timestamp = datetime.fromtimestamp(ses_event['timestamp']/1000).isoformat()
        
        # 更新数据库中的投递状态
        db_table.update_item(
            Key={'msg_id': msg_id},
            UpdateExpression='SET delivery_status = :status, updated_at = :ts',
            ExpressionAttributeValues={
                ':status': event_type,
                ':ts': timestamp
            }
        )
    return {'statusCode': 200}

内容的提问来源于stack exchange，提问作者Mirza