819 lines
36 KiB
Python
Executable File
819 lines
36 KiB
Python
Executable File
import asyncio
|
||
import logging
|
||
import math
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from dateutil import parser
|
||
from dotenv import load_dotenv
|
||
import os
|
||
|
||
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
|
||
from telegram.ext import (
|
||
ApplicationBuilder, CommandHandler, MessageHandler, CallbackQueryHandler,
|
||
ConversationHandler, ContextTypes, filters
|
||
)
|
||
|
||
from nezha_api import NezhaAPI
|
||
from database import Database
|
||
|
||
# 配置日志
|
||
logging.basicConfig(
|
||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||
level=logging.INFO
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
load_dotenv()
|
||
|
||
# 定义常量和配置
|
||
TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN')
|
||
DATABASE_PATH = 'users.db'
|
||
|
||
# 定义阶段
|
||
BIND_USERNAME, BIND_PASSWORD, BIND_DASHBOARD, BIND_ALIAS = range(4)
|
||
SEARCH_SERVER = range(1)
|
||
|
||
# 初始化数据库
|
||
db = Database(DATABASE_PATH)
|
||
|
||
# 添加 format_bytes 函数
|
||
def format_bytes(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
|
||
|
||
def is_online(server):
|
||
"""根据last_active判断服务器是否在线，如果最后活跃时间在10秒内则为在线。"""
|
||
now_utc = datetime.now(timezone.utc)
|
||
last_active_str = server.get('last_active')
|
||
if not last_active_str:
|
||
return False
|
||
try:
|
||
last_active_dt = parser.isoparse(last_active_str)
|
||
except ValueError:
|
||
return False
|
||
last_active_utc = last_active_dt.astimezone(timezone.utc)
|
||
diff = now_utc - last_active_utc
|
||
is_on = diff.total_seconds() < 10
|
||
logger.info("Checking online: diff=%s now=%s last=%s is_online=%s",
|
||
diff, now_utc, last_active_utc, is_on)
|
||
return is_on
|
||
|
||
# 添加 IP 地址掩码函数
|
||
def mask_ipv4(ipv4_address):
|
||
if ipv4_address == '未知' or ipv4_address == '❌':
|
||
return ipv4_address
|
||
parts = ipv4_address.split('.')
|
||
if len(parts) != 4:
|
||
return ipv4_address # 非法的 IPv4 地址，直接返回
|
||
# 将后两部分替换为 'xx'
|
||
masked_ip = f"{parts[0]}.{parts[1]}.xx.xx"
|
||
return masked_ip
|
||
|
||
def mask_ipv6(ipv6_address):
|
||
if ipv6_address == '未知' or ipv6_address == '❌':
|
||
return ipv6_address
|
||
parts = ipv6_address.split(':')
|
||
if len(parts) < 3:
|
||
return ipv6_address # 非法的 IPv6 地址，直接返回
|
||
# 只显示前两个部分，后面用 'xx' 替代
|
||
masked_ip = ':'.join(parts[:2]) + ':xx:xx:xx:xx'
|
||
return masked_ip
|
||
|
||
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
await update.message.reply_text(
|
||
"欢迎使用 Nezha 监控机器人！\n请使用 /bind 命令绑定您的账号。\n请注意，使用公共机器人有安全风险，用户名密码将会被记录用以鉴权，解绑删除。"
|
||
)
|
||
|
||
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
await update.message.reply_text("""
|
||
可用命令：
|
||
/bind - 绑定账号
|
||
/unbind - 解绑账号
|
||
/dashboard - 管理面板
|
||
/overview - 查看服务器状态总览
|
||
/server - 查看单台服务器状态
|
||
/cron - 执行计划任务
|
||
/services - 查看服务状态总览
|
||
/help - 获取帮助
|
||
""")
|
||
|
||
async def bind_start(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
# 检查当前对话类型
|
||
if update.effective_chat.type != "private":
|
||
await update.message.reply_text("请与机器人私聊进行绑定操作，\n避免机密信息泄露。")
|
||
return ConversationHandler.END
|
||
|
||
await update.message.reply_text("请输入您的用户名：")
|
||
return BIND_USERNAME
|
||
|
||
async def bind_username(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
context.user_data['username'] = update.message.text.strip()
|
||
await update.message.reply_text("请输入您的密码：")
|
||
return BIND_PASSWORD
|
||
|
||
async def bind_password(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
context.user_data['password'] = update.message.text.strip()
|
||
await update.message.reply_text("请输入您的 Dashboard 地址（例如：https://nezha.example.com）：")
|
||
return BIND_DASHBOARD
|
||
|
||
async def bind_dashboard(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
dashboard_url = update.message.text.strip()
|
||
context.user_data['dashboard_url'] = dashboard_url
|
||
await update.message.reply_text("请为这个面板设置一个别名（如：主面板、备用等）：")
|
||
return BIND_ALIAS
|
||
|
||
async def bind_alias(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
alias = update.message.text.strip()
|
||
context.user_data['alias'] = alias
|
||
telegram_id = update.effective_user.id
|
||
username = context.user_data['username']
|
||
password = context.user_data['password']
|
||
dashboard_url = context.user_data['dashboard_url']
|
||
|
||
# 测试连接
|
||
try:
|
||
api = NezhaAPI(dashboard_url, username, password)
|
||
await api.authenticate()
|
||
await api.close()
|
||
except Exception as e:
|
||
await update.message.reply_text(f"绑定失败：{e}\n请检查您的信息并重新绑定。")
|
||
return ConversationHandler.END
|
||
|
||
# 保存到数据库
|
||
await db.add_user(telegram_id, username, password, dashboard_url, alias)
|
||
await update.message.reply_text("绑定成功！您现在可以使用机器人的功能了。")
|
||
return ConversationHandler.END
|
||
|
||
async def unbind(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
dashboards = await db.get_all_dashboards(update.effective_user.id)
|
||
if not dashboards:
|
||
await update.message.reply_text("您尚未绑定任何面板。")
|
||
return
|
||
|
||
keyboard = []
|
||
# 添加每个 dashboard 的解绑选项
|
||
for dashboard in dashboards:
|
||
default_mark = "（默认）" if dashboard['is_default'] else ""
|
||
button_text = f"解绑 {dashboard['alias']}{default_mark}"
|
||
keyboard.append([InlineKeyboardButton(button_text, callback_data=f"unbind_{dashboard['id']}")])
|
||
|
||
# 添加解绑所有的选项
|
||
if len(dashboards) > 1:
|
||
keyboard.append([InlineKeyboardButton("解绑所有面板", callback_data="unbind_all")])
|
||
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await update.message.reply_text("请选择要解绑的面板：", reply_markup=reply_markup)
|
||
|
||
async def overview(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
user = await db.get_user(update.effective_user.id)
|
||
if not user:
|
||
await update.message.reply_text("请先使用 /bind 命令绑定您的账号。")
|
||
return
|
||
|
||
api = NezhaAPI(user['dashboard_url'], user['username'], user['password'])
|
||
try:
|
||
data = await api.get_overview()
|
||
except Exception as e:
|
||
await update.message.reply_text(f"获取数据失败：{e}")
|
||
await api.close()
|
||
return
|
||
# print("返回的服务数据:", data)
|
||
|
||
if data and data.get('success'):
|
||
servers = data['data']
|
||
online_servers = sum(1 for s in servers if is_online(s))
|
||
total_servers = len(servers)
|
||
total_mem = sum(s['host'].get('mem_total', 0) for s in servers if s.get('host'))
|
||
used_mem = sum(s['state'].get('mem_used', 0) for s in servers if s.get('state'))
|
||
total_swap = sum(s['host'].get('swap_total', 0) for s in servers if s.get('host'))
|
||
used_swap = sum(s['state'].get('swap_used', 0) for s in servers if s.get('state'))
|
||
total_disk = sum(s['host'].get('disk_total', 0) for s in servers if s.get('host'))
|
||
used_disk = sum(s['state'].get('disk_used', 0) for s in servers if s.get('state'))
|
||
net_in_speed = sum(s['state'].get('net_in_speed', 0) for s in servers if s.get('state'))
|
||
net_out_speed = sum(s['state'].get('net_out_speed', 0) for s in servers if s.get('state'))
|
||
net_in_transfer = sum(s['state'].get('net_in_transfer', 0) for s in servers if s.get('state'))
|
||
net_out_transfer = sum(s['state'].get('net_out_transfer', 0) for s in servers if s.get('state'))
|
||
transfer_ratio = (net_out_transfer / net_in_transfer * 100) if net_in_transfer else 0
|
||
|
||
response = f"""📊 **统计信息**
|
||
===========================
|
||
**服务器数量**： {total_servers}
|
||
**在线服务器**： {online_servers}
|
||
**内存**： {used_mem / total_mem * 100 if total_mem else 0:.1f}% [{format_bytes(used_mem)}/{format_bytes(total_mem)}]
|
||
**交换**： {used_swap / total_swap * 100 if total_swap else 0:.1f}% [{format_bytes(used_swap)}/{format_bytes(total_swap)}]
|
||
**磁盘**： {used_disk / total_disk * 100 if total_disk else 0:.1f}% [{format_bytes(used_disk)}/{format_bytes(total_disk)}]
|
||
**下行速度**： ↓{format_bytes(net_in_speed)}/s
|
||
**上行速度**： ↑{format_bytes(net_out_speed)}/s
|
||
**下行流量**： ↓{format_bytes(net_in_transfer)}
|
||
**上行流量**： ↑{format_bytes(net_out_transfer)}
|
||
**流量对等性**： {transfer_ratio:.1f}%
|
||
|
||
**更新于**： {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC
|
||
"""
|
||
keyboard = [[InlineKeyboardButton("刷新", callback_data="refresh_overview")]]
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await update.message.reply_text(response, parse_mode='Markdown', reply_markup=reply_markup)
|
||
else:
|
||
await update.message.reply_text("获取服务器信息失败。")
|
||
await api.close()
|
||
|
||
async def server_status(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
user = await db.get_user(update.effective_user.id)
|
||
if not user:
|
||
await update.message.reply_text("请先使用 /bind 命令绑定您的账号。")
|
||
return
|
||
|
||
await update.message.reply_text("请输入要查询的服务器名称（支持模糊搜索）：")
|
||
return SEARCH_SERVER
|
||
|
||
async def search_server(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
query_text = update.message.text.strip()
|
||
user = await db.get_user(update.effective_user.id)
|
||
api = NezhaAPI(user['dashboard_url'], user['username'], user['password'])
|
||
try:
|
||
results = await api.search_servers(query_text)
|
||
except Exception as e:
|
||
await update.message.reply_text(f"搜索失败：{e}")
|
||
await api.close()
|
||
return ConversationHandler.END
|
||
|
||
if not results:
|
||
await update.message.reply_text("未找到匹配的服务器。")
|
||
await api.close()
|
||
return ConversationHandler.END
|
||
|
||
keyboard = [
|
||
[InlineKeyboardButton(s['name'], callback_data=f"server_detail_{s['id']}")]
|
||
for s in results
|
||
]
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await update.message.reply_text("请选择服务器：", reply_markup=reply_markup)
|
||
await api.close()
|
||
return ConversationHandler.END
|
||
|
||
async def button_handler(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
query = update.callback_query
|
||
data = query.data
|
||
|
||
if data.startswith('unbind_'):
|
||
if data == 'unbind_all':
|
||
await db.delete_user(query.from_user.id)
|
||
await query.edit_message_text("已解绑所有面板，您可以使用 /bind 重新绑定。")
|
||
else:
|
||
dashboard_id = int(data.split('_')[-1])
|
||
# 获取当前面板信息，用于判断是否是默认面板
|
||
dashboards = await db.get_all_dashboards(query.from_user.id)
|
||
current_dashboard = next((d for d in dashboards if d['id'] == dashboard_id), None)
|
||
was_default = current_dashboard and current_dashboard['is_default']
|
||
|
||
has_remaining = await db.delete_dashboard(query.from_user.id, dashboard_id)
|
||
|
||
if not has_remaining:
|
||
await query.edit_message_text("已解绑最后一个面板，您可以使用 /bind 重新绑定。")
|
||
else:
|
||
# 新面板列表
|
||
dashboards = await db.get_all_dashboards(query.from_user.id)
|
||
keyboard = []
|
||
|
||
# 如果解绑的是默认面板，显示新的默认面板提示
|
||
if was_default:
|
||
new_default = next((d for d in dashboards if d['is_default']), None)
|
||
message = f"已解绑面板，新的默认面板已设置为：{new_default['alias']}\n\n请选择要解绑的面板："
|
||
else:
|
||
message = "请选择要解绑的面板："
|
||
|
||
for dashboard in dashboards:
|
||
default_mark = "（默认）" if dashboard['is_default'] else ""
|
||
button_text = f"解绑 {dashboard['alias']}{default_mark}"
|
||
keyboard.append([InlineKeyboardButton(button_text, callback_data=f"unbind_{dashboard['id']}")])
|
||
|
||
if len(dashboards) > 1:
|
||
keyboard.append([InlineKeyboardButton("解绑所有面板", callback_data="unbind_all")])
|
||
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await query.edit_message_text(message, reply_markup=reply_markup)
|
||
return
|
||
|
||
elif data.startswith('set_default_'):
|
||
dashboard_id = int(data.split('_')[-1])
|
||
dashboards = await db.get_all_dashboards(query.from_user.id)
|
||
selected_dashboard = next((d for d in dashboards if d['id'] == dashboard_id), None)
|
||
|
||
if not selected_dashboard:
|
||
await query.answer("未找到该面板", show_alert=True)
|
||
return
|
||
|
||
if selected_dashboard['is_default']:
|
||
await query.answer("这已经是默认面板了", show_alert=True)
|
||
return
|
||
|
||
# 直接切换默认面板
|
||
await db.set_default_dashboard(query.from_user.id, dashboard_id)
|
||
|
||
# 更新面板列表
|
||
dashboards = await db.get_all_dashboards(query.from_user.id)
|
||
keyboard = []
|
||
for dashboard in dashboards:
|
||
default_mark = "（当前默认）" if dashboard['is_default'] else ""
|
||
button_text = f"{dashboard['alias']}{default_mark}"
|
||
keyboard.append([InlineKeyboardButton(button_text, callback_data=f"set_default_{dashboard['id']}")])
|
||
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await query.edit_message_text("您的面板列表：", reply_markup=reply_markup)
|
||
return
|
||
|
||
user = await db.get_user(query.from_user.id)
|
||
if not user:
|
||
await query.answer("请先使用 /bind 命令绑定您的账号。", show_alert=True)
|
||
return
|
||
|
||
# 实现刷新频率限制
|
||
last_refresh_time = context.user_data.get('last_refresh_time', 0)
|
||
current_time = time.time()
|
||
if data.startswith('refresh_'):
|
||
if current_time - last_refresh_time < 1:
|
||
await query.answer("刷新太频繁，请稍后再试。", show_alert=True)
|
||
return
|
||
else:
|
||
context.user_data['last_refresh_time'] = current_time
|
||
|
||
await query.answer()
|
||
|
||
api = NezhaAPI(user['dashboard_url'], user['username'], user['password'])
|
||
|
||
if data.startswith('server_detail_'):
|
||
server_id = int(data.split('_')[-1])
|
||
try:
|
||
server = await api.get_server_detail(server_id)
|
||
except Exception as e:
|
||
await query.edit_message_text(f"获取服务器详情失败：{e}")
|
||
await api.close()
|
||
return
|
||
|
||
await api.close()
|
||
|
||
if not server:
|
||
await query.edit_message_text("未找到该服务器。")
|
||
return
|
||
|
||
name = server.get('name', '未知')
|
||
online_status = is_online(server)
|
||
status = "❇️在线" if online_status else "❌离线"
|
||
ipv4 = server.get('geoip', {}).get('ip', {}).get('ipv4_addr', '未知')
|
||
ipv6 = server.get('geoip', {}).get('ip', {}).get('ipv6_addr', '❌')
|
||
|
||
# 对 IP 地址进行掩码处理
|
||
ipv4 = mask_ipv4(ipv4)
|
||
ipv6 = mask_ipv6(ipv6)
|
||
|
||
platform = server.get('host', {}).get('platform', '未知')
|
||
cpu_info = ', '.join(server.get('host', {}).get('cpu', [])) if server.get('host') else '未知'
|
||
uptime_seconds = server.get('state', {}).get('uptime', 0)
|
||
uptime_days = uptime_seconds // 86400
|
||
uptime_hours = (uptime_seconds % 86400) // 3600
|
||
load_1 = server.get('state', {}).get('load_1', 0)
|
||
load_5 = server.get('state', {}).get('load_5', 0)
|
||
load_15 = server.get('state', {}).get('load_15', 0)
|
||
cpu_usage = server.get('state', {}).get('cpu', 0)
|
||
mem_used = server.get('state', {}).get('mem_used', 0)
|
||
mem_total = server.get('host', {}).get('mem_total', 1)
|
||
swap_used = server.get('state', {}).get('swap_used', 0)
|
||
swap_total = server.get('host', {}).get('swap_total', 1)
|
||
disk_used = server.get('state', {}).get('disk_used', 0)
|
||
disk_total = server.get('host', {}).get('disk_total', 1)
|
||
net_in_transfer = server.get('state', {}).get('net_in_transfer', 0)
|
||
net_out_transfer = server.get('state', {}).get('net_out_transfer', 0)
|
||
net_in_speed = server.get('state', {}).get('net_in_speed', 0)
|
||
net_out_speed = server.get('state', {}).get('net_out_speed', 0)
|
||
arch = server.get('host', {}).get('arch', '')
|
||
|
||
response = f"""**{name}** {status}
|
||
==========================
|
||
**ID**: {server.get('id', '未知')}
|
||
**IPv4**: {ipv4}
|
||
**IPv6**: {ipv6}
|
||
**平台**： {platform}
|
||
**CPU 信息**： {cpu_info}
|
||
**运行时间**： {uptime_days} 天 {uptime_hours} 小时
|
||
**负载**： {load_1:.2f} {load_5:.2f} {load_15:.2f}
|
||
**CPU**： {cpu_usage:.2f}% [{arch}]
|
||
**内存**： {mem_used / mem_total * 100 if mem_total else 0:.1f}% [{format_bytes(mem_used)}/{format_bytes(mem_total)}]
|
||
**交换**： {swap_used / swap_total * 100 if swap_total else 0:.1f}% [{format_bytes(swap_used)}/{format_bytes(swap_total)}]
|
||
**磁盘**： {disk_used / disk_total * 100 if disk_total else 0:.1f}% [{format_bytes(disk_used)}/{format_bytes(disk_total)}]
|
||
**流量**： ↓{format_bytes(net_in_transfer)} ↑{format_bytes(net_out_transfer)}
|
||
**网速**： ↓{format_bytes(net_in_speed)}/s ↑{format_bytes(net_out_speed)}/s
|
||
|
||
**更新于**： {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC
|
||
"""
|
||
# 添加刷新按钮
|
||
keyboard = [[InlineKeyboardButton("刷新", callback_data=f"refresh_server_{server_id}")]]
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await query.edit_message_text(response, parse_mode='Markdown', reply_markup=reply_markup)
|
||
|
||
elif data.startswith('refresh_server_'):
|
||
server_id = int(data.split('_')[-1])
|
||
# 重新获取服务器详情，与上面相同的代码
|
||
try:
|
||
server = await api.get_server_detail(server_id)
|
||
except Exception as e:
|
||
await query.edit_message_text(f"获取服务器详情失败：{e}")
|
||
await api.close()
|
||
return
|
||
|
||
await api.close()
|
||
|
||
if not server:
|
||
await query.edit_message_text("未找到该服务器。")
|
||
return
|
||
|
||
# 同上，构建响应和刷新按钮
|
||
name = server.get('name', '未知')
|
||
online_status = is_online(server)
|
||
status = "❇️在线" if online_status else "❌离线"
|
||
ipv4 = server.get('geoip', {}).get('ip', {}).get('ipv4_addr', '未知')
|
||
ipv6 = server.get('geoip', {}).get('ip', {}).get('ipv6_addr', '❌')
|
||
|
||
# 对 IP 地址进行掩码处理
|
||
ipv4 = mask_ipv4(ipv4)
|
||
ipv6 = mask_ipv6(ipv6)
|
||
|
||
platform = server.get('host', {}).get('platform', '未知')
|
||
cpu_info = ', '.join(server.get('host', {}).get('cpu', [])) if server.get('host') else '未知'
|
||
uptime_seconds = server.get('state', {}).get('uptime', 0)
|
||
uptime_days = uptime_seconds // 86400
|
||
uptime_hours = (uptime_seconds % 86400) // 3600
|
||
load_1 = server.get('state', {}).get('load_1', 0)
|
||
load_5 = server.get('state', {}).get('load_5', 0)
|
||
load_15 = server.get('state', {}).get('load_15', 0)
|
||
cpu_usage = server.get('state', {}).get('cpu', 0)
|
||
mem_used = server.get('state', {}).get('mem_used', 0)
|
||
mem_total = server.get('host', {}).get('mem_total', 1)
|
||
swap_used = server.get('state', {}).get('swap_used', 0)
|
||
swap_total = server.get('host', {}).get('swap_total', 1)
|
||
disk_used = server.get('state', {}).get('disk_used', 0)
|
||
disk_total = server.get('host', {}).get('disk_total', 1)
|
||
net_in_transfer = server.get('state', {}).get('net_in_transfer', 0)
|
||
net_out_transfer = server.get('state', {}).get('net_out_transfer', 0)
|
||
net_in_speed = server.get('state', {}).get('net_in_speed', 0)
|
||
net_out_speed = server.get('state', {}).get('net_out_speed', 0)
|
||
arch = server.get('host', {}).get('arch', '')
|
||
|
||
response = f"""**{name}** {status}
|
||
==========================
|
||
**ID**: {server.get('id', '未知')}
|
||
**IPv4**: {ipv4}
|
||
**IPv6**: {ipv6}
|
||
**平台**： {platform}
|
||
**CPU 信息**： {cpu_info}
|
||
**运行时间**： {uptime_days} 天 {uptime_hours} 小时
|
||
**负载**： {load_1:.2f} {load_5:.2f} {load_15:.2f}
|
||
**CPU**： {cpu_usage:.2f}% [{arch}]
|
||
**内存**： {mem_used / mem_total * 100 if mem_total else 0:.1f}% [{format_bytes(mem_used)}/{format_bytes(mem_total)}]
|
||
**交换**： {swap_used / swap_total * 100 if swap_total else 0:.1f}% [{format_bytes(swap_used)}/{format_bytes(swap_total)}]
|
||
**磁盘**： {disk_used / disk_total * 100 if disk_total else 0:.1f}% [{format_bytes(disk_used)}/{format_bytes(disk_total)}]
|
||
**流量**： ↓{format_bytes(net_in_transfer)} ↑{format_bytes(net_out_transfer)}
|
||
**网速**： ↓{format_bytes(net_in_speed)}/s ↑{format_bytes(net_out_speed)}/s
|
||
|
||
**更新于**： {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC
|
||
"""
|
||
keyboard = [[InlineKeyboardButton("刷新", callback_data=f"refresh_server_{server_id}")]]
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await query.edit_message_text(response, parse_mode='Markdown', reply_markup=reply_markup)
|
||
|
||
elif data == 'refresh_overview':
|
||
# 重新获取概览数据，与 overview 函数类似
|
||
try:
|
||
data = await api.get_overview()
|
||
except Exception as e:
|
||
await query.edit_message_text(f"获取数据失败：{e}")
|
||
await api.close()
|
||
return
|
||
|
||
if data and data.get('success'):
|
||
servers = data['data']
|
||
total_servers = len(servers)
|
||
online_servers = sum(1 for s in servers if is_online(s))
|
||
total_mem = sum(s['host'].get('mem_total', 0) for s in servers if s.get('host'))
|
||
used_mem = sum(s['state'].get('mem_used', 0) for s in servers if s.get('state'))
|
||
total_swap = sum(s['host'].get('swap_total', 0) for s in servers if s.get('host'))
|
||
used_swap = sum(s['state'].get('swap_used', 0) for s in servers if s.get('state'))
|
||
total_disk = sum(s['host'].get('disk_total', 0) for s in servers if s.get('host'))
|
||
used_disk = sum(s['state'].get('disk_used', 0) for s in servers if s.get('state'))
|
||
net_in_speed = sum(s['state'].get('net_in_speed', 0) for s in servers if s.get('state'))
|
||
net_out_speed = sum(s['state'].get('net_out_speed', 0) for s in servers if s.get('state'))
|
||
net_in_transfer = sum(s['state'].get('net_in_transfer', 0) for s in servers if s.get('state'))
|
||
net_out_transfer = sum(s['state'].get('net_out_transfer', 0) for s in servers if s.get('state'))
|
||
transfer_ratio = (net_out_transfer / net_in_transfer * 100) if net_in_transfer else 0
|
||
|
||
response = f"""📊 **统计信息**
|
||
===========================
|
||
**服务器数量**： {total_servers}
|
||
**在线服务器**： {online_servers}
|
||
**内存**： {used_mem / total_mem * 100 if total_mem else 0:.1f}% [{format_bytes(used_mem)}/{format_bytes(total_mem)}]
|
||
**交换**： {used_swap / total_swap * 100 if total_swap else 0:.1f}% [{format_bytes(used_swap)}/{format_bytes(total_swap)}]
|
||
**磁盘**： {used_disk / total_disk * 100 if total_disk else 0:.1f}% [{format_bytes(used_disk)}/{format_bytes(total_disk)}]
|
||
**下行速度**： ↓{format_bytes(net_in_speed)}/s
|
||
**上行速度**： ↑{format_bytes(net_out_speed)}/s
|
||
**下行流量**： ↓{format_bytes(net_in_transfer)}
|
||
**上行流量**： ↑{format_bytes(net_out_transfer)}
|
||
**流量对等性**： {transfer_ratio:.1f}%
|
||
|
||
**更新于**： {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC
|
||
"""
|
||
keyboard = [[InlineKeyboardButton("刷新", callback_data="refresh_overview")]]
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await query.edit_message_text(response, parse_mode='Markdown', reply_markup=reply_markup)
|
||
else:
|
||
await query.edit_message_text("获取服务器信息失败。")
|
||
await api.close()
|
||
|
||
elif data.startswith('cron_job_'):
|
||
cron_id = int(data.split('_')[-1])
|
||
keyboard = [
|
||
[InlineKeyboardButton("确认执行", callback_data=f"confirm_cron_{cron_id}")],
|
||
[InlineKeyboardButton("取消", callback_data="cancel")]
|
||
]
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await query.edit_message_text("您确定要执行此计划任务吗？", reply_markup=reply_markup)
|
||
|
||
elif data.startswith('confirm_cron_'):
|
||
cron_id = int(data.split('_')[-1])
|
||
try:
|
||
result = await api.run_cron_job(cron_id)
|
||
except Exception as e:
|
||
await query.edit_message_text(f"执行失败：{e}")
|
||
await api.close()
|
||
return
|
||
|
||
await api.close()
|
||
|
||
if result and result.get('success'):
|
||
await query.edit_message_text("计划任务已执行。")
|
||
else:
|
||
await query.edit_message_text("执行失败。")
|
||
|
||
elif data == 'cancel':
|
||
await query.edit_message_text("操作已取消。")
|
||
|
||
elif data == 'view_loop_traffic':
|
||
await view_loop_traffic(query, context, api)
|
||
|
||
elif data == 'refresh_loop_traffic':
|
||
await view_loop_traffic(query, context, api)
|
||
|
||
elif data == 'view_availability':
|
||
await view_availability(query, context, api)
|
||
|
||
elif data == 'refresh_availability':
|
||
await view_availability(query, context, api)
|
||
|
||
elif data.startswith('set_default_'):
|
||
dashboard_id = int(data.split('_')[-1])
|
||
await db.set_default_dashboard(query.from_user.id, dashboard_id)
|
||
await query.edit_message_text("已更新默认面板。")
|
||
return
|
||
|
||
elif data.startswith('dashboard_'):
|
||
dashboard_id = int(data.split('_')[-1])
|
||
dashboards = await db.get_all_dashboards(query.from_user.id)
|
||
selected_dashboard = next((d for d in dashboards if d['id'] == dashboard_id), None)
|
||
|
||
if not selected_dashboard:
|
||
await query.answer("未找到该面板", show_alert=True)
|
||
return
|
||
|
||
if selected_dashboard['is_default']:
|
||
await query.answer("这已经是默认面板了", show_alert=True)
|
||
return
|
||
|
||
# 直接切换默认面板
|
||
await db.set_default_dashboard(query.from_user.id, dashboard_id)
|
||
|
||
# 更新面板列表
|
||
dashboards = await db.get_all_dashboards(query.from_user.id)
|
||
keyboard = []
|
||
for dashboard in dashboards:
|
||
default_mark = "（当前默认）" if dashboard['is_default'] else ""
|
||
button_text = f"{dashboard['alias']}{default_mark}"
|
||
keyboard.append([InlineKeyboardButton(button_text, callback_data=f"set_default_{dashboard['id']}")])
|
||
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await query.edit_message_text("您的面板列表：", reply_markup=reply_markup)
|
||
return
|
||
|
||
elif data == "dashboard_back":
|
||
# 返回面板列表
|
||
dashboards = await db.get_all_dashboards(query.from_user.id)
|
||
keyboard = []
|
||
for dashboard in dashboards:
|
||
default_mark = "（当前默认）" if dashboard['is_default'] else ""
|
||
button_text = f"{dashboard['alias']}{default_mark}"
|
||
keyboard.append([InlineKeyboardButton(button_text, callback_data=f"set_default_{dashboard['id']}")])
|
||
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await query.edit_message_text("您的面板列表：", reply_markup=reply_markup)
|
||
return
|
||
|
||
async def view_loop_traffic(query, context, api):
|
||
# 获取服务状态
|
||
try:
|
||
services_data = await api.get_services_status()
|
||
except Exception as e:
|
||
await query.edit_message_text(f"获取服务信息失败：{e}")
|
||
await api.close()
|
||
return
|
||
|
||
if services_data and services_data.get('success'):
|
||
cycle_stats = services_data['data'].get('cycle_transfer_stats', {})
|
||
if not cycle_stats:
|
||
await query.edit_message_text("暂无循环流量信息。")
|
||
await api.close()
|
||
return
|
||
|
||
response = "**循环流量信息总览**\n==========================\n"
|
||
for stat_name, stats in cycle_stats.items():
|
||
rule_name = stats.get('name', '未知规则')
|
||
server_names = stats.get('server_name', {})
|
||
transfers = stats.get('transfer', {})
|
||
max_transfer = stats.get('max', 1) # 最大流量（字节）
|
||
|
||
response += f"**规则：{rule_name}**\n"
|
||
for server_id_str, transfer_value in transfers.items():
|
||
server_id = str(server_id_str)
|
||
server_name = server_names.get(server_id, f"服务器ID {server_id}")
|
||
transfer_formatted = format_bytes(transfer_value)
|
||
max_transfer_formatted = format_bytes(max_transfer)
|
||
percentage = (transfer_value / max_transfer * 100) if max_transfer else 0
|
||
response += f"服务器 **{server_name}**：已使用 {transfer_formatted} / {max_transfer_formatted}，已使用 {percentage:.2f}%\n"
|
||
response += "--------------------------\n"
|
||
|
||
response += f"**更新于**： {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC"
|
||
|
||
# 添加刷新按钮
|
||
keyboard = [[InlineKeyboardButton("刷新", callback_data="refresh_loop_traffic")]]
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await query.edit_message_text(response, parse_mode='Markdown', reply_markup=reply_markup)
|
||
else:
|
||
await query.edit_message_text("获取循环流量信息失败。")
|
||
await api.close()
|
||
|
||
async def view_availability(query, context, api):
|
||
# 获取服务状态
|
||
try:
|
||
services_data = await api.get_services_status()
|
||
except Exception as e:
|
||
await query.edit_message_text(f"获取服务信息失败：{e}")
|
||
await api.close()
|
||
return
|
||
# print("返回的服务数据:", services_data)
|
||
|
||
if services_data and services_data.get('success'):
|
||
services = services_data['data'].get('services', {})
|
||
if not services:
|
||
await query.edit_message_text("暂无可用性监测信息。")
|
||
await api.close()
|
||
return
|
||
|
||
response = "**可用性监测信息总览**\n==========================\n"
|
||
for service_id, service_info in services.items():
|
||
service = service_info.get('service', {})
|
||
name = service_info.get('service_name', '未知')
|
||
total_up = service_info.get('total_up', 0)
|
||
total_down = service_info.get('total_down', 0)
|
||
total = total_up + total_down
|
||
availability = (total_up / total * 100) if total else 0
|
||
status = "🟢 UP" if service_info.get('current_up', 0) else "🔴 DOWN"
|
||
# 计算平均延迟
|
||
delays = service_info.get('delay', [])
|
||
if delays:
|
||
avg_delay = sum(delays) / len(delays)
|
||
else:
|
||
avg_delay = None
|
||
if avg_delay is not None:
|
||
delay_text = f"，平均延迟 {avg_delay:.2f}ms"
|
||
else:
|
||
delay_text = ""
|
||
response += f"**{name}**：可用率 {availability:.2f}%，状态 {status}{delay_text}\n------------------\n"
|
||
response += f"\n**更新于**： {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC"
|
||
|
||
# 添加刷新按钮
|
||
keyboard = [[InlineKeyboardButton("刷新", callback_data="refresh_availability")]]
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await query.edit_message_text(response, parse_mode='Markdown', reply_markup=reply_markup)
|
||
else:
|
||
await query.edit_message_text("获取可用性监测信息失败。")
|
||
await api.close()
|
||
|
||
async def cron_jobs(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
user = await db.get_user(update.effective_user.id)
|
||
if not user:
|
||
await update.message.reply_text("请先使用 /bind 命令绑定您的账号。")
|
||
return
|
||
|
||
api = NezhaAPI(user['dashboard_url'], user['username'], user['password'])
|
||
try:
|
||
data = await api.get_cron_jobs()
|
||
except Exception as e:
|
||
await update.message.reply_text(f"获取计划任务失败：{e}")
|
||
await api.close()
|
||
return
|
||
|
||
if data and data.get('success'):
|
||
cron_jobs = data['data']
|
||
if not cron_jobs:
|
||
await update.message.reply_text("暂无计划任务。")
|
||
await api.close()
|
||
return
|
||
|
||
keyboard = [
|
||
[InlineKeyboardButton(job['name'], callback_data=f"cron_job_{job['id']}")]
|
||
for job in cron_jobs
|
||
]
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await update.message.reply_text("请选择要执行的计划任务：", reply_markup=reply_markup)
|
||
else:
|
||
await update.message.reply_text("获取计划任务失败。")
|
||
await api.close()
|
||
|
||
async def services_overview(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
user = await db.get_user(update.effective_user.id)
|
||
if not user:
|
||
await update.message.reply_text("请先使用 /bind 命令绑定您的账号。")
|
||
return
|
||
|
||
keyboard = [
|
||
[InlineKeyboardButton("查看循环流量信息", callback_data="view_loop_traffic")],
|
||
[InlineKeyboardButton("查看可用性监测信息", callback_data="view_availability")]
|
||
]
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await update.message.reply_text("请选择要查看的服务信息：", reply_markup=reply_markup)
|
||
|
||
async def dashboard(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
dashboards = await db.get_all_dashboards(update.effective_user.id)
|
||
if not dashboards:
|
||
await update.message.reply_text("您还没有绑定任何面板。")
|
||
return
|
||
|
||
keyboard = []
|
||
for dashboard in dashboards:
|
||
default_mark = "（当前默认）" if dashboard['is_default'] else ""
|
||
button_text = f"{dashboard['alias']}{default_mark}"
|
||
keyboard.append([InlineKeyboardButton(button_text, callback_data=f"set_default_{dashboard['id']}")])
|
||
|
||
reply_markup = InlineKeyboardMarkup(keyboard)
|
||
await update.message.reply_text("您的面板列表：", reply_markup=reply_markup)
|
||
|
||
def main():
|
||
application = ApplicationBuilder().token(TELEGRAM_TOKEN).build()
|
||
|
||
# 初始化数据库
|
||
loop = asyncio.get_event_loop()
|
||
loop.run_until_complete(db.initialize())
|
||
|
||
# 回调查询处理（放在最前面）
|
||
application.add_handler(CallbackQueryHandler(button_handler))
|
||
|
||
# 命令处理
|
||
application.add_handler(CommandHandler('start', start))
|
||
application.add_handler(CommandHandler('help', help_command))
|
||
application.add_handler(CommandHandler('unbind', unbind))
|
||
application.add_handler(CommandHandler('overview', overview))
|
||
application.add_handler(CommandHandler('cron', cron_jobs))
|
||
application.add_handler(CommandHandler('services', services_overview))
|
||
application.add_handler(CommandHandler('dashboard', dashboard))
|
||
|
||
# 绑定命令的会话处理
|
||
bind_handler = ConversationHandler(
|
||
entry_points=[CommandHandler('bind', bind_start)],
|
||
states={
|
||
BIND_USERNAME: [MessageHandler(filters.TEXT & ~filters.COMMAND, bind_username)],
|
||
BIND_PASSWORD: [MessageHandler(filters.TEXT & ~filters.COMMAND, bind_password)],
|
||
BIND_DASHBOARD: [MessageHandler(filters.TEXT & ~filters.COMMAND, bind_dashboard)],
|
||
BIND_ALIAS: [MessageHandler(filters.TEXT & ~filters.COMMAND, bind_alias)],
|
||
},
|
||
fallbacks=[]
|
||
)
|
||
application.add_handler(bind_handler)
|
||
|
||
# 查看单台服务器状态的会话处理
|
||
server_handler = ConversationHandler(
|
||
entry_points=[CommandHandler('server', server_status)],
|
||
states={
|
||
SEARCH_SERVER: [MessageHandler(filters.TEXT & ~filters.COMMAND, search_server)],
|
||
},
|
||
fallbacks=[]
|
||
)
|
||
application.add_handler(server_handler)
|
||
|
||
# 在 run_polling 中指定 allowed_updates
|
||
application.run_polling(allowed_updates=['message', 'callback_query'])
|
||
|
||
if __name__ == '__main__':
|
||
main()
|