使用 Amazon Rekognition (REK) 偵測行銷應用程式的標籤

本教學課程會引導您建置範例 Python 應用程式，該應用程式可根據上傳至網站的影像，用來傳送電子郵件給人員。範例應用程式旨在透過傳送關於旅遊交易的個人化電子郵件，以吸引使用者參與社交媒體行銷活動，如果他們的相片中辨識到界標。

解決方案整合了各種 AWS 服務，包括 Amazon Rekognition、Amazon S3、DynamoDB、CloudWatch 和 Amazon SES。應用程式使用 Amazon Rekognition 來偵測 DynamoDB 資料庫中辨識之帳戶上傳的映像中的標籤，然後會根據映像中偵測到的標籤傳送行銷電子郵件給帳戶持有人。完整的解決方案架構如下所示：

• 將使用者資料儲存在 DynamoDB 資料庫中。

• 使用者將映像資料和中繼資料 （使用者帳戶號碼） 上傳至 Amazon S3。

• Lambda 函數會叫用 DetectLabels 來識別和記錄上傳影像中的標籤、在 DynamoDB 中查詢使用者電子郵件，以及傳送行銷電子郵件給上傳影像的使用者。

• Lambda 函數的結果會記錄在 CloudWatch 中以供日後檢閱。

以下是教學中所有步驟的概觀：

1. 建立和填入 DynamoDB 資料表。

2. 使用記錄和通知撰寫 Lambda 函數。

3. 設定 Lambda 函數和許可。

4. 使用 Amazon S3 上傳映像和中繼資料。

5. 輪詢 CloudWatch 日誌。

先決條件

開始本教學課程之前，您將需要：

• AWS 帳戶和適當的 IAM 許可。

• 安裝在開發環境上的 Python 和 Boto3。

• 對 Lambda、Amazon S3、DynamoDB、Amazon SES 和 CloudWatch 的基本了解。

• 基本熟悉 Lambda 和 Amazon SES。

建立和填入 DynamoDB 資料表

若要開始本教學課程，我們將建立 DynamoDB 資料表來存放客戶資料，例如：電子郵件地址、年齡、電話號碼和成員資格狀態，並使用「AccountNumber」變數做為主索引鍵。我們也會將一些範例資料插入資料表。

建立 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': '[e-mail]',
            'Age': 25,
            'PhoneNumber': '098-765-4321',
            'MembershipStatus': 'Inactive'
        },
        {
            'AccountNumber': 'ACC1002',
            'EmailAddress': '[e-mail]',
            '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='[e-mail]',
        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 的許可，並在 AWS 管理主控台中建立 Lambda 函數的執行個體。

設定許可和 Lambda 函數

建立或更新 IAM 角色

在我們使用 Lambda 函數處理使用者上傳的任何映像之前，我們必須處理其許可。Lambda 函數需要一個範圍限定的 IAM 角色，具有允許其與 Amazon S3、DynamoDB、Amazon SES、Amazon Rekognition 和 CloudWatch 互動的政策。

若要設定 Lambda 函數的 IAM 角色：

1. 開啟 AWS 管理主控台。

2. 前往 IAM > 角色 > 建立角色。

   選擇「建立角色」。

3. 選取 Lambda 做為將使用此角色的服務。按一下「下一步」。

4. 連接下列政策 （請注意，這些政策僅用於示範目的，在實際生產設定中，您希望將許可限制為僅需要的許可。)：

   • AmazonS3ReadOnlyAccess

   • AmazonRekognitionReadOnlyAccess

   • AmazonDynamoDBFullAccess

   • AmazonSESFullAccess

   • AWSLambdaExecute

   • AWSLambdaBasicExecutionRole （用於 CloudWatch 記錄）

   按一下「下一步」。

5. 命名角色和描述，然後選擇「建立角色」來建立角色。

6. 設定 Lambda 的適當許可後，我們可以使用 AWS 管理主控台建立 Lambda 函數的執行個體。

   前往 AWS 管理主控台中的 Lambda 服務。

7. 按一下「建立函數」。選擇「從頭開始的作者」。

8. 輸入函數名稱，然後選取 Python 執行時間。在「變更預設執行角色」下，選取「使用現有角色」，然後選擇您先前建立的 IAM 角色。

9. 現在您必須在 lambda 函數索引標籤中建立和更新程式碼。前往名為 "lambda_function" 的索引標籤，並以繼續進行的 Lambda 程式碼範例取代其中的程式碼。

   儲存變更並部署變更。

10. 現在您必須將 Amazon S3 事件設定為 Lambda 觸發。

    在 Lambda 函數的組態索引標籤/頁面上，前往觸發，按一下「新增觸發」。

11. 從可用觸發條件清單中選取 Amazon S3。

12. 若要設定觸發：選取應觸發函數的儲存貯體。

    選擇事件類型 PUT。或者，如果您只想要處理具有特定名稱或類型的檔案，您可以指定字首或尾碼。

13. 按一下「新增」並儲存組態來啟用觸發。

驗證 Amazon SES 電子郵件地址

您必須先驗證寄件者和收件人的電子郵件地址，才能使用 Amazon SES 傳送電子郵件。若要執行此作業：

1. 前往 Amazon SES 主控台。導覽至「身分管理」，然後導覽至「電子郵件地址」。

2. 按一下「驗證新的電子郵件地址」。輸入您要驗證的電子郵件地址，然後按一下「驗證此電子郵件地址」。您將會收到一封包含驗證連結的電子郵件。按一下連結以完成驗證程序。

3. 在您驗證兩個帳戶的電子郵件地址、已使用正確的許可設定 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 Logs

檢查 Lambda 函數的 CloudWatch 日誌，以查看是否如預期觸發和執行。您可以在 CloudWatch > Logs > Logs 群組 > /AWS/lambda/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 儲存貯體的影像中的標籤，然後透過電子郵件將促銷訊息傳送給上傳影像的使用者。請務必刪除您不再需要的任何資源，這樣就不會向您收取這些資源的費用。