460 lines
21 KiB
Python
460 lines
21 KiB
Python
#[title: nazha]
|
||
#[language: python]
|
||
#[service: 10086] 售后联系方式
|
||
#[disable: false] 禁用开关，true表示禁用，false表示可用
|
||
#[admin: true] 是否为管理员指令
|
||
#[rule: ^哪吒查询$]
|
||
#[rule: ^哪吒管理$]
|
||
#[cron: 0 7 * * *]
|
||
#[priority: 300] 优先级，数字越大表示优先级越高
|
||
#[platform: qq,wx,tg,tb] 适用的平台
|
||
#[open_source: true]是否开源
|
||
#[icon: 图标url]图标链接地址，请使用48像素的正方形图标，支持http和https
|
||
#[version: 1.0.0]版本号
|
||
#[public: false] 是否发布？值为true或false，不设置则上传aut云时会自动设置为true，false时上传后不显示在市场中，但是搜索能搜索到，方便开发者测试
|
||
#[price: 0.01] 上架价格
|
||
# [description: 【简介】：哪吒监控查询 支持多账号管理，可查询服务器状态、服务状态等]
|
||
# [icon: https://cdn.jsdelivr.net/gh/naiba/nezha@master/resource/static/img/nezha.png]
|
||
# =======================
|
||
|
||
|
||
import importlib
|
||
import inspect
|
||
import json
|
||
import logging
|
||
import math
|
||
from math import fabs
|
||
import os
|
||
import random
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from multiprocessing import Pool
|
||
from multiprocessing.pool import ThreadPool
|
||
from urllib.parse import urlencode
|
||
from wsgiref import headers
|
||
|
||
import requests
|
||
import aiohttp
|
||
import asyncio
|
||
|
||
import middleware
|
||
|
||
plugin_pre = 'vhook_nezha_'
|
||
|
||
class TOOL:
|
||
def __init__(self):
|
||
self.plat = ["qq", "wx", "wb", "qb", "bt", "tg"]
|
||
self.platform_arr = self.platformArr()
|
||
self.sender = middleware.Sender(middleware.getSenderID())
|
||
self.plugin_title = self.sender.getPluginName()
|
||
self.plugin_ver = self.sender.getPluginVersion()
|
||
self.plugin_name = self.sender.getPluginName()
|
||
self.msg = self.sender.getMessage().strip('"')
|
||
self.imType = "fake" if self.sender.getImtype() == "cron" else self.sender.getImtype()
|
||
self.chatId = self.sender.getChatID()
|
||
self.userId = self.sender.getUserID()
|
||
self.isAdmin = self.sender.isAdmin()
|
||
self.remark = ""
|
||
self.timeStr = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
# 安装必须依赖
|
||
self.dependencies = ['colorlog', "requests", "aiohttp"]
|
||
self.install_dependencies()
|
||
self.logger = logging.getLogger()
|
||
self.color_formatter = sys.modules['colorlog'].ColoredFormatter(
|
||
'%(log_color)s%(levelname)s: %(message)s',
|
||
log_colors={
|
||
'DEBUG': 'cyan',
|
||
'INFO': 'green',
|
||
'WARNING': 'yellow',
|
||
'ERROR': 'red',
|
||
'CRITICAL': 'red,bg_white',
|
||
}
|
||
)
|
||
|
||
def install_dependencies(self):
|
||
for dependency in self.dependencies:
|
||
try:
|
||
importlib.import_module(dependency)
|
||
except ImportError as e:
|
||
self.pushErr2Master(f"【错误】：缺少【{dependency}】模块, 正在自动安装，请稍后....")
|
||
if self.check_pip_version("pip3"):
|
||
cmd = "pip3"
|
||
else:
|
||
cmd = "pip"
|
||
result = subprocess.run([cmd, 'install', dependency], capture_output=True, text=True)
|
||
if result.returncode == 0:
|
||
self.pushErr2Master(f"依赖【{dependency}】安装完成，请重新发送指令，运行插件")
|
||
|
||
def check_pip_version(self, pip_command):
|
||
try:
|
||
result = subprocess.run([pip_command, '--version'], capture_output=True, text=True)
|
||
if result.returncode == 0:
|
||
return True
|
||
except FileNotFoundError:
|
||
pass
|
||
return False
|
||
|
||
def get_log(self, level=logging.INFO):
|
||
self.logger.setLevel(level)
|
||
console_handler = logging.StreamHandler()
|
||
console_handler.setLevel(level)
|
||
console_handler.setFormatter(self.color_formatter)
|
||
for handler in self.logger.handlers:
|
||
self.logger.removeHandler(handler)
|
||
self.logger.addHandler(console_handler)
|
||
|
||
def log_info(self, msg):
|
||
self.get_log(logging.INFO)
|
||
return self.logger.info(f"{msg}")
|
||
|
||
def log_err(self, msg):
|
||
self.get_log(logging.ERROR)
|
||
return self.logger.error(f"{msg}")
|
||
|
||
def log_warn(self, msg):
|
||
self.get_log(logging.WARN)
|
||
return self.logger.warning(f"{msg}")
|
||
|
||
def platformArr(self):
|
||
plat_arr = []
|
||
for item in self.plat:
|
||
t = {"platform": f"{plugin_pre}{item}", "imType": item}
|
||
plat_arr.append(t)
|
||
return plat_arr
|
||
|
||
def pushMsg(self, userId, chatId, imType, title, content):
|
||
now = datetime.now()
|
||
formatted_datetime = now.strftime('%Y-%m-%d %H:%M:%S')
|
||
self.remark = ""
|
||
msg = f"*********{tool.plugin_title}*********"
|
||
msg += f'\n【任务时间】：{formatted_datetime}'
|
||
msg += f"\n{content}"
|
||
middleware.push(imType, chatId, userId, title, msg)
|
||
return
|
||
|
||
def pushGroup(self, imtype, groupCode, content):
|
||
self.pushMsg(0, groupCode, imtype, "", content)
|
||
|
||
def pushErr2Master(self, content):
|
||
middleware.notifyMasters(
|
||
f"【插件】：{self.plugin_title}\n {content}",
|
||
['qq', 'tg', 'wx', 'wb', ])
|
||
|
||
def replyMsg(self, content):
|
||
return self.sender.reply(content)
|
||
|
||
class ACCOUNT:
|
||
def __init__(self):
|
||
self.method = "account"
|
||
self.plugin_title = "哪吒监控"
|
||
self.attr_arr = [
|
||
{
|
||
"title": "备注",
|
||
"key": "name",
|
||
"timeOut": 60000
|
||
}, {
|
||
"title": "面板地址",
|
||
"key": "dashboard_url",
|
||
"timeOut": 60000
|
||
}, {
|
||
"title": "用户名",
|
||
"key": "username",
|
||
"timeOut": 60000
|
||
}, {
|
||
"title": "密码",
|
||
"key": "password",
|
||
"timeOut": 60000
|
||
}
|
||
]
|
||
|
||
def setVal(self, item, title, key, timeOut):
|
||
tool.log_info(item)
|
||
tool.replyMsg(f"请输入{title}：")
|
||
value = tool.sender.listen(timeOut)
|
||
tool.log_info(f"{key}==={value}")
|
||
|
||
if value == "q" or value == "error":
|
||
tool.replyMsg("输入有误/超时，已退出！")
|
||
return False
|
||
item[key] = value
|
||
return True
|
||
|
||
def addCount(self):
|
||
tool.replyMsg("哪吒监控配置说明：\n1. 备注：用于区分不同账号\n2. 面板地址：哪吒面板的完整URL\n3. 用户名：面板登录用户名\n4. 密码：面板登录密码")
|
||
item = {}
|
||
go = True
|
||
for attr in self.attr_arr:
|
||
if not self.setVal(item, attr['title'], attr['key'], attr['timeOut']):
|
||
go = False
|
||
break
|
||
if not go:
|
||
tool.log_info("数据配置有误，已退出！")
|
||
return tool.replyMsg("数据配置有误，已退出！")
|
||
item_arr = []
|
||
item_str = middleware.bucketGet(f"{plugin_pre}{tool.imType}", tool.userId)
|
||
if item_str:
|
||
item_arr = json.loads(item_str)
|
||
item_arr.append(item)
|
||
middleware.bucketSet(f"{plugin_pre}{tool.imType}", tool.userId, json.dumps(item_arr))
|
||
tool.replyMsg("数据已配置完成，开始自动任务")
|
||
item['push_user_id'] = tool.userId
|
||
item['push_im_type'] = tool.imType
|
||
item['index'] = 1
|
||
item['total'] = 1
|
||
READ(item).run()
|
||
|
||
def editCount(self, item_arr, no):
|
||
item = item_arr[no]
|
||
content = f"请在【2分钟】内输入 序号，编辑对应的属性（q：退出）"
|
||
content += f"\n--------------------"
|
||
content += f"\n输入数字：0 删除此账号！"
|
||
content += f"\n--------------------"
|
||
for index, attr in enumerate(self.attr_arr):
|
||
if attr['key'] not in item:
|
||
item[attr['key']] = ""
|
||
content += f"\n{index + 1}.【{attr['title']}】：{item[attr['key']]}"
|
||
tool.replyMsg(content)
|
||
value = tool.sender.listen(120000)
|
||
if value == "q" or value == "error":
|
||
middleware.bucketSet(f"{plugin_pre}{tool.imType}", tool.userId, json.dumps(item_arr))
|
||
return tool.replyMsg("已退出！")
|
||
if not value.isdigit():
|
||
return self.editCount(item_arr, no)
|
||
if 0 < int(value) <= len(self.attr_arr):
|
||
attr = self.attr_arr[int(value) - 1]
|
||
if self.setVal(item, attr['title'], attr['key'], 12000):
|
||
item_arr[no] = item
|
||
self.editCount(item_arr, no)
|
||
elif value == "0":
|
||
item_arr.pop(no)
|
||
if len(item_arr) == 0:
|
||
middleware.bucketDel(f"{plugin_pre}{tool.imType}", tool.userId)
|
||
else:
|
||
middleware.bucketSet(f"{plugin_pre}{tool.imType}", tool.userId, json.dumps(item_arr))
|
||
return tool.replyMsg(f"已删除第{no + 1}个账号信息！请重新发送：哪吒管理 ！")
|
||
else:
|
||
return self.editCount(item_arr, no)
|
||
|
||
def accoount_manager(self):
|
||
item_str = middleware.bucketGet(f"{plugin_pre}{tool.imType}", tool.userId)
|
||
if not item_str or item_str == "":
|
||
tool.replyMsg(f"未配置账号，请发送：哪吒查询")
|
||
exit(1)
|
||
item_arr = json.loads(item_str)
|
||
content = f"请选择要账号查看详情：（0增加， q退出）\n"
|
||
for index, item in enumerate(item_arr):
|
||
if not hasattr(item, ""):
|
||
item['disable'] = True
|
||
disable = item['disable']
|
||
status = '已启用' if disable else '已禁用'
|
||
content = "".join([content, f"\n{index + 1}、{item['name']}"])
|
||
tool.replyMsg(content)
|
||
value = tool.sender.listen(60000)
|
||
if value == "q" or value == "error":
|
||
tool.replyMsg("输入有误，已退出！")
|
||
elif value == "0":
|
||
self.addCount()
|
||
elif value.isdigit() and 0 < int(value) <= len(item_arr):
|
||
self.editCount(item_arr, int(value) - 1)
|
||
else:
|
||
tool.replyMsg(f"输入有误,请重新发送:哪吒管理，并输入正确的序号！")
|
||
|
||
def cron_account_arr(self):
|
||
account_arr = []
|
||
for plat in tool.platformArr():
|
||
p = plat["platform"]
|
||
user_id_arr = middleware.bucketAllKeys(p)
|
||
if not user_id_arr:
|
||
continue
|
||
for index, user_id in enumerate(user_id_arr):
|
||
user_data_str = middleware.bucketGet(p, user_id)
|
||
if not user_data_str or user_id == '':
|
||
continue
|
||
user_data_arr_temp = json.loads(user_data_str)
|
||
for n, account_data in enumerate(user_data_arr_temp):
|
||
account_data["push_user_id"] = user_id
|
||
account_data["push_im_type"] = plat["imType"]
|
||
account_data['index'] = n + 1
|
||
account_data['total'] = len(user_data_arr)
|
||
account_arr.append(account_data)
|
||
return account_arr
|
||
|
||
def account_task(self, item, no):
|
||
read = READ(item)
|
||
read.run()
|
||
|
||
class READ:
|
||
def __init__(self, account):
|
||
self.name = account['name']
|
||
self.dashboard_url = account['dashboard_url']
|
||
self.username = account['username']
|
||
self.password = account['password']
|
||
self.index = account['index']
|
||
self.total = account['total']
|
||
self.push_user_id = account['push_user_id']
|
||
self.push_im_type = account['push_im_type']
|
||
self.msg = ''
|
||
self.dateStr = datetime.now().strftime('%Y-%m-%d')
|
||
self.timeStr = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||
self.servers = [] # 初始化servers属性
|
||
self.chat_id = None # 初始化chat_id属性
|
||
|
||
def pushMsg(self, msg):
|
||
content = f"【账号备注】：{self.name}"
|
||
content += f'{msg}'
|
||
tool.pushMsg(self.push_user_id, None, self.push_im_type, "", content)
|
||
|
||
def format_bytes(self, size_in_bytes):
|
||
if size_in_bytes == 0:
|
||
return "0B"
|
||
units = ['B', 'KB', 'MB', 'GB', 'TB']
|
||
power = int(math.floor(math.log(size_in_bytes, 1024)))
|
||
power = min(power, len(units) - 1)
|
||
size = size_in_bytes / (1024 ** power)
|
||
formatted_size = f"{size:.2f}{units[power]}"
|
||
return formatted_size
|
||
|
||
async def get_server_info(self):
|
||
try:
|
||
async with aiohttp.ClientSession() as session:
|
||
# 登录获取token
|
||
login_url = f"{self.dashboard_url.rstrip('/')}/api/v1/login"
|
||
login_data = {
|
||
"username": self.username,
|
||
"password": self.password
|
||
}
|
||
async with session.post(login_url, json=login_data) as resp:
|
||
login_result = await resp.json()
|
||
if not login_result.get('success'):
|
||
return False, "登录失败，请检查用户名和密码"
|
||
token = login_result['data']['token']
|
||
|
||
# 获取服务器信息
|
||
headers = {"Authorization": f"Bearer {token}"}
|
||
server_url = f"{self.dashboard_url.rstrip('/')}/api/v1/server"
|
||
async with session.get(server_url, headers=headers) as resp:
|
||
server_result = await resp.json()
|
||
if not server_result.get('success'):
|
||
return False, "获取服务器信息失败"
|
||
|
||
self.servers = server_result['data'] # 填充servers属性
|
||
# 计算总体统计信息
|
||
online_count = sum(1 for server in self.servers if server.get('state', {}).get('last_active'))
|
||
total_mem = sum(server.get('host', {}).get('mem_total', 0) for server in self.servers)
|
||
used_mem = sum(server.get('state', {}).get('mem_used', 0) for server in self.servers)
|
||
total_disk = sum(server.get('host', {}).get('disk_total', 0) for server in self.servers)
|
||
used_disk = sum(server.get('state', {}).get('disk_used', 0) for server in self.servers)
|
||
total_swap = sum(server.get('host', {}).get('swap_total', 0) for server in self.servers)
|
||
used_swap = sum(server.get('state', {}).get('swap_used', 0) for server in self.servers)
|
||
total_net_in = sum(server.get('state', {}).get('net_in_transfer', 0) for server in self.servers)
|
||
total_net_out = sum(server.get('state', {}).get('net_out_transfer', 0) for server in self.servers)
|
||
|
||
# 返回总体统计信息
|
||
self.msg += f"\n【服务器数量】：{len(self.servers)}"
|
||
self.msg += f"\n【在线服务器】：{online_count}"
|
||
self.msg += f"\n【内存使用】：{self.format_bytes(used_mem)}/{self.format_bytes(total_mem)}"
|
||
self.msg += f"\n【磁盘使用】：{self.format_bytes(used_disk)}/{self.format_bytes(total_disk)}"
|
||
self.msg += f"\n【交换使用】：{self.format_bytes(used_swap)}/{self.format_bytes(total_swap)}"
|
||
self.msg += f"\n【总流量】：↓{self.format_bytes(total_net_in)} ↑{self.format_bytes(total_net_out)}"
|
||
|
||
return True, "获取成功"
|
||
except Exception as e:
|
||
return False, f"获取信息失败：{str(e)}"
|
||
|
||
def run(self):
|
||
loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(loop)
|
||
success, message = loop.run_until_complete(self.get_server_info())
|
||
loop.close()
|
||
|
||
if not success:
|
||
self.msg += f"\n【错误】：{message}"
|
||
self.pushMsg(self.msg)
|
||
return
|
||
|
||
self.display_menu()
|
||
|
||
def display_menu(self):
|
||
# 只返回总体数据和选单
|
||
self.msg = f"\n【服务器数量】：{len(self.servers)}"
|
||
self.msg += f"\n【在线服务器】：{sum(1 for server in self.servers if server.get('state', {}).get('last_active'))}"
|
||
self.msg += f"\n【内存使用】：{self.format_bytes(sum(server.get('state', {}).get('mem_used', 0) for server in self.servers))}/{self.format_bytes(sum(server.get('host', {}).get('mem_total', 0) for server in self.servers))}"
|
||
self.msg += f"\n【磁盘使用】：{self.format_bytes(sum(server.get('state', {}).get('disk_used', 0) for server in self.servers))}/{self.format_bytes(sum(server.get('host', {}).get('disk_total', 0) for server in self.servers))}"
|
||
self.msg += f"\n【交换使用】：{self.format_bytes(sum(server.get('state', {}).get('swap_used', 0) for server in self.servers))}/{self.format_bytes(sum(server.get('host', {}).get('swap_total', 0) for server in self.servers))}"
|
||
self.msg += f"\n【总流量】：↓{self.format_bytes(sum(server.get('state', {}).get('net_in_transfer', 0) for server in self.servers))} ↑{self.format_bytes(sum(server.get('state', {}).get('net_out_transfer', 0) for server in self.servers))}"
|
||
|
||
self.msg += "\n\n【服务器编号选单】："
|
||
for index, server in enumerate(self.servers):
|
||
server_name = server.get('name', '未知')
|
||
self.msg += f"\n{index + 1}. {server_name}"
|
||
self.msg += "\n0. 查询所有服务器信息"
|
||
|
||
self.pushMsg(self.msg)
|
||
|
||
# 等待用户输入
|
||
user_input = tool.sender.listen(30000) # 30秒超时
|
||
if user_input.lower() == 'q':
|
||
tool.replyMsg("对话已结束。")
|
||
elif user_input.isdigit():
|
||
index = int(user_input) - 1
|
||
if index == -1:
|
||
# 查询所有服务器信息
|
||
all_details = ""
|
||
for i in range(len(self.servers)):
|
||
all_details += self.query_server_by_index(i) + "\n"
|
||
tool.replyMsg(all_details)
|
||
#tool.replyMsg("查询结束。")
|
||
elif 0 <= index < len(self.servers):
|
||
details = self.query_server_by_index(index)
|
||
tool.replyMsg(details)
|
||
self.display_menu() # 返回到二级菜单
|
||
else:
|
||
tool.replyMsg("无效的编号，请重新输入。")
|
||
self.display_menu()
|
||
else:
|
||
tool.replyMsg("超时或无效输入，已自动退出。")
|
||
|
||
def query_server_by_index(self, index):
|
||
if index < 0 or index >= len(self.servers):
|
||
return "无效的服务器编号。"
|
||
|
||
server = self.servers[index]
|
||
server_name = server.get('name', '未知')
|
||
status = "在线" if server.get('state', {}).get('last_active') else ""
|
||
details = f"\n【{server_name}】 {status}"
|
||
details += f"\nCPU使用率：{server.get('state', {}).get('cpu', 0):.2f}%"
|
||
details += f"\n内存使用：{self.format_bytes(server.get('state', {}).get('mem_used', 0))}/{self.format_bytes(server.get('host', {}).get('mem_total', 0))}"
|
||
details += f"\n磁盘使用：{self.format_bytes(server.get('state', {}).get('disk_used', 0))}/{self.format_bytes(server.get('host', {}).get('disk_total', 0))}"
|
||
details += f"\n流量统计：↓{self.format_bytes(server.get('state', {}).get('net_in_transfer', 0))} ↑{self.format_bytes(server.get('state', {}).get('net_out_transfer', 0))}"
|
||
details += f"\n当前下载速率：↓{self.format_bytes(server.get('state', {}).get('net_in_speed', 0))}/s"
|
||
details += f"\n当前上传速率：↑{self.format_bytes(server.get('state', {}).get('net_out_speed', 0))}/s"
|
||
|
||
return details
|
||
|
||
if __name__ == "__main__":
|
||
tool = TOOL()
|
||
account = ACCOUNT()
|
||
if tool.msg == '哪吒管理':
|
||
account.accoount_manager()
|
||
exit(1)
|
||
if tool.imType == 'fake':
|
||
user_data_arr = account.cron_account_arr()
|
||
else:
|
||
user_str = middleware.bucketGet(f"{plugin_pre}{tool.imType}", tool.userId)
|
||
if not user_str or user_str == "":
|
||
account.addCount()
|
||
exit(1)
|
||
user_data_arr = json.loads(user_str)
|
||
for i, user_data in enumerate(user_data_arr):
|
||
user_data["push_user_id"] = tool.userId
|
||
user_data["push_im_type"] = tool.imType
|
||
user_data['index'] = i + 1
|
||
user_data['total'] = len(user_data_arr)
|
||
if tool.sender.param(1):
|
||
user_data_arr = [user_data_arr[int(tool.sender.param(1))-1]]
|
||
for account_data in user_data_arr:
|
||
account.account_task(account_data, account_data['index'])
|
||
exit(1)
|