商品入荷情報を定期的にスクレイピングしてSlack通知する(Lambda with serverless framework)

公開:2021/01/31
更新:2021/02/01
7 min読了の目安(約6400字TECH技術記事

欲しいディスプレイが品切れだったので「なんとか再入荷時に購入したい! !」と思い立ち、Python の勉強がてら商品ページをスクレイピングして Slack に流すスクリプトを書きました。そして serverless framework を使って、AWS Lambda の定期実行環境を組んだので、その過程をまとめます。

環境構築

AWS のコンソールでポチポチしたくないので serverless frameworkを使って、Lambda を構築するようにします。

まず serverless framework をインストール。

npm install -g serverless

任意の作業ディレクトリで以下を実行。

sls create \
  --template aws-python3 \
  --name my-scraping-app \
  --path my-scraping-app

my-scraping-app ディレクトリ内に serverless framework 関連のファイルが生成されます。
その後 venv の設定や、serverless framework で AWS にデプロイするための credentials の設定をします(本記事では省略)。
以下 credentials 設定の参考ページです。

https://www.serverless.com/framework/docs/providers/aws/guide/credentials/

スクレイピング & slack通知スクリプトの実装

スクレピングは様々な方法があると思うのですが、今回は該当商品の商品ページに出ている「現在品切れ中」というボタンの有無を確認することで、入荷状況を判断することとします。

依存モジュールを追加して、handler.py にスクレピングコードと Slack 通知コードを書いていきます。

pip install requests beautifulsoup4 slack_sdk
handler.py
import requests
import re
import os
from bs4 import BeautifulSoup
from slack_sdk.webhook import WebhookClient


def scraping(event, context):
    TARGET_URL = "https://www.dell.com/ja-jp/shop/dell-%E3%83%87%E3%82%B8%E3%82%BF%E3%83%AB%E3%83%8F%E3%82%A4%E3%82" \
                 "%A8%E3%83%B3%E3%83%89%E3%82%B7%E3%83%AA%E3%83%BC%E3%82%BA-u4021qw-40%E3%82%A4%E3%83%B3%E3%83%81%E3" \
                 "%83%AF%E3%82%A4%E3%83%89%E6%9B%B2%E9%9D%A2usb-c-hub-%E3%83%A2%E3%83%8B%E3%82%BF/apd/210-aypy/%E3%83" \
                 "%A2%E3%83%8B%E3%82%BF%E3%83%BC-%E3%83%A2%E3%83%8B%E3%82%BF%E3%83%BC%E3%82%A2%E3%82%AF%E3%82%BB%E3" \
                 "%82%B5%E3%83%AA%E3%83%BC"
    html = requests.get(TARGET_URL)
    soup = BeautifulSoup(html.content, "html.parser")

    search = re.compile('.*現在品切れ中.*')
    find_result = soup.find("div", string=search)

    if find_result is None:
        _send_slack(TARGET_URL)
    else:
        print('まだ品切れ中😭')

    return {"statusCode": 200}


def _send_slack(target_url):
    url = os.environ['SLACK_WEBHOOK_URL']
    webhook = WebhookClient(url)
    res = webhook.send(
        text="探しものが見つかったよ🔎",
        blocks=[{
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": "探しものが見つかったよ🔎"
            },
            "accessory": {
                "type": "button",
                "text": {
                    "type": "plain_text",
                    "text": "サイトを開く",
                    "emoji": True
                },
                "url": target_url
            }
        }],
    )
    print(res)

現在品切れ中ボタンの判定は以下コードです。
BeautifulSoup にて正規表現で対象文字列を検索しています。そして、文字列がない場合のみ Slack 通知のメソッドを実行しています。

2021/01/31 20:40 追記
この部分は BeautifulSoup を使わず、request 単体でレスポンスに対して文字列マッチしたほうが低コストだとアドバイス頂きました。
確かにそうのとおりなので後で書き直します。

    search = re.compile('.*現在品切れ中.*')
    find_result = soup.find("div", string=search)

    if find_result is None:
        _send_slack(TARGET_URL)
    else:
        print('まだ品切れ中😭')

Slack 通知部分は以下のコードです。
環境変数から Incoming Webhook の URL を取得して、その URL に対して SDK 経由でメッセージを送信しています。

    url = os.environ['SLACK_WEBHOOK_URL']
    webhook = WebhookClient(url)
    res = webhook.send(
        text="探しものが見つかったよ🔎",
        blocks=[{
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": "探しものが見つかったよ🔎"
            },
            "accessory": {
                "type": "button",
                "text": {
                    "type": "plain_text",
                    "text": "サイトを開く",
                    "emoji": True
                },
                "url": target_url
            }
        }],
    )

serverless frameworkでデプロイ

serverless framework の設定をして、スクリプトを Lambda にデプロイします。

外部モジュールのデプロイ設定

まず、lambda で Python の外部モジュールを使うために serverless framework のプラグインserverless-python-requirementsを追加します。
以下コマンド実行すれば OK です。これでpackage.jsonの追加と、serverless.ymlのプラグイン設定の追記が完了します。

sls plugin install -n serverless-python-requirements

その後、依存モジュールの情報を requirements.txt にまとめます。

pip freeze > requirements.txt

※ こちらの作業の詳細は以下記事にまとめました。

https://zenn.dev/ryo_kawamata/articles/python-exclude-package-on-serverless-framework

環境変数の設定

続いて Incoming Webhook の URL で参照している環境変数の設定をします。
Git 管理しない.envファイルで値を保持したいので、serverless-dotenv-plugin を使います。

以下コマンドでインストールします。

sls plugin install -n serverless-dotenv-plugin

.envファイルを作成し、Slack の Incoming Webhook URL を設定します。

SLACK_WEBHOOK_URL=xxxxxxxxxxxxxxxxxxxxxxxx

URL の取得方法・設定方法はこちらをご確認ください。

https://slack.com/intl/ja-jp/help/articles/115005265063-Slack-での-Incoming-Webhook-の利用

handlerの指定、Lambdaの起動イベントの設定

商品入荷情報をいちはやく知るためには定期的な実行が必要です。
今回は CloudWatch Events を使って、Lambda の定期実行を実現します。

serverless framework では events に schedule を指定するだけでその環境が構築されます。
今回は以下のように指定しました。

7 時から 21 時の間、1 時間に 1 回スクリプトが定期実行されます。
※ reasion をap-northeast-1にするのも忘れずに。

serverless.yml
provider:
  name: aws
  runtime: python3.8
  lambdaHashingVersion: 20201221
  region: ap-northeast-1

functions:
  scraping:
    handler: handler.scraping
    events:
      - schedule: cron(0 7-21 * * ? *)

デプロイ

最終的な serverless.yml はこちらです。

serverless.yml
service: my-scraping-app
frameworkVersion: '2'

provider:
  name: aws
  runtime: python3.8
  lambdaHashingVersion: 20201221
  region: ap-northeast-1

functions:
  scraping:
    handler: handler.scraping
    events:
      - schedule: cron(0 7-21 * * ? *)

plugins:
  - serverless-python-requirements
  - serverless-dotenv-plugin

最後にデプロイコマンドを実行して完了です。

sls deploy

AWS にリソースが作成されます。

以下テスト実行の結果です。もし商品が入荷されれば Slack に通知がきます。

終わりに

これで、いちはやくディスプレイをゲットできる(はず。..)💪

参考