Spaces:

SZFSPC
/

apple

Paused

apple / bt-source /panel /class /push /database_push.py

feat: deploy to hf space | model=mistral-21461b

2bfa191 verified 5 months ago

14 kB

# coding: utf-8
# +-------------------------------------------------------------------
# | 宝塔Windows面板
# +-------------------------------------------------------------------
# | Copyright (c) 2015-2020 宝塔软件(https://www.bt.cn) All rights reserved.
# +-------------------------------------------------------------------
# | Author: baozi <1191604998@qq.com>
# +-------------------------------------------------------------------

import sys
import os
import json
import datetime

class_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# /www/server/panel/class
bt_panel_path = os.path.dirname(class_path)
sys.path.insert(0, class_path)
sys.path.insert(0, bt_panel_path)
import public
import panelPush
from push.base_push import base_push, WxAccountMsg
from panel_msg.collector import DatabasePushMsgCollect
import db_mysql
import panelMysql


class database_push(base_push):
    __TEMPLATE_PATH = os.path.join(public.get_panel_path(), "class/push/scripts/database_push.json")
    __push_model = ['dingding', 'weixin', 'mail', 'sms', 'wx_account', 'feishu']
    __push_conf = os.path.join(public.get_panel_path(), "class/push/push.json")

    def __init__(self):
        self.__push = panelPush.panelPush()


    def get_version_info(self, get=None):
        data = {
            'ps': '宝塔系统告警',
            'version': '1.0',
            'date': '2023-07-16',
            'author': '宝塔',
            'help': 'http://www.bt.cn/bbs'
        }
        return data

    # 格式化返回执行周期， 目前无作用
    def get_push_cycle(self, data: dict):
        return data

    # 获取模块推送参数
    def get_module_config(self, get: public.dict_obj):
        data = []
        item = self.__push.format_push_data(
            push=["mail", 'dingding', 'weixin', "feishu", "wx_account"],
            project='system', type='')
        item['cycle'] = 15
        item['title'] = '磁盘容量限额'
        data.append(item)
        return data

    # 获取模块配置项
    def get_push_config(self, get: public.dict_obj):
        task_id = get.id
        push_list = self.__push._get_conf()

        if task_id not in push_list["database_push"]:
            res_data = public.returnMsg(False, '未找到指定配置.')
            res_data['code'] = 100
            return res_data
        result = push_list["database_push"][task_id]
        return result

    # 清除推送次数
    @staticmethod
    def clear_push_count(task_id):
        try:
            tip_file = os.path.join(public.get_panel_path(), "data/push/tips", task_id)
            if os.path.exists(tip_file):
                os.remove(tip_file)
        except:
            pass

    # 写入推送配置文件
    def set_push_config(self, get: public.dict_obj):
        id = get.id
        module = get.name
        pdata = json.loads(get.data)
        data = self.__push._get_conf()
        if module not in data: data[module] = {}

        self.clear_push_count(id)

        if pdata["type"] == "mysql_replicate_status":
            if not pdata["project"]:
                return public.returnMsg(False, "请选择告警的从库！")
            data[module][pdata["project"]] = pdata
        elif pdata["type"] == "mysql_pwd_endtime":
            if not pdata["project"]:
                return public.returnMsg(False, "请选择告警的MySQL用户！")
            data[module][id] = pdata
        else:
            data[module][id] = pdata

        return data

    # 删除推送配置
    def del_push_config(self, get: public.dict_obj):
        # 从配置中删除信息，并做一些您想做的事，如记日志
        task_id = get.id
        module = get.name

        data = self.__push._get_conf()
        if str(task_id).strip() in data[module]:
            del data[module][task_id]
        public.writeFile(self.__push_conf, json.dumps(data))
        return public.returnMsg(True, "删除成功!")

    def get_total(self):
        return True

    # 检查并获取推送消息，返回空时，不做推送, 传入的data是配置项
    def get_push_data(self, data, total):

        current_time = datetime.datetime.now()
        stime = current_time.timestamp()
        result = {'index': stime, 'push_keys': []}

        if data['type'] in ['mysql_pwd_endtime']:
            if stime < data['index'] + 86400:
                return public.returnMsg(False, "一天推送一次，跳过.")

            sid = data["project"][0]
            username = data["project"][1]
            host = data["project"][2]

            if sid == 0:
                try:
                    db_port = int(panelMysql.panelMysql().query("show global variables like 'port'")[0][1])
                    if db_port == 0:
                        db_port = 3306
                except:
                    db_port = 3306
                conn_config = {
                    "db_host": "localhost",
                    "db_port": db_port,
                    "db_user": "root",
                    "db_password": public.M("config").where("id=?", (1,)).getField("mysql_root"),
                    "ps": "本地服务器",
                }
            else:
                conn_config = public.M("database_servers").where("id=? AND LOWER(db_type)=LOWER('mysql')", (sid,)).find()
            if not conn_config:
                return result

            mysql_obj = db_mysql.panelMysql().set_host(conn_config["db_host"], conn_config["db_port"], None, conn_config["db_user"], conn_config["db_password"])
            if isinstance(mysql_obj, bool):
                return result

            data_list = mysql_obj.query("SELECT password_last_changed FROM mysql.user WHERE user='{}' AND host='{}';".format(username, host))
            if not isinstance(data_list, list) or not data_list:
                return result

            last_time = data_list[0][0]
            expire_time = last_time + datetime.timedelta(days=data["cycle"])

            if current_time > expire_time:
                s_list = [
                    ">告警类型：MySQL密码即将到期",
                    ">告警内容：{} {}@{} 密码过期时间<font color=#ff0000>{} 天</font>".format(conn_config["ps"], username, host, expire_time.strftime("%Y-%m-%d %H:%M:%S"))
                ]
                msg = DatabasePushMsgCollect.mysql_pwd_endtime(s_list, data["title"])
                for m_module in data['module'].split(','):
                    if m_module == 'sms':
                        continue
                    elif m_module == 'wx_account':
                        result[m_module] = ToWechatAccountMsg.mysql_pwd_endtime("MySQL密码已到期")
                        continue

                    sdata = public.get_push_info('MySQL密码到期提醒', s_list)
                    sdata["push_type"] = 'MySQL密码到期'
                    result[m_module] = sdata

            elif data["cycle"] > 3 and current_time > last_time + datetime.timedelta(days=data["cycle"] - 3):
                expire_day = (current_time - (last_time + datetime.timedelta(days=data["cycle"] - 3))).days
                s_list = [
                    ">告警类型：MySQL密码即将到期",
                    ">告警内容：{} {}@{} 密码剩余天数:<font color=#ff0000>{} 天</font>".format(conn_config["ps"], username, host,expire_day)
                ]
                msg = DatabasePushMsgCollect.mysql_pwd_endtime(s_list, data["title"])
                for m_module in data['module'].split(','):
                    if m_module == 'sms':
                        continue
                    elif m_module == 'wx_account':
                        result[m_module] = ToWechatAccountMsg.mysql_pwd_endtime("MySQL密码{}天后到期".format(expire_day))
                        continue

                    sdata = public.get_push_info('MySQL密码到期提醒', s_list)
                    sdata["push_type"] = 'MySQL密码到期'
                    result[m_module] = sdata
            return result

        # MySQL 主从复制异常告警
        elif data['type'] in ['mysql_replicate_status']:
            if stime < data["index"] + data["interval"]:
                return public.returnMsg(False, "未达到间隔时间，跳过.")
            
            import PluginLoader

            args = public.dict_obj()
            args.slave_ip = data["project"]
            res = PluginLoader.plugin_run("mysql_replicate", "get_replicate_status", args)
            if res.get("status", False) is False:
                return public.returnMsg(False, "获取主从信息失败！")

            if len(res.get("data")) == 0:
                s_list = [">告警类型：MySQL主从复制异常告警",
                          ">告警内容：<font color=#ff0000>从库 {} 主从复制已停止，请尽快登录面板查看详情</font>".format(data["project"])]
                msg = DatabasePushMsgCollect.mysql_replicate_status(s_list, data["title"])
                for m_module in data['module'].split(','):
                    if m_module == 'sms':
                        continue
                    elif m_module == 'wx_account':
                        result[m_module] = ToWechatAccountMsg.mysql_replicate_status("MySQL主从复制异常告警")
                        continue

                    sdata = public.get_push_info('MySQL主从复制异常告警', s_list)
                    sdata["push_type"] = 'MySQL主从复制异常告警'
                    result[m_module] = sdata
                    return result

            io_status = False
            sql_status = False
            for item in res.get("data"):
                if item["name"] == "Slave_IO_Running" and item["value"] == "Yes":
                    io_status = True
                if item["name"] == "Slave_SQL_Running" and item["value"] == "Yes":
                    sql_status = True
                if io_status is True and sql_status is True:
                    break

            if io_status is False or sql_status is False:
                repair_txt = "请尽快登录面板查看详情"
                if data["count"] == 1: # 自动修复
                    PluginLoader.plugin_run("mysql_replicate", "repair_replicate", args)
                    repair_txt = "，正在尝试修复"

                s_list = [">告警类型：MySQL主从复制异常告警",
                          ">告警内容：<font color=#ff0000>从库 {} 主从复制发生异常{}</font>".format(data["project"], repair_txt)]
                msg = DatabasePushMsgCollect.mysql_replicate_status(s_list, data["title"])
                for m_module in data['module'].split(','):
                    if m_module == 'sms':
                        continue
                    elif m_module == 'wx_account':
                        slave_ip = data["project"][:3] if len(data["project"]) >= 3 else data["project"]
                        result[m_module] = ToWechatAccountMsg.mysql_replicate_status("从库 {} 主从复制发生异常{}".format(slave_ip, repair_txt))
                        continue
    
                    sdata = public.get_push_info('MySQL主从复制异常告警', s_list)
                    sdata["push_type"] = 'MySQL主从复制异常告警'
                    result[m_module] = sdata
                return result

        return public.returnMsg(False, "未达到阈值，跳过.")

    # 返回到前端信息的钩子, 默认为返回传入信息（即：当前设置的任务的信息）
    @staticmethod
    def get_view_msg(task_id, task_data):
        task_data["tid"] = view_msg.get_tid(task_data)
        task_data["view_msg"] = view_msg.get_msg_by_type(task_data)
        return task_data

    def get_task_template(self):
        res = public.readFile(self.__TEMPLATE_PATH)
        if not res:
            return "数据库告警设置", [{}]
        res_data = json.loads(res)

        res_data["mysql_replicate_status"]["field"][0]["items"] = self.__get_mysql_replicate()
        return "数据库警设置", [v for v in res_data.values()]

    @classmethod
    def __get_mysql_replicate(cls) -> list:
        slave_list = []
        mysql_replicate_path = os.path.join(public.get_plugin_path(), "mysql_replicate", "config.json")
        if os.path.isfile(mysql_replicate_path):
            conf = public.readFile(mysql_replicate_path)
            try:
                conf = json.loads(conf)
                slave_list = [{"title": slave_ip, "value": slave_ip} for slave_ip  in conf["slave"].keys()]
            except:
                pass
        return slave_list


