擷取文字並傳送至 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。在下面的程式碼範例中，將 的值取代roleArn為您設定用於 Amazon Textract 之角色的 ARN。將 的值取代region_name為您帳戶操作所在的區域。最後，將值取代bucket_name為包含文件的 S3 儲存貯體名稱。

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 檔案並執行。