eos-toolkit/utils/notify.py at master · antpool/eos-toolkit · GitHub

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
import argparse

import init_work_home

init_work_home.init()
from config.config import NotifyConfig
import http
from logger import logger

beary_id = NotifyConfig.get_beary_id()
beary_token = NotifyConfig.get_beary_token()
ding_talk_token = NotifyConfig.get_ding_talk_token()
err_beary_id = NotifyConfig.get_err_beary_id()
err_beary_token = NotifyConfig.get_err_beary_token()
err_ding_talk_token = NotifyConfig.get_err_ding_talk_token()
telegram_chat_id = NotifyConfig.get_telegram_chat_id()
telegram_token = NotifyConfig.get_telegram_token()


class Notify:
    @staticmethod
    def notify_status(*args):
        Notify.all_notify(False, *args)

    @staticmethod
    def notify_error(*args):
        Notify.all_notify(True, *args)

    @staticmethod
    def all_notify(is_error, *args):
        # add other like slack, sms...
        if args is None or len(args) < 1:
            return
        msg = '\n'.join(args)
        if is_error:
            Beary.beary_notify(msg, err_beary_id, err_beary_token)
            DingTalk.ding_talk_notify(msg, err_ding_talk_token)
            Telegram.telegram_notify(msg, telegram_chat_id, telegram_token)
        else:
            Beary.beary_notify(msg, beary_id, beary_token)
            DingTalk.ding_talk_notify(msg, ding_talk_token)
            Telegram.telegram_notify(msg, telegram_chat_id, telegram_token)


class DingTalk:
    @staticmethod
    def ding_talk_notify(message, token):
        if token is None or token is "":
            return
        try:
            ding_talk_hook_url = "https://oapi.dingtalk.com/robot/send?access_token=" + token
            body = ('{"msgtype":"text","text":{"content":"%s"}}' % (message))
            http.post('ding_talk', ding_talk_hook_url, data=body, headers=http.def_headers)
        except Exception as e:
            logger.error("dingTalk notify error, msg:%s, exception:%s", message, e)


class Beary:
    @staticmethod
    def beary_notify(message, chat_id, token):
        if chat_id is None or chat_id is "":
            return
        if token is None or token is "":
            return
        try:
            beary_hook_url = ("https://hook.bearychat.com/%s/incoming/%s" % (chat_id, token))
            body = ('{"text":"%s"}' % (message))
            http.post('beary', beary_hook_url, data=body, headers=http.def_headers)
        except Exception as e:
            logger.error("beary notify error, msg:%s, exception:%s", message, e)


class Telegram:
    @staticmethod
    def telegram_notify(message, chat_id, token):
        if chat_id is None or chat_id is "":
            return
        if token is None or token is "":
            return
        try:
            telegram_url = "https://api.telegram.org/bot%s/sendMessage" % (token)
            param = {"chat_id": chat_id, "text": message}
            http.post('telegram', telegram_url, params=param)
        except Exception as e:
            logger.error("telegram notify error, msg:%s, exception:%s", message, e)


def usage():
    global msg, is_error
    parser = argparse.ArgumentParser(description='notify tool.')
    parser.add_argument('-m', '--msg', default=None, help='notify msg')
    parser.add_argument('-e', '--error', default=None, help='None is error data')
    args = parser.parse_args()
    msg = args.msg
    is_error = args.error


if __name__ == "__main__":
    usage()
    if is_error is None:
        Notify.notify_error(msg)
    else:
        Notify.notify_status(msg)