class ViewMsgFormat(object):
    _FORMAT = {
        "mysql_pwd_endtime": (
            lambda x: "<span>剩余时间小于{}天{}</span>".format(
                x.get("cycle"),
                "(如未处理，次日会重新发送1次，持续%d天)" % x.get("push_count", 0) if x.get("push_count", 0) else ""
            )
        ),
        "mysql_replicate_status": (lambda x: "<span>MySQL主从复制异常告警</span>".format()),
    }

    _TID = {
        "mysql_pwd_endtime": "database_push@0",
        "mysql_replicate_status": "database_push@1",
    }

    def get_msg_by_type(self, data):
        return self._FORMAT[data["type"]](data)

    def get_tid(self, data):
        return self._TID[data["type"]]


view_msg = ViewMsgFormat()


class ToWechatAccountMsg:
    @staticmethod
    def mysql_pwd_endtime(message: str):
        msg = WxAccountMsg.new_msg()
        msg.thing_type = "MySQL数据库密码到期提醒"
        msg.msg = message
        msg.next_msg = "请登录面板，在[数据库]中修改密码"
        return msg

    @staticmethod
    def mysql_replicate_status(message: str):
        msg = WxAccountMsg.new_msg()
        msg.thing_type = "MySQL主从复制异常告警"
        msg.msg = message
        msg.next_msg = "请登录面板，在[软件商店-MySQL主从复制(重构版)]中查看"
        return msg

	# coding: utf-8
	# +-------------------------------------------------------------------
	# \| 宝塔Windows面板
	# +-------------------------------------------------------------------
	# \| Copyright (c) 2015-2020 宝塔软件(https://www.bt.cn) All rights reserved.
	# +-------------------------------------------------------------------
	# \| Author: baozi <1191604998@qq.com>
	# +-------------------------------------------------------------------

	import sys
	import os
	import json
	import datetime

	class_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
	# /www/server/panel/class
	bt_panel_path = os.path.dirname(class_path)
	sys.path.insert(0, class_path)
	sys.path.insert(0, bt_panel_path)
	import public
	import panelPush
	from push.base_push import base_push, WxAccountMsg
	from panel_msg.collector import DatabasePushMsgCollect
	import db_mysql
	import panelMysql


	class database_push(base_push):
	__TEMPLATE_PATH = os.path.join(public.get_panel_path(), "class/push/scripts/database_push.json")
	__push_model = ['dingding', 'weixin', 'mail', 'sms', 'wx_account', 'feishu']
	__push_conf = os.path.join(public.get_panel_path(), "class/push/push.json")

	def __init__(self):
	self.__push = panelPush.panelPush()


	def get_version_info(self, get=None):
	data = {
	'ps': '宝塔系统告警',
	'version': '1.0',
	'date': '2023-07-16',
	'author': '宝塔',
	'help': 'http://www.bt.cn/bbs'
	}
	return data

	# 格式化返回执行周期，目前无作用
	def get_push_cycle(self, data: dict):
	return data

	# 获取模块推送参数
	def get_module_config(self, get: public.dict_obj):
	data = []
	item = self.__push.format_push_data(
	push=["mail", 'dingding', 'weixin', "feishu", "wx_account"],
	project='system', type='')
	item['cycle'] = 15
	item['title'] = '磁盘容量限额'
	data.append(item)
	return data

	# 获取模块配置项
	def get_push_config(self, get: public.dict_obj):
	task_id = get.id
	push_list = self.__push._get_conf()

	if task_id not in push_list["database_push"]:
	res_data = public.returnMsg(False, '未找到指定配置.')
	res_data['code'] = 100
	return res_data
	result = push_list["database_push"][task_id]
	return result

	# 清除推送次数
	@staticmethod
	def clear_push_count(task_id):
	try:
	tip_file = os.path.join(public.get_panel_path(), "data/push/tips", task_id)
	if os.path.exists(tip_file):
	os.remove(tip_file)
	except:
	pass

	# 写入推送配置文件
	def set_push_config(self, get: public.dict_obj):
	id = get.id
	module = get.name
	pdata = json.loads(get.data)
	data = self.__push._get_conf()
	if module not in data: data[module] = {}

	self.clear_push_count(id)

	if pdata["type"] == "mysql_replicate_status":
	if not pdata["project"]:
	return public.returnMsg(False, "请选择告警的从库！")
	data[module][pdata["project"]] = pdata
	elif pdata["type"] == "mysql_pwd_endtime":
	if not pdata["project"]:
	return public.returnMsg(False, "请选择告警的MySQL用户！")
	data[module][id] = pdata
	else:
	data[module][id] = pdata

	return data

	# 删除推送配置
	def del_push_config(self, get: public.dict_obj):
	# 从配置中删除信息，并做一些您想做的事，如记日志
	task_id = get.id
	module = get.name

	data = self.__push._get_conf()
	if str(task_id).strip() in data[module]:
	del data[module][task_id]
	public.writeFile(self.__push_conf, json.dumps(data))
	return public.returnMsg(True, "删除成功!")

	def get_total(self):
	return True

	# 检查并获取推送消息，返回空时，不做推送, 传入的data是配置项
	def get_push_data(self, data, total):

	current_time = datetime.datetime.now()
	stime = current_time.timestamp()
	result = {'index': stime, 'push_keys': []}

	if data['type'] in ['mysql_pwd_endtime']:
	if stime < data['index'] + 86400:
	return public.returnMsg(False, "一天推送一次，跳过.")

	sid = data["project"][0]
	username = data["project"][1]
	host = data["project"][2]

	if sid == 0:
	try:
	db_port = int(panelMysql.panelMysql().query("show global variables like 'port'")[0][1])
	if db_port == 0:
	db_port = 3306
	except:
	db_port = 3306
	conn_config = {
	"db_host": "localhost",
	"db_port": db_port,
	"db_user": "root",
	"db_password": public.M("config").where("id=?", (1,)).getField("mysql_root"),
	"ps": "本地服务器",
	}
	else:
	conn_config = public.M("database_servers").where("id=? AND LOWER(db_type)=LOWER('mysql')", (sid,)).find()
	if not conn_config:
	return result

	mysql_obj = db_mysql.panelMysql().set_host(conn_config["db_host"], conn_config["db_port"], None, conn_config["db_user"], conn_config["db_password"])
	if isinstance(mysql_obj, bool):
	return result

	data_list = mysql_obj.query("SELECT password_last_changed FROM mysql.user WHERE user='{}' AND host='{}';".format(username, host))
	if not isinstance(data_list, list) or not data_list:
	return result

	last_time = data_list[0][0]
	expire_time = last_time + datetime.timedelta(days=data["cycle"])

	if current_time > expire_time:
	s_list = [
	">告警类型：MySQL密码即将到期",
	">告警内容：{} {}@{} 密码过期时间<font color=#ff0000>{} 天</font>".format(conn_config["ps"], username, host, expire_time.strftime("%Y-%m-%d %H:%M:%S"))
	]
	msg = DatabasePushMsgCollect.mysql_pwd_endtime(s_list, data["title"])
	for m_module in data['module'].split(','):
	if m_module == 'sms':
	continue
	elif m_module == 'wx_account':
	result[m_module] = ToWechatAccountMsg.mysql_pwd_endtime("MySQL密码已到期")
	continue

	sdata = public.get_push_info('MySQL密码到期提醒', s_list)
	sdata["push_type"] = 'MySQL密码到期'
	result[m_module] = sdata

	elif data["cycle"] > 3 and current_time > last_time + datetime.timedelta(days=data["cycle"] - 3):
	expire_day = (current_time - (last_time + datetime.timedelta(days=data["cycle"] - 3))).days
	s_list = [
	">告警类型：MySQL密码即将到期",
	">告警内容：{} {}@{} 密码剩余天数:<font color=#ff0000>{} 天</font>".format(conn_config["ps"], username, host,expire_day)
	]
	msg = DatabasePushMsgCollect.mysql_pwd_endtime(s_list, data["title"])
	for m_module in data['module'].split(','):
	if m_module == 'sms':
	continue
	elif m_module == 'wx_account':
	result[m_module] = ToWechatAccountMsg.mysql_pwd_endtime("MySQL密码{}天后到期".format(expire_day))
	continue

	sdata = public.get_push_info('MySQL密码到期提醒', s_list)
	sdata["push_type"] = 'MySQL密码到期'
	result[m_module] = sdata
	return result

	# MySQL 主从复制异常告警
	elif data['type'] in ['mysql_replicate_status']:
	if stime < data["index"] + data["interval"]:
	return public.returnMsg(False, "未达到间隔时间，跳过.")

	import PluginLoader

	args = public.dict_obj()
	args.slave_ip = data["project"]
	res = PluginLoader.plugin_run("mysql_replicate", "get_replicate_status", args)
	if res.get("status", False) is False:
	return public.returnMsg(False, "获取主从信息失败！")

	if len(res.get("data")) == 0:
	s_list = [">告警类型：MySQL主从复制异常告警",
	">告警内容：<font color=#ff0000>从库 {} 主从复制已停止，请尽快登录面板查看详情</font>".format(data["project"])]
	msg = DatabasePushMsgCollect.mysql_replicate_status(s_list, data["title"])
	for m_module in data['module'].split(','):
	if m_module == 'sms':
	continue
	elif m_module == 'wx_account':
	result[m_module] = ToWechatAccountMsg.mysql_replicate_status("MySQL主从复制异常告警")
	continue

	sdata = public.get_push_info('MySQL主从复制异常告警', s_list)
	sdata["push_type"] = 'MySQL主从复制异常告警'
	result[m_module] = sdata
	return result

	io_status = False
	sql_status = False
	for item in res.get("data"):
	if item["name"] == "Slave_IO_Running" and item["value"] == "Yes":
	io_status = True
	if item["name"] == "Slave_SQL_Running" and item["value"] == "Yes":
	sql_status = True
	if io_status is True and sql_status is True:
	break

	if io_status is False or sql_status is False:
	repair_txt = "请尽快登录面板查看详情"
	if data["count"] == 1: # 自动修复
	PluginLoader.plugin_run("mysql_replicate", "repair_replicate", args)
	repair_txt = "，正在尝试修复"

	s_list = [">告警类型：MySQL主从复制异常告警",
	">告警内容：<font color=#ff0000>从库 {} 主从复制发生异常{}</font>".format(data["project"], repair_txt)]
	msg = DatabasePushMsgCollect.mysql_replicate_status(s_list, data["title"])
	for m_module in data['module'].split(','):
	if m_module == 'sms':
	continue
	elif m_module == 'wx_account':
	slave_ip = data["project"][:3] if len(data["project"]) >= 3 else data["project"]
	result[m_module] = ToWechatAccountMsg.mysql_replicate_status("从库 {} 主从复制发生异常{}".format(slave_ip, repair_txt))
	continue

	sdata = public.get_push_info('MySQL主从复制异常告警', s_list)
	sdata["push_type"] = 'MySQL主从复制异常告警'
	result[m_module] = sdata
	return result

	return public.returnMsg(False, "未达到阈值，跳过.")

	# 返回到前端信息的钩子, 默认为返回传入信息（即：当前设置的任务的信息）
	@staticmethod
	def get_view_msg(task_id, task_data):
	task_data["tid"] = view_msg.get_tid(task_data)
	task_data["view_msg"] = view_msg.get_msg_by_type(task_data)
	return task_data

	def get_task_template(self):
	res = public.readFile(self.__TEMPLATE_PATH)
	if not res:
	return "数据库告警设置", [{}]
	res_data = json.loads(res)

	res_data["mysql_replicate_status"]["field"][0]["items"] = self.__get_mysql_replicate()
	return "数据库警设置", [v for v in res_data.values()]

	@classmethod
	def __get_mysql_replicate(cls) -> list:
	slave_list = []
	mysql_replicate_path = os.path.join(public.get_plugin_path(), "mysql_replicate", "config.json")
	if os.path.isfile(mysql_replicate_path):
	conf = public.readFile(mysql_replicate_path)
	try:
	conf = json.loads(conf)
	slave_list = [{"title": slave_ip, "value": slave_ip} for slave_ip in conf["slave"].keys()]
	except:
	pass
	return slave_list


	class ViewMsgFormat(object):
	_FORMAT = {
	"mysql_pwd_endtime": (
	lambda x: "<span>剩余时间小于{}天{}</span>".format(
	x.get("cycle"),
	"(如未处理，次日会重新发送1次，持续%d天)" % x.get("push_count", 0) if x.get("push_count", 0) else ""
	)
	),
	"mysql_replicate_status": (lambda x: "<span>MySQL主从复制异常告警</span>".format()),
	}

	_TID = {
	"mysql_pwd_endtime": "database_push@0",
	"mysql_replicate_status": "database_push@1",
	}

	def get_msg_by_type(self, data):
	return self._FORMAT[data["type"]](data)

	def get_tid(self, data):
	return self._TID[data["type"]]


	view_msg = ViewMsgFormat()


	class ToWechatAccountMsg:
	@staticmethod
	def mysql_pwd_endtime(message: str):
	msg = WxAccountMsg.new_msg()
	msg.thing_type = "MySQL数据库密码到期提醒"
	msg.msg = message
	msg.next_msg = "请登录面板，在[数据库]中修改密码"
	return msg

	@staticmethod
	def mysql_replicate_status(message: str):
	msg = WxAccountMsg.new_msg()
	msg.thing_type = "MySQL主从复制异常告警"
	msg.msg = message
	msg.next_msg = "请登录面板，在[软件商店-MySQL主从复制(重构版)]中查看"
	return msg