初学者向け「Amazon Comprehend」（AI自然言語処理サービス）をPythonで利用するには：AWSチートシート

reviews = ['My wife really enjoyed this book and so did I!',
           'This book becomes less interesting towards the end. '
           'この本は終盤に向かうにつれてつまらなくなりました。',
           'Cathedral, the last story collected in this book, is one of my '
           'favorite short stories by Raymond Carver.',
           'すごくわかりやすい！要点が簡潔にまとめてある。1ヶ月で無事合格できた。',
           '出題範囲を一通り網羅してるのは良いと思います。'
           'ただ、この本だけで対策するのは厳しいと思います。',
           '他の方のレビューにもありますが、著者の意見は賛否が分かれる内容だと思いました。',
           'ベストセラー作品なので期待して購入しましたが、私にはイマイチでした。',
           'この本に収録されている背の曲がった男は、シャーロック・ホームズシリーズの中でも'
           '特に好きな作品の一つです。']
 
import boto3
 
# リージョンを（東京リージョンに）指定
region = 'ap-northeast-1'
 
# クライアントを作成
comprehend = boto3.client('comprehend', region)
 
# 主要言語を特定するバッチ処理を実行
response = comprehend.batch_detect_dominant_language(TextList=reviews)
 
# ドキュメントごとに結果を取得
for result in response['ResultList']:
    # 原文を表示
    print(result['Index'] + 1, reviews[result['Index']])
    # 検出された言語情報を取得して表示
    for language in result['Languages']:
        print(f"言語コード:{language['LanguageCode']} "
              f"信頼度:{language['Score']:.2f}")
    print()

# エンティティを検出するバッチ処理を実行
response = comprehend.batch_detect_entities(TextList=reviews,
                                            LanguageCode='ja')
 
# ドキュメントごとに結果を取得
for result in response['ResultList']:
    # エンティティが検出さた場合
    entities = result['Entities']
    if entities:
        # 原文を表示
        print(result['Index'] + 1, reviews[result['Index']])
        for entity in entities:
            # エンティティを表示
            print(f"Score:{entity['Score']:.2f} "
                  f"Type:{entity['Type'].ljust(8)} "
                  f"Text:{entity['Text']}")
        print()

# 感情分析のバッチ処理を実行
response = comprehend.batch_detect_sentiment(TextList=reviews,
                                             LanguageCode='ja')
 
# ドキュメントごとに結果を取得
for result in response['ResultList']:
    # 原文を表示
    print(result['Index'] + 1, reviews[result['Index']])
    # 結果を表示
    score = result['SentimentScore']
    print(f"総合:{result['Sentiment'].ljust(8)} ("
          f"Positive:{score['Positive']:.2f} "
          f"Negative:{score['Negative']:.2f} "
          f"Neutral:{score['Neutral']:.2f} "
          f"Mixed:{score['Mixed']:.2f})\n")

import gzip
import json
import pandas as pd
import re
 
# レビューデータをロード
beauty_data = []
with gzip.open('All_Beauty_5.json.gz') as f:
    for l in f:
        beauty_data.append(json.loads(l.strip()))
 
magazine_data = []
with gzip.open('Magazine_Subscriptions_5.json.gz') as f:
    for l in f:
        magazine_data.append(json.loads(l.strip()))
 
# データをPandasのデータフレームに変換
df_beauty = pd.DataFrame.from_records(beauty_data)
df_magazine = pd.DataFrame.from_records(magazine_data)
 
# 「トピック」列を追加
df_beauty['topic'] = 'beauty'
df_magazine['topic'] = 'magazine'
 
# データフレームを結合
df = pd.concat([df_beauty, df_magazine])
 
# レビューが重複しているレコードを削除
df.drop_duplicates(subset='reviewText', inplace=True)
 
# レビュー中の改行文字を削除
df['reviewText'] = df['reviewText'].apply(lambda x: str(x).replace('\n', ' '))
 
# 「文章数」列を追加して、文章数が3以上のレコードのみ残す
eos = r'\.+|\?!|!\?|\?+|!+'
df['numSent'] = df['reviewText'].apply(lambda x: len(re.split(eos, str(x))) - 1)
df = df[df['numSent'] >= 3]
 
