获取钉钉机器人
- 创建钉钉群组(要求至少三个成员)
- 进入群组 设置>机器人>添加机器人
- 选择自定义机器人 按照要求填写完获取到 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)
target_url = 'http://XXXX'
target_args = {'question': 'XXX'}
start_time = "08:30"
end_time = "19:00"
sleep_time = 60 * 60 * 2def chat_service_check(url, body):try:response = requests.post(url, json=body)if response.status_code != 200:logging.info(f'响应结果异常:响应值={response.text}')return Falseif json.loads(response.text)['code'] != 200:logging.info(f'调用接口失败:{response.text}')return Falseelse:return Trueexcept requests.RequestException as e1:logging.error(f'响应结果异常', e1)return Falsedef dingmessage(msg):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
nohup python3 service_monitor.py > monitor.log &
sh start.sh