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 : domain_blcheck_push.py

import glob
import hashlib
import json
import os
import re
import sys
import time

import psutil
from datetime import datetime
from importlib import import_module
from typing import Tuple, Union, Optional, List

from .send_tool import WxAccountMsg, WxAccountLoginMsg
from .base_task import BaseTask
from .mods import PUSH_DATA_PATH, TaskConfig, SenderConfig, PANEL_PATH
from .util import read_file, DB, write_file, check_site_status,GET_CLASS, ExecShell, get_config_value, public_get_cache_func, \
    public_set_cache_func, get_network_ip, public_get_user_info, public_http_post, panel_version
from mod.base.web_conf import RealSSLManger

# 邮局域名进入黑名单
class MailDomainBlcheck(BaseTask):
    push_tip_file = "/www/server/panel/data/mail_domain_blcheck_send_type.pl"

def __init__(self):
        super().__init__()
        self.source_name = "mail_domain_black"
        self.template_name = "Your IP is on the email blacklist"
        self.title = "Your IP is on the email blacklist"

def check_task_data(self, task_data: dict) -> Union[dict, str]:
        return {}

def get_keyword(self, task_data: dict) -> str:
        return "mail_domain_black"

def get_push_data(self, task_id: str, task_data: dict) -> Optional[dict]:
        return None

def filter_template(self, template) -> dict:
        return template

def to_sms_msg(self, push_data: dict, push_public_data: dict) -> Tuple[str, dict]:
        return "", {}

def task_config_update_hook(self, task: dict) -> None:

sender = task["sender"]
        if len(sender) > 0:
            send_id = sender[0]
        else:
            return

sender_data = SenderConfig().get_by_id(send_id)
        if sender_data:
            write_file(self.push_tip_file, sender_data["sender_type"])

def task_config_create_hook(self, task: dict) -> None:
        return self.task_config_update_hook(task)

def task_config_remove_hook(self, task: dict) -> None:
        if os.path.exists(self.push_tip_file):
            os.remove(self.push_tip_file)

# 邮局服务异常告警
class MailServerDown(BaseTask):
    push_tip_file = "/www/server/panel/data/mail_server_down_send_type.pl"

def __init__(self):
        super().__init__()
        self.source_name = "mail_server_status"
        self.template_name = "Your Mail Service is down"
        self.title = "Your Mail Service is down"

def check_task_data(self, task_data: dict) -> Union[dict, str]:
        return {}

def get_keyword(self, task_data: dict) -> str:
        return "mail_server_status"

def get_push_data(self, task_id: str, task_data: dict) -> Optional[dict]:
        return None

def filter_template(self, template) -> dict:
        return template

def to_sms_msg(self, push_data: dict, push_public_data: dict) -> Tuple[str, dict]:
        return "", {}

def task_config_update_hook(self, task: dict) -> None:

sender = task["sender"]
        if len(sender) > 0:
            send_id = sender[0]
        else:
            return

sender_data = SenderConfig().get_by_id(send_id)
        if sender_data:
            write_file(self.push_tip_file, sender_data["sender_type"])

def task_config_create_hook(self, task: dict) -> None:
        return self.task_config_update_hook(task)

def task_config_remove_hook(self, task: dict) -> None:
        if os.path.exists(self.push_tip_file):
            os.remove(self.push_tip_file)

# 邮局服务异常告警
class CertExpiry(BaseTask):
    push_tip_file = "/www/server/panel/data/mail_cert_expiry_send_type.pl"

def __init__(self):
        super().__init__()
        self.source_name = "mail_server_certificate_expired"
        self.template_name = "Mail Server certificate expired"
        self.title = "Your Mail Server certificate expired"

def check_task_data(self, task_data: dict) -> Union[dict, str]:
        return {}

def get_keyword(self, task_data: dict) -> str:
        return "mail_server_certificate_expired"

def get_push_data(self, task_id: str, task_data: dict) -> Optional[dict]:
        return None

def filter_template(self, template) -> dict:
        return template

def to_sms_msg(self, push_data: dict, push_public_data: dict) -> Tuple[str, dict]:
        return "", {}

def task_config_update_hook(self, task: dict) -> None:

sender = task["sender"]
        if len(sender) > 0:
            send_id = sender[0]
        else:
            return

sender_data = SenderConfig().get_by_id(send_id)
        if sender_data:
            write_file(self.push_tip_file, sender_data["sender_type"])

def task_config_create_hook(self, task: dict) -> None:
        return self.task_config_update_hook(task)

def task_config_remove_hook(self, task: dict) -> None:
        if os.path.exists(self.push_tip_file):
            os.remove(self.push_tip_file)

class ViewMsgFormat(object):
 _FORMAT = {
 "1": (
 lambda x: "When your MailServer domain is blacklisted, an alarm is generated"
 ),
 "2": (
 lambda x: "When your Mail Service is down, an alarm is generated"
 ),
 "3": (
 lambda x: "When your Mail Server certificate expired, an alarm is generated"
 )
 }

def get_msg(self, task: dict) -> Optional[str]:
        if task["template_id"] in ["80"]:
            return self._FORMAT["1"](task)
        if task["template_id"] in ["81"]:
            return self._FORMAT["2"](task)
        if task["template_id"] in ["82"]:
            return self._FORMAT["3"](task)
        if task["template_id"] in self._FORMAT:
            return self._FORMAT[task["template_id"]](task)
        return None

${this.title}

Code Editor : domain_blcheck_push.py