マーケティングアプリケーション用に Amazon Rekognition (REK) を使用してラベルを検出する

このチュートリアルでは、ウェブサイトにアップロードされた画像に基づいてユーザーに E メールを送信する、Python サンプルアプリケーションの構築方法について説明します。このサンプルアプリケーションの目的は、ユーザーをソーシャルメディアのマーケティングキャンペーンに参加させることです。写真内でランドマークが認識された場合に、旅行のお得情報に関するパーソナライズされた E メールがユーザーに送信されます。

このソリューションは、Amazon Rekognition、Amazon S3、DynamoDB DynamoDB 、CloudWatch、Amazon SES などのさまざまな AWS サービスを統合します。DynamoDB データベースに登録されているアカウントから画像がアップロードされると、このアプリケーションは Amazon Rekognition を使用してその画像のラベルを検出します。その後、検出したラベルに基づいて、アカウント所有者にマーケティング E メールを送信します。このソリューションのアーキテクチャの全容は以下のとおりです。

• ユーザーデータを DynamoDB データベースに保存します。

• ユーザーによって画像データとメタデータ (ユーザーアカウント番号) が Amazon S3 にアップロードされます。

• Lambda 関数が DetectLabels を呼び出して、アップロードされた画像内のラベルを識別して記録し、DynamoDB 内のユーザー E メールを検索して、画像をアップロードしたユーザーにマーケティング E メールを送信します。

• Lambda 関数の結果が、後で確認できるように、CloudWatch に記録されます。

このチュートリアルのすべての手順の概要は以下のとおりです。

1. DynamoDB テーブルを作成してデータを入力します。

2. ログと通知の機能を備えた Lambda 関数を作成します。

3. Lambda 関数とアクセス許可を設定します。

4. 画像とメタデータを Amazon S3 にアップロードします。

5. CloudWatch Logs をポーリングします。

前提条件

このチュートリアルを開始する前に、以下が必要です。

• AWS アカウントと適切な IAM アクセス許可。

• 開発環境にインストールされた Python と Boto3。

• Lambda、Amazon S3、DynamoDB、Amazon SES、CloudWatch の基本的な理解。

• Lambda と Amazon SES の基本的な知識。

DynamoDB テーブルの作成とデータ入力

このチュートリアルを開始するにあたり、顧客データ (E メールアドレス、年齢、電話番号、会員ステータスなど) を保存するための 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()
     

このスクリプトを実行すると、AccountNumber をプライマリキーとする CustomerDataTable という名前の DynamoDB テーブルが設定されます。

サンプルデータの挿入

次は、テーブルにサンプルデータを挿入します。このサンプルデータを使用して、アプリケーションの機能を一通り確認できます。

先ほど作成した CustomerDataTable にサンプルデータを追加する 2 つ目の関数を作成します。以下のコードによって 3 つのサンプルエントリが作成されます。各エントリには、アカウント番号、E メールアドレス、年齢、電話番号、会員ステータスが含まれています。以下のコードサンプルで、region_name の値をお客様のアカウントが運用されているリージョンの名前に置き換えてから、コードセルを実行してテーブルを作成します。アプリケーションの E メール配信部分をテストする場合は、以下の最初の顧客エントリの EmailAddress の値を、受信可能な E メールアドレスに置き換えます。コードを保存して実行します。

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 から該当する E メールアドレスを取得する関数も必要です。さらに、Amazon SES を使用してその E メールアドレスに E メールを送信する関数も必要です。最後に、ロガーを使用してこのプロセスに関する情報を記録します。このデータは CloudWatch によって表示されます。

Lambda 関数の作成

ここでは、前述の要件を処理する Lambda 関数の概要を示します。テーブル名として CustomerDataTable 以外を使用した場合は、以下のコードサンプルで、正しい DynamoDB テーブルを指定していることを確認してください。さらに、「send_marketing_email」関数では、「Source」引数の値を、実際に送信元として使用できる E メールアドレスに置き換える必要があります。

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 関数には、Amazon S3、DynamoDB、Amazon SES、Amazon Rekognition、CloudWatch とのやり取りを許可するポリシーを設定した IAM ロールが必要です。

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. 次は、Lambda 関数のトリガーとして Amazon S3 イベントを設定する必要があります。

    Lambda 関数の設定タブ/ページで、[トリガー] に移動し、[トリガーを追加] をクリックします。

11. 使用可能なトリガーのリストから [Amazon S3] を選択します。

12. トリガーを設定するには、関数をトリガーするバケットを選択します。

    イベントタイプとして [PUT] を選択します。特定の名前やタイプのファイルのみを処理する場合は、オプションでプレフィックスまたはサフィックスを指定できます。

13. [追加] をクリックしてトリガーを有効にし、設定を保存します。

Amazon SES E メールアドレスの検証

Amazon SES を使用して E メールを送信する前に、送信者と受取人の両方の E メールアドレスを検証する必要があります。これを実行するには:

1. Amazon SES コンソールに移動します。[ID 管理]、[E メールアドレス] の順に移動します。

2. [新しい E メールアドレスを検証する] をクリックします。検証する E メールアドレスを入力し、[この E メールアドレスを検証する] をクリックします。検証用リンクが記載された E メールが届きます。リンクをクリックして、検証プロセスを完了します。

3. 両方のアカウントの E メールアドレスを検証し、Lambda 関数に正しいアクセス許可を設定し、サンプルの顧客データを作成したら、選択した Amazon S3 バケットにテスト画像をアップロードすることで Lambda 関数をテストできます。

画像をメタデータとともに Amazon S3 にアップロードする

これで、 AWS コンソールまたは前に準備したスクリプトを使用して、選択した Amazon S3 バケットにテストイメージをアップロードできます。先ほど Lambda 関数で指定した Amazon S3 バケットに画像をアップロードするスクリプトを作成します。以下のコードサンプルでは、「upload_image_to_s3」への最初の引数として画像のパスを、2 番目の引数としてバケット名を、最後の引数として画像をアップロードするユーザーのアカウント番号を指定します。

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 されて、それに応じてイメージを処理できます。

スクリプトを保存し、実行します。これにより、画像がアップロードされます。

画像をアップロードしてから数分後、ここで指定したアカウントに先ほど関連付けたアドレスに、E メールが届きます。

クライアントからの CloudWatch Logs のポーリング

CloudWatch のログで、Lambda 関数が想定どおりにトリガーされ、実行されたかどうかを確認します。ログは [CloudWatch] > [ログ] > [ロググループ] > /AWS/lambda/your_function_name にあります。また、ログにアクセスして表示するスクリプトを作成することもできます。以下のコードサンプルは、Lambda グループのログをポーリングし、過去 1 時間に生成されたログを出力します。コードを保存して実行します。

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'])
     

このコードを実行するとログが表示され、関連する旅行情報を含む E メールがユーザーに通知されたことがわかります。

これで、Amazon S3 バケットにアップロードされた画像のラベルを検出し、画像をアップロードしたユーザーにプロモーションメッセージを含む E メールを送信するアプリケーションが完成しました。不要な課金が発生しないように、使用しなくなったリソースは必ず削除してください。