钉钉消息异常通知
支持多线程
支持多进程
支持@多人
支持@所有人
from pprint import pprintimport requests
import json
import threading
import multiprocessing
from datetime import datetimeclass DingTalkAlert:def __init__(self, webhook_url, secret=None):self.webhook_url = webhook_urlself.secret = secretdef send_message(self, title, error_message, **kwargs):at_all = kwargs.get('at_all', False)at_mobiles = kwargs.get('at_mobiles', [])timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')message_content = f"**标题**: {title}\n\n**异常信息**: {error_message}\n\n**发生时间**: {timestamp}"# headers = {# 'Content-Type': 'application/json'# }data = {"msgtype": "text","text": {"title": title,"content": message_content},"at": {"isAtAll": at_all,"atUserIds": [],"atMobiles": at_mobiles}}pprint(data)response = requests.post(self.webhook_url, data=json.dumps(data), headers={'Content-Type': 'application/json'})print(response.status_code)return response.json()def send_message_thread(self, title, error_message, **kwargs):thread = threading.Thread(target=self.send_message, args=(title, error_message), kwargs=kwargs)thread.start()thread.join()def send_message_process(self, title, error_message, **kwargs):process = multiprocessing.Process(target=self.send_message, args=(title, error_message), kwargs=kwargs)process.start()process.join()# 使用示例
if __name__ == "__main__":webhook_url = 'https://oapi.dingtalk.com/robot/send?access_token='alert = DingTalkAlert(webhook_url)# 发送消息，@所有人alert.send_message("告警通知", "这是一个告警消息,测试阶段，消息打扰，请多多包涵", at_all=True)# 发送消息，@具体的多个负责人alert.send_message("i告警通知", "这是一个告警消息,测试阶段，消息打扰，请多多包涵", at_mobiles=["1881xxxx363", ])# 多线程发送消息，@所有人alert.send_message_thread("告警通知", "这是一个多线程告警消息,测试阶段，消息打扰，请多多包涵", at_all=True)# 多进程发送消息，@具体的多个负责人alert.send_message_process("i告警通知", "这是一个多进程告警消息,测试阶段，消息打扰，请多多包涵", at_mobiles=["1881xxxx363", "13900000000"])