File size: 3,698 Bytes
020c337 | 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 | #coding: utf-8
#-------------------------------------------------------------------
# 宝塔Linux面板
#-------------------------------------------------------------------
# Copyright (c) 2015-2099 宝塔软件(http://bt.cn) All rights reserved.
#-------------------------------------------------------------------
# Author: baozi <baozi@bt.cn>
#-------------------------------------------------------------------
# 告警
#------------------------------
import os
import json
from importlib import import_module
from panelModel.base import panelBase
import public
class main(panelBase):
def __init__(self):
pass
def get_list(self, args):
"""
@name 获取推送列表
"""
path = '{}/class/push/scripts/'.format(public.get_panel_path())
if not os.path.exists(path):
return []
data = []
for fname in os.listdir(path):
if fname.find('.json') == -1: continue
try:
item = json.loads(public.readFile('{}/{}'.format(path,fname)))
data.append(item)
except json.JSONDecodeError:
pass
return data
@staticmethod
def get_msg_configs(get=None):
import config
data = config.config().get_msg_configs(None)
web_hook = public.init_msg("web_hook")
if web_hook is False:
return data
data["web_hook"] = {
"name": "web_hook",
"title": "WEB HOOK",
"version": "1.0",
"date": "2023-10-30",
"help": "https://www.bt.cn/bbs/thread-121791-1-1.html",
"ps": "宝塔自定义API信息通道，用于接收面板消息推送",
"setup": True,
"info": web_hook.get_version_info(),
"data": web_hook.get_config()
}
return data
@staticmethod
def set_webhook_config(get):
web_hook = import_module('.web_hook_msg', package="msg")
web_hook_obj = getattr(web_hook, "web_hook_msg", None)()
return web_hook_obj.set_config(get)
@staticmethod
def set_webhook_status(get):
name = None
if "name" in get:
name = get.name.strip()
if name == '':
name = None
status = getattr(get, "status", None)
if status is None:
return public.returnMsg(False, '没有状态参数')
if status in ('true', True, 1, "1"):
status = True
else:
status = False
web_hook = import_module('.web_hook_msg', package="msg")
web_hook_obj = getattr(web_hook, "web_hook_msg", None)()
web_hook_obj.set_status(status, name)
return public.returnMsg(True, '操作成功')
@staticmethod
def remove_hook(get):
name = None
if "name" in get:
name = get.name.strip()
if name == '':
name = None
web_hook = import_module('.web_hook_msg', package="msg")
web_hook_obj = getattr(web_hook, "web_hook_msg", None)()
web_hook_obj.del_hook_by_name(name)
return public.returnMsg(True, '操作成功')
@staticmethod
def unbind_wx_account(get):
wx_account = import_module('.wx_account_msg', package="msg")
wx_account_obj = getattr(wx_account, "wx_account_msg", None)()
return wx_account_obj.unbind()
@staticmethod
def system_push_next(get):
system_push = import_module(".system_push", package="push")
system_push_obj = getattr(system_push, "system_push", None)()
data_list, _ = getattr(system_push_obj, "_get_can_next", None)()
return data_list
|