File size: 3,475 Bytes
511c1dc
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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()