こんにちは、ゆーるるです。
我が家の家計簿は、デビット連携させたKyash・クレジットカードを使って、マネーフォワードで完全に自動化しています（現金支払いのみは除く）。
毎月の予算もマネーフォワードの機能で立てて管理しているので、マネフォさえ見に行けば使いすぎることはないのですが、どうしても途中からアプリを開くことすら面倒になってしまって使いすぎることが多々あり。。
毎日Slackに自動投稿すれば、節約意識を高められそうだなと思って、勉強も兼ねてPython、selenium、AWS Lambdaを使って実現してみました。
コードの解説は、スクリプト内にコメントを入れているので、デプロイ周りの解説をメインにまとめました。
Slack投稿イメージ
最終的な完成図はこちら
メイン口座の当日の残高と、今日使える食費、今日使える変動費を、毎朝7時にチェックするように自動実行しています。
合わせて、予算ページのグラフのキャプチャも取得しています。
これを始めたおかげか、食費は年間の平均額から30%削減達成しました！
スクリプトファイル
処理部分の全コードはこちらです。
# handler.py
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.select import Select
import time
import os
import requests
import re
os.environ["HOME"] = "/var/task"
SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
CHANNEL_ID = os.environ["CHANNEL_ID"]
LOGIN_ID = os.environ["LOGIN_ID"]
LOGIN_PASSWORD = os.environ["LOGIN_PASSWORD"]
UA = os.environ["UA"]
ACOUNT_SBI1_NAME = os.environ["ACOUNT_SBI1_NAME"]
ACOUNT_SBI2_NAME = os.environ["ACOUNT_SBI2_NAME"]
ACOUNT_BUSINESS_NAME = os.environ["ACOUNT_BUSINESS_NAME"]
SBI1_NAME = os.environ["SBI1_NAME"]
SBI2_NAME = os.environ["SBI2_NAME"]
BUSINESS_NAME = os.environ["BUSINESS_NAME"]
def main(event, context):
# driverをセット
driver = set_driver()
# 口座一覧ページを開く
driver.get("https://moneyforward.com/accounts")
# 正しいURLが表示されてるか確認
assert "moneyforward.com" in driver.current_url
# マネーフォワードにログイン
login(driver)
# 口座一覧を一括更新する
accounts_update = WebDriverWait(driver, 60).until(EC.visibility_of_element_located((By.CLASS_NAME, "btn-warning")))
accounts_update.click()
# ステータスが「更新中」に変わるのを待つ
time.sleep(3)
# ステータス「正常」が表示されるまで待つ
status = WebDriverWait(driver, 300).until(EC.visibility_of_element_located((By.ID, "js-status-sentence-span-cNHmiFwd2QoSX5MiHCFs_w")))
assert status.text == "正常"
# スクレイピングした口座残高データ等を変数に格納
acount_remaining_list = acount_table_scraping(driver)
# 予算ページを表示
driver.get("https://moneyforward.com/spending_summaries")
# 念の為URL確認
assert "spending_summaries" in driver.current_url
# Select要素を取得
groups = get_select("group_id_hash", driver)
# プライベートグループじゃなかったらグループを変更する
if groups.first_selected_option.text != "プライベートの収支":
groups.select_by_visible_text("プライベートの収支")
time.sleep(1)
# 集計期間をスクレイピング
period = (
WebDriverWait(driver, 60)
.until(EC.visibility_of_element_located((By.CSS_SELECTOR, "#budgets-progress > div > section > div > div > div")))
.text
)
# 残り日数をスクレイピングして数値だけに変換
days_left = driver.find_element(
By.CSS_SELECTOR, "#budgets-progress > div > section > table > thead > tr.budget_sub_header > th:nth-child(2) > div > div"
).text
days_left_int = int(re.sub(r"\D", "", days_left))
# 今日の残高に計算した結果を変数に格納
food_remaining_per_day = calc_remaining_per_day(
"#budgets-progress > div > section > table > tbody > tr:nth-child(6) > td.remaining", days_left_int, driver
)
total_remaining_per_day = calc_remaining_per_day(
"#budgets-progress > div > section > table > tbody > tr.budget_type_total_expense.variable_type > td.remaining", days_left_int, driver
)
# 今月のトータル残高を取得
total_remaining = driver.find_element(
By.CSS_SELECTOR, "#budgets-progress > div > section > table > tbody > tr.budget_type_total_expense.variable_type > td.remaining"
).text
# slackに送るテキストを整形
balance = f"＊——プライベートの残高——＊\n今日使える食費：{food_remaining_per_day}\n今日使える総額：{total_remaining_per_day}\n今月の予算残高総額：{total_remaining}"
# グラフのスクショを撮る
png = screenshot(driver)
# Select要素を再度取得（ページ更新でリセットされるため）
groups = get_select("group_id_hash", driver)
# 業務委託にグループを変更する
groups.select_by_visible_text("業務委託の収支")
time.sleep(1)
# 今月のトータル残高を取得
business_total_remaining = driver.find_element(
By.CSS_SELECTOR, "#budgets-progress > div > section > table > tbody > tr.budget_type_total_expense.variable_type > td.remaining"
).text
# slackに送るテキストを整形
business_balance = f"＊——事業用の残高——＊\n今月の予算残高総額：{business_total_remaining}\n\n集計期間：{period}"
# slackに送信
send_message(f"{acount_remaining_list}\n\n{balance}\n\n{business_balance}")
upload_img(png)
# Select要素を再度取得（ページ更新でリセットされるため）
groups = get_select("group_id_hash", driver)
# アプリから確認のときにプライベートが開いてて欲しいので選択状態にしておく
groups.select_by_visible_text("プライベートの収支")
time.sleep(1)
driver.quit()
def set_driver():
options = webdriver.ChromeOptions()
options.binary_location = "/opt/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(f"--user-agent={UA}")
# Headless Chromeをreturnする
return webdriver.Chrome(executable_path="/opt/chromedriver", chrome_options=options)
def login(driver):
"""[summary]
マネーフォワードにログインする処理
"""
# メールアドレスでログインをクリック
login_mail_address = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.LINK_TEXT, "メールアドレスでログイン")))
login_mail_address.send_keys(Keys.RETURN)
# ログインIDを入力
login_id = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.NAME, "mfid_user[email]")))
assert "email" in driver.current_url
login_id.send_keys(LOGIN_ID, Keys.RETURN)
# パスワードを入力
password = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.NAME, "mfid_user[password]")))
assert "password" in driver.current_url
password.send_keys(LOGIN_PASSWORD, Keys.RETURN)
def get_select(id: str, driver):
"""[summary]
Args:
id (str): select要素を取得したい場所のid名
driver ([type]): driverを渡す
Returns:
select要素
"""
group_list = WebDriverWait(driver, 60).until(EC.visibility_of_element_located((By.ID, id)))
return Select(group_list)
def screenshot(driver):
"""[summary]
Args:
driver ([type]): driverを渡す
Returns:
pngのバイナリデータ
"""
# 念の為待機
time.sleep(1)
# ページ全体の横幅と縦幅を取得してウィンドウサイズとしてセット。これをやらないと画面外が見切れる。
page_width = driver.execute_script("return document.body.scrollWidth")
page_height = driver.execute_script("return document.body.scrollHeight")
driver.set_window_size(page_width, page_height)
# スクリーンショットをとる
png = driver.find_element(By.CSS_SELECTOR, "#budgets-progress > div > section").screenshot_as_png
return png
def calc_remaining_per_day(webElement: str, days_left: int, driver) -> str:
"""[summary]
Args:
webElement (str): スクレイピングしたい残高のCSS_SELECTORを渡す
days_left (int): 残り日数を渡す
driver ([type]): driverを渡す
Returns:
[str]: 残高を残り日数で割った金額をカンマつきのstrにして返す
"""
time.sleep(1)
remaining = driver.find_element(By.CSS_SELECTOR, webElement).text
return "{:,d}円".format(int(remaining[0:-1].replace(",", "")) // days_left)
def acount_table_scraping(driver) -> str:
"""[summary]
特定の口座残高の情報をスクレイピングして文字列に変換して返す
Args:
driver ([type]): driverを渡す
Returns:
str: slack送信用に整形された文字列
"""
tableElem = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.TAG_NAME, "table")))
trs = tableElem.find_elements(By.TAG_NAME, "tr")
# ヘッダ行は除いて取得
for i in range(1, len(trs)):
tds = trs[i].find_elements(By.TAG_NAME, "td")
line = []
for j in range(0, len(tds)):
if j < len(tds) - 1:
line.append(f"{tds[j].text}\t")
else:
line.append(tds[j].text)
acount_name = line[0]
if ACOUNT_SBI1_NAME in acount_name:
acount_sbi_1_balance = line[1]
acount_sbi_1_latest_date = line[2][-13:-2]
acount_sbi_1_status = line[3]
if ACOUNT_SBI2_NAME in acount_name:
acount_sbi_2_balance = line[1]
acount_sbi_2_latest_date = line[2][-13:-2]
acount_sbi_2_status = line[3]
if ACOUNT_BUSINESS_NAME in acount_name:
acount_business_balance = line[1]
acount_business_latest_date = line[2][-13:-2]
acount_business_status = line[3]
acount1 = f"{SBI1_NAME}：{acount_sbi_1_balance}\n{acount_sbi_1_latest_date} {acount_sbi_1_status}"
acount2 = f"{SBI2_NAME}：{acount_sbi_2_balance}\n{acount_sbi_2_latest_date} {acount_sbi_2_status}"
acount3 = f"{BUSINESS_NAME}：{acount_business_balance}\n{acount_business_latest_date} {acount_business_status}"
return f"＊——最新の口座残高（生活・事業）——＊\n{acount1}\n\n{acount2}\n\n{acount3}"
def send_message(text: str):
"""[summary]
Slackにテキストを送る
Args:
text (str): Slackに送りたい文字列
"""
headers = {"Authorization": "Bearer" + SLACK_BOT_TOKEN}
data = {
"token": SLACK_BOT_TOKEN,
"channel": CHANNEL_ID,
"text": text,
"icon_emoji": ":moneybag:",
}
res = requests.post("https://slack.com/api/chat.postMessage", headers=headers, data=data)
return res
def upload_img(png):
"""[summary]
Slackにサマリー画像のキャプチャを送る
Returns:
なし
"""
files = {"file": png}
data = {
"token": SLACK_BOT_TOKEN,
"channels": CHANNEL_ID,
"filename": "sammary.png",
"initial_comment": "プライベート予算のサマリー",
"title": "プライベート予算のサマリー",
"icon_emoji": ":moneybag:",
}
res = requests.post("https://slack.com/api/files.upload", data=data, files=files)
return res
テーブルのスクレイピングをしているacount_table_scraping(driver) のコードは、以下サイトを参考にさせていただきました。
PythonでSeleniumを使ってWebページ内のテーブルの内容を取得する - 管理人Kのひとりごと
AWS Lambdaにデプロイ
以上のコードを、AWS Lambdaにデプロイします。
※事前にローカルでSlack投稿まで動作確認してます。
※ローカルで動かすときはドライバライブラリの読み込み方が違うので、set_driver()関数の中身を以下に変えたものを使ってます。
def set_driver():
options = webdriver.ChromeOptions()
# 以下のコメントアウトを外すとブラウザ非表示で実行
# options.headless = True
# ユーザーエージェントを偽装（ヘッドレスモードで実行するため）
options.add_argument(f"--user-agent={UA}")
return webdriver.Chrome(executable_path="chromedriver", options=options)
Serverless FrameworkでAWS Lambdaにデプロイする
デプロイのやり方はいろいろあるようですが、夫に教えてもらってserverless-frameworkを採用しました。
Serverless Frameworkのインストールは以下を参考にしました。
Serverless FrameworkでLambdaをPython 3.6で使ってみた | DevelopersIO
ymlファイルは以下の内容です。
# serverless.yml
service: crawler-with-selenium
frameworkVersion: "2"
provider:
name: aws
runtime: python3.6
stage: dev
region: ap-northeast-1
timeout: 900
lambdaHashingVersion: 20201221
functions:
main:
handler: handler.main
events:
- schedule: cron(0 22 * * ? *)
environment:
LOGIN_ID: ${env:LOGIN_ID}
LOGIN_PASSWORD: ${env:LOGIN_PASSWORD}
UA: ${env:UA}
SLACK_BOT_TOKEN: ${env:SLACK_BOT_TOKEN}
CHANNEL_ID_TEST: ${env:CHANNEL_ID_TEST}
CHANNEL_ID: ${env:CHANNEL_ID}
AWS_PROFILE: ${env:AWS_PROFILE}
ACOUNT_SBI1_NAME: ${env:ACOUNT_SBI1_NAME}
ACOUNT_SBI2_NAME: ${env:ACOUNT_SBI2_NAME}
ACOUNT_BUSINESS_NAME: ${env:ACOUNT_BUSINESS_NAME}
SBI1_NAME: ${env:SBI1_NAME}
SBI2_NAME: ${env:SBI2_NAME}
BUSINESS_NAME: ${env:BUSINESS_NAME}
package:
patterns:
- ".fonts/**"
layers:
- ${env:CHROME_DRIVER_LAYER}
- ${env:SERVERLESS_CHROME_LAYER}
plugins:
- serverless-python-requirements
${env:[環境変数名]} でenvファイルで設定している環境変数を読み込んでいます。
参考サイト
【Serverless Framework】簡単デプロイとserverless.ymlの記載について その２ - echo("備忘録");
python3でserverless frameworkの環境作成するコマンド
$ sls create -t aws-python3 -p {好きなフォルダ名}
デプロイコマンド。-vはデプロイの詳細が見れるオプション
$ serverless deploy -v
seleniumはrequirements.txtでインストール
seleniumと、slack通知に使っているrequestsは、requirements.txtに記載してまとめてインストールしました。
serverless-frameworkでrequirements.txtを使うために、rerserverless-python-requirementsを利用しています。
参考サイト
Serverless で Python のパッケージを使った Lambda 関数をデプロイ
chromeDriverとserverless chromeをダウンロードする
AWS Lambda上でseleniumを使うためには、Lambda上でブラウザ操作ができるライブラリが必要です。
chromedriverとserverless chromeが良く利用されているようなので、私もそれに倣いました。
上述のように、seleniumはpip installで使用できますが、chromedriverとserverless chromeはそれができないので、zipファイルをダウンロードして、何らかの方法でLambdaに上げる必要があります。
私は、AWSのコンソール画面からレイヤーに上げました。
なお、バージョンが揃ってないとうまく動かないため、注意が必要です。以下は2021/08/08動作確認済みのバージョンです。
Chrome Driver = 2.43
Serverless Chrome = 1.0.0-55
ダウンロードは以下サイトを参考に、コマンドで実行しました。
Lambda(Python)でSeleniumが動かないのでバージョンを調整して解決した件
# ダウンロード Chrome Driver 2.43
$ wget https://chromedriver.storage.googleapis.com/2.43/chromedriver_linux64.zip
# ダウンロード Serverless Chrome v1.0.0-55
$ wget https://github.com/adieuadieu/serverless-chrome/releases/download/v1.0.0-55/stable-headless-chromium-amazonlinux-2017-03.zip
chromeDriverとserverless chromeをレイヤーに上げる
以下のページからレイヤーの作成でZipファイルを読み込めます。
ランタイムはymlファイルで指定しているPythonバージョンと合わせました。
名前と説明は任意ですが、説明なしで入れたとき、どのバージョンのライブラリを入れたかわからなくなってしまったので、説明文にバージョンを入れるようにしました。
レイヤーを上げると、レイヤー一覧画面に「バージョンARN」が表示されるので、それをymlファイルに記載して関連づけます。
layers:
- ${env:CHROME_DRIVER_LAYER}
- ${env:SERVERLESS_CHROME_LAYER}
環境変数で隠してる部分は、こんなコードです。
arn:aws:lambda:ap-northeast-1:[自分のAWSアカウントID]:layer:ChromeDriver:1
ライブラリ使用のためのコードは以下です。
レイヤーに上げたものは/opt/以下に配置になります。ライブラリを任意のディレクトリにまとめてた場合は/opt/任意のディレクトリ名/ライブラリ名のような指定になるので注意が必要です。
def set_driver():
options = webdriver.ChromeOptions()
options.binary_location = "/opt/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(f"--user-agent={UA}")
# Headless Chromeをreturnする
return webdriver.Chrome(executable_path="/opt/chromedriver", chrome_options=options)
文字化けを直す
Lambda上に日本語フォントがない関係で、そのままだとキャプチャ画面の日本語が文字化けしてしまいます。
IPAが配布しているフォントファイルをダウンロードして利用しました。
Lambdaから.fontsフォルダを参照させるためのコード。
# handler.py
os.environ["HOME"] = "/var/task"
ymlファイルのfunctions.main 以下に記載。
※以前はincludeだったが、次のメジャーアップデートでincludeからpatternsに変わるため、patternsで記載。
#serverless.yml
package:
patterns:
- ".fonts/**"
参考サイト
Headless ChromeとSeleniumをLambdaで動かす
エラーの記録
"errorMessage": "Unable to import module 'handler'"
LambdaでRuntime.ImportModuleErrorが発生した時の対処 | Oji-Cloud
ymlに記載するhandler名は、{実行したいスクリプトを書いたファイル名}.{関数名}
ファイル名にスペルミスがあると、このエラーが表示されます。
外部モジュールが正しくインポートされてない場合も出ました。
"provider.lambdaHashingVersion" to "20201221"
ymlファイルのprovider以下にlambdaHashingVersion: 20201221 の記述が必要でした。
ページ遷移しない
自動処理で、ID入力後、パスワード入力ページに遷移するはずなのに遷移しない問題が起こりました。（トップ画面からID入力画面には遷移していたので、エラーのポイントが分かりづらかった）
Chrome DriverとServerless Chromeのバージョンを再確認して、レイヤーを使ってインポートし直したら解決されました。
当初は見よう見まねでymlファイルを通してレイヤーに上げていたので、インポート時のディレクトリ記載が間違っていたのかもしれません。。
今後の課題
スクレイピング前に、口座一覧を一括更新する処理を入れている影響で、口座の一部がメンテナンス中の場合に処理が正常に完了しなかったことがありました。
- 更新が完了しなくても一定時間経過でその後のスクレイピング処理が進むようにする
- エラーをキャッチしてどういうエラーだったかをSlackに投げる
あたりの処理が必要だなと思ってますが、最近は正常に動いていて困っていないので、追々やる気が出たら追加しようかなと思っています。
追記
東急カードのメンテナンスで口座情報が取得できないことが頻繁でやっぱり気になってきたので、改修しました。
口座情報更新のところにtry/except/finallyをいれて、更新が終わらなくてもそのまま口座情報をスクレイピングする処理にしました。
try:
status = WebDriverWait(driver, 300).until(EC.visibility_of_element_located((By.ID, "js-status-sentence-span-cNHmiFwd2QoSX5MiHCFs_w")))
assert status.text == "正常"
# タイムアウトをキャッチして無視
except TimeoutException:
pass
# タイムアウトしてもしなくても口座情報スクレイピングする
finally:
# スクレイピングした口座残高データ等を変数に格納
acount_remaining_list = acount_table_scraping(driver)
あと、今回調べて気づいたんですが、デフォルトだと自動的に2回リトライ（初回を入れると3回）実行されるようで。
自動再試行入ってるっぽいなーやだなーとは思ってたんですがこれでした。
AWS Lambda が非同期呼び出しの最大イベント経過時間と最大再試行回数をサポートするようになりました
except文でキャッチしたので、たぶん大丈夫かなと思ったんですが、念のためリトライ回数0にしました。
（最悪1日くらい送信されなくても良いし、無駄に再ログインさせたくなかったので。）
functions:
main:
handler: handler.main
maximumEventAge: 7200
maximumRetryAttempts: 0
serverless.ymlへの書き方は以下参考にさせていただきました。
Serverless Framework で最大イベント経過時間と最大再試行回数をカスタマイズする
気になること
AWSコンソール上のテストでは一回だけ実行されたんでとりあえずいいかなと思ってるんですが、ローカルのターミナル上でsls invoke -f mainでテストすると、何度も実行されちゃって強制終了するまで止まらないのでちょっと困りました。sls invokeコマンドの理解が浅いせいなんでしょうが。。
とりあえず定期実行には問題ないとこまでできたと思うので、また気になった時に調査して直せたら直そうかと思ってます。
まとめ
以上、簡単にですがまとめてみました。
この記事を見ていただき、ご指摘や質問などありましたら、コメントを送っていただけると大変ありがたいです。
最後まで読んでいただき、ありがとうございました。
0 件のコメント:
コメントを投稿