快捷導(dǎo)航

詳解Python實(shí)現(xiàn)URL監(jiān)測(cè)與即時(shí)推送

更新時(shí)間：2021年11月17日 08:42:24 作者：董式小爬蟲

這篇文章主要為大家介紹了Python實(shí)現(xiàn)URL監(jiān)測(cè)與即時(shí)推送，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下，希望能夠給你帶來(lái)幫助

原理

采用Python requests發(fā)起請(qǐng)求監(jiān)測(cè)的URL，檢測(cè)Http響應(yīng)狀態(tài)及是否超時(shí)，如果Http狀態(tài)異?；蝽憫?yīng)超時(shí)，則通過(guò)聚合云推的消息推送API將預(yù)警消息發(fā)送至郵箱、釘釘機(jī)器人、企業(yè)微信機(jī)器人、微信公眾號(hào)等，服務(wù)端通過(guò)crontab定時(shí)(每分鐘)執(zhí)行代碼，實(shí)現(xiàn)動(dòng)態(tài)監(jiān)測(cè)功能。

環(huán)境

操作系統(tǒng): CentOS 7.x

Python版本: 3.6

消息推送服務(wù): tui.juhe.cn

代碼

#!/usr/bin/python3
import requests
import time
import json

# 監(jiān)測(cè)URL是否正常響應(yīng)
def url_check(url):
    # 當(dāng)前時(shí)間
    check_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    print("開始監(jiān)測(cè)：%s -- %s" % (url, check_time))
    try:
        # 請(qǐng)求URL， 設(shè)置3s超時(shí)
        r = requests.get(url, timeout=3)
        if r.status_code != 200:
            # 請(qǐng)求響應(yīng)狀態(tài)異常
            msg = "監(jiān)控的URL：%s%sHttp狀態(tài)異常：%s%s監(jiān)測(cè)時(shí)間：%s" % (url, "\n\n", r.status_code, "\n\n", check_time)
            print("監(jiān)測(cè)結(jié)果：異常（Http狀態(tài)異常:%s） -- %s" % (r.status_code, check_time))
            # 通過(guò)云推推送消息
            yuntui_push(msg)
        else:
            # 請(qǐng)求響應(yīng)正常
            print("監(jiān)測(cè)結(jié)果：正常 -- %s" % check_time)
    except requests.exceptions.ConnectTimeout:
        # 請(qǐng)求響應(yīng)超時(shí)
        msg = "監(jiān)控的URL：%s%s請(qǐng)求異常：%s%s監(jiān)測(cè)時(shí)間：%s" % (url, "\n\n", "請(qǐng)求超時(shí)", "\n\n", check_time)
        print("監(jiān)測(cè)結(jié)果：超時(shí) -- %s" % check_time)
        # 通過(guò)云推推送消息
        yuntui_push(msg)

# 將預(yù)警消息通過(guò)云推推送
def yuntui_push(content):
    # 當(dāng)前時(shí)間
    push_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    # 云推接口的信息配置，可以通過(guò) https://tui.juhe.cn 自行注冊(cè)創(chuàng)建
    # (可以配置郵件、釘釘機(jī)器人、微信公眾號(hào)等接收方式)
    token = "*****************"
    service_id = "******"
    title = "URL可用性監(jiān)控預(yù)警"
    doc_type = "markdown"
    body = {"token": token, "service_id": service_id, "title": title, "content": content, "doc_type": doc_type}
    try:
        r = requests.post("https://tui.juhe.cn/api/plus/pushApi", data=body, timeout=15)
        push_res = json.loads(r.content)
        code = push_res['code']
        if code == 200:
            print("推送結(jié)果：成功 -- %s" % push_time)
        else:
            print("推送結(jié)果：失敗（%s） -- %s" % (push_res['reason'], push_time))
    except requests.exceptions.ConnectTimeout:
        print("推送結(jié)果：超時(shí) -- %s" % push_time)

# 執(zhí)行URL可用性監(jiān)測(cè)
if __name__ == '__main__':
    # 監(jiān)控URL可用性
    # url_check("https://www.test.com")
    url_check("https://www.baidu.com/")

如果requests未安裝可以執(zhí)行以下命令安裝

pip3 install requests

crontab計(jì)劃任務(wù)配置

命令行輸入crontab -e進(jìn)入計(jì)劃任務(wù)配置

# 每分鐘執(zhí)行一次
*/1 * * * * /usr/bin/python3 /data/check_url/main.py >> /data/log.txt

查看日志

cat /data/log.txt

開始監(jiān)測(cè)：https://www.baidu.com/ -- 2021-11-16 15:04:01
監(jiān)測(cè)結(jié)果：正常 -- 2021-11-16 15:04:01
開始監(jiān)測(cè)：https://www.baidu.com/ -- 2021-11-16 15:05:02
監(jiān)測(cè)結(jié)果：正常 -- 2021-11-16 15:05:02
開始監(jiān)測(cè)：https://www.baidu.com/ -- 2021-11-16 15:06:01
監(jiān)測(cè)結(jié)果：正常 -- 2021-11-16 15:06:01
開始監(jiān)測(cè)：https://www.baidu.com/ -- 2021-11-16 15:07:01
監(jiān)測(cè)結(jié)果：正常 -- 2021-11-16 15:07:01
開始監(jiān)測(cè)：https://www.baidu.com/ -- 2021-11-16 15:08:01
監(jiān)測(cè)結(jié)果：正常 -- 2021-11-16 15:08:01
開始監(jiān)測(cè)：https://www.test.com -- 2021-11-16 15:11:01
監(jiān)測(cè)結(jié)果：超時(shí) -- 2021-11-16 15:11:01
推送結(jié)果：成功 -- 2021-11-16 15:11:04
開始監(jiān)測(cè)：https://www.test.com -- 2021-11-16 15:12:01
監(jiān)測(cè)結(jié)果：超時(shí) -- 2021-11-16 15:12:01
推送結(jié)果：成功 -- 2021-11-16 15:12:04