はじめに
Seleniumを使って、Webサイトが正常に動作しているかを監視し、サイトが落ちていたりしてエラーとなる場合にLINEに通知を送るということをやってみます。
実装にあたっては、@watti さんの「Selenium(Python)でオンラインストアを自動検索して結果をLINEに通知する」の記事を参考にさせていただきました。
やること
- Seleniumで対象サイトにアクセス
- 対象サイトが正しく表示できたかを判定
- 異常と判断された場合には、スクリーンショットを取得し、メッセージとともにLINEに通知
スクリーンショットをLINEに通知するコード
まずは、Seleniumでスクリーンショットを撮って、LINEに通知するという部分だけを行います。
LINE Notifyの準備等は以下をご参照ください。
○参考：PythonでLINEにメッセージを送る
import os
import requests
from selenium import webdriver
from webdriver_manager.chrome import ChromeDriverManager
# LINE Notifyのアクセストークン
ACCESS_TOKEN = "YOUR ACCESS TOKEN"
LINE_API = "https://notify-api.line.me/api/notify"
# WebDriverの設定・取得
def get_driver():
options = webdriver.ChromeOptions()
options.add_argument('--headless')
options.add_argument('--no-sandbox')
options.add_argument('--disable-gpu')
options.add_argument('--window-size=1024,768')
# Webdriver ManagerでChromeDriverを取得
driver = webdriver.Chrome(
executable_path=ChromeDriverManager().install(),
options=options)
driver.set_page_load_timeout(15)
driver.implicitly_wait(15)
return driver
# 結果をLINEへ送信
def notify_to_line(message, image_file_path=None):
headers = {'Authorization': 'Bearer ' + ACCESS_TOKEN}
payload = {'message': message}
files = {}
if ((image_file_path is not None) and os.path.exists(image_file_path)):
files = {'imageFile': open(image_file_path, 'rb')}
response = requests.post(LINE_API, headers=headers, params=payload, files=files)
def main():
driver = get_driver()
driver.get('https://qiita.com/')
driver.save_screenshot('qiita.png')
# テキストのみ送信
notify_to_line('テストメッセージ')
# テキスト + 画像を送信
notify_to_line('スクリーンショット付きメッセージ', 'qiita.png')
driver.close()
driver.quit()
if __name__ == "__main__":
main()
-
notify_to_lineという関数内でLINE NotifyのAPIを実行しています。関数の第1引数はメッセージで、第2引数は省略可で画像ファイルを指定します。画像ファイルが指定された場合は、filesにセットします。 - WebDriverの取得には、
webdriver_managerを利用しています。
実行すると以下のようにLINEに通知が届きました。
Webサイトの正常性確認（リトライ処理あり）
次に監視対象Webサイトにアクセスして、サイトが正常に動作することを確認する処理に関連するコードを一部抜粋して紹介します。
ここでは、Qiitaを対象サイトにしています。ログインページでID・パスワードを入力してログインします。ログイン後のページに「投稿する」という文字列が表示されていることでサイトが正常に動作していると判断します。
from retrying import retry, RetryError
USER_ID = "YOUR ID"
PASSWORD = "YOUR PASSWORD"
ASSERTION_STRING = "投稿する"
# Webサイトが正常かをチェックする
@retry(stop_max_attempt_number=3, wait_fixed=5000, wrap_exception=True)
def check_site(driver):
r = random.random()
print_log(r)
if r < 0.5:
driver.get('https://qiita.com/login')
else:
driver.get('http://getstatuscode.com/503')
# ログインを試みる
driver.find_element_by_name("identity").send_keys(USER_ID)
driver.find_element_by_name("password").send_keys(PASSWORD)
driver.find_element_by_name("commit").click()
# 期待した内容が表示されていることを確認
assert ASSERTION_STRING in driver.page_source
def main():
#・・・
try:
check_site(driver)
except RetryError:
driver.save_screenshot(screenshot_path)
notify_to_line('Webサイトで異常が発生しています', screenshot_path)
- ここでは、Webサイトが503エラーを返す状態になっていることを擬似的に実現するために、アクセスすると503エラーを返してくれるサイトにランダムにアクセスするように実装しています。
- また、1度の失敗のみで異常と判断せずに3回アクセスを実施して、全て失敗した場合に異常とするようにリトライ処理を導入しています。
- リトライ処理は、retryingというライブラリを利用しています。詳細は以下の記事をご参照ください。
- 参考；[Pythonのretryingによるリトライ] (https://blog.imind.jp/entry/2019/01/15/144542)
- 3回全て失敗した場合は、
RetryErrorという例外が発生するので、この時のスクリーンショットを取得し、LINEに通知します。
コード全体
コード全体は、下記の通りです。上記で一部抜粋して紹介した以外にもいくつかの補助的な処理が入っています。
selenium-line-notify.py
import os
import requests
import random
from datetime import datetime as dt
from selenium import webdriver
from webdriver_manager.chrome import ChromeDriverManager
from retrying import retry, RetryError
ACCESS_TOKEN = "YOUR ACCESS TOKEN"
LINE_API = "https://notify-api.line.me/api/notify"
USER_ID = "YOUR ID"
PASSWORD = "YOUR PASSWORD"
ASSERTION_STRING = "投稿する"
# ログ用に日時を付与する関数
def print_log(*log_messages):
print(dt.now().strftime('[%Y%m%d-%H%M%S] '), end='')
print(*log_messages)
# スクリーンショットのファイル名を取得
def get_filepath_with_timestamp():
img_path = os.path.join(
os.path.abspath(os.path.dirname(__file__)),
"screenshot_" + dt.now().strftime('%Y%m%d%H%M%S') + ".png")
return img_path
# WebDriverの設定・取得
def get_driver():
options = webdriver.ChromeOptions()
options.add_argument('--headless')
options.add_argument('--no-sandbox')
options.add_argument('--disable-gpu')
options.add_argument('--window-size=1024,768')
# Webdriver ManagerでChromeDriverを取得
driver = webdriver.Chrome(
executable_path=ChromeDriverManager().install(),
options=options)
driver.set_page_load_timeout(15)
driver.implicitly_wait(15)
return driver
# 結果をLINEへ送信
def notify_to_line(message, image_file_path=None):
headers = {'Authorization': 'Bearer ' + ACCESS_TOKEN}
payload = {'message': message}
files = {}
if ((image_file_path is not None) and os.path.exists(image_file_path)):
files = {'imageFile': open(image_file_path, 'rb')}
response = requests.post(LINE_API, headers=headers, params=payload, files=files)
print_log('notify response code:', response.status_code)
# Webサイトが正常かをチェックする
@retry(stop_max_attempt_number=3, wait_fixed=5000, wrap_exception=True)
def check_site(driver):
r = random.random()
print_log(r)
if r < 0.5:
driver.get('https://qiita.com/login')
else:
driver.get('http://getstatuscode.com/503')
# ログインを試みる
driver.find_element_by_name("identity").send_keys(USER_ID)
driver.find_element_by_name("password").send_keys(PASSWORD)
driver.find_element_by_name("commit").click()
# 期待した内容が表示されていることを確認
assert ASSERTION_STRING in driver.page_source
def main():
try:
# スクリーンショットのファイル名を取得
screenshot_path = get_filepath_with_timestamp()
# WebDriverを取得
driver = get_driver()
try:
# Webサイトの正常性確認
check_site(driver)
except RetryError: # 異常時の処理
driver.save_screenshot(screenshot_path)
notify_to_line('Webサイトで異常が発生しています', screenshot_path)
else: # 正常時の処理
driver.save_screenshot(screenshot_path)
notify_to_line('Webサイトは正常です', screenshot_path)
except Exception as e:
print_log('Error', e)
finally:
driver.close()
driver.quit()
print_log('Driver is closed')
if __name__ == "__main__":
main()
cronで定期実行
cron以外にも方法はいろいろとあると思いますが、定期的に実行して監視するために、cronを使います。とりあえず15分毎に実行するのであれば、以下のような感じになると思います。
*/15 * * * * (pythonコマンドのパス) (スクリプトのパス) >> (ログファイル) 2>&1
注意！
- 本記事で紹介しているコードは、本番運用に堪えるようなきちんと検証されたコードではありません。
- 利用・参考にされる際には、利用者の責任において必要な修正・検証を行ってください。
- 本記事の内容を利用、参照されたことにより発生したいかなる損害も著者は一切責任を負いません。