Amazon SES 电商邮件运营进阶：大规模营销活动和邮件高可用

# 计算所需配额
def calculate_quota(user_count, email_per_user, campaign_days):
    """
    user_count: 目标用户数
    email_per_user: 每用户邮件数（预热+正式）
    campaign_days: 活动天数
    """
    total_emails = user_count * email_per_user
    daily_quota = total_emails / campaign_days * 1.2  # 20% 缓冲
    
    # 计算所需发送速率（假设集中在 8 小时发送）
    sending_hours = 8
    required_rate = daily_quota / (sending_hours * 3600)
    
    return {
        'daily_quota': int(daily_quota),
        'sending_rate': int(required_rate) + 1
    }

# 示例：双十一活动
quota = calculate_quota(
    user_count=1_000_000,      # 100 万用户
    email_per_user=3,          # 预热 2 封 + 正式 1 封
    campaign_days=5            # 5 天活动期
)
print(f"所需配额：{quota['daily_quota']:,} 封/天")
print(f"所需速率：{quota['sending_rate']} 封/秒")

import boto3
from datetime import datetime, timedelta

class StandardAutoWarmupManager:
    """标准自动预热管理器"""
    
    def __init__(self, dedicated_ip):
        self.ses_client = boto3.client('sesv2')
        self.dedicated_ip = dedicated_ip
    
    def get_warmup_status(self):
        """获取预热状态和百分比"""
        response = self.ses_client.get_dedicated_ip(Ip=self.dedicated_ip)
        ip_info = response['DedicatedIp']
        
        return {
            'ip': ip_info['Ip'],
            'warmup_status': ip_info['WarmupStatus'],
            'warmup_percentage': ip_info.get('WarmupPercentage', 0),
            'pool_name': ip_info.get('PoolName', 'default')
        }
    
    def calculate_sending_capacity(self, target_volume):
        """根据预热百分比计算当前可发送量"""
        status = self.get_warmup_status()
        percentage = status['warmup_percentage']
        
        # 当前可通过 DIP 发送的量
        dip_capacity = int(target_volume * (percentage / 100))
        
        # 剩余流量会通过 Public IP 发送
        public_ip_volume = target_volume - dip_capacity
        
        return {
            'warmup_percentage': percentage,
            'dip_capacity': dip_capacity,
            'public_ip_volume': public_ip_volume,
            'total_volume': target_volume
        }
    
    def adjust_warmup_percentage(self, new_percentage):
        """手动调整预热百分比（高级功能）"""
        try:
            self.ses_client.put_dedicated_ip_warmup_attributes(
                Ip=self.dedicated_ip,
                WarmupPercentage=new_percentage
            )
            print(f"✅ 预热百分比已调整为 {new_percentage}%")
            return True
        except Exception as e:
            print(f"❌ 调整失败：{e}")
            return False
    
    def monitor_warmup_progress(self):
        """监控预热进度"""
        status = self.get_warmup_status()
        
        print(f"IP: {status['ip']}")
        print(f"预热状态: {status['warmup_status']}")
        print(f"预热进度: {status['warmup_percentage']}%")
        
        # 根据进度给出建议
        percentage = status['warmup_percentage']
        if percentage < 25:
            print("📊 建议：当前处于预热初期，保持低发送量")
        elif percentage < 50:
            print("📊 建议：可以逐步增加发送量")
        elif percentage < 75:
            print("📊 建议：接近预热完成，可以加大发送量")
        else:
            print("📊 建议：预热接近完成，可以全力发送")
        
        return status

# 使用示例
manager = StandardAutoWarmupManager('192.0.2.1')

# 监控预热进度
status = manager.monitor_warmup_progress()

# 计算当前可发送量（目标 100 万封）
capacity = manager.calculate_sending_capacity(1_000_000)
print(f"\n当前可通过 DIP 发送：{capacity['dip_capacity']:,} 封")
print(f"通过 Public IP 发送：{capacity['public_ip_volume']:,} 封")

# 如果需要加速预热（谨慎使用）
# manager.adjust_warmup_percentage(60)

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

