AWS Lambda 関数を使用したテキストの検出

にサインイン AWS マネジメントコンソール し、https://console.aws.amazon.com/lambda/ で AWS Lambda コンソールを開きます。

[関数の作成] を選択してください。詳細については、「コンソールで Lambda の関数の作成」を参照してください。

次のオプションを選択します。

[関数の作成] を選択し、 AWS Lambda 関数を作成します。

関数ページで、[設定] タブを選択します。

[アクセス権限] タブの [実行ロール] でロール名を選択し、このロールを IAM コンソールで開きます。

[アクセス権限] タブで、[アクセス権限を追加]、次に [インラインポリシーを作成] をクリックします。

[JSON] タブを選択し、ポリシーを次のポリシーに置き換えます。

[ポリシーの確認] を選択します。

DetectDocumentText-access など、ポリシーの名前を入力します。

[Create policy] (ポリシーの作成) を選択します。

分析用のドキュメントを Amazon S3 バケットに保存する場合は、Amazon S3 アクセスポリシーを追加する必要があります。追加するには、 AWS Lambda コンソールでステップ 7～11 を繰り返し、以下のように変更を加えます。

1. ステップ 8 で、以下のポリシーを使用します。bucket/folder path を、Amazon S3 バケットと、分析するドキュメントのフォルダパスに置き換えます。

   JSON

   {
       "Version":"2012-10-17",
       "Statement": [
           {
               "Sid": "S3Access",
               "Effect": "Allow",
               "Action": "s3:GetObject",
               "Resource": "arn:aws:s3:::bucket/folder path/*"
           }
       ]
   }
   

2. ステップ 10 で、異なるポリシー名 (S3Bucket-access など) を選択します。

コマンドプロンプトを開き、次のコマンドを入力して、最新バージョンの AWS SDK でデプロイパッケージを作成します。

pip install boto3 --target python/.
zip boto3-layer.zip -r python/
                

zip ファイルの名前 (boto3-layer.zip) を書き留めます。この名前は、この手順内のステップ 8 で使用します。

https://console.aws.amazon.com/lambda/ で AWS Lambda コンソールを開きます。

ナビゲーションペインで [レイヤー] を選択します。

[レイヤーを作成] を選択します。

[名前] と [説明] の値を入力します。

[コードエントリタイプ] で、[.zip ファイルをアップロード] を選択し、[アップロード] を選択します。

ダイアログボックスで、この手順のステップ 1 で作成した .zip ファイルアーカイブ (boto3-layer.zip) を選択します。

互換性のあるランタイムで、Python 3.9 を選択します。

[作成] を選択してレイヤーを作成します。

ナビゲーションペインのメニューアイコンを選択します。

ナビゲーションペインで、[関数] を選択します。

リソースリストの中から、「ステップ 1: AWS Lambda 関数を作成する (コンソール)」で作成した関数を選択します。

[コード] タブを選択します。

[レイヤー] セクションで、[レイヤーの追加] を選択します。

[カスタムレイヤー] を選択します。

[カスタムレイヤー] で、ステップ 6 で入力したレイヤー名を選択します。

[バージョン] でレイヤーバージョンを選択します。バージョンは「1」です。

[追加] を選択します。

[コード] タブに移動します。

コードエディタで、lambda_function.py のコードを以下のコードに置き換えます。

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0

"""
Purpose
An AWS lambda function that analyzes documents with Amazon Textract.
"""
import json
import base64
import logging
import boto3

from botocore.exceptions import ClientError

# Set up logging.
logger = logging.getLogger(__name__)

# Get the boto3 client.
textract_client = boto3.client('textract')


def lambda_handler(event, context):
    """
    Lambda handler function
    param: event: The event object for the Lambda function.
    param: context: The context object for the lambda function.
    return: The list of Block objects recognized in the document
    passed in the event object.
    """

    try:

        # Determine document source.
        if 'image' in event:
            # Decode the image
            image_bytes = event['image'].encode('utf-8')
            img_b64decoded = base64.b64decode(image_bytes)
            image = {'Bytes': img_b64decoded}


        elif 'S3Object' in event:
            image = {'S3Object':
                     {'Bucket':  event['S3Object']['Bucket'],
                      'Name': event['S3Object']['Name']}
                     }

        else:
            raise ValueError(
                'Invalid source. Only image base 64 encoded image bytes or S3Object are supported.')


        # Analyze the document.
        response = textract_client.detect_document_text(Document=image)

        # Get the Blocks
        blocks = response['Blocks']

        lambda_response = {
            "statusCode": 200,
            "body": json.dumps(blocks)
        }

    except ClientError as err:
        error_message = "Couldn't analyze image. " + \
            err.response['Error']['Message']

        lambda_response = {
            'statusCode': 400,
            'body': {
                "Error": err.response['Error']['Code'],
                "ErrorMessage": error_message
            }
        }
        logger.error("Error function %s: %s",
            context.invoked_function_arn, error_message)

    except ValueError as val_error:
        lambda_response = {
            'statusCode': 400,
            'body': {
                "Error": "ValueError",
                "ErrorMessage": format(val_error)
            }
        }
        logger.error("Error function %s: %s",
            context.invoked_function_arn, format(val_error))

    return lambda_response

