mirror of
https://github.com/aaPanel/aaPanel.git
synced 2026-08-17 21:25:47 +02:00
Jack
a24f2e23c6
1. Added Mail Marketing Automation trigger tasks 2. Added Mail Marketing Groups Import, Export, Merge 3. Added Mail Marketing Subscribers to support paste import 4. Added Mail Marketing Template Import Export Duplicate 5. Added Mail server add timed automatic reply 6. Added HTTPS Protection function for Website (disables automatic HTTP to HTTPS redirection when enabled) 7. Optimize Website Interface Button Integration 8. Optimize the response speed of WP Toolkit interface
134 lines
3.3 KiB
Python
134 lines
3.3 KiB
Python
import ipaddress
|
||
import re
|
||
|
||
from .util import get_config_value
|
||
|
||
|
||
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.thing_type: str = ""
|
||
self.login_type: str = ""
|
||
self.address: str = ""
|
||
self._server_name: str = ""
|
||
|
||
def set_ip_address(self, server_ip, local_ip):
|
||
if self._server_name == "":
|
||
self._server_name = "服务器IP{}".format(server_ip)
|
||
|
||
def _get_server_name(self):
|
||
data = get_config_value("title") # 若获得别名，则使用别名
|
||
if data != "":
|
||
self._server_name = data
|
||
|
||
def to_send_data(self):
|
||
self._get_server_name()
|
||
if self.address.startswith(">Place of Ownership:"):
|
||
self.address = self.address[5:]
|
||
if self.address == "":
|
||
self.address = "Unknown place of ownership"
|
||
|
||
if not _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
|
||
|
||
|
||
# 处理短信告警信息的不规范问题
|
||
def sms_msg_normalize(sm_args: dict) -> dict:
|
||
for key, val in sm_args.items():
|
||
sm_args[key] = _norm_sms_push_argv(str(val))
|
||
return sm_args
|
||
|
||
|
||
def _norm_sms_push_argv(data):
|
||
"""
|
||
@处理短信参数，否则会被拦截
|
||
"""
|
||
if _is_ipv4(data):
|
||
tmp1 = data.split('.')
|
||
return '{}_***_***_{}'.format(tmp1[0], tmp1[3])
|
||
|
||
data = data.replace(".", "_").replace("+", "＋")
|
||
return data
|
||
|
||
|
||
def _is_ipv4(data: str) -> bool:
|
||
try:
|
||
ipaddress.IPv4Address(data)
|
||
except:
|
||
return False
|
||
return True
|
||
|
||
|
||
def _is_domain(domain):
|
||
rep_domain = re.compile(r"^([\w\-*]{1,100}\.){1,10}([\w\-]{1,24}|[\w\-]{1,24}\.[\w\-]{1,24})$")
|
||
if rep_domain.match(domain):
|
||
return True
|
||
return False
|