# ピックごとに550件ずつサンプリング
df = pd.concat([df[df.topic == 'beauty'].sample(550, random_state=1), 
                df[df.topic == 'magazine'].sample(550, random_state=1)])
 
# データを確認
print(df.groupby('topic').agg(reviewText_count=('reviewText', 'count'),
                              overall_mean=('overall', 'mean'),
                              numSent_mean=('numSent', 'mean')
                              ).round(2))
 
# 以降のトピックモデリングで使用する入力ファイル名
input_file = 'amazon_reviews.txt'
 
# レビューをファイルに書き出し
df.reviewText.to_csv(input_file, header=None, index=None)

import uuid
 
# S3クライアントを作成
s3 = boto3.client('s3', region)
 
# 一意識別子を利用してバケット名を作成
bucket = 'comprehend-' + str(uuid.uuid1())
 
# S3にバケットを作成
response = s3.create_bucket(
                Bucket=bucket,
                CreateBucketConfiguration={'LocationConstraint': region})
 
# 入力ファイルのオブジェクトキー
input_object_key = f'input/{input_file}'
 
# S3にファイルをアップロード
s3.upload_file(input_file, bucket, input_object_key)

import time
 
# 入出力データの構成を設定
input_s3_url = f's3://{bucket}/{input_object_key}'
input_doc_format = 'ONE_DOC_PER_LINE'
output_s3_url = f's3://{bucket}/output'
input_data_config = {'S3Uri': input_s3_url, 'InputFormat': input_doc_format}
output_data_config = {'S3Uri': output_s3_url}
 
# データアクセスロールARNを設定
data_access_role_arn = '<data_access_role_arn>'
# ジョブ名を指定
job_name = 'topic-detection-sample-job'
# トピック数を指定
number_of_topics = 2
 
# トピック検出ジョブを開始
response = comprehend.start_topics_detection_job(
                            InputDataConfig=input_data_config,
                            OutputDataConfig=output_data_config,
                            DataAccessRoleArn=data_access_role_arn,
                            JobName=job_name,
                            NumberOfTopics=number_of_topics)
 
# ジョブIDを取得 
job_id = response['JobId']
 
while True:
    # ジョブの情報を取得
    response = comprehend.describe_topics_detection_job(
        JobId=job_id
    )
    # ステータスを取得
    status = response['TopicsDetectionJobProperties']['JobStatus']
    
    # ステータスが次のいずれかになったらwhile文を抜け出す
    if status in ['COMPLETED', 'FAILED', 'STOP_REQUESTED', 'STOPPED']:
        print(' Status:', status)
        break
 
    print('>', end='')
    time.sleep(60)

# 出力ファイルのオブジェクトキーを取得
output_uri = response['TopicsDetectionJobProperties']['OutputDataConfig']['S3Uri']
output_object_key = ''.join(output_uri.partition('output')[1:])
 
# S3から結果ファイルをダウンロード
output_file = 'output.tar.gz'
s3.download_file(bucket, output_object_key, output_file)

import tarfile
# ファイルを抽出
with tarfile.open(output_file, 'r:gz') as t:
    t.extractall(path='output')
# topic-terms.csvをデータフレームとして読み込み
df_terms = pd.read_csv('output/topic-terms.csv')
df_terms

# doc-topics.csvをデータフレームとして読み込み
df_topics = pd.read_csv('output/doc-topics.csv')
 
# ドキュメント名(docname)列からドキュメント番号(doc_num)列を作成
df_topics['doc_num'] = df_topics.docname.apply(lambda x: int(x.split(':')[1]))
 
# ドキュメント番号順にデータをソート
df_topics.sort_values('doc_num', inplace=True)
df_topics.head(10)

# 「正解データ」列を作成
df_topics.loc[df_topics.doc_num<550, 'true'] = 1
df_topics.loc[df_topics.doc_num>=550, 'true'] = 0
 
# proportionが0.5より大きいレコードだけ残す
df_topics = df_topics[df_topics.proportion.round() > 0].reset_index(drop=True)
 
# 「正解率」
print(f'正解率:{(df_topics.topic == df_topics.true).sum()/len(df_topics):.2f}')

# オブジェクトを削除
response = s3.list_objects_v2(Bucket=bucket)
for content in response['Contents']:
    s3.delete_object(Bucket=bucket, Key=content['Key'])
 
# バケットを削除
s3.delete_bucket(Bucket=bucket)