AWS Lambdaで列車運行情報を定期的にLINEへ通知してみた【Python】

背景

近年、電車の運行情報は各事業者が、メール通知サービスや公式twitterアカウントなどで発信していたり、公式アプリでも情報がリアルタイムに見れたりしますね。
しかし、twitterや公式アプリは利用者側が情報を見に行く手間がかかります。メール通知サービスは、メールが多い人はすぐにメールボックス内に埋もれてしまい、情報が取り出しにくいというデメリットも潜在します。
そこで、LINEで定期的にプッシュ通知できるようにしてみました。

電車の運行情報だけでなく、他のことにも応用できると思いましたため、ここにまとめます。

先人たちの知恵をお借りするなどして解決できたことを、この場をお借りして感謝するとともに、大変恐縮ですが自分のメモとして、こちらへまとめておきます。

◆◆◆◆◆◆◆◆◆◆◆

特に参考となった記事：
また、執筆者の @nsuhara様 には何度も質問させていただき、都度ご丁寧に回答いただきましたことを、大変感謝しております。この場をお借りして御礼申し上げます。

• AWS-Lambda + Python + CronでWEBスクレイピングを定期的に実行する　(@nsuhara様)

◆◆◆◆◆◆◆◆◆◆◆

はじめに

Windows環境での手順ですが、Mac環境でも同様かと思います。環境に関する部分は、必要に応じて読み替えてくださいませ。

1. 目的

この記事をお読みいただくと、Python によるスクレイピング情報を、AWS Lambda を用いたサーバレス環境で定期的に LINE へ通知することができるようになります。幅広い応用が可能です。

2. AWS Lambda とは

AWS Lambda がサポートするいずれかの言語でコードを指定するだけで、サーバーのプロビジョニングや管理の必要なしにコードを実行できるコンピューティングサービスです。
必要時にのみコードを実行し、1 日あたり数個のリクエストから 1 秒あたり数千のリクエストまで自動的にスケーリングします。
コンピューティング時間に対してのみ課金され、コードが実行中でなければ料金はかかりません。

詳細は、公式サイトの紹介ページをご参照ください。

• AWS Lambda とは
  • 特徴
  • 料金

3. 開発

概略

3-1. Pythonコードを作成する

3-2. Lambdaへアップロードするためのzipファイルを作成する

3-3. Lambda関数を作成する

3-4. Lambda関数へzipファイルをアップロードする

3-5. Lambda関数の環境変数を設定する

3-6. 定期的に実行するためスケジューリングする（cron）

3-7. ロギングを設定する

3-1. Pythonコードを作成する

実際の実装内容やソースコードを見ると理解が深まると思います。公開しておりますため、是非ご活用ください。

• GitHub

※本来はスクレイピングではなく、WEB APIが利用可能ならばそれを利用すべきです。
　　⇒ 2021/02/02現在、東京メトロオープンデータ開発者サイトにてAPIの利用を申請中。

実行環境

(本番)

• Aamazon Web Services
• Lambda
• Event Bridge (旧 Cloud Watch Event)
• Simple Notification Service
• Cloud Trail
• Python 3.7
• LINE Notify

(開発)

• Python 3.9.0, 3.8.5

ソースコード（全体）

• クローリングの対象は、東京メトロの全9路線としています。

app/lambda_function.py

"""
lambda_function.py
"""
# postリクエストをline notify APIに送るためにrequestsのimport
import os
import time
import requests
import shutil
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from datetime import datetime, timezone
import pytz

url_list = [
    "https://www.tokyometro.jp/unkou/history/ginza.html",
    "https://www.tokyometro.jp/unkou/history/marunouchi.html",
    "https://www.tokyometro.jp/unkou/history/hibiya.html",
    "https://www.tokyometro.jp/unkou/history/touzai.html",
    "https://www.tokyometro.jp/unkou/history/chiyoda.html",
    "https://www.tokyometro.jp/unkou/history/yurakucho.html",
    "https://www.tokyometro.jp/unkou/history/hanzoumon.html",
    "https://www.tokyometro.jp/unkou/history/nanboku.html",
    "https://www.tokyometro.jp/unkou/history/fukutoshin.html"
]

# line notify APIのトークン
line_notify_token = os.getenv("LINE_NOTIFY_TOKEN")
# line notify APIのエンドポイントの設定
line_notify_api = 'https://notify-api.line.me/api/notify'


