AWS Lambda 함수를 사용하여 텍스트 감지

에 로그인 AWS Management Console 하고 https://console.aws.amazon.com/lambda/ AWS Lambda 콘솔을 엽니다.

함수 생성을 선택합니다. 자세한 내용은 콘솔로 Lambda 함수 생성을 참조하세요.

다음과 같은 옵션을 선택하세요.

함수 생성을 선택하여 AWS Lambda 함수를 생성합니다.

함수 페이지에서 구성 탭을 선택합니다.

권한 창의 실행 역할에서 IAM 콘솔의 해당 역할을 열 역할 이름을 선택합니다.

권한 탭에서 권한 추가, 그 다음 인라인 정책 생성을 선택합니다.

JSON을 선택하고 정책을 다음 정책으로 바꿉니다.

정책 검토를 선택합니다.

DetectDocumentText-access와 같은 정책의 이름을 입력합니다.

정책 생성을 선택합니다.

Amazon S3 버킷에 분석용 문서를 저장하는 경우 반드시 Amazon S3 액세스 정책을 추가해야 합니다. 이를 위해서는 AWS Lambda 콘솔에서 7~11단계를 반복하고 다음과 같이 변경합니다.

1. 8단계에서는 다음 정책을 사용하세요. 버킷/폴더 경로를 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/
                

이 절차의 8단계에서 사용하는 zip 파일(boto3-layer.zip)의 이름을 기록해 둡니다.

https://console.aws.amazon.com/lambda/ AWS Lambda 콘솔을 엽니다.

탐색 창에서 계층을 선택합니다.

계층 생성을 선택합니다.

Name(이름)과 Description(설명)을 입력합니다.

코드 항목 유형에서 .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 콘솔의 함수 개요에서 Lambda 함수에 대한 ARN을 가져올 수 있습니다.

   액세스 권한을 제공하려면 사용자, 그룹 또는 역할에 권한을 추가하세요.

   • 의 사용자 및 그룹 AWS IAM Identity Center:

     권한 세트를 생성합니다. AWS IAM Identity Center 사용자 안내서에서 권한 세트 생성의 지침을 따릅니다.

   • ID 제공업체를 통해 IAM에서 관리되는 사용자:

     ID 페더레이션을 위한 역할을 생성합니다. IAM 사용자 설명서의 Create a role for a third-party identity provider (federation)의 지침을 따릅니다.

   • IAM 사용자:

     • 사용자가 맡을 수 있는 역할을 생성합니다. IAM 사용자 설명서에서 Create a role for an IAM user의 지침을 따릅니다.

     • (권장되지 않음) 정책을 사용자에게 직접 연결하거나 사용자를 사용자 그룹에 추가합니다. 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 응답을 반환합니다.