Lambda 関数をデプロイするには、[デプロイ] を選択します。

まだの場合は、 以下を実行してください:

1. ユーザーに lambda:InvokeFunction のアクセス許可があることを確認します。以下のポリシーを使用できます。

   Lambda 関数の ARN は、Lambda コンソールにある [関数の概要] から取得できます。

   アクセス許可を付与するにはユーザー、グループ、またはロールにアクセス許可を追加します。

   • 以下のユーザーとグループ AWS IAM アイデンティティセンター:

     アクセス許可セットを作成します。「AWS IAM アイデンティティセンター ユーザーガイド」の「アクセス許可セットを作成する」の手順に従ってください。

   • IAM 内で、ID プロバイダーによって管理されているユーザー:

     ID フェデレーションのロールを作成します。詳細については IAM ユーザーガイド の サードパーティー ID プロバイダー (フェデレーション) 用のロールを作成する を参照してください。

   • IAM ユーザー:

     • ユーザーが担当できるロールを作成します。手順については IAM ユーザーガイド の IAM ユーザーのロールの作成 を参照してください。

     • (お奨めできない方法) ポリシーをユーザーに直接アタッチするか、ユーザーをユーザーグループに追加します。「IAM ユーザーガイド」の「ユーザー (コンソール) へのアクセス許可の追加」の指示に従います。

2. AWS SDK for Python をインストールして設定します。詳細については、「ステップ 2: AWS CLI と AWS SDKs」を参照してください。

次のコードを client.py という名前のファイルに保存します。

# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: Apache-2.0

"""
Purpose
Test code for running the Amazon Textract Lambda
function example code.
"""

import argparse
import logging
import base64
import json
import io
import boto3

from botocore.exceptions import ClientError
from PIL import Image, ImageDraw


logger = logging.getLogger(__name__)


def analyze_image(function_name, image):
    """Analyzes a document with an AWS Lambda function.
    :param image: The document that you want to analyze.
    :return The list of Block objects in JSON format.
    """

    lambda_client = boto3.client('lambda')

    lambda_payload = {}

    if image.startswith('s3://'):
        logger.info("Analyzing document from S3 bucket: %s", image)
        bucket, key = image.replace("s3://", "").split("/", 1)
        s3_object = {
            'Bucket': bucket,
            'Name': key
        }

        lambda_payload = {"S3Object": s3_object}

    else:
        with open(image, 'rb') as image_file:
            logger.info("Analyzing local document: %s ", image)
            image_bytes = image_file.read()
            data = base64.b64encode(image_bytes).decode("utf8")

            lambda_payload = {"image": data}

    # Call the lambda function with the document.

    response = lambda_client.invoke(FunctionName=function_name,
                                    Payload=json.dumps(lambda_payload))

    return json.loads(response['Payload'].read().decode())


def add_arguments(parser):
    """
    Adds command line arguments to the parser.
    :param parser: The command line parser.
    """

    parser.add_argument(
        "function", help="The name of the AWS Lambda function that you want " \
        "to use to analyze the document.")
    parser.add_argument(
        "image", help="The document that you want to analyze.") 


def main():
    """
    Entrypoint for script.
    """
    try:
        logging.basicConfig(level=logging.INFO,
                            format="%(levelname)s: %(message)s")

        # Get command line arguments.
        parser = argparse.ArgumentParser(usage=argparse.SUPPRESS)
        add_arguments(parser)
        args = parser.parse_args()

        # Get analysis results.
        result = analyze_image(args.function, args.image)
        status = result['statusCode']

        blocks = result['body']
        blocks = json.loads(blocks)

        if status == 200:

            for block in blocks:
                print('Type: ' + block['BlockType'])
                if block['BlockType'] != 'PAGE':
                    print('Detected: ' + block['Text'])
                    print('Confidence: ' + "{:.2f}".format(block['Confidence']) + "%")

                print('Id: {}'.format(block['Id']))
                if 'Relationships' in block:
                    print('Relationships: {}'.format(block['Relationships']))
                print('Bounding Box: {}'.format(block['Geometry']['BoundingBox']))
                print('Polygon: {}'.format(block['Geometry']['Polygon']))
                print()
            print("Blocks detected: " + str(len(blocks)))
        else:
            print(f"Error: {result['statusCode']}")
            print(f"Message: {result['body']}")

    except ClientError as error:
        logging.error(error)
        print(error)


if __name__ == "__main__":
    main()

コードを実行します。コマンドライン引数には、Lambda 関数名と分析するドキュメントを指定します。ローカルドキュメントへのパスを指定するか、Amazon S3 バケットに保存されている ドキュメントへの Amazon S3 パスを使用できます。例えば、次のようになります。

python client.py function_name s3://bucket/path/document.jpg

ドキュメントが Amazon S3 バケットにある場合は、「ステップ 1: AWS Lambda 関数を作成する (コンソール)」のステップ 12 で指定したバケットと同じバケットであることを確認します。

正常に実行されると、コードはドキュメント内で検出された各ブロックタイプについて JSON レスポンスの一部を返します。