분석을 위해 AWS Comprehend로 텍스트 추출 및 전송

먼저 코드를 작성하여 Amazon SNS 주제와 Amazon SQS 대기열을 생성합니다.

다음 코드 샘플은 세 가지 필수 서비스에 연결하는 DocumentProcessor 클래스를 생성한 다음 Amazon SQS 대기열과 Amazon SNS 주제를 모두 생성합니다. Amazon SNS 주제는 작업 완료 상태에 대한 정보를 Amazon SQS 대기열에 제공하는 데 사용되며,이 대기열은 작업의 완료 상태를 얻기 위해 폴링됩니다. 작업이 완료되고 리소스가 더 이상 필요하지 않은 경우 Amazon SQS 대기열 및 Amazon SNS 주제를 삭제하는 방법도 있습니다.

import boto3
import json
import sys
import time

class DocumentProcessor:

    jobId = ''
    region_name = ''

    roleArn = ''
    bucket = ''
    document = ''

    sqsQueueUrl = ''
    snsTopicArn = ''
    processType = ''

    def __init__(self, role, bucket, document, region):
        self.roleArn = role
        self.bucket = bucket
        self.document = document
        self.region_name = region

        # Instantiates necessary AWS clients
        session = boto3.Session(profile_name='profile-name',
        region_name='self.region_name')
        self.textract = session.client('textract', region_name=self.region_name)
        self.sqs = session.client('sqs', region_name=self.region_name)
        self.sns = session.client('sns', region_name=self.region_name)

    def CreateTopicandQueue(self):

        millis = str(int(round(time.time() * 1000)))

        # Create SNS topic
        snsTopicName = "AmazonTextractTopic" + millis

        topicResponse = self.sns.create_topic(Name=snsTopicName)
        self.snsTopicArn = topicResponse['TopicArn']

        # create SQS queue
        sqsQueueName = "AmazonTextractQueue" + millis
        self.sqs.create_queue(QueueName=sqsQueueName)
        self.sqsQueueUrl = self.sqs.get_queue_url(QueueName=sqsQueueName)['QueueUrl']

        attribs = self.sqs.get_queue_attributes(QueueUrl=self.sqsQueueUrl,
                                                AttributeNames=['QueueArn'])['Attributes']

        sqsQueueArn = attribs['QueueArn']

        # Subscribe SQS queue to SNS topic
        self.sns.subscribe(
            TopicArn=self.snsTopicArn,
            Protocol='sqs',
            Endpoint=sqsQueueArn)

        # Authorize SNS to write SQS queue
        policy = """{{
  "Version": "2012-10-17",
  "Statement":[
    {{
      "Sid":"MyPolicy",
      "Effect":"Allow",
      "Principal" : {{"AWS" : "*"}},
      "Action":"SQS:SendMessage",
      "Resource": "{}",
      "Condition":{{
        "ArnEquals":{{
          "aws:SourceArn": "{}"
        }}
      }}
    }}
  ]
}}""".format(sqsQueueArn, self.snsTopicArn)

        response = self.sqs.set_queue_attributes(
            QueueUrl=self.sqsQueueUrl,
            Attributes={
                'Policy': policy
            })

    def DeleteTopicandQueue(self):
        self.sqs.delete_queue(QueueUrl=self.sqsQueueUrl)
        self.sns.delete_topic(TopicArn=self.snsTopicArn)
    

코드를 작성하여 StartDocumentTextDetection 작업을 호출하고 작업 결과를 가져옵니다.

DocumentProcessor 클래스에는 다음과 같은 방법도 필요합니다.