def move_bin(
    fname: str, src_dir: str = "/var/task/bin", dest_dir: str = "/tmp/bin"
) -> None:
    if not os.path.exists(dest_dir):
        os.makedirs(dest_dir)
    dest_file = os.path.join(dest_dir, fname)
    shutil.copy2(os.path.join(src_dir, fname), dest_file)
    os.chmod(dest_file, 0o775)


def create_driver(
    options: webdriver.chrome.options.Options,
) -> webdriver.chrome.webdriver:
    driver = webdriver.Chrome(
        executable_path="/tmp/bin/chromedriver", chrome_options=options
    )
    return driver


def lambda_handler(event, context):
    """
    lambda_handler
    """
    print('event: {}'.format(event))
    print('context: {}'.format(context))

    move_bin("headless-chromium")
    move_bin("chromedriver")

    #headless_chromium = os.getenv('HEADLESS_CHROMIUM', '')
    #chromedriver = os.getenv('CHROMEDRIVER', '')
    # webdriverの設定
    options = Options()
    #options.binary_location = headless_chromium
    options.binary_location = "/tmp/bin/headless-chromium"
    options.add_argument('--headless')
    options.add_argument('--no-sandbox')
    options.add_argument('--single-process')
    options.add_argument('--disable-dev-shm-usage')
    options.add_argument("--disable-gpu")
    options.add_argument("--window-size=1280x1696")
    options.add_argument("--disable-application-cache")
    options.add_argument("--disable-infobars")
    options.add_argument("--hide-scrollbars")
    options.add_argument("--enable-logging")
    options.add_argument("--log-level=0")
    options.add_argument("--ignore-certificate-errors")
    options.add_argument("--homedir=/tmp")

    driver = create_driver(options)
    # driver = webdriver.Chrome(executable_path=chromedriver, options=options)

    # 現在時刻
    now = datetime.now(tz=timezone.utc)
    tokyo = pytz.timezone('Asia/Tokyo')
    # 東京のローカル時間に変換
    jst_now = tokyo.normalize(now.astimezone(tokyo))
    content0 = jst_now.strftime("%m月%d日 %H:%M現在")

    info_list = []
    for url in url_list:
        content = []
        driver.get(url)
        time.sleep(1)
        # 路線名
        content1 = driver.find_element_by_css_selector("#v2_contents > div.v2_contents > div > div.v2_headingH1.v2_headingRoute > h1")
        # 運行状況（概要）
        content2 = driver.find_element_by_css_selector("#v2_contents > div.v2_contents > div > div.v2_gridC.v2_section.v2_clear.v2_stationUnkouMap.v3_stationUnkouMap > div.v2_sectionS.v2_gridCRow > div.v2_unkouReportInfo > div > div > div.v2_unkouReportTxtCaption.v3_unkouReportTxtCaption > p")
        # 運行状況（詳細）
        content3 = driver.find_elements_by_css_selector("#v2_contents > div.v2_contents > div > div.v2_gridC.v2_section.v2_clear.v2_stationUnkouMap.v3_stationUnkouMap > div.v2_sectionS.v2_gridCRow > div.v2_unkouReportInfo > div > p")

        # lineに通知するメッセージを組み立て
        content.append("●" + content1.text[:-5])
        content.append(content2.text)
        for content3s in content3:
            content.append(content3s.text)

        info_list.append(content)

    content_text = []
    for i in range(9):
        content_text.append('\n'.join(info_list[i]))

    notification_message = content0 +'\n' + '\n\n'.join(content_text)

    driver.close()
    driver.quit()

    # ヘッダーの指定
    headers = {'Authorization': f'Bearer {line_notify_token}'}
    # 送信するデータの指定
    data = {'message': f'{notification_message}'}
    # line notify apiにpostリクエストを送る
    requests.post(line_notify_api, headers=headers, data=data)

    return {
        'status_code': 200
    }

if __name__ == "__main__":
    print(lambda_handler(event=None, context=None))

chromedriver-binary-auto==0.1
requests==2.25.1
selenium==3.141.0
chardet==3.0.4
pytz==2021.1

解説：

3-1-1. Selenium : optionsの設定

• クローリングは Selenium で。Lambda での動作には options の設定が必要です。

from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By

    # webdriverの設定
    options = Options()
    #options.binary_location = headless_chromium
    options.binary_location = "/tmp/bin/headless-chromium"
    options.add_argument('--headless')
    options.add_argument('--no-sandbox')
    options.add_argument('--single-process')
    options.add_argument('--disable-dev-shm-usage')
    options.add_argument("--disable-gpu")
    options.add_argument("--window-size=1280x1696")
    options.add_argument("--disable-application-cache")
    options.add_argument("--disable-infobars")
    options.add_argument("--hide-scrollbars")
    options.add_argument("--enable-logging")
    options.add_argument("--log-level=0")
    options.add_argument("--ignore-certificate-errors")
    options.add_argument("--homedir=/tmp")

