我们使用 AWS SES 发送邮件。Amazon SES 通过电子邮件或 AWS SNS发送退回和投诉通知。我们希望自动处理退回和投诉通知（来自电子邮件或 AWS SNS）以提取电子邮件 ID，以便可以从原始列表中删除这些电子邮件。
一种自动化方法是将这些通知发送到 AWS SNS 中的主题，然后使用 AWS SQS 订阅该主题，最后读取 AWS SQS 中的消息。SNS 支持通过以下协议订阅 - HTTP/HTTPS/EMail/EMail(JSON)/SMS/SQS。这是可行的，但我发现它对于自动处理退回和投诉通知的简单任务来说太麻烦了。
有什么优雅的方法可以解决这个问题吗？
我在亚马逊上找到了一篇博客文章，其中包含 C# 中的代码。有更好的解决方案吗？
我发现使用 HTTP 端点直接订阅 SNS 是最直接的方法。您实际上只需要编写几行代码。这是我的 Django 示例：
def process(request):
json = request.raw_post_data
js = simplejson.loads(json)
info = simplejson.loads(js["Message"])
type = info["notificationType"] # "Complaint" or "Bounce"
email = info["mail"]["destination"][0]
# do whatever you want with the email
通过试验，我想出了一个错误——它适用于 Django，但对我来说做得不错。
首先是模型，然后是请求处理程序......
class ComplaintType:
ABUSE = 'abuse'
AUTH_FAILURE = 'auth-failure'
FRAUD = 'fraud'
NOT_SPAM = 'not-spam'
OTHER = 'other'
VIRUS = 'virus'
COMPLAINT_FEEDBACK_TYPE_CHOICES = [
[ComplaintType.ABUSE, _('Unsolicited email or some other kind of email abuse')],
[ComplaintType.AUTH_FAILURE, _('Unsolicited email or some other kind of email abuse')],
[ComplaintType.FRAUD, _('Some kind of fraud or phishing activity')],
[ComplaintType.NOT_SPAM, _('Entity providing the report does not consider the message to be spam')],
[ComplaintType.OTHER, _('Feedback does not fit into any other registered type')],
[ComplaintType.VIRUS, _('A virus was found in the originating message')]
]
class SES_Complaint(models.Model):
subject = models.CharField(max_length=255)
message = models.TextField()
email_address = models.EmailField(db_index=True)
user_agent = models.CharField(max_length=255)
complaint_feedback_type = models.CharField(max_length=255, choices=COMPLAINT_FEEDBACK_TYPE_CHOICES)
arrival_date = models.DateTimeField()
timestamp = models.DateTimeField()
feedback_id = models.CharField(max_length=255)
created = models.DateTimeField(auto_now_add=True)
modified = models.DateTimeField(auto_now=True)
class Meta:
verbose_name = 'SES Complaint'
verbose_name_plural = 'SES Complaints'
def get_reason(self):
return self.get_complaint_feedback_type_display()
class BounceType:
UNDETERMINED = 'Undetermined'
PERMANENT = 'Permanent'
TRANSIENT = 'Transient'
class BounceSubType:
UNDETERMINED = 'Undetermined'
GENERAL = 'General'
NO_EMAIL = 'NoEmail'
SUPPRESSED = 'Suppressed'
MAILBOX_FULL = 'MailboxFull'
MESSAGE_TOO_LARGE = 'MessageToolarge'
CONTENT_REJECTED = 'ContentRejected'
ATTACHMENT_REJECTED = 'AttachmentRejected'
BOUNCE_TYPE_CHOICES = [
[BounceType.UNDETERMINED, _('Unable to determine a specific bounce reason')],
[BounceType.PERMANENT, _('Unable to successfully send')],
[BounceType.TRANSIENT, _('All retry attempts have been exhausted')],
]
BOUNCE_SUB_TYPE_CHOICES = [
[BounceSubType.UNDETERMINED, _('Unable to determine a specific bounce reason')],
[BounceSubType.GENERAL, _('General bounce. You may be able to successfully retry sending to that recipient in the future.')],
[BounceSubType.NO_EMAIL, _('Permanent hard bounce. The target email address does not exist.')],
[BounceSubType.SUPPRESSED, _('Address has a recent history of bouncing as invalid.')],
[BounceSubType.MAILBOX_FULL, _('Mailbox full')],
[BounceSubType.MESSAGE_TOO_LARGE, _('Message too large')],
[BounceSubType.CONTENT_REJECTED, _('Content rejected')],
[BounceSubType.ATTACHMENT_REJECTED, _('Attachment rejected')]
]
class SES_Bounce(models.Model):
subject = models.CharField(max_length=255)
message = models.TextField()
bounce_type = models.CharField(max_length=255, choices=BOUNCE_TYPE_CHOICES)
bounce_sub_type = models.CharField(max_length=255, choices=BOUNCE_SUB_TYPE_CHOICES)
timestamp = models.DateTimeField()
feedback_id = models.CharField(max_length=255)
status = models.CharField(max_length=255)
action = models.CharField(max_length=255)
diagnostic_code = models.CharField(max_length=255)
email_address = models.EmailField(db_index=True)
created = models.DateTimeField(auto_now_add=True)
modified = models.DateTimeField(auto_now=True, db_index=True)
class Meta:
verbose_name = 'SES Bounce'
verbose_name_plural = 'SES Bounces'
def get_reason(self):
return '%s - %s' % (self.get_bounce_type_display(), self.get_bounce_sub_type_display())
这是请求处理程序：
@csrf_exempt
def aws_sns(request):
logger.debug('Incoming SNS')
if request.method == 'POST':
logger.debug('Incoming SNS is POST')
sns_message_type = request.META.get('HTTP_X_AMZ_SNS_MESSAGE_TYPE', None)
if sns_message_type is not None:
logger.debug('Incoming SNS - %s', sns_message_type)
json_body = request.body
json_body = json_body.replace('\n', '')
js = loads(json_body)
if sns_message_type == "SubscriptionConfirmation":
subscribe_url = js["SubscribeURL"]
logger.debug('Incoming subscription - %s', subscribe_url)
urllib.urlopen(subscribe_url)
elif sns_message_type == "Notification":
message = js.get("Message", None)
message = message.replace('\n', '')
message = loads(message)
notification_type = message.get("notificationType", None)
if notification_type == 'AmazonSnsSubscriptionSucceeded':
logger.debug('Subscription succeeded')
elif notification_type == 'Bounce':
logger.debug('Incoming bounce')
bounce = message['bounce']
bounce_type = bounce['bounceType']
bounce_sub_type = bounce['bounceSubType']
timestamp = bounce['timestamp']
feedback_id = bounce['feedbackId']
bounce_recipients = bounce['bouncedRecipients']
for recipient in bounce_recipients:
status = recipient.get('status')
action = recipient.get('action')
#diagnostic_code = recipient['diagnosticCode']
email_address = recipient['emailAddress']
SES_Bounce.objects.filter(email_address=email_address).delete()
SES_Bounce.objects.create(
message=message,
bounce_type=bounce_type,
bounce_sub_type=bounce_sub_type,
timestamp=timestamp,
feedback_id=feedback_id,
status=status,
action=action,
#diagnostic_code=diagnostic_code,
email_address=email_address
)
elif notification_type == 'Complaint':
logger.debug('Incoming complaint')
complaint = message['complaint']
user_agent = complaint.get('userAgent')
complaint_feedback_type = complaint.get('complaintFeedbackType')
arrival_date = complaint.get('arrivalDate')
timestamp = complaint['timestamp']
feedback_id = complaint['feedbackId']
recipients = complaint['complainedRecipients']
for recipient in recipients:
email_address = recipient['emailAddress']
SES_Complaint.objects.filter(email_address=email_address).delete()
SES_Complaint.objects.create(
#subject=subject,
message=message,
email_address=email_address,
user_agent=user_agent,
complaint_feedback_type=complaint_feedback_type,
arrival_date=arrival_date,
timestamp=timestamp,
feedback_id=feedback_id
)
else:
logger.exception('Incoming Notification SNS is not supported: %s', notification_type)
return HttpResponse()
else:
logger.exception('Incoming SNS did not have the right header')
for key, value in request.META.items():
logger.debug('Key: %s - %s', key, value)
else:
logger.exception('Incoming SNS was not a POST')
return HttpResponseBadRequest()
我认为你描述的方式可能是最优雅的方式。您已经在 SNS 和 SQS 中拥有非常合适的服务，这些服务与大多数主要语言的 SDK 相关联，让您可以轻松地做您需要的事情。最困难的部分是编写代码来更新/删除邮件列表中的记录。
最近，我能够通过 SNS 使用 HTTP 端点来完成这项工作。我使用 python/django 来使用通知。在使用通知之前，您必须先处理订阅消息；您可以在 SNS 文档中阅读有关订阅的信息。
我想如果您有一个较小的应用程序，它不会发送很多电子邮件；http端点应该可以正常工作。此代码要求您创建通知模型。
#process an amazon sns http endpoint notification for amazon ses bounces and complaints
@csrf_exempt
def process_ses_notification(request):
if request.POST:
json_body = request.body
#remove this control character(throws an error) thats present inside the test subscription confirmation
js = loads(json_body.replace('\n', ''))
if js["Type"] == "SubscriptionConfirmation":
subscribe_url = js["SubscribeURL"]
urllib.urlopen(subscribe_url)
return HttpResponse(status=200)
elif js["Type"] == "Notification":
#process message from amazon sns
arg_info = loads(js["Message"]) # may need to use loads(js["Message"]) after testing with amazon
arg_notification_type = arg_info["notificationType"]
if arg_notification_type == 'Bounce':
#required bounce object fields
arg_emails=arg_info["bounce"]["bouncedRecipients"]
arg_notification_subtype=arg_info["bounce"]["bounceType"]
arg_feedback_id=arg_info["bounce"]["feedbackId"]
arg_date_recorded=arg_info["bounce"]["timestamp"]
elif arg_notification_type == 'Complaint':
#required complaint object fields
arg_emails=arg_info["complaint"]["complainedRecipients"]
arg_feedback_id=arg_info["complaint"]["feedbackId"]
arg_date_recorded=arg_info["complaint"]["timestamp"]
#check if feedback type is inside optional field name
if "complaintFeedbackType" in arg_info["complaint"]:
arg_notification_subtype=arg_info["complaint"]["complaintFeedbackType"]
else:
arg_notification_subtype=""
else:
HttpResponse(status=400)
#save notifications for multiple emails
for arg_email in arg_emails:
notification = SES_Notification(info=json_body, notification_type=arg_notification_type,
email=arg_email["emailAddress"], notification_subtype=arg_notification_subtype,
date_recorded=arg_date_recorded, feedback_id=arg_feedback_id)
notification.save()
return HttpResponse(status=200)
return HttpResponse(status=400)
上面的所有答案都很棒，但只是一个小而重要的补充：
您首先需要验证请求是否来自 amazon sns：（如验证 Amazon SNS 消息的签名中所述）
对于验证签名的python代码 - 一个很好的例子是here