다음 코드는 각각를 호출StartDocumentTextDetection하고 추출된 텍스트를 가져오는 ProcessDocument 및 GetResults 메서드를 생성합니다.

    def ProcessDocument(self):
    
            # Checks if job found
            jobFound = False
    
            # Starts the text detection operation on the documents in the provided bucket
            # Sends status to supplied SNS topic arn
            response = self.textract.start_document_text_detection(
                    DocumentLocation={'S3Object': {'Bucket': self.bucket, 'Name': self.document}},
                    NotificationChannel={'RoleArn': self.roleArn, 'SNSTopicArn': self.snsTopicArn})
            print('Processing type: Detection')
    
            print('Start Job Id: ' + response['JobId'])
            dotLine = 0
            while jobFound == False:
                sqsResponse = self.sqs.receive_message(QueueUrl=self.sqsQueueUrl, MessageAttributeNames=['ALL'],
                                                    MaxNumberOfMessages=10)
    
                # Waits until messages are found in the SQS queue
                if sqsResponse:
                    if 'Messages' not in sqsResponse:
                        if dotLine < 40:
                            print('.', end='')
                            dotLine = dotLine + 1
                        else:
                            print()
                            dotLine = 0
                        sys.stdout.flush()
                        time.sleep(5)
                        continue
    
                    # Checks for a completed job that matches the jobID in the response from
                    # StartDocumentTextDetection
                    for message in sqsResponse['Messages']:
                        notification = json.loads(message['Body'])
                        textMessage = json.loads(notification['Message'])
                        if str(textMessage['JobId']) == response['JobId']:
                            print('Matching Job Found:' + textMessage['JobId'])
                            jobFound = True
                            text_data = self.GetResults(textMessage['JobId'])
                            self.sqs.delete_message(QueueUrl=self.sqsQueueUrl,
                                                    ReceiptHandle=message['ReceiptHandle'])
                            return text_data
                        else:
                            print("Job didn't match:" +
                                str(textMessage['JobId']) + ' : ' + str(response['JobId']))
                        # Delete the unknown message. Consider sending to dead letter queue
                        self.sqs.delete_message(QueueUrl=self.sqsQueueUrl,
                                                ReceiptHandle=message['ReceiptHandle'])
    
            print('Done!')
        
    # gets the results of the completed text detection job
    # checks for pagination tokens to determine if there are multiple pages in the input doc
    def GetResults(self, jobId):
        maxResults = 1000
        paginationToken = None
        finished = False

        while finished == False:
            response = None
            if paginationToken == None:
                response = self.textract.get_document_text_detection(JobId=jobId,
                                                                         MaxResults=maxResults)
            else:
                response = self.textract.get_document_text_detection(JobId=jobId,
                                                                         MaxResults=maxResults,
                                                                         NextToken=paginationToken)

            blocks = response['Blocks']

            # List to hold detected text
            detected_text = []

            # Display block information and add detected text to list
            for block in blocks:
                if 'Text' in block and block['BlockType'] == "LINE":
                    detected_text.append(block['Text'])

            # If response contains a next token, update pagination token
            if 'NextToken' in response:
                paginationToken = response['NextToken']
            else:
                finished = True

            return detected_text
    

위의 코드를 라는 파일에 저장합니다detectFileAsync.py.

다음 섹션에서이 파일을 사용하여 입력 문서의 텍스트 감지를 처리합니다.

코드를 작성하여 감지된 텍스트에 대한 감정 분석을 수행합니다.

from detectFileAsync import DocumentProcessor
import boto3
import pandas as pd

# Detect sentiment
def sentiment_analysis(detected_text, lang):

    comprehend = boto3.client("comprehend")

    detect_sent_response = comprehend.batch_detect_sentiment(
            TextList=detected_text, LanguageCode=lang)

    # Lists to hold sentiment labels and sentiment scores
    sentiments = []
    pos_score = []
    neg_score = []
    neutral_score = []
    mixed_score = []

    # for all results add the Sentiment label and sentiment scores to lists
    for res in detect_sent_response['ResultList']:
        sentiments.append(res['Sentiment'])
        print(res['SentimentScore'])
        print(type(res['SentimentScore']))
        for key, val in res['SentimentScore'].items():
            if key == "Positive":
                pos_score.append(val)
            if key == "Negative":
                neg_score.append(val)
            if key == "Neutral":
                neutral_score.append(val)
            if key == "Mixed":
                mixed_score.append(val)

    return sentiments, pos_score, neg_score, neutral_score, mixed_score
    