3-1-2. Headless Chrome の配置先

• Headless Chrome の配置先で "Pathが通っていない！" と弾かれることから調べたところ、配置先を /tmp/bin とする必要があることが分かり対処しました。コードの該当箇所は下記の通りです。

Headless Chrome の配置先には注意が必要である。Lambda requirements for container imagesに以下の記述がある。

The container image must be able to run on a read-only file system. Your function code can access a writable /tmp directory with 512 MB of storage. If you are using an image that requires a writable directory outside of /tmp, configure it to write to a directory under the /tmp directory.

Lambda 関数から書き込み可能な領域は /tmp (512 MB) のみで, /var/task や /opt への配置を試したところ Headless Chrome の起動に失敗した。このエラーは [2] でも報告されており, [2] を参考に Lambda 関数実行時に Headless Chrome を /tmp/bin に移動する move_bin() を追加した。

import shutil

def move_bin(
    fname: str, src_dir: str = "/var/task/bin", dest_dir: str = "/tmp/bin"
) -> None:
    if not os.path.exists(dest_dir):
        os.makedirs(dest_dir)
    dest_file = os.path.join(dest_dir, fname)
    shutil.copy2(os.path.join(src_dir, fname), dest_file)
    os.chmod(dest_file, 0o775)


def create_driver(
    options: webdriver.chrome.options.Options,
) -> webdriver.chrome.webdriver:
    driver = webdriver.Chrome(
        executable_path="/tmp/bin/chromedriver", chrome_options=options
    )
    return driver

def lambda_handler(event, context):
    ....
    move_bin("headless-chromium")
    move_bin("chromedriver")

3-1-3. タイムゾーンの変換

AWS Lambda の実行環境におけるタイムゾーンは UTC（協定世界時）のため、出力時には JST（日本標準時）のタイムゾーンで出力するよう対処しました。
Python でのタイムゾーン変換を pytz で行ないます。コードの該当箇所は下記の通りです。

参照：pytz - PyPI

pytz は必要となるすべてのタイムゾーンの完全なデータベースを含んでいるため、任意のタイムゾーンに変換することが可能

import pytz

    # 現在時刻
    now = datetime.now(tz=timezone.utc)
    tokyo = pytz.timezone('Asia/Tokyo')
    # 東京のローカル時間に変換
    jst_now = tokyo.normalize(now.astimezone(tokyo))
    content0 = jst_now.strftime("%m月%d日 %H:%M現在")

3-1-4. LINEへの通知

• 事前に トークンと、API key の取得が必要です。
  • LINE Notify へアクセスし、ログインします。
  • ログイン後、マイページへ遷移。
  • ページ下部にある「トークンを発行する」をクリック。
    ・トークン名：　任意　※通知の際に表示されます
    ・通知を送信するトークルーム：　1:1でLINE Notifyから通知を受け取る を選択　※LINEグループも選択可能
    ・「発行する」をクリック
    発行されたトークンとAPI keyは大切に保管してください。
• ソースの該当箇所は以下の通りです。requets で post します。

import requests


# line notify APIのトークン
line_notify_token = os.getenv("LINE_NOTIFY_TOKEN")
# line notify APIのエンドポイントの設定
line_notify_api = 'https://notify-api.line.me/api/notify'


    # ヘッダーの指定
    headers = {'Authorization': f'Bearer {line_notify_token}'}
    # 送信するデータの指定
    data = {'message': f'{notification_message}'}
    # line notify apiにpostリクエストを送る
    requests.post(line_notify_api, headers=headers, data=data)

3-2. Lambdaへアップロードするためのzipファイルを作成する

スクリプト作成

• 環境に合わせて chromedriver , headless-chromium のバージョン／パスを変更してください。
  　2021/02/08現在、Lambda上のpython3.7では下記の設定で稼働させることができます。

rm upload.zip
rm -r upload/
rm -r download/

mkdir -p download/bin
curl -L https://chromedriver.storage.googleapis.com/2.41/chromedriver_linux64.zip -o download/chromedriver.zip
curl -L https://github.com/adieuadieu/serverless-chrome/releases/download/v1.0.0-55/stable-headless-chromium-amazonlinux-2017-03.zip -o download/headless-chromium.zip
unzip download/chromedriver.zip -d download/bin
unzip download/headless-chromium.zip -d download/bin

