"""LinuxAwesomeBot 入口：启动 OneBot 反向 WS 服务端 + 命令处理 + 高负载告警。""" import asyncio import logging import sys import monitor import templates 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) # ---------- 消息格式化（模板见 templates/） ---------- 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 _power_str(p: dict) -> str: if p["total_watts"] is None: return "不支持(需要电池或Intel RAPL)" 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 "" return f"{p['total_watts']}W [{src}]{extra}" def _hottest(temps: dict): flat = [t for entries in temps.values() for t in entries] return max(flat, key=lambda t: t["current"]) if flat else None def fmt_status(s: dict) -> str: c, m = s["cpu"], s["memory"] hot = _hottest(s["temps"]) net = s["net"] d = s["disk"][0] if s["disk"] else None return templates.render( "status", power=_power_str(s["power"]), cpu_total=f"{sum(c['per_core']):.1f}%", cpu_max=f"{c['cores'] * 100}%", cpu_avg=f"{c['percent']}", cores=c["cores"], load=f"{c['load'][0]:.2f}/{c['load'][1]:.2f}/{c['load'][2]:.2f}", memory=f"{m['used_gb']}/{m['total_gb']}GB ({m['percent']}%) Swap {m['swap_percent']}%", temp=f"{hot['label']} {hot['current']}C" if hot else "无数据", net=(f"上行 {net['up_mbps']}Mbps / 下行 {net['down_mbps']}Mbps" if net["up_mbps"] is not None else "无数据"), disk_mount=d["mount"] if d else "-", disk_used=f"{d['used_gb']}/{d['total_gb']}GB" if d else "-", disk_percent=f"{d['percent']}%" if d else "-", uptime=_fmt_uptime(s["uptime_s"]), ) def fmt_power(s: dict) -> str: p = s["power"] if p["total_watts"] is None and p["battery"] is None: return templates.render( "power", power=("未获取到功耗数据。支持的来源：笔记本电池 /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 域:") lines += [f" {name}: {watts}W" for name, watts in p["domains"].items()] return templates.render("power", power="\n".join(lines) or "无数据") def fmt_top(s: dict) -> str: if not s["top"]: return templates.render("top", top_n=10, top_list="无数据") lines = [ f"{i}. {p['name']} (PID {p['pid']}) CPU {p['cpu']}% 内存 {p['mem_mb']}MB" for i, p in enumerate(s["top"], 1) ] return templates.render("top", top_n=len(s["top"]), top_list="\n".join(lines)) def fmt_temps(s: dict) -> str: if not s["temps"]: return templates.render("temps", temp_list="未获取到 (需要 lm-sensors，虚拟机通常不支持)") lines = [ f" {name}/{t['label']}: {t['current']}C" for name, entries in s["temps"].items() for t in entries ] return templates.render("temps", temp_list="\n".join(lines)) def fmt_net(net: dict) -> str: return templates.render( "net", up=f"{net['up_mbps']}Mbps", down=f"{net['down_mbps']}Mbps") # ---------- 命令处理 ---------- 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", "网络"): return fmt_net(monitor.get_net(interval=1.0)) if cmd in ("help", "帮助"): return templates.render("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()