毎日の電気量を通知する仕組みをLambdaで作った

目的

最近電気使用料が高くなってますよね？
サイトにログインして、前日使用した電気量を確認することが日課となっていましたが
面倒くさい

バッチ処理で毎日LINEに通知してくれたら楽だし
そういった仕組みのようなものを作ったこともなかったので
勉強になるかな～と思ってそれっぽいものを作ろうと重い腰を上げました
この記事はその備忘録的な感じです

構成図

仕組みとしては簡単です

LINE NOTIFYの設定

こちらのサイトのやり方を拝借しました。簡単簡単
https://zenn.dev/protoout/articles/18-line-notify-setup

問題なくテスト通知できました
[ ]で囲われているのはトークン名です

Lambda関数の作成

ここが一番時間を食いました。
なぜなら業務で一度も作成したことがないからです。

上の構成図にあるように
スクレイピング用、LINE通知用の二つを作ることにしました
どちらもPythonで作れそうです

まずはスクレイピングのLambdaから。
seleniumというPythonのライブラリを使うとスクレイピングが可能になるということが分かったので
ChatGPT君に相談しながら作成

# seleniumの必要なライブラリをインポート
from selenium import webdriver

# 要素を取得する際に必要
from selenium.webdriver.common.by import By
import time
import datetime
import boto3

# TokyoGasログインページを起動
driver = webdriver.Chrome()
driver.get('https://members.tokyo-gas.co.jp/')
time.sleep(10)

# NAME属性が”identity”であるHTML要素を取得し、ログインID文字列をキーボード送信
driver.find_element(By.NAME,"loginId").send_keys("")
# NAME属性が”password”であるHTML要素を取得し、パスワード文字列をキーボード送信
driver.find_element(By.NAME,"password").send_keys("")
# CLASS属性が”sessions_button--wide”であるHTML要素を取得してクリック
driver.find_element(By.ID,"submit-btn").click()

time.sleep(5)

# 電気使用量のページに移動
driver.get('https://members.tokyo-gas.co.jp/usage?tab=electricity')
time.sleep(5)
driver.find_element(By.ID,"segmented-button-daily").click()

time.sleep(5)

# 昨日の電気使用量を取得
daily = driver.find_element(By.XPATH,'//*[@id="mtg-karte-electricity"]/div/div/div[2]/div/div[1]/div[2]/div[2]/div/div[1]/div').text

# 昨日の電気使用量を出力
print(daily)

# Webドライバー の セッションを終了
driver.quit()

# 今日の日付を取得して年月を取り出す
today = datetime.datetime.now()
year_month = today.strftime("%Y%m")
day = today.strftime("%d")
key = f'{year_month}/{day}.txt'

# S3 への保存
s3 = boto3.client('s3')
s3.put_object(Bucket='tokyo-gas-electricity', Key=key, Body=daily)

ローカルではもんだなく動いたのでLambdaでも念のため動くか確認したところ、、、

なんやこれ？ローカルでは動いたで？

調べたらLambdaに用意されているPython実行環境に
インストールされているライブラリの中にseleniumが入っていないらしい
なるほど。。。。

使用したいライブラリはレイヤーに追加することで使用できることが分かったので早速設定。

なんで？

エラー内容的にchromeを見つけることができてないっぽいですね
いろいろなサイトを見て試してみましたがうまくいかず
Twitter(X)で調べてみると前までは使えたけどRuntimeの更新をしたところ使えなくなったという情報がちらほら

もう少し掘ってみるとこんな記事が
https://tks2.co.jp/2023/12/02/aws-lambda-selenium-python312/

どうやらpython3.12ではseleniumが使えないらしい
しかしこうも書いてある

Amazon ECR のイメージを利用した AWS Lambda が最も適した方法だと考えられます。 Amazon ECR とは Docker コンテナレジストリであり、 Docker のイメージを Amazon ECR リポジトリに配置することができます。

???

あまり深く理解はできなかったがECRのイメージからLambdaをデプロイすることができるらしい

・環境の一貫性
・ランタイムのカスタマイズ
・容量とサイズの制限緩和

などの利点があるらしい。

とりあえずECRやDockerやら初めてのことばかり試行錯誤してみた結果。。。。

動いた！！

そんでもってLine通知用のLambdaは以下のような感じで作成

import json
import boto3
import datetime
import urllib.request

def lambda_handler(event, context):
    # S3クライアントを作成
    s3 = boto3.client('s3')
    
    # LINE Notify

    TOKEN = '作成したアクセストークン'
    api_url = 'https://notify-api.line.me/api/notify'
    
    BUCKET_NAME = 'tokyo-gas-electricity'

    # 前日の日付を取得
    yesterday = datetime.datetime.now() - datetime.timedelta(days=1)
    
    # 年と月と曜日を取得
    year = str(yesterday.year)
    month = str(yesterday.month).zfill(2)
    day = str(yesterday.day).zfill(2)
    
    # 曜日を取得して日本語の曜日名に変換
    weekday_num = yesterday.strftime("%w")
    weekdays_jp = ["日", "月", "火", "水", "木", "金", "土"]
    weekday_jp = weekdays_jp[int(weekday_num)]
    
    object_key = f'{year}/{month}/{day}.txt'

    try:
        response = s3.get_object(Bucket=BUCKET_NAME, Key=object_key)
        file_content = response['Body'].read().decode('utf-8')
        
        # kWhの数値を取得
        kWh_value = float(''.join(filter(str.isdigit, file_content))) / 10  # 数字以外を取り除き、10で除算して kWh に変換
        # コメントを作成
        if kWh_value <= 5:
            comment = "この調子です！\nこのまま節電を意識しましょう！"
        elif kWh_value <= 15:
            comment = "使いすぎではないですが\n節約を意識しましょう！"
        else:
            comment = "使いすぎじゃい！\n気を付けぃ！"
        
        # 送信するメッセージ
        message = f"\n{month}/{day}({weekday_jp})の電気使用量は\n{file_content}です\n\n{comment}"
        
        # LINE Notifyに送信
        send_line_notify(api_url, TOKEN, message)  # api_url を引数として渡す
        
        return {
            'statusCode': 200,
            'body': 'Message sent successfully'
        }
    except Exception as e:
        return {
            'statusCode': 400,
            'body': str(e)
        }

def send_line_notify(api_url, token, message):  # api_url を引数に追加
    request_headers = {
        'Content-Type': 'application/x-www-form-urlencoded',
        'Authorization': 'Bearer' + ' ' + token
    }
    params = {'message': message}
    data = urllib.parse.urlencode(params).encode('ascii')
    req = urllib.request.Request(api_url, headers=request_headers, data=data, method='POST')
    conn = urllib.request.urlopen(req)

動かしてみる

素材はそろったのでEvent_Bridgeの設定を行い動かしてみると、

でけた！！

スクレイピング用のLambdaを作るのにかなり手こずりましたが思っていた通りのものが作れました！
この仕組みに加えて当日までの使用量や、電気料などもわかるようにできればより便利だなと考えた次第です！

以上！