# 将 Configuration Set 关联到托管池
aws sesv2 put-configuration-set-delivery-options \
  --configuration-set-name marketing-campaigns \
  --sending-pool-name ecommerce-managed-pool

def select_warmup_users(all_users, warmup_percentage):
    """根据预热百分比选择发送用户"""
    # 预热初期（< 25%）：只发送给最活跃用户
    if warmup_percentage < 25:
        return [u for u in all_users if u['engagement_score'] > 80]
    
    # 预热中期（25-50%）：活跃 + 中等活跃用户
    elif warmup_percentage < 50:
        return [u for u in all_users if u['engagement_score'] > 50]
    
    # 预热后期（50-75%）：大部分用户
    elif warmup_percentage < 75:
        return [u for u in all_users if u['engagement_score'] > 30]
    
    # 预热完成（> 75%）：所有用户
    else:
        return all_users

def monitor_warmup_health(dedicated_ip):
    """监控预热健康度"""
    cloudwatch = boto3.client('cloudwatch')
    
    # 获取过去 1 小时的指标
    metrics = cloudwatch.get_metric_statistics(
        Namespace='AWS/SES',
        MetricName='Reputation.BounceRate',
        Dimensions=[{'Name': 'DedicatedIp', 'Value': dedicated_ip}],
        StartTime=datetime.now() - timedelta(hours=1),
        EndTime=datetime.now(),
        Period=3600,
        Statistics=['Average']
    )
    
    bounce_rate = metrics['Datapoints'][0]['Average'] if metrics['Datapoints'] else 0
    
    # 告警阈值
    if bounce_rate > 0.05:  # 5%
        send_alert(f"⚠️ 退信率过高：{bounce_rate:.2%}")
        return False
    
    return True

class MultiRegionSender:
    def __init__(self):
        self.regions = [
            {'name': 'us-east-1', 'priority': 1, 'client': boto3.client('ses', region_name='us-east-1')},
            {'name': 'us-west-2', 'priority': 2, 'client': boto3.client('ses', region_name='us-west-2')}
        ]
        self.current_region = self.regions[0]
    
    def send_email_with_failover(self, email_data, max_retries=2):
        """带故障转移的邮件发送"""
        for region in self.regions:
            try:
                response = region['client'].send_email(**email_data)
                print(f"✅ 发送成功 via {region['name']}")
                return response
            
            except Exception as e:
                print(f"❌ {region['name']} 发送失败: {e}")
                
                # 如果不是最后一个区域，尝试下一个
                if region != self.regions[-1]:
                    print(f"🔄 切换到 {self.regions[self.regions.index(region) + 1]['name']}")
                    continue
                else:
                    # 所有区域都失败，记录到死信队列
                    self.send_to_dlq(email_data)
                    raise
    
    def send_to_dlq(self, email_data):
        """发送到死信队列，稍后重试"""
        sqs = boto3.client('sqs')
        sqs.send_message(
            QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/email-dlq',
            MessageBody=json.dumps(email_data)
        )

def check_ses_health(region):
    """检查 SES 服务健康状态"""
    ses_client = boto3.client('ses', region_name=region)
    cloudwatch = boto3.client('cloudwatch', region_name=region)
    
    try:
        # 检查账户状态
        account_info = ses_client.get_account()
        
        # 检查发送配额
        quota = account_info['SendQuota']
        usage_percent = (quota['SentLast24Hours'] / quota['Max24HourSend']) * 100
        
        # 检查最近的错误率
        metrics = cloudwatch.get_metric_statistics(
            Namespace='AWS/SES',
            MetricName='Send',
            StartTime=datetime.now() - timedelta(minutes=5),
            EndTime=datetime.now(),
            Period=300,
            Statistics=['Sum']
        )
        
        return {
            'healthy': usage_percent < 90,  # 使用率 < 90%
            'usage_percent': usage_percent,
            'quota_remaining': quota['Max24HourSend'] - quota['SentLast24Hours']
        }
    
    except Exception as e:
        return {'healthy': False, 'error': str(e)}

