欲しいディスプレイが品切れだったので「なんとか再入荷時に購入したい！ !」と思い立ち、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 設定の参考ページです。

スクレイピング & 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 通知のメソッドを実行しています。

    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

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

環境変数の設定

続いて 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 の取得方法・設定方法はこちらをご確認ください。

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 に通知がきます。

終わりに

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

参考

Discussion