Gecko [ www.jdcpatala.org ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
__init__.py	17.01 KB	-rw-r--r--
base_task.py	9.72 KB	-rw-r--r--
compatible.py	787 B	-rw-r--r--
database_push.py	9.24 KB	-rw-r--r--
database_push_template.json	1.55 KB	-rw-r--r--
domain_blcheck_push.py	5.62 KB	-rw-r--r--
domain_blcheck_push_template.j...	1.82 KB	-rw-r--r--
load_push.py	10.69 KB	-rw-r--r--
manager.py	17.54 KB	-rw-r--r--
mods.py	12.87 KB	-rw-r--r--
rsync_push.py	10.03 KB	-rw-r--r--
send_tool.py	3.29 KB	-rw-r--r--
site_push.py	55.03 KB	-rw-r--r--
site_push_template.json	7.78 KB	-rw-r--r--
ssl_push.py	13.28 KB	-rw-r--r--
ssl_push_template.json	2.27 KB	-rw-r--r--
system.py	17.35 KB	-rw-r--r--
system_push.py	15.3 KB	-rw-r--r--
system_push_template.json	5.5 KB	-rw-r--r--
task_manager_push.py	17.17 KB	-rw-r--r--
tool.py	2.24 KB	-rw-r--r--
util.py	3.31 KB	-rw-r--r--

Code Editor : send_tool.py

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

${this.title}

Code Editor : send_tool.py