pythonで、urlの死活監視を行い、NGであれば、sendmail を外部コマンドで通知

以下の通りで、ポイントは

• urllib.request による HTTP GET
• 「/usr/sbin/sendmail -t」によるmail送信
• subprocess による標準入力渡し
• python3 の標準moduleで実現OK

かと思います。

#!/usr/bin/python3
# -*- coding: utf-8 -*-

import json
import os
import ssl
ssl._create_default_https_context = ssl._create_unverified_context
import sys
import time, datetime
import urllib.request
import subprocess
from subprocess import PIPE

ret_urls = [
    "https://ないしょ1.jp",
    "https://ないしょ2.jp" ]

http_timeout = 10    #sec
retry_sleep =  10    #sec
retry_max   =   5    #times

chk_history_file \
= os.path.dirname(os.path.abspath(__file__)) + "/chk_alive_urls.history"

mail_cond = {
    "cmd" :"/usr/sbin/sendmail -t",
    "from":"ないしょ@ないしょ.com",
    # 宛先複数は「,」で接続
    "to"  :"ないしょ@ないしょ.com",
    "subject": "!!!! server has DOWN URL !!!!" }

def main():
    # 過去のcheck結果をload
    chk_history = load_chk_history()

    # 実際にaccessすることで、check
    for req_url in ret_urls:
        if not req_url in chk_history:
            chk_history[req_url] = {'last_check': '',
                                    'repeat_ok' : 0,
                                    'repeat_ng' : 0  }
        i = 0
        while i < retry_max:
            i += 1
            time.sleep( retry_sleep )
            
            datetime_str = \
                datetime.datetime.fromtimestamp(time.time()).strftime(
                    '%Y-%m-%d %H:%M' )
            res = http_get(req_url)
            if res:
                break
            
            
        chk_history[req_url]['last_check'] = datetime_str
        
        if res:
            chk_history[req_url]['repeat_ok'] += 1
            chk_history[req_url]['repeat_ng']  = 0
        else:
            chk_history[req_url]['repeat_ok']  = 0
            chk_history[req_url]['repeat_ng'] += 1
        
    # check結果を保存
    save_chk_history(chk_history)

    # NGなurlが1つでもあれば、通知メール通知
    for req_url in chk_history:
        if chk_history[req_url]['repeat_ng']:
            return notify_chk_result(chk_history)
    
        
def notify_chk_result(chk_history):

    msg_head_body = ""
    msg_head_body += "From: "   +mail_cond["from"]+"\n"
    msg_head_body += "To: "     +mail_cond["to"]+"\n"
    msg_head_body += "Subject: "+mail_cond["subject"]+"\n"
    msg_head_body += "\n"
    msg_head_body += "PLEASE CHECK BELOW REPEAT_NG !!\n"
    msg_head_body += json.dumps(chk_history, indent=2)

    proc = subprocess.run(mail_cond["cmd"],
                          shell=True,
                          input=msg_head_body.encode(), #★要byte列化
                          stdout=PIPE,
                          stderr=PIPE)
    
        
def save_chk_history(chk_history):
    with open(chk_history_file, mode='w') as file:
        json.dump(chk_history, file, indent=2)
  
def load_chk_history():

    if not os.path.exists(chk_history_file):
        return {}
    
    fh = open(chk_history_file,'r')
    fh_content = fh.read()
    if len(fh_content) == 0:
        return {}

    return json.loads(fh_content)

def http_get(req_url):
    req = urllib.request.Request(req_url)
    try:
        res = urllib.request.urlopen(req,
                                     data=None,
                                     timeout=http_timeout)
    except urllib.error.HTTPError as err:
        print("WARN",err.code, req_url,file=sys.stderr)
        return None
    except urllib.error.URLError as err:
        print("ERROR",err.reason, req_url,file=sys.stderr)
        return None

    # print("DONE HTTP",res.getcode(), res.geturl() )
    # print(res.read() )
    # res.close()
    return res
    

if __name__ == '__main__':
    main()