212 lines
7.4 KiB
Python
212 lines
7.4 KiB
Python
"""LinuxAwesomeBot 入口：启动 OneBot 反向 WS 服务端 + 命令处理 + 高负载告警。"""
|
||
import asyncio
|
||
import logging
|
||
import sys
|
||
|
||
import monitor
|
||
from alerter import Alerter
|
||
from onebot import OneBotServer
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
|
||
datefmt="%H:%M:%S",
|
||
)
|
||
log = logging.getLogger("bot")
|
||
|
||
|
||
# ---------- 配置 ----------
|
||
|
||
def load_config() -> dict:
|
||
import tomllib
|
||
import shutil
|
||
try:
|
||
with open("config.toml", "rb") as f:
|
||
return tomllib.load(f)
|
||
except FileNotFoundError:
|
||
shutil.copy("config.example.toml", "config.toml")
|
||
log.warning("未找到 config.toml，已从模板生成，请修改后再运行（先退出）")
|
||
sys.exit(1)
|
||
|
||
|
||
# ---------- 消息格式化 ----------
|
||
|
||
def _fmt_uptime(seconds: float) -> str:
|
||
d, r = divmod(int(seconds), 86400)
|
||
h, r = divmod(r, 3600)
|
||
m = r // 60
|
||
return f"{d}天{h}小时{m}分" if d else f"{h}小时{m}分"
|
||
|
||
|
||
def fmt_status(s: dict) -> str:
|
||
p, c, m = s["power"], s["cpu"], s["memory"]
|
||
lines = ["📊 服务器状态"]
|
||
power_str = "不支持"
|
||
if p["total_watts"] is not None:
|
||
src = {"battery": "🔋电池", "rapl": "RAPL"}.get(p["source"], p["source"])
|
||
extra = f" (电池 {p['battery']}W)" if p["battery"] is not None and p["source"] == "rapl" else ""
|
||
power_str = f"{p['total_watts']}W [{src}]{extra}"
|
||
lines.append(f"⚡ 功耗: {power_str}")
|
||
load = c["load"]
|
||
total_pct = sum(c["per_core"])
|
||
lines.append(f"🖥 CPU: {total_pct:.1f}% / {c['cores'] * 100}% "
|
||
f"(单核均值 {c['percent']}%) 核心 {c['cores']} "
|
||
f"负载 {load[0]:.2f}/{load[1]:.2f}/{load[2]:.2f}")
|
||
lines.append(f"💾 内存: {m['used_gb']}/{m['total_gb']}GB ({m['percent']}%) Swap {m['swap_percent']}%")
|
||
if s["temps"]:
|
||
flat = [t for entries in s["temps"].values() for t in entries]
|
||
hottest = max(flat, key=lambda t: t["current"])
|
||
lines.append(f"🌡 温度: 最高 {hottest['label']} {hottest['current']}°C")
|
||
if s["net"]["up_mbps"] is not None:
|
||
lines.append(f"🌐 网络: ↑{s['net']['up_mbps']}Mbps ↓{s['net']['down_mbps']}Mbps")
|
||
lines.append(f"⏱ 运行: {_fmt_uptime(s['uptime_s'])}")
|
||
if s["disk"]:
|
||
d = s["disk"][0]
|
||
lines.append(f"🗄 磁盘 {d['mount']}: {d['used_gb']}/{d['total_gb']}GB ({d['percent']}%)")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def fmt_power(s: dict) -> str:
|
||
p = s["power"]
|
||
if p["total_watts"] is None and p["battery"] is None:
|
||
return ("⚡ 未获取到功耗数据。\n"
|
||
"支持的来源：笔记本电池 /sys/class/power_supply，"
|
||
"Intel RAPL /sys/class/powercap（AMD/部分虚拟机不支持）。")
|
||
lines = ["⚡ 功耗详情"]
|
||
if p["battery"] is not None:
|
||
lines.append(f"🔋 电池功率: {p['battery']}W")
|
||
if p["domains"]:
|
||
lines.append("RAPL 域:")
|
||
for name, watts in p["domains"].items():
|
||
lines.append(f" {name}: {watts}W")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def fmt_top(s: dict) -> str:
|
||
if not s["top"]:
|
||
return "🔝 未获取到进程数据"
|
||
lines = ["🔝 CPU 占用 TOP10:"]
|
||
for i, p in enumerate(s["top"], 1):
|
||
lines.append(f"{i}. {p['name']} (PID {p['pid']}) CPU {p['cpu']}% 内存 {p['mem_mb']}MB")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def fmt_temps(s: dict) -> str:
|
||
if not s["temps"]:
|
||
return "🌡 未获取到温度数据（需要 lm-sensors 或 psutil 支持，虚拟机通常不支持）"
|
||
lines = ["🌡 温度:"]
|
||
for name, entries in s["temps"].items():
|
||
for t in entries:
|
||
lines.append(f" {name}/{t['label']}: {t['current']}°C")
|
||
return "\n".join(lines)
|
||
|
||
|
||
HELP = """📖 可用命令（/ 或 ! 前缀均可）:
|
||
/status 状态 - 总览(功耗/CPU/内存/温度/网络)
|
||
/power 功耗 - 功耗详情(RAPL/电池)
|
||
/top 进程 - CPU TOP10 进程
|
||
/temp 温度 - 传感器温度
|
||
/net 网络 - 网络速率
|
||
/help 帮助 - 本帮助"""
|
||
|
||
|
||
# ---------- 命令处理 ----------
|
||
|
||
class Bot:
|
||
def __init__(self, cfg: dict):
|
||
self.cfg = cfg
|
||
self.ob = OneBotServer(
|
||
cfg["server"]["host"], cfg["server"]["port"],
|
||
cfg["server"].get("access_token", ""),
|
||
)
|
||
|
||
async def on_event(self, ev: dict):
|
||
if ev.get("post_type") != "message":
|
||
return
|
||
if ev.get("message_type") == "group":
|
||
text = self._extract_text(ev.get("message"))
|
||
# 群聊需 @ 或带前缀
|
||
if not self._is_command(text) and not self._at_me(ev.get("message")):
|
||
return
|
||
elif ev.get("message_type") == "private":
|
||
text = self._extract_text(ev.get("message"))
|
||
else:
|
||
return
|
||
|
||
reply = await self.handle_command(text)
|
||
if reply is None:
|
||
return
|
||
if ev.get("message_type") == "group":
|
||
await self.ob.send_group_msg(ev["group_id"], reply)
|
||
else:
|
||
await self.ob.send_private_msg(ev["user_id"], reply)
|
||
|
||
@staticmethod
|
||
def _extract_text(message) -> str:
|
||
if isinstance(message, str):
|
||
return message.strip()
|
||
if isinstance(message, list): # CQ 段数组
|
||
return "".join(
|
||
seg.get("data", {}).get("text", "")
|
||
for seg in message if isinstance(seg, dict) and seg.get("type") == "text"
|
||
).strip()
|
||
return ""
|
||
|
||
@staticmethod
|
||
def _at_me(message) -> bool:
|
||
if isinstance(message, list):
|
||
return any(
|
||
seg.get("type") == "at" for seg in message if isinstance(seg, dict)
|
||
)
|
||
return "[CQ:at" in (message or "")
|
||
|
||
@staticmethod
|
||
def _is_command(text: str) -> bool:
|
||
return text.startswith("/") or text.startswith("!")
|
||
|
||
async def handle_command(self, text: str) -> str | None:
|
||
cmd = text.lstrip("/!").strip().lower()
|
||
cmd = cmd.split()[0] if cmd else ""
|
||
try:
|
||
if cmd in ("status", "状态", ""):
|
||
return fmt_status(monitor.get_status())
|
||
if cmd in ("power", "功耗"):
|
||
return fmt_power({"power": monitor.get_power()})
|
||
if cmd in ("top", "进程"):
|
||
return fmt_top({"top": monitor.get_top_processes(10)})
|
||
if cmd in ("temp", "温度"):
|
||
return fmt_temps({"temps": monitor.get_temps()})
|
||
if cmd in ("net", "网络"):
|
||
net = monitor.get_net(interval=1.0)
|
||
return f"🌐 网络速率: ↑{net['up_mbps']}Mbps ↓{net['down_mbps']}Mbps"
|
||
if cmd in ("help", "帮助"):
|
||
return HELP
|
||
return None
|
||
except Exception:
|
||
log.exception("处理命令失败: %s", text)
|
||
return "❌ 命令执行出错，请查看服务端日志。"
|
||
|
||
async def run(self):
|
||
await self.ob.start(self.on_event)
|
||
alerter = Alerter(self.ob, self.cfg.get("alert", {}))
|
||
alerter.start()
|
||
log.info("机器人就绪，等待 OneBot 实现端连接...")
|
||
try:
|
||
await asyncio.Event().wait()
|
||
except (KeyboardInterrupt, asyncio.CancelledError):
|
||
pass
|
||
finally:
|
||
await alerter.stop()
|
||
await self.ob.stop()
|
||
|
||
|
||
def main():
|
||
if sys.platform == "win32":
|
||
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
|
||
cfg = load_config()
|
||
asyncio.run(Bot(cfg).run())
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|