電車の運行情報（遅延・運転見合・運休など）を毎朝Slackに通知してみた

はじめに

サーバーレス開発部の藤井元貴です。

駅に到着してから電車の運行情報を知ると、わりと焦りますよね。 遅延ならまだしも、一時見合わせや運休の場合は「どうしよう……」となります。

家を出る前に電車の運行情報を知ることができれば、少し早めに家を出たり、別のルートを使ったり、効率よく行動できます。

クラスメソッドでは、無理な出社をせず、リモートワークに切り替えることもできます。 （お客様の来訪など、仕事状況で変わります）

そこで、普段使用している電車について、何らかの運行情報があればSlackに通知する仕組みをサーバーレスで作ってみました！ （正常な場合は通知しない）

ここではSlackに通知していますが、スマートスピーカーに発声させると、より気づきやすいと思います。

なお、本ブログの半分は下記と同じです。

• AWSサービス毎の請求額を毎日Slackに通知してみた

おすすめの方

• 電車の運行情報（遅延など）を毎朝知りたい
• AWS SAMでLambdaの環境変数を使いたい
• サーバーレスに興味がある

電車の運行情報

電車の運行情報については、下記のサイトを使用させていただきました。 使用にあたっては、下記サイトの「お約束」をご確認ください。

• 鉄道遅延情報のjson

全体概要

「毎日、指定時刻になるとLambdaを起動し、Lambdaが電車の運行情報を取得＆SlackにPostする」というサーバーレスな構成です。 デプロイ等にAWS SAMを使用します。

環境

項目	バージョン
macOS	High Sierra 10.13.6
AWS CLI	aws-cli/1.16.89 Python/3.6.1 Darwin/17.7.0 botocore/1.12.79
AWS SAM CLI	0.10.0
Python	3.6

事前準備

下記の設定を行います。

• Slackの設定

Slackの設定

チャンネルの作成

通知先のチャンネルを作成します。ここでは、チャンネル名を#train-delayとしています。

Incoming Webhookの追加

Incoming Webhookの設定を行います。

通知先チャンネルから「アプリを追加する」を選択します。

アプリとしてIncoming Webhookを検索します。

「設定を追加」を選択します。初回であれば画面は違うかもしれません。

投稿先のチャンネルを選択し、「incomming Webhookインテグレーションの追加」を選択します。

作成されたWebhook URLをメモしておきます。このURLに対して、特定フォーマットでPOSTすれば、Slackのチャンネルに投稿されます。

投稿時のアイコンなども設定できるので、必要に応じて設定します。

やってみる

プロジェクトフォルダの作成

AWS SAMでプロジェクトフォルダを作成します。

sam init --runtime python3.6 --name TrainDelay

templateファイル

AWS SAMのtemplate.yamlは下記です。

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Notify Slack Train Delay

Globals:
  Function:
    Timeout: 10

Parameters:
  SlackWebhookUrl:
    Type: String
    Default: hoge

Resources:

  HelloWorldFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: hello_world/
      Handler: app.lambda_handler
      Runtime: python3.6
      Environment:
        Variables:
          # このURLはコミット＆公開したくないため、デプロイ時にコマンドで設定する
          SLACK_WEBHOOK_URL: !Ref SlackWebhookUrl
      Events:
        NotifySlack:
          Type: Schedule
          Properties:
            Schedule: cron(0 23 * * ? *) # 日本時間AM8時に毎日通知する

Outputs:

  HelloWorldFunction:
    Description: "Hello World Lambda Function ARN"
    Value: !GetAtt HelloWorldFunction.Arn

  HelloWorldFunctionIamRole:
    Description: "Implicit IAM Role created for Hello World function"
    Value: !GetAtt HelloWorldFunctionRole.Arn

Lambda関数の作成

Lambda関数のコードは下記です。

import os
import json
import requests


# ここを任意に変更してください。
CHECK_LIST = [
    {
        'name': '常磐線各駅停車',
        'company': 'JR東日本',
        'website': 'https://traininfo.jreast.co.jp/train_info/kanto.aspx'
    },
    {
        'name': '東西線',
        'company': '東京メトロ',
        'website': 'https://www.tokyometro.jp/unkou/history/touzai.html'
    },
    {
        'name': '京急線',
        'company': '京急電鉄',
        'website': 'https://unkou.keikyu.co.jp/?from=top'
    },
]