감지된 텍스트에 대해 개체 감지 또는 키 구문 감지와 같은 다른 분석 작업을 수행할 수도 있습니다. 진행 중인 감정 분석 작업과 마찬가지로 텍스트에서 이러한 분석 작업을 수행하는 함수를 작성할 수 있습니다.

코드를 작성하여 감지된 텍스트에서 개체 감지를 수행합니다.

# detect entities
def entity_detection(detected_text, lang):

    comprehend = boto3.client("comprehend")

    # convert and handle string here
    # do string handling
    detect_ent_response = comprehend.batch_detect_entities(
        TextList=detected_text, LanguageCode=lang)

    # To fold detected entities and entity types
    ents = []
    types = []

    # Get detected entities and types from the response returned by Comprehend
    for i in detect_ent_response['ResultList']:
        if len(i['Entities']) == 0:
            ents.append("N/A")
            types.append("N/A")
        else:
            sentence_ents = []
            sentence_types = []
            for entities in i['Entities']:
                sentence_ents.append(entities['Text'])
                sentence_types.append(entities['Type'])
            ents.append(sentence_ents)
            types.append(sentence_types)

    return ents, types

감지된 텍스트에서 키 구문 감지를 수행하는 코드를 작성합니다.

# Detect key phrases
def key_phrases_detection(detected_text, lang):

    comprehend = boto3.client("comprehend")

    key_phrases = []
    detect_phrases_response = comprehend.batch_detect_key_phrases(
        TextList=detected_text, LanguageCode=lang)
    for i in detect_phrases_response['ResultList']:
        if len(i['KeyPhrases']) == 0:
            key_phrases.append("N/A")
        else:
            phrases = []
            for phrase in i['KeyPhrases']:
                phrases.append(phrase['Text'])
            key_phrases.append(phrases)

    return key_phrases     
    

지금까지 생성한 모든 코드를 호출하는 함수를 생성해야 합니다. 함수는 DetectAnalyzeFileAsync.py 파일에서 생성한 DocumentProcessor 클래스를 사용한 다음 이전에 작성한 Amazon Comprehend를 사용하여 세 함수에 입력할 수 있도록 감지된 텍스트를 변수에 저장합니다. 또한 함수는 감지된 텍스트 및 분석 데이터가 삽입될 Pandas 데이터 프레임을 구성해야 합니다. 마지막으로 Pandas 데이터 프레임은 CSV 파일로 저장됩니다.

코드를 작성하여 Textract로 입력 문서를 처리하고 감지된 텍스트를 Comprehend에 전달합니다.

