LambdaとCloud Vision APIで日本語の文字抽出をしてみた

リサリサ

2021.06.17

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、リサリサです。

とあるアプリのスコアを毎日手入力で記録しているのですが、文字を打ち込む作業がとてもめんどくさく、画像を送ったらいい感じに認識して記録してくれたらいいなと思ったので、作ってみました。

AWSにはAmazon Textractという文字抽出できるサービスがあるのですが、日本語にまだ対応してくれておらず、今回やりたい事は実現できそうになかったので、GoogleCloudのサービスのGoogle Cloud Visionを使ってみることにしました。

作ったもの

インプット用のS3バケットに画像が置かれたらLambdaがトリガーされて、Cloud Vision APIを呼び出し、返ってきた文字をtxtでアウトプット用のS3バケットに保存します。

こんな感じになります。よく見ると絵を文字と認識してしまったりしていますが、今回欲しいのは数字なので、これだけの精度があれば十分です。

作ってみた

Google Cloudのプロジェクトを作成する

Cloud Vision APIのクイックスタートの通りに進めていきます。詳しくはこちらの記事にまとめたので、こちらをご覧頂ければと思います。

ここで取得したJSONファイルのキーをLambdaに置くことで、作ったプロジェクトを呼び出せるようになります。

S3バケットを作成する

手順は割愛しますが、インプット用のバケットとアウトプット用のバケットを作成します。

PUTイベントをトリガーにする時は、ミスって無限ループにならないように、インプット用のバケットとアウトプット用のバケットは分けた方がよいみたいです。

Lambdaを作成する

SAMでやる場合は、こちらの記事をどうぞ

私はSAMはあまりよく分からないので、ライブラリをzipに固めてレイヤーにしてみました。

レイヤーを作る

python使える環境で、レイヤー用のzipを作ります。

今回はローカル環境で作ってしまいましたが、環境が違うとエラーになったりしますので、Lambdaのランタイム環境と同じEC2で作ると確実だと思います。

mkdir google-cloud-vision
cd google-cloud-vision
mkdir python
pip3 install -t ./python google-cloud-vision
zip -r google-cloud-vision.zip python

Lambdaを作る

ランタイムはちょっと古いのですが、Python3.6にしました。3.8ではうまく動かず…。

ロールはインプットのS3に読み込み権限があるもの、アウトプットのS3に書き込み権限があるものを設定します。

コードはこちら

import io
import os
import boto3
import json
from datetime import datetime

# Imports the Google Cloud client library
from google.cloud import vision
output_bucket = 'アウトプット用のバケット名'

    
def lambda_handler(event, context):

    # Instantiates a client
    client = vision.ImageAnnotatorClient()
    
    # s3から画像取得
    s3 = boto3.client('s3')
    input_bucket=event['Records'][0]['s3']['bucket']['name']
    input_key=event['Records'][0]['s3']['object']['key']
    content = s3.get_object(Bucket=input_bucket, Key=input_key)['Body'].read()
    
    # Cloud Vision API呼び出し
    image = vision.Image(content=content)
    response = client.document_text_detection(
        image=image,
        image_context={'language_hints': ['ja']}
    )
    
    # レスポンスからテキストデータを抽出
    output_text = ''
    for page in response.full_text_annotation.pages:
        for block in page.blocks:
            for paragraph in block.paragraphs:
                for word in paragraph.words:
                    output_text += ''.join([
                        symbol.text for symbol in word.symbols
                    ])
                output_text += '\n'

    # S3に書き込み
    s3 = boto3.resource('s3')
    output_key = f"{input_key}_{datetime.now().strftime('%Y-%m-%d-%H-%M-%S')}.txt"
    file_contents = output_text
    
    obj = s3.Object(output_bucket,output_key)
    obj.put( Body=file_contents )

    return {
        'statusCode': 200
    }