JSON_ADDR = 'https://rti-giken.jp/fhc/api/train_tetsudo/delay.json'

SLACK_WEBHOOK_URL = os.environ['SLACK_WEBHOOK_URL']


def lambda_handler(event, context):

    notify_delays = get_notify_delays()

    if not notify_delays:
        # 遅延が無ければ通知しない
        return

    # Slack用のメッセージを作成して投げる
    (title, detail) = get_message(notify_delays)
    post_slack(title, detail)

    return


def get_notify_delays():

    current_delays = get_current_delays()

    notify_delays = []

    for delay_item in current_delays:
        for check_item in CHECK_LIST:
            if delay_item['name'] == check_item['name'] and delay_item['company'] == check_item['company']:
                notify_delays.append(check_item)

    return notify_delays


def get_current_delays():
    try:
        res = requests.get(JSON_ADDR)
    except requests.RequestException as e:
        print(e)
        raise e

    if res.status_code == 200:
        return json.loads(res.text)
    return []


def get_message(delays):
    title = "電車の遅延があります。"

    details = []

    for item in delays:
        company = item['company']
        name = item['name']
        website = item['website']
        details.append(f'・{company}： {name}： <{website}|こちら>')

    return title, '\n'.join(details)


def post_slack(title, detail):
    # https://api.slack.com/incoming-webhooks
    # https://api.slack.com/docs/message-formatting
    # https://api.slack.com/docs/messages/builder
    payload = {
        'attachments': [
            {
                'color': '#36a64f',
                'pretext': title,
                'text': detail
            }
        ]
    }

    # http://requests-docs-ja.readthedocs.io/en/latest/user/quickstart/
    try:
        response = requests.post(SLACK_WEBHOOK_URL, data=json.dumps(payload))
    except requests.exceptions.RequestException as e:
        print(e)
    else:
        print(response.status_code)

環境変数

SlackのPOST先のURLをコードに直接記載したくないため、Lambdaの環境変数を利用します。

SLACK_WEBHOOK_URL = os.environ['SLACK_WEBHOOK_URL']

Lambdaの環境変数を設定するためには、template.yamlで下記を記載します。

      Environment:
        Variables:
          # このURLはコミット＆公開したくないため、デプロイ時にコマンドで設定する
          SLACK_WEBHOOK_URL: !Ref SlackWebhookUrl

上記に直接記載しても良いのですが、さらに変数化させています。 template.yamlで下記のようにパラメータの設定を行い、デプロイ時にこの値を上書きしています（後述）。

Parameters:
  SlackWebhookUrl:
    Type: String
    Default: hoge

S3バケットの作成

コード等を格納するためのS3バケットを作成します。作成済みの場合は飛ばします。

aws s3 mb s3://gnk263-lambda-bucket

ビルド

下記コマンドでビルドします。

sam build

package

続いてコード一式をS3バケットにアップロードします。

sam package \
    --output-template-file packaged.yaml \
    --s3-bucket gnk263-lambda-bucket

deploy

最後にデプロイします。template.yamlの環境変数をオーバーライドし、ここでSlackのWebhook URLを設定します。

sam deploy \
    --template-file packaged.yaml \
    --stack-name NotifyTrainDelayToSlack \
    --capabilities CAPABILITY_IAM \
    --parameter-overrides SlackWebhookUrl=https://hooks.slack.com/services/xxxxxxxxxxxxx

あとは時間になればSlackに通知されます。

すぐに試したい場合は、ブラウザでAWSにログインしてLambdaを手動テストするか、AWS SAMでLambda関数をローカル実行すればOKです。 （Lambda関数をローカル実行する場合は、実行時に環境変数でWebhook URLを指定します）

日本時間　AM8時

無事に通知が来ました！

さいごに

以前からやりたいと思っていましたが、今週月曜日の電車遅延をきっかけに試してみました。 どちらかと言うと、役立たない（通知がこない）ほうが嬉しいです。