def process_document(roleArn, bucket, document, region_name):

    # Create analyzer class from DocumentProcessor, create a topic and queue, use Textract to get text,
    # then delete topica and queue
    analyzer = DocumentProcessor(roleArn, bucket, document, region_name)
    analyzer.CreateTopicandQueue()
    extracted_text = analyzer.ProcessDocument()
    analyzer.DeleteTopicandQueue()

    # detect dominant language
    comprehend = boto3.client("comprehend")
    response = comprehend.detect_dominant_language(Text=str(extracted_text[:10]))
    print(response)
    print(type(response))
    lang = ""
    for i in response['Languages']:
        lang = i['LanguageCode']
    print(lang)

    # or you can enter language code below
    # lang = "en"

    print("Lines in detected text:" + str(len(extracted_text)))
    sliced_list = []
    start = 0
    end = 24
    while end < len(extracted_text):
        sliced_list.append(extracted_text[start:end])
        start += 25
        end += 25
    print(sliced_list)

    # Create lists to hold analytics data, these will be turned into columns
    all_sents = []
    all_scores = []
    all_ents = []
    all_types = []
    all_key_phrases = []
    all_pos_ratings = []
    all_neg_ratings = []
    all_neutral_ratings = []
    all_mixed_ratings = []

    # For every slice, get sentiment analysis, entity detection and key phrases, append results to lists
    for slice in sliced_list:
        slice_labels, pos_ratings, neg_ratings, neutral_ratings, mixed_ratings = sentiment_analysis(slice, lang)
        all_sents.append(slice_labels)
        all_pos_ratings.append(pos_ratings)
        all_neg_ratings.append(neg_ratings)
        all_neutral_ratings.append(neutral_ratings)
        all_mixed_ratings.append(mixed_ratings)
        slice_ents, slice_types = entity_detection(slice, lang)
        all_ents.append(slice_ents)
        all_types.append(slice_types)
        key_phrases = key_phrases_detection(slice, lang)
        all_key_phrases.append(key_phrases)

    # List comprehension to flatten multiple lists into a single list
    extracted_text = [line for sublist in sliced_list for line in sublist]
    all_sents = [sent for sublist in all_sents for sent in sublist]
    all_scores = [score for sublist in all_scores for score in sublist]
    all_ents = [ents for sublist in all_ents for ents in sublist]
    all_types = [types for sublist in all_types for types in sublist]
    all_key_phrases = [kp for sublist in all_key_phrases for kp in sublist]
    all_mixed_ratings = [kp for sublist in all_mixed_ratings for kp in sublist]
    all_pos_ratings = [kp for sublist in all_pos_ratings for kp in sublist]
    all_neg_ratings = [kp for sublist in all_neg_ratings for kp in sublist]
    all_neutral_ratings = [kp for sublist in all_neutral_ratings for kp in sublist]

    print(len(extracted_text))
    print(len(all_sents))
    print(len(all_ents))
    print(len(all_types))
    print(len(all_key_phrases))

    print("List of Recognized Entities:")

    # Create dataframe and save as CSV
    df = pd.DataFrame({'Sentences':extracted_text, 'Sentiment':all_sents, 'SentPosScore':all_pos_ratings,
                       'SentNegScore':all_neg_ratings, 'SentNeutralScore':all_neutral_ratings, 'SentMixedRatings':all_mixed_ratings,
                       'Entities':all_ents, 'EntityTypes':all_types,'KeyPhrases:':all_key_phrases})
    analysis_results = str(document.replace(".","_") + "_" + "analysis" + ".csv")
    df.to_csv(analysis_results, index=False)

    print(df)
    print("Data written to file!")

    return extracted_text, analysis_results     
    

코드를 작성하여 문서를 처리하고 결과 데이터를 S3에 업로드합니다. 아래 코드 샘플에서의 값을 Amazon TextractroleArn와 함께 사용하도록 구성한 역할의 ARN으로 바꿉니다. 의 값을 계정이 운영 중인 리전region_name으로 바꿉니다. 마지막으로 값을 문서가 포함된 S3 버킷의 bucket_name 이름으로 바꿉니다.

def main():

    # Initialize S3 client and set RoleArn, region name, and bucket name
    s3 = boto3.client("s3")
    roleArn = ''
    region_name = ''
    bucket_name = ''

    # initialize global corpus
    full_corpus = []

    # to hold all docs in bucket
    docs_list = []

    # loop through docs in bucket, get names of all docs
    s3_resource = boto3.resource("s3")
    bucket = s3_resource.Bucket(bucket_name)
    for bucket_object in bucket.objects.all():
        docs_list.append(bucket_object.key)
    print(docs_list)

    # For all the docs in the bucket, invoke document processing function,
    # add detected text to corpus of all text in batch docs,
    # and save CSV of comprehend analysis data and textract detected to S3
    for i in docs_list:
        detected_text, analysis_results = process_document(roleArn, bucket_name, i, region_name)
        full_corpus.append(detected_text)
        print("Uploading file: {}".format(str(analysis_results)))
        name_of_file = str(analysis_results)
        s3.upload_file(name_of_file, bucket_name, name_of_file)

    # print the global corpus
    print(full_corpus)

if __name__ == "__main__":
    main()     
    

섹션의 진행 코드를 Python 파일에 넣고 실행합니다.