def send_bulk_campaign(recipients, template_name):
    """批量发送模板化邮件"""
    ses_client = boto3.client('sesv2')
    
    # 准备批量目标（最多 50 个/批次）
    bulk_destinations = []
    for recipient in recipients:
        bulk_destinations.append({
            'Destination': {
                'ToAddresses': [recipient['email']]
            },
            'ReplacementTemplateData': json.dumps({
                'name': recipient['name'],
                'product': recipient['recommended_product'],
                'discount': recipient['discount_code']
            })
        })
    
    # 分批发送
    batch_size = 50
    for i in range(0, len(bulk_destinations), batch_size):
        batch = bulk_destinations[i:i + batch_size]
        
        try:
            response = ses_client.send_bulk_email(
                FromEmailAddress='marketing@example.com',
                DefaultContent={
                    'Template': {
                        'TemplateName': template_name,
                        'TemplateData': json.dumps({'default': 'value'})
                    }
                },
                BulkEmailEntries=batch,
                ConfigurationSetName='marketing-campaign'
            )
            
            print(f"✅ 批次 {i//batch_size + 1} 发送成功：{len(batch)} 封")
            
        except Exception as e:
            print(f"❌ 批次 {i//batch_size + 1} 发送失败：{e}")

import asyncio
import aioboto3
from asyncio import Semaphore

class RateLimitedSender:
    def __init__(self, max_rate_per_second=50):
        self.max_rate = max_rate_per_second
        self.semaphore = Semaphore(max_rate_per_second)
        self.sent_count = 0
        self.start_time = None
    
    async def send_with_rate_limit(self, email_data):
        """带速率限制的发送"""
        async with self.semaphore:
            if self.start_time is None:
                self.start_time = asyncio.get_event_loop().time()
            
            # 检查是否需要等待
            elapsed = asyncio.get_event_loop().time() - self.start_time
            if elapsed < 1.0 and self.sent_count >= self.max_rate:
                await asyncio.sleep(1.0 - elapsed)
                self.sent_count = 0
                self.start_time = asyncio.get_event_loop().time()
            
            # 发送邮件
            async with aioboto3.Session().client('ses', region_name='us-east-1') as ses:
                response = await ses.send_email(**email_data)
                self.sent_count += 1
                return response

# 使用示例
async def send_campaign(recipients):
    sender = RateLimitedSender(max_rate_per_second=50)
    tasks = [sender.send_with_rate_limit(r) for r in recipients]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

from botocore.config import Config

# 配置连接池
config = Config(
    max_pool_connections=100,  # 最大连接数
    retries={'max_attempts': 3, 'mode': 'adaptive'}
)

ses_client = boto3.client('ses', config=config)

# CloudWatch 告警配置
CRITICAL_ALARMS = {
    'bounce_rate': {
        'threshold': 0.05,  # 5%
        'evaluation_periods': 1,
        'action': 'pause_sending'
    },
    'complaint_rate': {
        'threshold': 0.001,  # 0.1%
        'evaluation_periods': 1,
        'action': 'pause_sending'
    },
    'send_error_rate': {
        'threshold': 0.10,  # 10%
        'evaluation_periods': 2,
        'action': 'switch_region'
    },
    'quota_usage': {
        'threshold': 0.90,  # 90%
        'evaluation_periods': 1,
        'action': 'request_increase'
    }
}

import boto3

cloudwatch = boto3.client('cloudwatch')

# 创建送达率告警
cloudwatch.put_metric_alarm(
    AlarmName='ses-low-delivery-rate',
    ComparisonOperator='LessThanThreshold',
    EvaluationPeriods=1,
    MetricName='Reputation.DeliveryRate',
    Namespace='AWS/SES',
    Period=300,  # 5 分钟
    Statistic='Average',
    Threshold=0.95,  # 95%
    ActionsEnabled=True,
    AlarmActions=[
        'arn:aws:sns:us-east-1:123456789012:ses-alerts'
    ],
    AlarmDescription='SES 送达率低于 95%'
)