mkdir upload
cp -r download/bin upload/bin
cp app/lambda_function.py upload/
pip install -r app/requirements.txt -t upload/
cd upload/
zip -r ../upload.zip --exclude=__pycache__/* .
cd ../

rm -r upload/
rm -r download/

upload.zip 作成

sh make_upload.sh

3-3. Lambda関数を作成する

AWS マネジメントコンソールから、以下の手順を実行します。

• サービスメニューの Lambda を選択
  • サブメニューの「関数」を選択して「関数の作成」をクリック
  • 「一から作成」を選択
  • "関数名"、"ランタイム"、"実行ロール" を入力して関数の作成をクリック

3-4. Lambda関数へzipファイルをアップロードする

• zipファイルをアップロードします。

  • Lambda関数の関数コードセクションを表示する。

  • 10MiB未満は直接アップロードが可能です。
    ・Lambda関数の関数コードセクションを表示
    ・「.zipファイルをアップロード」を選択
    ・アップロードから upload.zip を選択して「保存」をクリック

  • 10MiB以上は S3 (Simple Storage Service) バケット経由でのアップロードとなります。事前にバケットの作成が必要。
    ・S3 へ upload.zipファイルをアップロード
    ・Lambda関数の関数コードセクションを表示
    ・「Amazon S3 からファイルをアップロードする」を選択
    ・S3 のリンクURL を入力して「保存」をクリック

3-5. Lambda関数の環境変数を設定する

環境変数設定

• Lambda関数の環境変数セクションを表示する
  • 下記設定を行なう

キー	値	備考
LINE_NOTIFY_TOKEN	********************************* (43文字)
CHROMEDRIVER	/var/task/bin/chromedriver	* 3-1-2. Headless Chromeの配置先変更対処で設定不要になります
HEADLESS_CHROMIUM	/var/task/bin/headless-chromium	* 同上

3-6. 定期的に実行するためスケジューリングする（cron）

• トリガーに Event Bridge (Cloud Watch Events) を設定
  • Lambda関数の Designerセクション から「トリガーを追加」をクリック
  • トリガーの設定：　Event Bridge (Cloud Watch Events) を選択
  • 「新規ルールの作成」を選択
  • ルール名：　任意のルール名を入力
  • ルールタイプ：　「スケジュール式」を選択する（cronまたはrate式）
  • スケジュール式：　cron(0/30 0-13,21-23 * * ? *) を入力し、「追加」をクリック
    ・下記例は、JST（日本標準時）06:00～22:00の間、毎時00分と30分に稼働します。
    ・cronは、UTC（協定世界時）での指定となるため、▲9時間で記載しています。

※cron式の設定については、【Linux】【AWS Lambda】Cron形式の設定マニュアル をご参照。

3-7. ロギングを設定する

• 送信先に Simple Notification Service を設定
  • Lambda関数の Designerセクション から「トリガーを追加」をクリック
  • ソース：　非同期呼び出しを選択
  • 条件：
    ・失効時
    ・正常
    いずれにもログ出力を以下の通り設定
  • 送信先タイプ：　SNSトピックを選択
  • 送信先：　"cloudtrail-logs" を選択

4. 実行

• 実行結果：

---

参考

• AWS-Lambda + Python + CronでWEBスクレイピングを定期的に実行する　(@nsuhara様)

⇒ ご質問にも快く回答いただきました。誠にありがとうございました。🙇

• 【AWS】lambdaファンクションを定期的に実行する　(@Toshinori_Hayashi様)

• AWS Lambda上のheadless chromeをPythonで動かす　(@nabehide様)

• AWS Lambdaでpythonプログラムの定期実行　(@hiiii)様

• AWS LambdaでPythonスクレイピングを定期実行させる　(@eisu26)様

• 【Python】AWS Lambda の Container Image Support を使い Selenium を動かす　(FiS Project - t2sy:@fisproject)様

• AWS Lambda でのタイムゾーン変換　(Serverworks ENGINEER BLOG)

---

(編集後記)

（お願い）APIを利用できる情報提供者からは、必ずAPIを使いましょう。

天気予報、株価の設定値到達アラート、服薬の飲み忘れ防止通知、等々に応用が利くことですので、みなさんのアイディアで様々な利用方法を検討いただけたら幸いです。
今回は LINE Notify を利用しましたが、Slack などのチャットサービスへの通知もそれに関するロジックを変えるのみで実現可能です。
私も他のいろいろなサービスと連携することを考えております。