File size: 5,647 Bytes
17e971c | 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 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 | # coding: utf-8
# +-------------------------------------------------------------------
# | 宝塔Linux面板
# +-------------------------------------------------------------------
# | Copyright (c) 2015-2020 宝塔软件(https://www.bt.cn) All rights reserved.
# +-------------------------------------------------------------------
# | Author: baozi <baozi@bt.cn>
# | Author: baozi
# +-------------------------------------------------------------------
from typing import List, Dict, Tuple
import public, panelPush
try:
from BTPanel import cache
except:
from cachelib import SimpleCache
cache = SimpleCache()
class metaclass(type):
def __new__(cls, name, *args, **kwargs):
push_cls = super().__new__(cls, name, *args, **kwargs)
if name == "base_push":
return push_cls
else:
push_cls.all_push_model.append(push_cls)
return push_cls
class base_push(metaclass=metaclass):
all_push_model = []
# 版本信息 目前无作用
def get_version_info(self, get=None):
raise NotImplementedError
# 格式化返回执行周期， 目前无作用
def get_push_cycle(self, data: dict):
return data
# 获取模块推送参数
def get_module_config(self, get: public.dict_obj):
raise NotImplementedError
# 获取模块配置项
def get_push_config(self, get: public.dict_obj):
# 其实就是配置信息，没有也会从全局配置文件push.json中读取
raise NotImplementedError
# 写入推送配置文件
def set_push_config(self, get: public.dict_obj):
raise NotImplementedError
# 删除推送配置
def del_push_config(self, get: public.dict_obj):
# 从配置中删除信息，并做一些您想做的事，如记日志
raise NotImplementedError
# 无意义？？？
def get_total(self):
return True
# 检查并获取推送消息，返回空时，不做推送, 传入的data是配置项
def get_push_data(self, data, total):
# data 内容
# index : 时间戳 time.time()
# 消息 以类型为key， 以内容为value， 内容中包含title 和msg
# push_keys： 列表，发送了信息的推送任务的id，用来验证推送任务次数（） 意义不大
raise NotImplementedError
# 返回这个告警类型可以设置的任务模板，返回形式为列表套用字典
def get_task_template(self) -> Tuple[str, List[Dict]]:
# 返回这个告警类型可以设置的任务模板，返回形式为列表套用字典
raise NotImplementedError
# 返回到前端信息的钩子, 默认为返回传入信息（即：当前设置的任务的信息）
def get_view_msg(self, task_id: str, task_data: dict) -> dict:
return task_data
class WxAccountMsgBase:
@classmethod
def new_msg(cls):
return cls()
def set_ip_address(self, server_ip, local_ip):
pass
def to_send_data(self):
return "", {}
class WxAccountMsg(WxAccountMsgBase):
def __init__(self):
self.ip_address: str = ""
self.thing_type: str = ""
self.msg: str = ""
self.next_msg: str = ""
def set_ip_address(self, server_ip, local_ip):
self.ip_address = "{}({})".format(server_ip, local_ip)
if len(self.ip_address) > 32:
self.ip_address = self.ip_address[:29] + "..."
def to_send_data(self):
res = {
"first": {},
"keyword1": {
"value": self.ip_address,
},
"keyword2": {
"value": self.thing_type,
},
"keyword3": {
"value": self.msg,
}
}
if self.next_msg != "":
res["keyword4"] = {"value": self.next_msg}
return "", res
class WxAccountLoginMsg(WxAccountMsgBase):
tid = "RJNG8dBZ5Tb9EK6j6gOlcAgGs2Fjn5Fb07vZIsYg1P4"
def __init__(self):
self.login_name: str = ""
self.login_ip: str = ""
self.login_type: str = ""
self.address: str = ""
self._server_name: str = ""
self._server_ip = ""
def set_ip_address(self, server_ip, local_ip: str):
if local_ip != "127.0.0.1" and local_ip.startswith("192.168"):
self._server_ip = local_ip
else:
self._server_ip = server_ip
if self._server_name == "":
self._server_name = "服务器IP{}".format(server_ip)
def _get_server_name(self):
data = public.GetConfigValue("title") # 若获得别名，则使用别名
if data == "宝塔Linux面板":
self._server_name = "宝塔面板({})".format(".".join(self._server_ip.split(".")[2:]))
elif data != "":
self._server_name = data
def to_send_data(self):
self._get_server_name()
if self.address.startswith(">归属地："):
self.address = self.address[5:]
if self.address == "":
self.address = "未知的归属地"
if not public.is_ipv4(self.login_ip):
self.login_ip = "ipv6-can not show"
res = {
"thing10": {
"value": self._server_name,
},
"character_string9": {
"value": self.login_ip,
},
"thing7": {
"value": self.login_type,
},
"thing11": {
"value": self.address,
},
"thing2": {
"value": self.login_name,
}
}
return self.tid, res
|