使用检测文本 AWS Lambda 函数

登录 AWS 管理控制台 并打开 AWS Lambda 控制台，网址为https://console.aws.amazon.com/lambda/。

选择创建函数。有关更多信息，请参阅使用控制台创建 Lambda 函数。

选择以下选项：

选择创建函数以创建 AWS Lambda 函数。

在函数页面上，选择配置选项卡。

在权限窗格的执行角色下，选择角色名称以在 IAM 控制台中打开该角色。

在权限选项卡中，依次选择添加权限和创建内联策略。

选择 JSON 选项卡，并将该策略替换为以下策略：

选择查看策略。

例如，输入策略的名称DetectDocumentText-access。

选择创建策略。

如果您要在 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 开发工具包创建部署包。

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

记下您在本过程的步骤 8 中使用的压缩文件 (boto3-layer.zip) 的名称。

打开 AWS Lambda 控制台，网址为https://console.aws.amazon.com/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 控制台中的函数概述，获取 Lambda 函数的 ARN。

   要提供访问权限，请为您的用户、组或角色添加权限：

   • 中的用户和群组 AWS IAM Identity Center：

     创建权限集合。按照《AWS IAM Identity Center 用户指南》中创建权限集的说明进行操作。

   • 通过身份提供者在 IAM 中托管的用户：

     创建适用于身份联合验证的角色。按照《IAM 用户指南》中针对第三方身份提供者创建角色（联合身份验证）的说明进行操作。

   • IAM 用户：

     • 创建您的用户可以担任的角色。按照《IAM 用户指南》中为 IAM 用户创建角色的说明进行操作。

     • （不推荐使用）将策略直接附加到用户或将用户添加到用户组。按照《IAM 用户指南》中向用户添加权限（控制台）中的说明进行操作。

2. 安装和配置 AWS 适用于 Python 的开发工具包。有关更多信息，请参阅 步骤 2：设置 AWS CLI and AWS 软件开发工具包。

将以下代码保存到名为 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 响应。