(半強制的に)人を笑顔にさせる余興アプリを作ってみた #Python – Qiita

import boto3
import json
import os
from decimal import Decimal
from linebot import LineBotApi, WebhookHandler
from linebot.exceptions import InvalidSignatureError
from linebot.models import MessageEvent, ImageMessage, TextSendMessage

LINE_CHANNEL_SECRET = os.environ['LINE_CHANNEL_SECRET']
LINE_CHANNEL_ACCESS_TOKEN = os.environ['LINE_CHANNEL_ACCESS_TOKEN']

line_bot_api = LineBotApi(LINE_CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(LINE_CHANNEL_SECRET)

s3 = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
rekognition = boto3.client('rekognition')

BUCKET_NAME = os.environ['S3_BUCKET_NAME']
TABLE_NAME = os.environ['DYNAMODB_TABLE_NAME']

@handler.add(MessageEvent, message=ImageMessage)
def handle_image_message(event):
    message_content = line_bot_api.get_message_content(event.message.id)

    file_name = datetime.now().strftime("%Y%m%d%H%M%S%f") + ".jpg"
    s3.put_object(
        Bucket=BUCKET_NAME,
        Key=file_name,
        Body=message_content.content,
        ContentType='image/jpeg'
    )

    # 一時返答
    line_bot_api.reply_message(
        event.reply_token,
        TextSendMessage(text="ピッ！(画像を受け取ったようだ)")
    )

    user_id = event.source.user_id
    profile = line_bot_api.get_profile(user_id)
    user_name = profile.display_name

    table = dynamodb.Table(TABLE_NAME)
    response = rekognition.detect_faces(
        Image={
            'S3Object': {
                'Bucket': BUCKET_NAME,
                'Name': file_name
            }
        },
        Attributes=['ALL']
    )

    # 写っている人単位
    faces = response['FaceDetails']
    smile_scores = []
    reply_message = ""

    # 顔が検出されなかった場合の処理
    if not faces:
        reply_message = "ビィ!!!!!(画像に人物が含まれていなくて怒っている)"
    else:
        # 顔ごとの情報を処理
        for face in faces:
            if 'Smile' in face and face['Smile']['Value']:
                score = face['Smile']['Confidence']
                for emotion in face['Emotions'] :
                    match emotion['Type']:
                        case 'ANGRY' | 'DISGUSTED' | 'CONFUSED' | 'SAD' | 'FEAR':
                            score -= emotion['Confidence'] * 1000
                        case 'HAPPY' :
                            score += emotion['Confidence'] / 100
                smile_scores.append(score)

        if smile_scores:
            ave_smile = sum(smile_scores) / len(smile_scores)
            res_smile = ave_smile if ave_smile > 70 else 70
            reply_message = f"ピィー！（今回のスマイルポイントは、{res_smile:.2f}点だそうです）"

            # Dynamodbに情報を保存
            table.put_item(
                Item={
                    'id': file_name,
                    'user_id': event.source.user_id,
                    'user_name': user_name,
                    'score': Decimal(str(ave_smile))
                }
            )
        else:
            reply_message = "ぴぃ…(笑顔の人物が見つからず悲しんでいる)"

    line_bot_api.push_message(user_id, TextSendMessage(text=reply_message))

def lambda_handler(event, context):

    signature = event["headers"]["x-line-signature"]
    body = event["body"]

    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        return {"statusCode": 400, "body": "Invalid signature"}

    return {"statusCode": 200, "body": "OK"}

import boto3
import json
from decimal import Decimal
import os

s3 = boto3.client('s3')
dynamodb = boto3.resource('dynamodb')
BUCKET_NAME = os.environ['S3_BUCKET_NAME']
TABLE_NAME = os.environ['DYNAMODB_TABLE_NAME']

def lambda_handler(event, context):
    table = dynamodb.Table(TABLE_NAME)
    # DynamoDBからスコアでソートして上位3件を取得
    data = table.scan()
    items = data['Items']
    sorted_items = sorted(items, key=lambda x: x['score'], reverse=True)[:3]

    for item in sorted_items:
        image_key = item['id']
        item['presigned_url'] = s3.generate_presigned_url('get_object',
                                                          Params={'Bucket': BUCKET_NAME, 'Key': image_key},
                                                          ExpiresIn=3600)

    def convert_decimals(obj):
        if isinstance(obj, list):
            return [convert_decimals(x) for x in obj]
        elif isinstance(obj, dict):
            return {k: convert_decimals(v) for k, v in obj.items()}
        elif isinstance(obj, Decimal):
            if obj % 1 == 0:
                return int(obj)
            else:
                return float(obj)
        else:
            return obj

    converted_items = convert_decimals(sorted_items)
    json_response = json.dumps(converted_items)

    return {
        'statusCode': 200,
        'headers': {'Content-Type': 'application/json'},
        'body': json_response
    }