创建三个文件
- main.py
import time
import tools
import webhook
# 定义需要监控的服务关系字典
service_dict = {
"test-baidu": "https://www.baidu.com/",
"test-sina": "https://www.sina.com.cn/"
}
# 根据服务字典进行巡检
while True:
for name, service in service_dict.items():
# 获取状态码
http_code = tools.get_status_code(service)
# print(f"{name}的状态码为:{http_code}")
# 判断有无异常,有则报警 无则结束
if http_code == 200:
print(f"{
name}服务正常,状态码为:{
http_code}")
elif http_code == 401:
print(f"{
name}服务正常,状态码为:{
http_code}")
elif http_code == 502:
webhook.send_msg(name, http_code, service)
else:
print(f"{
name}服务正常,状态码为:{
http_code}")
time.sleep(360)
- webhook.py
import requests
import json
wx_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=4794df11-557e-4690-xxxa-9bfaf4aca474"
def send_msg(name, http_code, service):
"""
发送指定消息
:param name:
:param http_code:
:param service:
:return:
"""
send_message = f"<font color=\"warning\"> {
name} </font>服务离线,请相关同事注意。\n> 状态码:<font color=\"comment\"> {
http_code} </font>\n> 链接:<font color=\"comment\"> [{
service}]({
service}) </font>"
data = json.dumps({
"msgtype": "markdown", "markdown": {
"content": send_message}})
r = requests.post(wx_url, data, auth=('Content-Type', 'application/json'))
- tools.py
import requests
def get_status_code(url):
"""
根据输入的url,获取对应的http状态码
:param url: 需要测试的url
:return: 状态码
"""
res = requests.get(url)
return res.status_code
启动main.py
即可