File size: 5,630 Bytes
3a5cf48 | 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 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | # 临时解决，处理旧版告警问题
import re
from .weixin_msg import WeiXinMsg
from .mail_msg import MailMsg
from .web_hook_msg import WebHookMsg
from .feishu_msg import FeiShuMsg
from .dingding_msg import DingDingMsg
from .wx_account_msg import WeChatAccountMsg
from mod.base.push_mod import SenderConfig, WxAccountMsg
from typing import Union, Dict, Optional, Tuple
class WarpToOld:
def __init__(self,
sender: Union[WeiXinMsg, MailMsg, WebHookMsg, FeiShuMsg, DingDingMsg, WeChatAccountMsg],
default_title: str):
self.sender = sender
self.default_title = default_title
def send_msg(self, msg: str, title: str = None) -> Dict:
msg, tmp_title = self._get_title_by_msg(msg)
if tmp_title is not None:
title = tmp_title
if title is None:
title = self.default_title
res = self.sender.send_msg(msg, title)
if res is None:
return {"status": True, "msg": "发送成功"}
else:
return {"status": False, "msg": res}
def push_data(self, data: dict) -> Dict:
if "msg" not in data:
return {"status": False, "msg": "消息内容不能为空"}
return self.sender.send_msg(data["msg"], data.get("title", None))
@staticmethod
def _get_title_by_msg(msg: str) -> Tuple[str, Optional[str]]:
title = None
try:
if msg.find("####") >= 0:
try:
title = re.search(r"####(.+)", msg).groups()[0]
except:
pass
msg = msg.replace("####", ">").replace("\n\n", "\n").strip()
s_list = msg.split('\n')
if len(s_list) > 3:
s_title = s_list[0].replace(" ", "")
s_list = s_list[3:]
s_list.insert(0, s_title)
msg = '\n'.join(s_list)
s_list = []
regexp = re.compile(r'<font.+>(.+)</font>')
for msg_info in msg.split('\n'):
tmp = regexp.search(msg_info)
if tmp:
tmp = tmp.groups()[0]
msg_info = regexp.sub(tmp, msg_info)
s_list.append(msg_info)
msg = '\n'.join(s_list)
except:
pass
return msg, title
def get_config(self, get=None):
if hasattr(self.sender, "config"):
return getattr(self.sender, "config")
if hasattr(self.sender, "data"):
return getattr(self.sender, "data")
return {}
class WarpWebHookMsg(WarpToOld):
def __init__(self,
sender: WebHookMsg,
default_title: str):
super().__init__(sender, default_title)
def send_msg(self, msg: str, title: str = None, push_type: str = None) -> Dict:
msg, tmp_title = self._get_title_by_msg(msg)
if tmp_title is not None:
title = tmp_title
if title is None:
title = self.default_title
if push_type is None:
push_type = "unknown"
res = self.sender.send_msg(msg, title, push_type)
if res is None:
return {"status": True, "msg": "发送成功"}
else:
return {"status": False, "msg": res}
def push_data(self, data: dict) -> Dict:
if "msg" not in data:
return {"status": False, "msg": "消息内容不能为空"}
return self.send_msg(data["msg"], data.get("title", None), data.get("push_type", None))
class WarpWeChatAccountMsg(WarpToOld):
def __init__(self,
sender: WeChatAccountMsg,
default_title: str):
super().__init__(sender, default_title)
def send_msg(self, msg: str, title: str = None) -> Dict:
msg, tmp_title = self._get_title_by_msg(msg)
if tmp_title is not None:
title = tmp_title
if title is None:
title = self.default_title
wxmsg = WxAccountMsg.new_msg()
wxmsg.thing_type = title
wxmsg.msg = msg
res = self.sender.send_msg(wxmsg)
if res is None:
return {"status": True, "msg": "发送成功"}
else:
return {"status": False, "msg": res}
def push_data(self, data: dict) -> Dict:
if "msg" not in data:
return {"status": False, "msg": "消息内容不能为空"}
return self.send_msg(data["msg"], data.get("title", None))
def get_sender_by_id(channel_id: str) -> Optional[WarpToOld]:
sc = SenderConfig()
if re.match(r"^[0-9a-f]{16}$", channel_id):
sender_config = sc.get_by_id(channel_id)
if sender_config is None:
return None
if sender_config["sender_type"] == "webhook":
return WarpWebHookMsg(WebHookMsg(sender_config), "宝塔Web Hook告警通知")
elif sender_config["sender_type"] == "mail":
return WarpToOld(MailMsg(sender_config), "宝塔邮件告警通知")
elif sender_config["sender_type"] == "weixin":
return WarpToOld(WeiXinMsg(sender_config), "宝塔微信告警通知")
elif sender_config["sender_type"] == "feishu":
return WarpToOld(FeiShuMsg(sender_config), "宝塔飞书告警通知")
elif sender_config["sender_type"] == "dingding":
return WarpToOld(DingDingMsg(sender_config), "宝塔钉钉告警通知")
elif sender_config["sender_type"] == "wx_account":
return WarpWeChatAccountMsg(WeChatAccountMsg(sender_config), "宝塔告警通知")
return None
return None
|