import requests import json import socket from loguru import logger from enum import Enum class MSGTYPE(Enum): INCIDENT = 1 ERROR = 2 REPORT = 3 INFO = 4 SUCCESS = 0 icons = { MSGTYPE.SUCCESS: "https://img.icons8.com/ios-glyphs/30/000000/rocket.png", MSGTYPE.INCIDENT: "https://img.icons8.com/ios-filled/30/car-accident.png", MSGTYPE.ERROR: "https://img.icons8.com/ios-glyphs/30/000000/error--v2.png", MSGTYPE.REPORT: "https://img.icons8.com/ios-glyphs/90/000000/checkmark--v2.png", MSGTYPE.INFO: "https://img.icons8.com/ios-glyphs/30/000000/info--v2.png", } class Message: host_name = socket.gethostname() host_ip = socket.gethostbyname(socket.gethostname()) type = MSGTYPE.INCIDENT type_hint = "" content = "" class dummyRequest: def __init__(self) -> None: self.status_code = 200 class Bot: def __init__( self, app_name="", bot_key="", enabled=True, ): self.Bot_URL = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key={bot_key}" self.template = { "msgtype": "template_card", "template_card": { "card_type": "text_notice", "source": { "icon_url": "", "desc": app_name, "desc_color": 0 }, "main_title": { "title": "", "desc": "" }, "emphasis_content": { "title": "", "desc": "" }, "horizontal_content_list": [ ], "card_action": { "type": 1, "url": "https://huggingface.co/myscale" }, } } self.app_name = app_name self.enabled = enabled def __constuct_msg(self, msg: Message): _dict = self.template.copy() _dict["template_card"]["source"]["icon_url"] = icons[msg.type] _dict["template_card"]["main_title"]["title"] = msg.type.name[:13] _dict["template_card"]["main_title"]["desc"] = msg.type_hint[:15] _dict["template_card"]["horizontal_content_list"].extend( self.__convert_dict2clist(msg)[:6]) _dict["template_card"]["sub_title_text"] = msg.content return _dict def __convert_dict2clist(self, msg: Message): return [ {"keyname": "App Name", "value": self.app_name}, {"keyname": "Host Name", "value": msg.host_name}, {"keyname": "Host IP", "value": str(msg.host_ip)}, ] def __send_md_msg(self, msg: Message): msg = json.dumps(self.__constuct_msg(msg), ensure_ascii=True) return requests.post( self.Bot_URL, data=msg, headers={ "Content-Type": "application/json"} ) def error(self, msg: Message): msg.type = MSGTYPE.ERROR if self.enabled: return self.__send_md_msg(msg) else: return dummyRequest() def incident(self, msg: Message): msg.type = MSGTYPE.INCIDENT if self.enabled: return self.__send_md_msg(msg) else: return dummyRequest() def report(self, msg: Message): msg.type = MSGTYPE.REPORT if self.enabled: return self.__send_md_msg(msg) else: return dummyRequest()