# 创建退信率告警
cloudwatch.put_metric_alarm(
    AlarmName='ses-high-bounce-rate',
    ComparisonOperator='GreaterThanThreshold',
    EvaluationPeriods=1,
    MetricName='Reputation.BounceRate',
    Namespace='AWS/SES',
    Period=300,
    Statistic='Average',
    Threshold=0.05,  # 5%
    ActionsEnabled=True,
    AlarmActions=[
        'arn:aws:sns:us-east-1:123456789012:ses-alerts'
    ],
    AlarmDescription='SES 退信率超过 5%'
)

# 每小时导出一次 VDM 数据
aws sesv2 create-export-job \
  --export-data-source '{
    "MetricsDataSource": {
      "Namespace": "VDM",
      "Dimensions": {
        "CONFIGURATION_SET": ["black-friday-campaign"]
      },
      "Metrics": [
        {"Name": "SEND"},
        {"Name": "DELIVERY"},
        {"Name": "BOUNCE"},
        {"Name": "COMPLAINT"}
      ],
      "StartDate": "2024-11-24T00:00:00Z",
      "EndDate": "2024-11-24T23:59:59Z"
    }
  }' \
  --export-destination '{
    "DataFormat": "CSV",
    "S3Url": "s3://campaign-metrics/black-friday/"
  }'

def lambda_handler(event, context):
    """CloudWatch 告警触发的自动响应"""
    alarm_name = event['detail']['alarmName']
    alarm_state = event['detail']['state']['value']
    
    if alarm_state == 'ALARM':
        if 'bounce_rate' in alarm_name or 'complaint_rate' in alarm_name:
            # 暂停发送，等待人工介入
            pause_all_campaigns()
            send_urgent_notification("🚨 发送已暂停：声誉指标异常")
        
        elif 'send_error_rate' in alarm_name:
            # 切换到备用区域
            switch_to_backup_region()
            send_notification("🔄 已切换到备用区域")
        
        elif 'quota_usage' in alarm_name:
            # 自动申请配额提升
            request_quota_increase()
            send_notification("📈 已提交配额提升申请")

def pause_all_campaigns():
    """暂停所有活动发送"""
    # 停止 SQS 消费
    # 标记活动状态为暂停
    pass

def switch_to_backup_region():
    """切换到备用区域"""
    # 更新 Route 53 记录
    # 重定向流量到备用区域
    pass

def manage_dedicated_ip_lifecycle(campaign_schedule):
    """管理专用 IP 生命周期"""
    ses_client = boto3.client('sesv2')
    
    # 大促前 4 周申请
    if days_until_campaign(campaign_schedule) == 28:
        response = ses_client.request_dedicated_ips(
            PoolName='campaign-pool',
            WarmupEnabled=True
        )
        print(f"✅ 已申请专用 IP：{response['DedicatedIps']}")
    
    # 大促结束后 1 周释放
    elif days_after_campaign(campaign_schedule) == 7:
        # 释放专用 IP
        for ip in get_campaign_ips():
            ses_client.delete_dedicated_ip_pool(PoolName='campaign-pool')
        print("✅ 已释放专用 IP，节省成本")

def optimize_email_with_s3(email_content, attachments):
    """将附件上传到 S3，邮件中使用链接"""
    s3_client = boto3.client('s3')
    
    attachment_links = []
    for attachment in attachments:
        # 上传到 S3
        key = f"attachments/{uuid.uuid4()}/{attachment['filename']}"
        s3_client.upload_fileobj(
            attachment['file'],
            'email-attachments-bucket',
            key
        )
        
        # 生成预签名 URL（7 天有效）
        url = s3_client.generate_presigned_url(
            'get_object',
            Params={'Bucket': 'email-attachments-bucket', 'Key': key},
            ExpiresIn=604800  # 7 days
        )
        attachment_links.append({
            'name': attachment['filename'],
            'url': url
        })
    
    # 在邮件中插入下载链接
    email_html = email_content + "<h3>附件下载：</h3><ul>"
    for link in attachment_links:
        email_html += f"<li><a href='{link['url']}'>{link['name']}</a></li>"
    email_html += "</ul>"
    
    return email_html