[AWS]Rekognition과 Comprehend를 이용한 텍스트 정보 추출 애플리케이션

앞서 pipenv --three가 실행되지 않던 문제는 버전 등 pipenv에 어떤 복잡한 문제가 있기 때문에 많이 사용되지 않는 추세라고 하기도 해서 일단은 스킵했다. 대신 pipenv --python 3.7과 같은 방식으로 원하는 파이썬 버전을 골라 가상환경을 구축할 수 있다.

 

텍스트 정보 추출에서 각 AWS 서비스들은 S3 - 스토리지, Rekognition - 인식, Comprehend(Medical) - 추출, DynamoDB - 저장소의 역할을 도맡아서 서비스 구현부를 구성한다. Chalice는 마찬가지로 boto3의 도움을 받아 오케스트레이션 계층을 형성하고 각 서비스를 UI 계층으로 내보내는 엔드포인트의 역할을 수행한다.

 

프로젝트 구조는 다음과 같다.

노란 글씨들은... 색깔이 이쁘니까 무시해주자(...). 이번 프로젝트도 클론 코딩을 하면서 애플리케이션이 어떻게 돌아가는지 확인해보는 것에 중점을 뒀는데 나름 두번째 하는거라고 파일 간의 연관관계가 꽤나 머리에 잘 들어왔다. 책에서 로컬-웹페이지-배포의 과정을 거치면서 애플리케이션을 테스트하도록 하는데, 로컬에서 테스트 해볼 때 가상환경에 모듈을 설치해야 한다는 언급이 없다든지, 에러 메시지에 대한 언급이 없어 똑같은 화면을 구현하는 데에 너무 애를 먹는다...ㅜㅜ

 

오늘 먹은 애는 아래의 app.py에서 불러와야 할 storage_location이 문자열 자료형을 받지 않는다는 거였는데 버킷명으로 표시하든 url로 표시하든 boto3를 써서 버킷 자체를 불러오든 저놈을 선언할 수가 없어서 결국 로컬이고 배포고 아무것도 못해봤다.

 

- 구현부 소스코드들

# contact_store.py

import boto3

class ContactStore:
    def __init__(self, store_location):
        self.table = boto3.resource('dynamodb').Table(store_location)
    
    def save_contact(self, contact_info):
        response = self.table.put_item(
            Item = contact_info
        )
    
    def get_all_contacts(self):
        response = self.table.scan()

        contact_info_list = []
        for item in response['Items']:
            contact_info_list.append(item)
        
        return contact_info_list
    
    def get_contact_by_name(self, name):
        response = self.table.get_item(
            Key = {'name': name}
        )

        if 'Item' in response:
            contact_info = response['Item']
        else:
            contact_info = {}
        
        return contact_info

연락처를 저장하는 곳은 AWS의 DynamoDB인데 NoSQL 기반 서비스로 성능이 좋은데다가 DB 관리를 할 필요가 없다고 한다.

# extraction_service.py

import boto3
from collections import defaultdict
import usaddress

class ExtractionService:
    def __init__(self):
        self.comprehend = boto3.client('comprehend')
        self.comprehend_med = boto3.client('comprehendmedical')
    
    def extract_contact_info(self, contact_string):
        contact_info = defaultdict(list)

        # Comprehend
        response = self.comprehend.detect_entities(
            Text = contact_string,
            LanguageCode = 'en'
        )

        for entity in response['Entities']:
            if entity['Type'] == 'PERSON':
                contact_info['name'].append(entity['Text'])
            elif entity['Type'] == 'ORGANIZATION':
                contact_info['organization'].append(entity['Text'])
        
        # Comprehend Medical
        response = self.comprehend_med.detect_phi(
            Text = contact_string
        )

        for entity in response['Entities']:
            if entity['Type'] == 'EMAIL':
                contact_info['email'].append(entity['Text'])
            elif entity['Type'] == 'PHONE_OR_FAX':
                contact_info['phone'].append(entity['Text'])
            elif entity['Type'] == 'PROFESSION':
                contact_info['title'].append(entity['Text'])
            elif entity['Type'] == 'ADDRESS':
                contact_info['address'].append(entity['Text'])
        
        address_string = ''.join(contact_info['address'])
        address_parts = usaddress.parse(address_string)

        for part in address_parts:
            if part[1] == 'PlaceName':
                contact_info['city'].append(part[0])
            elif part[1] == 'StateName':
                contact_info['state'].append(part[0])
            elif part[1] == 'ZipCode':
                contact_info['zip'].append(part[0])
        
        return dict(contact_info)

