基于python实现的监听服务接口是否正常，发送异常消息到钉钉群

获取钉钉机器人

• 创建钉钉群组(要求至少三个成员)
• 进入群组 设置>机器人>添加机器人
• 选择自定义机器人 按照要求填写完获取到 Webhook的链接

实现代码

from time import sleep
import requests
import json
from datetime import datetime
import logging

# 配置日志记录的格式
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)

# 需要检测的目标url
target_url = 'http://XXXX'
# 参数
target_args = {'question': 'XXX'}
# 定义任务运行的时间段
start_time = "08:30"
end_time = "19:00"
# 定义任务运行的间隔时间
sleep_time = 60 * 60 * 2

def chat_service_check(url, body):
    try:
        response = requests.post(url, json=body)
        if response.status_code != 200:
            logging.info(f'响应结果异常：响应值={response.text}')
            return False
        if json.loads(response.text)['code'] != 200:
            logging.info(f'调用接口失败:{response.text}')
            return False
        else:
            return True
    except requests.RequestException as e1:
        logging.error(f'响应结果异常', e1)
        return False


def dingmessage(msg):
	# todo: 替换自己的钉钉token
    webhook_url = 'https://oapi.dingtalk.com/robot/send?access_token=XXXXX'
    data = {
        'msgtype': 'text',
        'text': {
            'content': '监控通知:' + msg
        }
    }
    res = requests.post(webhook_url , json=data)
    logging.info(f'通知返回结果:{res.text}')


def task():
    logging.info(f'当前时间:{datetime.now().strftime("%Y-%m-%d %H:%M:%S")},调用目标接口')
    chat_check = chat_service_check(target_url, body=target_args)
    logging.info(f'当前时间:{datetime.now().strftime("%Y-%m-%d %H:%M:%S")},调用目标接口结束')
    if not chat_check:
        logging.info(f'当前时间:{datetime.now().strftime("%Y-%m-%d %H:%M:%S")},调用目标接口失败')
        dingmessage('chat 接口异常请即使处理')


if __name__ == '__main__':

    while True:
        current_time = datetime.now()
        start_time1 = datetime.strptime(str(datetime.now().date()) + start_time, "%Y-%m-%d%H:%M")
        end_time1 = datetime.strptime(str(datetime.now().date()) + end_time, "%Y-%m-%d%H:%M")
        # 检查当前时间是否在指定范围内
        if start_time1 <= current_time < end_time1:
            logging.info("当前时间在早上八点三十到下午七点之间,执行task任务")
            try:
                task()
            except Exception as e:
                logging.info('执行任务失败....', e)
        else:
            logging.info("当前时间不在早上八点三十到下午七点之间，不执行task任务")
        logging.info(f'程序睡眠时间{sleep_time}秒')
        sleep(sleep_time)

linux 服务器启动脚本

vim start.sh
#!/bin/bash
nohup python3 service_monitor.py >  monitor.log &

• 执行脚本

sh start.sh