Amazon Rekognition(REK)을 사용하여 마케팅 애플리케이션의 레이블 감지

이 자습서에서는 웹 사이트에 업로드된 이미지를 기반으로 사람들에게 이메일을 보내는 데 사용할 수 있는 샘플 Python 애플리케이션을 구축하는 방법을 안내합니다. 샘플 애플리케이션은 사진에서 랜드마크가 인식되는 경우 여행 상품에 대한 개인 맞춤형 이메일을 전송하여 소셜 미디어 마케팅 캠페인에 사용자를 참여시키도록 설계되었습니다.

이 솔루션은 Amazon Rekognition, Amazon S3, DynamoDB, CloudWatch 및 Amazon SES를 비롯한 다양한 AWS 서비스를 통합합니다. 애플리케이션은 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 테이블 생성 및 채우기

이 자습서를 시작하려면 프라이머리 키로 ‘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()
     

이 스크립트를 실행하면 AccountNumber이 있는 CustomerDataTable이라는 DynamoDB 테이블을 프라이머리 키로 설정합니다.

샘플 데이터 삽입

이제 일부 샘플 데이터를 테이블에 삽입하겠습니다. 이 샘플 데이터는 애플리케이션의 전체 기능을 테스트하는 데 도움이 됩니다.

이전에 생성한 것과 동일한 CustomerDataTable에 샘플 데이터를 추가하는 두 번째 함수를 생성합니다. 아래 코드로 3개의 샘플 항목이 생성되며 각 항목에는 계정 번호, 이메일 주소, 연령, 전화번호 및 멤버십 상태가 포함됩니다. 아래 코드 샘플에서 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 함수에는 Amazon S3, DynamoDB, Amazon SES, Amazon Rekognition 및 CloudWatch와 상호 작용할 수 있는 정책이 있는 범위가 지정된 IAM 역할이 필요합니다.

Lambda 함수용 IAM 역할을 구성하려면

1. AWS Management Console을 엽니다.

2. IAM > 역할 > 역할 생성으로 이동합니다.

   ‘역할 생성’을 선택합니다.

3. 이 역할을 사용할 서비스로 Lambda를 선택합니다. ‘다음’을 클릭합니다.

4. 다음 정책을 연결합니다(이러한 정책은 실제 프로덕션 설정에서 필요한 정책으로만 권한을 제한하려는 데모용으로만 선택됩니다).

   • AmazonS3ReadOnlyAccess

   • AmazonRekognitionReadOnlyAccess

   • AmazonDynamoDBFullAccess

   • AmazonSESFullAccess

   • AWSLambdaExecute

   • AWSLambdaBasicExecutionRole(CloudWatch 로깅용)

   ‘다음’을 클릭합니다.

5. 역할의 이름과 설명을 지정한 다음 '역할 생성'을 선택하여 역할을 생성합니다.

6. Lambda에 대한 적절한 권한을 구성한 후에는 AWS 관리 콘솔을 사용하여 Lambda 함수의 인스턴스를 생성할 수 있습니다.

   AWS Management Console에서 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 콘솔로 이동합니다. 'ID 관리'로 이동한 다음 '이메일 주소'로 이동합니다.

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 로그 폴링

Lambda 함수의 CloudWatch 로그를 확인하여 예상대로 트리거되고 실행되었는지 확인합니다. 로그는 CloudWatch > 로그 > 로그 그룹 > /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 버킷에 업로드된 이미지의 레이블을 감지한 다음 이미지를 업로드한 사용자에게 프로모션 메시지를 이메일로 보낼 수 있는 애플리케이션을 성공적으로 생성했습니다. 더 이상 필요하지 않은 리소스는 삭제하여 요금이 부과되지 않도록 하세요.