comprehend도 일반 버전이랑 메디컬 버전이 나뉘는데, 각자 특화된 부분이 다르다고 한다. 이 프로젝트에서는 명함을 인식하는게 목적인데 이름이나 회사는 일반 버전으로, 연락처나 주소는 메디컬로 받고 있다. 주소 부분은 미국식 주소에 맞춰서 재가공 하는 부분이 또 들어가 있다.

# recognition_service.py

import boto3

class RecognitionService:
    def __init__(self, storage_service):
        self.client = boto3.client('rekognition')
        self.bucket_name = storage_service.get_storage_location()
    
    def detect_text(self, file_name):
        response = self.client.detect_text(
            Image = {
                'S3Object': {
                    'Bucket': self.bucket_name,
                    'Name': file_name
                }
            }
        )

        lines = []
        for detection in response['TextDetections']:
            if detection['Type'] == 'LINE':
                lines.append({
                    'text': detection['DetectedText'],
                    'confidence': detection['Confidence'],
                    'boundingBox': detection['Geometry']['BoundingBox']
                })
        
        return lines

# storage_service.py

import boto3

class StorageService:
    def __init__(self, storage_location):
        self.client = boto3.client('s3')
        self.bucket_name = storage_location
    
    def get_storage_location(self):
        return self.bucket_name

    def upload_file(self, file_bytes, file_name):
        self.client.put_object(Bucket = self.bucket_name,
                               Body = file_bytes,
                               Key = file_name,
                               ACL = 'public-read')
        
        return {'fileId': file_name,
                'fileUrl': "http://" + self.bucket_name + ".s3.amazonaws.com/" + file_name}

from chalice import Chalice
from chalicelib import storage_service
from chalicelib import recognition_service
from chalicelib import extraction_service
from chalicelib import contact_store

import boto3
import base64
import json

app = Chalice(app_name='Capabilities')
app.debug = True

# 외 않 되
s3r = boto3.resource('s3')
storage_location = s3r.Bucket('contents.stubborngastropod.ai')
# 외
storage_service = storage_service.StorageService(storage_location)
recognition_service = recognition_service.RecognitionService(storage_location)
extraction_service = extraction_service.ExtractionService()
store_location ='Contacts'
contact_store = contact_store.ContactStore(store_location)

@app.route('/images', methods = ['POST'], cors = True)
def upload_image():
    request_data = json.loads(app.current_request.raw_body)
    file_name = request_data['filename']
    file_bytes = base64.b64decode(request_data['filebytes'])

    file_info = storage_service.upload_file(file_bytes, file_name)

    return file_info


@app.route('/images/{image_id}/extract-info', methods = ['POST'], cors = True)
def extract_image_info(image_id):
    MIN_CONFIDENCE = 70.0

    text_lines = recognition_service.detect_text(image_id)

    contact_lines = []
    for line in text_lines:
        if float(line['confidence']) >= MIN_CONFIDENCE:
            contact_lines.append(line['text'])

    contact_string = '   '.join(contact_lines)
    contact_info = extraction_service.extract_contact_info(contact_string)

    return contact_info


@app.route('/contacts', methods = ['POST'], cors = True)
def save_contact():
    request_data = json.loads(app.current_request.raw_body)

    contact = contact_store.save_contact(request_data)

    return contact


@app.route('/contacts', methods = ['GET'], cors = True)
def get_all_contacts():
    contacts = contact_store.get_all_contacts()

    return contacts

막힌 부분은 여기... storage_location을 받아와야 하는데 app.py에서 자꾸 거부를 한다 ㅠ 앞에서 스킵한 챕터에 관련 내용이 있는데 시키는대로 내 버킷명도 집어넣어보고 이것저것 해봐도 안되는 이유를 모르겠다... 이건 스터디에서 물어볼 내용인듯

 

버그를 하나하나 잡아내기에는 시간이 너무 오래걸리고, aws를 이용한 ai 애플리케이션의 아키텍처를 이해하는 게 우선이므로 일단은 킵해두는 게 맞다고 판단했다. 이후에는 자바스크립트 배우고 연동 연습해보는걸로!

 

참고자료: http://www.yes24.com/Product/Goods/101806234

AWS 기반 AI 애플리케이션 개발 - YES24