本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
# 使用 Amazon Rekognition（REK）检测用于营销应用程序的标签
本教程将指导您根据上传到网站的图像，构建一个可用于向他人发送电子邮件的示例 Python 应用程序。该示例应用程序旨在通过向用户发送有关旅行优惠的个性化电子邮件（如果照片中识别出地标），从而吸引用户参与社交媒体营销活动。
该解决方案集成了各种 AWS 服务，包括亚马逊 Rekognition、Amazon S3 CloudWatch、DynamoDB 和亚马逊 SES。该应用程序使用 Amazon Rekognition 来检测由 DynamoDB 数据库中识别的账户上传的图像中的标签，然后根据图像中检测到的标签向账户持有人发送一封营销电子邮件。完整的解决方案架构如下：
+ 将用户数据存储在 DynamoDB 数据库中。
+ 用户将图像数据和元数据（用户账号）上传到 Amazon S3。
+ 调用 Lambda 函数[DetectLabels](https://docs.aws.amazon.com/rekognition/latest/APIReference/API_DetectLabels.html)来识别和记录上传图像中的标签，在 DynamoDB 中查找用户电子邮件，并向上传图像的用户发送营销电子邮件。
+ Lambda 函数的结果已登录以 CloudWatch 供日后查看。
本教程中所有步骤概述如下：
1. 创建和填充 DynamoDB 表。
1. 编写具有日志记录和通知功能的 Lambda 函数。
1. 设置 Lambda 函数和权限。
1. 使用 Amazon S3 上传图像和元数据。
1. 民意调查 CloudWatch 日志。
**Topics**
+ [先决条件](#label-marketing-prereq)
+ [创建和填充 DynamoDB 表](#label-marketing-tutorial-ddb)
+ [创建具有日志记录和通知功能的 Lambda 函数](#label-marketing-tutorial-lambda)
+ [设置权限和 Lambda 函数](#label-marketing-tutorial-permissions)
+ [使用元数据将图像上传到 Amazon S3](#label-marketing-tutorial-uploading)
+ [来自客户端的轮询 CloudWatch 日志](#label-marketing-tutorial-cloudwatch)
## 先决条件
开始本教程之前，您将需要：
+ AWS 账户和相应的 IAM 权限。
+ 已在开发环境中安装 Python 和 Boto3。
+ 对 Lambda、亚马逊 S3、DynamoDB、Amazon SES 和. CloudWatch
+ 基本熟悉 Lambda 和 Amazon SES。
## 创建和填充 DynamoDB 表
在本教程开始时，我们将使用 AccountNumber “” 变量作为主键创建一个 DynamoDB 表来存储客户数据，例如：电子邮件地址、年龄、电话号码和会员状态。我们还将在表中插入一些示例数据。
### 创建 DynamoDB 表
首先，我们创建 DynamoDB 表。下面介绍了如何使用 Python 和 Boto3 进行设置。我们创建了一个名为 `create_user_table` 的函数并连接到其中的 DynamoDB 资源。在下面的代码示例中，将“region\_name”的值替换为您的账户运营所在区域的名称，然后运行代码单元来创建您的表。
```
import boto3
def create_user_table(dynamodb=None):
if not dynamodb:
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.create_table(
TableName='CustomerDataTable',
KeySchema=[
{
'AttributeName': 'AccountNumber',
'KeyType': 'HASH' # Partition key
},
],
AttributeDefinitions=[
{
'AttributeName': 'AccountNumber',
'AttributeType': 'S'
},
],
ProvisionedThroughput={
'ReadCapacityUnits': 10,
'WriteCapacityUnits': 10
}
)
# Wait until the table exists.
table.wait_until_exists()
print("Table status:", table.table_status)
# Create the DynamoDB table.
create_user_table()
```
运行此脚本已设置了一个名为 `CustomerDataTable` 的 DynamoDB 表，以 `AccountNumber` 作为主键。
### 插入示例数据
现在，我们要在表中插入一些示例数据。此示例数据将帮助我们测试应用程序的完整功能。
我们将创建第二个函数，该函数将示例数据添加到我们之前创建的 `CustomerDataTable` 中。以下代码创建了三个示例条目，每个条目都包括账号、电子邮件地址、年龄、电话号码和成员身份状态。在下面的代码示例中，将 `region_name` 的值替换为您的账户运营所在区域的名称，然后运行代码单元来创建您的表。如果您想测试应用程序的电子邮件发送部分，请将下面第一个客户条目中 `EmailAddress` 的值替换为您可以接收电子邮件的电子邮件地址。保存并运行代码。
```
import boto3
def insert_sample_data(dynamodb=None):
if not dynamodb:
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
table = dynamodb.Table('CustomerDataTable')
# Sample data
customers = [
{
'AccountNumber': 'ACC1000',
'EmailAddress': 'email-for-delivery-here',
'Age': 30,
'PhoneNumber': '123-456-7890',
'MembershipStatus': 'Active'
},
{
'AccountNumber': 'ACC1001',
'EmailAddress': 'jane.doe@example.com',
'Age': 25,
'PhoneNumber': '098-765-4321',
'MembershipStatus': 'Inactive'
},
{
'AccountNumber': 'ACC1002',
'EmailAddress': 'pat.candella@example.com',
'Age': 35,
'PhoneNumber': '555-555-5555',
'MembershipStatus': 'Active'
}
]
# Inserting data
for customer in customers:
print(f"Adding customer: {customer['AccountNumber']}")
table.put_item(Item=customer)
# Insert sample data into DynamoDB
insert_sample_data()
```
设置并填充 DynamoDB 表后，我们现在可以继续将此数据检索集成到 Lambda 函数中。这样，我们的应用程序就可以根据在即将进行的图像处理工作流中识别的账号来检索用户详细信息。
## 创建具有日志记录和通知功能的 Lambda 函数
现在，我们可以创建一个 Lambda 函数。我们希望确保图像上传时触发的 Lambda 函数可以读取图像数据和账户元数据，并用它在 DynamoDB 表中查找关联的用户数据。这意味着我们不仅需要调用 Amazon Rekognition 的 DetectLabels 函数，还需要一个接收 AccountNumber 并使用它从 DynamoDB 检索关联电子邮件地址的函数。此外，我们还需要一个函数来使用 Amazon SES 向电子邮件地址发送电子邮件。最后，我们使用记录器来记录有关该过程的信息。这些数据将由显示 CloudWatch。
### 创建 Lambda 函数
下面概述了 Lambda 函数可能如何满足这些要求。在以下代码示例中，如果您使用了 CustomerDataTable 其他名称作为表名，请确保指定了正确的 DynamoDB 表。此外，在“send\_marketing\_email”函数中，应将“Source”参数的值替换为您具有访问权限的电子邮件地址，该电子邮件地址将用于发送电子邮件。
```
import json
import boto3
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
s3_bucket = event['Records'][0]['s3']['bucket']['name']
s3_object_key = event['Records'][0]['s3']['object']['key']
print(s3_bucket)
print(s3_object_key)
s3 = boto3.client('s3')
try:
s3_response = s3.head_object(Bucket=s3_bucket, Key=s3_object_key)
account_number = s3_response['Metadata']['account_number']
except Exception as e:
logger.error(f"Failed to retrieve object or metadata: {str(e)}")
raise e # Optionally re-raise to handle the error upstream or signal a failure
rekognition = boto3.client('rekognition')
try:
labels_response = rekognition.detect_labels(Image={'S3Object': {'Bucket': s3_bucket, 'Name': s3_object_key}})
#logger.info(f"Detected labels: {json.dumps(labels_response['Labels'], indent=4)}")
except Exception as e:
#logger.info(f"Detected label: {label['Name']}")
raise e
#logger.info(f"Detected labels: {json.dumps(labels_response['Labels'], indent=4)}")
landmark_detected = any(label['Name'] == 'Landmark' and label['Confidence'] > 20 for label in labels_response['Labels'])
if landmark_detected:
result = notify_user_based_on_landmark(account_number)
print(result)
#logger.info(f"Detected label: {label['Name']}")
#logger.info(f"Notification sent: {result}")
return {
'statusCode': 200,
'body': json.dumps('Process completed successfully!')
}
def notify_user_based_on_landmark(account_number):
# Retrieve user data from DynamoDB
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('CustomerDataTable')
user_info = table.get_item(Key={'AccountNumber': account_number})
# Send email if user is found
if 'Item' in user_info:
send_marketing_email(user_info['Item']['EmailAddress'])
return "User notified with relevant travel information."
def send_marketing_email(email):
ses = boto3.client('ses')
response = ses.send_email(
Source='your-email@example.com',
Destination={'ToAddresses': [email]},
Message={
'Subject': {'Data': 'Explore New Destinations!'},
'Body': {
'Text': {'Data': 'Check out our exclusive travel packages inspired by the landmark in your uploaded image!'}
}
}
)
return f"Email sent to {email} with status {response['ResponseMetadata']['HTTPStatusCode']}"
print("succeess")
```
现在我们已经编写了 Lambda 函数，我们需要为 Lambda 配置权限并在管理控制台中创建 Lambda 函数的实例。 AWS
## 设置权限和 Lambda 函数
**创建或更新 IAM 角色**
我们必须处理 Lambda 函数的权限，才能使用该函数处理用户上传的任何图像。Lambda 函数需要一个 IAM 角色的作用域，该角色具有允许其与亚马逊 S3、DynamoDB、Amazon SES、Amazon Rekognition 和进行交互的策略。 CloudWatch
为 Lambda 函数配置 IAM 角色：
1. 打开 AWS 管理控制台。
1. 前往“IAM”>“角色”>“创建角色”。
选择“创建角色”。
1. 选择 Lambda 作为将使用此角色的服务。单击“下一步”。
1. 附加以下策略（请注意，选择这些策略仅用于演示目的，在实际生产环境中，您可能希望仅对需要的策略授予权限。）：
+ 亚马逊 3 ReadOnlyAccess
+ AmazonRekognitionReadOnlyAccess
+ AmazonDynamoDBFull访问权限
+ 亚马逊SESFull访问权限
+ AWSLambda执行
+ AWSLambdaBasicExecutionRole （用于 CloudWatch 记录）
单击“下一步”。
1. 为角色命名并进行描述，然后通过选择“创建角色”来创建角色。
1. 一旦我们为 Lambda 配置了适当的权限，我们就可以使用管理控制台创建 Lambda 函数的实例。 AWS
在 AWS 管理控制台中，转到 Lambda 服务。
1. 单击“创建函数”。选择“从头开始创作”。
1. 输入一个函数名称并选择 Python 运行时。在“更改默认执行角色”下，选择“使用现有角色”，然后选择您之前创建的 IAM 角色。
1. 现在，您必须在 Lambda 函数选项卡中创建和更新代码。转到名为“lambda\_function”的选项卡，然后将那里的代码替换为后续的 Lambda 代码示例。
保存更改并部署更改。
1. 现在，您必须将 Amazon S3 事件配置为 Lambda 触发器。
在 Lambda 函数的配置选项卡/页面上，转到“触发器”，单击“添加触发器”。
1. 从可用触发器列表中选择 Amazon S3。
1. 要配置触发器，请选择相应的存储桶以从中触发函数。
选择事件类型，即 PUT。或者，如果您只想处理具有特定名称或类型的文件，则可以指定前缀或后缀。
1. 单击“添加”激活该触发器并保存配置。
### 验证 Amazon SES 电子邮件地址
您必须先验证发件人和收件人的电子邮件地址，然后才能使用 Amazon SES 发送电子邮件。要实现此目的，应按照以下步骤进行：
1. 前往 Amazon SES 控制台。导航到“身份管理”，然后导航到“电子邮件地址”。
1. 单击“验证新的电子邮件地址”。输入要验证的电子邮件地址，然后单击“验证此电子邮件地址”。您将收到一封包含验证链接的电子邮件。单击该链接即可完成验证过程。
1. 在您验证了两个账户的电子邮件地址、设置了具有正确权限的 Lambda 函数并创建了一些示例客户数据之后，便可通过将测试图像上传到所选的 Amazon S3 存储桶来测试 Lambda 函数。
## 使用元数据将图像上传到 Amazon S3
现在，我们可以使用 AWS 控制台或之前准备的脚本将测试图像上传到我们选择的 Amazon S3 存储桶。编写脚本，以将图像上传到之前在 Lambda 函数中指定的 Amazon S3 存储桶。在下面的代码示例中，将图像的路径指定为“upload\_image\_to\_s3”的第一个参数，将存储桶名称指定为第二个参数，将上传图像的用户的账号指定为最后一个参数。
```
import boto3
def upload_image_to_s3(file_name, bucket, account_number):
s3 = boto3.client('s3')
try:
with open(file_name, 'rb') as data:
s3.upload_fileobj(
Fileobj=data,
Bucket=bucket,
Key=file_name,
ExtraArgs={
'Metadata': {'account_number': account_number}
}
)
print("Image uploaded successfully with metadata.")
except Exception as e:
print("Failed to upload image")
print(e)
# Usage
upload_image_to_s3('path-to-image-here', 's3-bucket-name-here', 'user-account-number-here')
```
在此函数中，`upload_fileobj` 方法中的 `ExtraArgs` 参数用于在图像中包含用户定义的元数据 (`account_number`)。稍后可以使用此元数据 AWS 来相应地处理图像。
保存并运行该脚本。这将上传图像。
上传图像几分钟后，您应该会收到一封电子邮件，收件地址是您之前与此处指定的账户关联的地址。
## 来自客户端的轮询 CloudWatch 日志
查看您的 Lambda 函数的 CloudWatch 日志，查看该函数是否按预期触发和执行。日志可以在 > 日志 > 日志组 CloudWatch > //lambda AWS/your\_function\_name 下找到。您也可以编写脚本来访问和打印日志。以下代码示例在 Lambda 组中轮询日志，打印出过去一小时内生成的日志。保存并运行代码。
```
import boto3
import time
def fetch_lambda_logs(log_group_name, start_time):
client = boto3.client('logs')
query = "fields @timestamp, @message | sort @timestamp desc | limit 20"
start_query_response = client.start_query(
logGroupName=log_group_name,
startTime=int(start_time),
endTime=int(time.time()),
queryString=query,
)
query_id = start_query_response['queryId']
response = None
while response is None or response['status'] == 'Running':
time.sleep(1) # Wait for 1 second before checking the query status again
response = client.get_query_results(queryId=query_id)
return response['results']
# Usage
log_group = '/aws/lambda/RekMediaFunction'
logs = fetch_lambda_logs(log_group, int(time.time()) - 3600) # Fetch logs from the last hour
print("Retrieving logs")
for log in logs:
#print(log)
print(log[1]['value'])
```
运行代码应该会打印出日志，并且您应该能够看到用户已通过包含相关旅行信息的电子邮件收到通知。
您已成功创建了一个应用程序，该应用程序能够检测上传到 Amazon S3 存储桶的图像中的标签，然后通过电子邮件向上传图像的用户发送促销信息。请务必删除任何不再需要的资源，这样就不会为它们付费。