Zenn
💸

LINEbot, AWS Lambda(APIgateway, EventBridge)にまとめて入門しつつ6億円を狙う

2021/06/01に公開
LINE
Python
Lambda
Amazon API Gateway
#
スクレイピング
tech

0.はじめに

0.1.概要

ロト6の当選判定をしてくれるLINEbotを2種類作った

  • 返信型：購入した番号を送信したら直近の番号で当選判定を返信してくれるやつ
  • 定期プッシュ通知型：自動購入番号を元にした当選結果を毎回自動通知してくれるやつ

0.2.こんな人におすすめ

  • Lambda＋APIgatewayでサクッとAPIを作ってみたい(→返信型)
  • Lambda＋EventBridgeでサクッと定期実行バッチを作ってみたい(→通知型)
  • LINEbotを作ってみたい
  • 6億円欲しい

1.返信型作成

1.1.ローカルでやること

  1. lambda関数zipの用意のためにテキトーにディレクトリを作成(今回はlinebotと命名)
  2. linebot/lambda_function.pyを作成
lambda_function.py ソースコード
    import os
    import sys
    import bs4
    import requests
    from linebot import (
        LineBotApi, WebhookHandler
    )
    from linebot.models import (
        MessageEvent, TextMessage, TextSendMessage,
    )
    from linebot.exceptions import (
        LineBotApiError, InvalidSignatureError
    )
    
    CHANNEL_SECRET = os.getenv('LINE_CHANNEL_SECRET', None)
    CHANNEL_ACCESS_TOKEN = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', None)
    
    LINE_USER_ID = os.getenv('LINE_USER_ID', None)
    LINE_BOT_API = LineBotApi(CHANNEL_ACCESS_TOKEN)
    LINE_HANDLER = WebhookHandler(CHANNEL_SECRET)
    
    LOT_RESULT_URL = "https://takarakuji.rakuten.co.jp/backnumber/loto6/"
    SUCCESS_RES = {"statusCode": 200, "body": "Success"}
    FAILDED_RES = {"statusCode": 500, "body": "Error"}
    
    def lambda_handler(event, context):
    
        @LINE_HANDLER.add(MessageEvent, message=TextMessage)
        def message(line_event):
            # 当選番号をスクレイピングで取得
            res = requests.get(LOT_RESULT_URL)
            soup = bs4.BeautifulSoup(res.text, "html.parser")
            recent_loto_numbers = [s.text for s in soup.find_all(
                'span', class_='loto-font-large')[:7]]
            recent_loto_numbers_str = " ".join(recent_loto_numbers)
            
            # ユーザーから受信した当選番号の取得  入力例：「1 2 25 30 40 41」
            user_text = line_event.message.text
            my_numbers = user_text.split(" ")
            
            # 当選判定
            matched_numbers = set(recent_loto_numbers) & set(my_numbers)
            matched_numbers_length = len(matched_numbers)
            bonus_matched = recent_loto_numbers[-1] in [f"({x})" for x in my_numbers]
            if matched_numbers_length == 6:
                send_message = "1等！！！！！"
            elif matched_numbers_length == 5 and bonus_matched:
                send_message = "2等！！！！"
            elif matched_numbers_length == 5:
                send_message = "3等！！！"
            elif matched_numbers_length == 4:
                send_message = "4等！！"
            elif matched_numbers_length == 3:
                send_message = "5等！"
            else:
                send_message = "ハズレ"
            
            # 返信
            rep = f"{recent_loto_numbers_str}\n{send_message}"
            LINE_BOT_API.reply_message(
                line_event.reply_token, TextSendMessage(text=rep))
    
        try:
            body = event["body"]
            signature = event["headers"]["x-line-signature"]
            LINE_HANDLER.handle(body, signature)
        except LineBotApiError as e:
            print("Got exception from LINE Messaging API: %s\n" % e.message)
            for m in e.error.details:
                print("  %s: %s" % (m.property, m.message))
            return FAILDED_RES
        except InvalidSignatureError:
            return FAILDED_RES
        return SUCCESS_RES
    ```
  1. lambda関数zipを作成。pip install beautifulsoup4 requests line-bot-sdk -t ./linebot で直接ディレクトリを指定してインストールするとこんな感じになるのでlamda_function.pyもろとも全部圧縮。

1.2.AWSでやること

lambda

  1. コンソールから新規関数を作成(今回はlotNotificationと命名)(python3.8)
  2. 「アップロード元」というプルダウンからローカルで作ったlambda関数zipをアップ
  3. lambda関数の環境変数にトークンと自分のユーザーIDを設定

APIgateway

  1. 新規APIgatewayを作成(REST API)

  2. リソースを作成(lot-notification)し、メソッド(POST)も作成。

  3. メソッドリクエスト設定を編集。今回はLINE Messaging APIと通信するのに必要だが用途によっては必要ない。

  4. 「アクション」→「APIのデプロイ」から作成したAPIをデプロイ。これで↑のURLにPOSTリクエストを送ったらlambdaのlotNotification関数が実行される状態になった。

1.3.LINEでやること

  1. 公式ドキュメントに沿って色々作成
  2. 開発者コンソールから作成したchannelを開く
  3. 「Basic settings」からchannel secret, your user idをコピーしてlambdaの環境変数へ
  4. 「Messaging API」からchannel access token(long-lived)をコピーしてlambdaの環境変数へ
  5. AWS側の最後に得たAPIgatewayのURL(リソース名まで含めたURL)を開発者コンソール「→channel→massagingAPI→webhookURLに設定。

2.定期プッシュ通知型作成

2.1.やること

  1. 基本的に返信型と同じ
  2. 返信型のlambda_function.pyを以下に差し替え
lambda_function.py ソースコード
    import os
    import sys
    import bs4
    import requests
    from linebot import (
        LineBotApi, WebhookHandler
    )
    from linebot.models import (
        MessageEvent, TextMessage, TextSendMessage,
    )
    from linebot.exceptions import (
        LineBotApiError, InvalidSignatureError
    )
    
    
    CHANNEL_ACCESS_TOKEN = os.getenv('LINE_CHANNEL_ACCESS_TOKEN', None)
    LINE_USER_ID = os.getenv('LINE_USER_ID', None)
    LINE_BOT_API = LineBotApi(CHANNEL_ACCESS_TOKEN)
    
    LOT_RESULT_URL = "https://takarakuji.rakuten.co.jp/backnumber/loto6/"
    LOT_MY_NUMBER = ["10","12","1","29","35","42"]  #自動購入してるマイナンバー例 
            
    SUCCESS_RES = {"statusCode": 200, "body": "Success"}
    FAILDED_RES = {"statusCode": 500, "body": "Error"}
    
    
    def lambda_handler(event, context):
    
        # 当選番号をスクレイピングで取得
        res = requests.get(LOT_RESULT_URL)
        soup = bs4.BeautifulSoup(res.text, "html.parser")
        recent_loto_numbers = [s.text for s in soup.find_all(
            'span', class_='loto-font-large')[:7]]
        recent_loto_numbers_str = " ".join(recent_loto_numbers)
        
        # 当選判定
        matched_numbers = set(recent_loto_numbers) & set(LOT_MY_NUMBER)
        matched_numbers_length = len(matched_numbers)
        bonus_matched = recent_loto_numbers[-1] in [f"({x})" for x in LOT_MY_NUMBER]
        if matched_numbers_length == 6:
            send_message = "1等！！！！！"
        elif matched_numbers_length == 5 and bonus_matched:
            send_message = "2等！！！！"
        elif matched_numbers_length == 5:
            send_message = "3等！！！"
        elif matched_numbers_length == 4:
            send_message = "4等！！"
        elif matched_numbers_length == 3:
            send_message = "5等！"
        else:
            send_message = "ハズレ"
        
        # プッシュ送信
        message = f"当選番号：{recent_loto_numbers_str}\nあなたの番号：{" ".join(LOT_MY_NUMBER)}\n{send_message}"
        try: 
            LINE_BOT_API.push_message(
                LINE_USER_ID, TextSendMessage(text=message))
        except LineBotApiError as e:
            print("Got exception from LINE Messaging API: %s\n" % e.message)
            return FAILDED_RES
        return SUCCESS_RES
  1. LambdaのAPIgatewayトリガーをEvent Bridge(cloud watch)トリガーに差し替え。
  1. 楽天のロト6自動購入ページで マイナンバー(固定購入ナンバー)を作成してそのナンバーで自動購入する

3.さいごに

初めて技術記事を外に出したので緊張しました。この記事をきっかけにLambdaやLINEbotに入門する方や6億円をとる方が現れたら幸いです。

ちなみに僕は通知型を採用しているのですが、少しソースコードを改変して5等以上のときはLINE通知が20回鳴るように設定しています。その方が「ただごとじゃない感」を演出できて脳汁が出ると思ったからです。

Discussion