| 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() |