feat: template-based message display, remove all emoji

This commit is contained in:
snowy-yang committed 2026-09-12 19:49:13 +08:00
1 parent 570eb0df70
commit 2e077eab33
11 files changed
+124 -66

No files matched your search

+13 -11
View File
@@ -4,6 +4,7 @@ import logging
import time import time
import monitor import monitor
import templates
log = logging.getLogger("alerter") log = logging.getLogger("alerter")
@@ -60,8 +61,7 @@ class Alerter:
self._consecutive = 0 self._consecutive = 0
if self._alerting: if self._alerting:
self._alerting = False self._alerting = False
msg = (f"✅ [恢复] 服务器 CPU 已回落到 {cpu_percent:.1f}%," msg = templates.render("recovered", cpu=f"{cpu_percent:.1f}")
f"负载恢复正常。")
log.info(msg) log.info(msg)
await self.ob.broadcast(msg, self.groups, self.friends) await self.ob.broadcast(msg, self.groups, self.friends)
@@ -69,16 +69,18 @@ class Alerter:
top = monitor.get_top_processes(self.top_n) top = monitor.get_top_processes(self.top_n)
cpu = monitor.get_cpu() cpu = monitor.get_cpu()
load = cpu["load"] load = cpu["load"]
lines = [ top_list = "\n".join(
f"⚠️ [高负载告警] CPU 持续超过 {self.threshold:.0f}%,当前 {cpu_percent:.1f}%",
f"负载: {load[0]:.2f} / {load[1]:.2f} / {load[2]:.2f} 核心: {cpu['cores']}",
"",
f"TOP{self.top_n} 高负载进程:",
]
for i, p in enumerate(top, 1):
lines.append(
f"{i}. {p['name']} (PID {p['pid']}) CPU {p['cpu']}% 内存 {p['mem_mb']}MB" f"{i}. {p['name']} (PID {p['pid']}) CPU {p['cpu']}% 内存 {p['mem_mb']}MB"
for i, p in enumerate(top, 1)
)
msg = templates.render(
"alert",
threshold=f"{self.threshold:.0f}",
cpu=f"{cpu_percent:.1f}",
cores=cpu["cores"],
load=f"{load[0]:.2f} / {load[1]:.2f} / {load[2]:.2f}",
top_n=self.top_n,
top_list=top_list or "无数据",
) )
msg = "\n".join(lines)
log.warning("触发高负载告警: %.1f%%", cpu_percent) log.warning("触发高负载告警: %.1f%%", cpu_percent)
await self.ob.broadcast(msg, self.groups, self.friends) await self.ob.broadcast(msg, self.groups, self.friends)
+62 -55
View File
@@ -4,6 +4,7 @@ import logging
import sys import sys
import monitor import monitor
import templates
from alerter import Alerter from alerter import Alerter
from onebot import OneBotServer from onebot import OneBotServer
@@ -29,7 +30,7 @@ def load_config() -> dict:
sys.exit(1) sys.exit(1)
# ---------- 消息格式化 ---------- # ---------- 消息格式化(模板见 templates/) ----------
def _fmt_uptime(seconds: float) -> str: def _fmt_uptime(seconds: float) -> str:
d, r = divmod(int(seconds), 86400) d, r = divmod(int(seconds), 86400)
@@ -38,76 +39,83 @@ def _fmt_uptime(seconds: float) -> str:
return f"{d}天{h}小时{m}分" if d else f"{h}小时{m}分" return f"{d}天{h}小时{m}分" if d else f"{h}小时{m}分"
def fmt_status(s: dict) -> str: def _power_str(p: dict) -> str:
p, c, m = s["power"], s["cpu"], s["memory"] if p["total_watts"] is None:
lines = ["📊 服务器状态"] return "不支持(需要电池或Intel RAPL)"
power_str = "不支持" src = {"battery": "电池", "rapl": "RAPL"}.get(p["source"], p["source"])
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 "" 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}" return f"{p['total_watts']}W [{src}]{extra}"
lines.append(f"⚡ 功耗: {power_str}")
load = c["load"]
total_pct = sum(c["per_core"]) def _hottest(temps: dict):
lines.append(f"🖥 CPU: {total_pct:.1f}% / {c['cores'] * 100}% " flat = [t for entries in temps.values() for t in entries]
f"(单核均值 {c['percent']}%) 核心 {c['cores']} " return max(flat, key=lambda t: t["current"]) if flat else None
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"]: def fmt_status(s: dict) -> str:
flat = [t for entries in s["temps"].values() for t in entries] c, m = s["cpu"], s["memory"]
hottest = max(flat, key=lambda t: t["current"]) hot = _hottest(s["temps"])
lines.append(f"🌡 温度: 最高 {hottest['label']} {hottest['current']}°C") net = s["net"]
if s["net"]["up_mbps"] is not None: d = s["disk"][0] if s["disk"] else None
lines.append(f"🌐 网络: ↑{s['net']['up_mbps']}Mbps ↓{s['net']['down_mbps']}Mbps") return templates.render(
lines.append(f"⏱ 运行: {_fmt_uptime(s['uptime_s'])}") "status",
if s["disk"]: power=_power_str(s["power"]),
d = s["disk"][0] cpu_total=f"{sum(c['per_core']):.1f}%",
lines.append(f"🗄 磁盘 {d['mount']}: {d['used_gb']}/{d['total_gb']}GB ({d['percent']}%)") cpu_max=f"{c['cores'] * 100}%",
return "\n".join(lines) 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: def fmt_power(s: dict) -> str:
p = s["power"] p = s["power"]
if p["total_watts"] is None and p["battery"] is None: if p["total_watts"] is None and p["battery"] is None:
return ("⚡ 未获取到功耗数据。\n" return templates.render(
"支持的来源:笔记本电池 /sys/class/power_supply," "power",
"Intel RAPL /sys/class/powercap(AMD/部分虚拟机不支持)。") power=("未获取到功耗数据。支持的来源:笔记本电池 /sys/class/power_supply,"
lines = ["⚡ 功耗详情"] "Intel RAPL /sys/class/powercap (AMD/多数虚拟机不支持)。"),
)
lines = []
if p["battery"] is not None: if p["battery"] is not None:
lines.append(f"🔋 电池功率: {p['battery']}W") lines.append(f"电池功率: {p['battery']}W")
if p["domains"]: if p["domains"]:
lines.append("RAPL 域:") lines.append("RAPL 域:")
for name, watts in p["domains"].items(): lines += [f" {name}: {watts}W" for name, watts in p["domains"].items()]
lines.append(f" {name}: {watts}W") return templates.render("power", power="\n".join(lines) or "无数据")
return "\n".join(lines)
def fmt_top(s: dict) -> str: def fmt_top(s: dict) -> str:
if not s["top"]: if not s["top"]:
return "🔝 未获取到进程数据" return templates.render("top", top_n=10, top_list="无数据")
lines = ["🔝 CPU 占用 TOP10:"] lines = [
for i, p in enumerate(s["top"], 1): f"{i}. {p['name']} (PID {p['pid']}) CPU {p['cpu']}% 内存 {p['mem_mb']}MB"
lines.append(f"{i}. {p['name']} (PID {p['pid']}) CPU {p['cpu']}% 内存 {p['mem_mb']}MB") for i, p in enumerate(s["top"], 1)
return "\n".join(lines) ]
return templates.render("top", top_n=len(s["top"]), top_list="\n".join(lines))
def fmt_temps(s: dict) -> str: def fmt_temps(s: dict) -> str:
if not s["temps"]: if not s["temps"]:
return "🌡 未获取到温度数据(需要 lm-sensors 或 psutil 支持,虚拟机通常不支持)" return templates.render("temps", temp_list="未获取到 (需要 lm-sensors,虚拟机通常不支持)")
lines = ["🌡 温度:"] lines = [
for name, entries in s["temps"].items(): f" {name}/{t['label']}: {t['current']}C"
for t in entries: for name, entries in s["temps"].items() for t in entries
lines.append(f" {name}/{t['label']}: {t['current']}°C") ]
return "\n".join(lines) return templates.render("temps", temp_list="\n".join(lines))
HELP = """📖 可用命令(/ 或 ! 前缀均可): def fmt_net(net: dict) -> str:
/status 状态 - 总览(功耗/CPU/内存/温度/网络) return templates.render(
/power 功耗 - 功耗详情(RAPL/电池) "net", up=f"{net['up_mbps']}Mbps", down=f"{net['down_mbps']}Mbps")
/top 进程 - CPU TOP10 进程
/temp 温度 - 传感器温度
/net 网络 - 网络速率
/help 帮助 - 本帮助"""
# ---------- 命令处理 ---------- # ---------- 命令处理 ----------
@@ -177,10 +185,9 @@ class Bot:
if cmd in ("temp", "温度"): if cmd in ("temp", "温度"):
return fmt_temps({"temps": monitor.get_temps()}) return fmt_temps({"temps": monitor.get_temps()})
if cmd in ("net", "网络"): if cmd in ("net", "网络"):
net = monitor.get_net(interval=1.0) return fmt_net(monitor.get_net(interval=1.0))
return f"🌐 网络速率: ↑{net['up_mbps']}Mbps ↓{net['down_mbps']}Mbps"
if cmd in ("help", "帮助"): if cmd in ("help", "帮助"):
return HELP return templates.render("help")
return None return None
except Exception: except Exception:
log.exception("处理命令失败: %s", text) log.exception("处理命令失败: %s", text)
+21
View File
@@ -0,0 +1,21 @@
"""模板渲染:从 templates/ 目录加载 .txt 模板并用 str.format 填充。
查找顺序:本模块所在目录、当前工作目录。
"""
from pathlib import Path
_CANDIDATE_DIRS = [Path(__file__).parent / "templates", Path("templates")]
_cache: dict[str, str] = {}
def _dir() -> Path:
for d in _CANDIDATE_DIRS:
if d.is_dir():
return d
raise FileNotFoundError("未找到 templates/ 目录")
def render(name: str, **kwargs) -> str:
if name not in _cache:
_cache[name] = (_dir() / f"{name}.txt").read_text(encoding="utf-8")
return _cache[name].format(**kwargs)
+5
View File
@@ -0,0 +1,5 @@
[高负载告警] CPU 持续超过 {threshold}%,当前 {cpu}%
核心: {cores} 负载: {load}
TOP{top_n} 高负载进程:
{top_list}
+7
View File
@@ -0,0 +1,7 @@
可用命令(/ 或 ! 前缀均可,群聊可 @机器人):
/status 状态 - 总览(功耗/CPU/内存/温度/网络)
/power 功耗 - 功耗详情(RAPL/电池)
/top 进程 - CPU TOP10 进程
/temp 温度 - 传感器温度
/net 网络 - 网络速率
/help 帮助 - 本帮助
+1
View File
@@ -0,0 +1 @@
网络速率: 上行 {up} / 下行 {down}
+2
View File
@@ -0,0 +1,2 @@
功耗详情
{power}
+1
View File
@@ -0,0 +1 @@
[恢复] 服务器 CPU 已回落到 {cpu}%,负载恢复正常。
+8
View File
@@ -0,0 +1,8 @@
服务器状态
功耗: {power}
CPU: {cpu_total} / {cpu_max} (单核均值 {cpu_avg}%) 核心 {cores} 负载 {load}
内存: {memory}
最高温度: {temp}
网络: {net}
磁盘 {disk_mount}: {disk_used} ({disk_percent}%)
运行: {uptime}
+2
View File
@@ -0,0 +1,2 @@
传感器温度:
{temp_list}
+2
View File
@@ -0,0 +1,2 @@
CPU 占用 TOP{top_n}:
{top_list}