Twitter のタイムラインで流れる話題を AWS の AI サービスで分析してみる ~後編

{
    
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObject",
                "s3:ListBucket",
                "s3:PutBucketNotification",
                "s3:DeleteBucket"
            ],
            "Resource": [
                "<バケットのARN>",
                "<バケットのARN>/*",
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:<リージョン>:<AWSアカウントID>:log-group:/aws/lambda/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "firehose:ListDeliveryStreams",
                "firehose:PutRecord",
                "firehose:PutRecordBatch"
            ],
            "Resource": [
                "arn:aws:firehose:<リージョン>:<AWSアカウントID>:deliverystream/Twitter-Dashboard-Entities",
                "arn:aws:firehose:<リージョン>:<AWSアカウントID>:deliverystream/Twitter-Dashboard-Sentiment"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "comprehend:DetectEntities",
                "comprehend:DetectSentiment"
            ],
            "Resource": "*"
        }
    ]
}

import json
import boto3
import os
import re

s3 = boto3.resource('s3')
comprehend = boto3.client('comprehend')
firehose = boto3.client('firehose')

entity_should_be_filtered = re.compile('^[\d#@]$')

def lambda_handler(event, context):
    print(event)
    
    for record in event['Records']:
        s3_bucket = record['s3']['bucket']['name']
        s3_key = record['s3']['object']['key']
        
        obj = s3.Object(s3_bucket, s3_key)
        tweets_as_string = obj.get()['Body'].read().decode('utf-8') 
        
        # S3に置かれているファイルは、改行コードでレコードが区切られる形式の構造となっています。
        # そのため、改行コードでsplitしてからJSONとしてパースします。
        tweets = tweets_as_string.split('\n')
        for tweet_string in tweets:
            if len(tweet_string) < 1:
                continue
            
            tweet = json.loads(tweet_string)

            # Tweetの感情分析をComprehendに依頼します。
            sentiment_response = comprehend.detect_sentiment(
                    Text=tweet['text'],
                    LanguageCode=tweet['lang']
                )
            print(sentiment_response)

            sentiment_record = {
                'tweetid': tweet['id'],
                'text': tweet['text'],
                'sentiment': sentiment_response['Sentiment'],
                'sentimentposscore': sentiment_response['SentimentScore']['Positive'],
                'sentimentnegscore': sentiment_response['SentimentScore']['Negative'],
                'sentimentneuscore': sentiment_response['SentimentScore']['Neutral'],
                'sentimentmixedscore': sentiment_response['SentimentScore']['Mixed']
            }
            
            firehose.put_record(
                DeliveryStreamName=os.environ['SENTIMENT_STREAM'],
                Record={
                    'Data': json.dumps(sentiment_record) + '\n'
                }
            )
            
            # Tweetの名前付きエンティティ分析(人名、場所など)をComprehendに依頼します。
            entities_response = comprehend.detect_entities(
                    Text=tweet['text'],
                    LanguageCode=tweet['lang']
                )
            print(entities_response)
            
            seen_entities = []
            for entity in entities_response['Entities']:
                # 数字の0-9, 記号の#, @の1文字だけだった場合は意味を見出しにくいので除外します。
                if (entity_should_be_filtered.match(entity['Text'])):
                    continue

                id = entity['Text'] + '-' + entity['Type']
                if (id in seen_entities) == False:
                    entity_record = {
                        'tweetid': tweet['id'],
                        'entity': entity['Text'],
                        'type': entity['Type'],
                        'score': entity['Score']
                    }
                    
                    firehose.put_record(
                        DeliveryStreamName=os.environ['ENTITY_STREAM'],
                        Record={
                            'Data': json.dumps(entity_record) + '\n'
                        }
                    )
                    seen_entities.append(id)

    return 'true'

SELECT  s.*,
    e.entity,
    e.type,
    e.score,
    t.lang as language,
    coordinates.coordinates[1] AS lon,
    coordinates.coordinates[2] AS lat ,
    place.name,
    place.country,
    (t.timestamp_ms / 1000) + (9 * 60 * 60) AS timestamp_in_seconds,
    regexp_replace(source,
    '\<.+?\>', '') AS src
FROM twitter_timeline_dashboard.tweets t
JOIN twitter_timeline_dashboard.tweet_sentiments s
    ON (s.tweetid = t.id)
JOIN twitter_timeline_dashboard.tweet_entities e
    ON (e.tweetid = t.id)