diff --git a/bot.py b/bot.py index 53238b1..147a4c4 100644 --- a/bot.py +++ b/bot.py @@ -1,6 +1,7 @@ """LinuxAwesomeBot 入口：启动 OneBot 反向 WS 服务端 + 命令处理 + 高负载告警。""" import asyncio import logging +import re import sys from pathlib import Path @@ -128,6 +129,17 @@ def fmt_net(net: dict) -> str: # ---------- 命令处理 ---------- +# 命令别名 -> 动作；只有此表内的词会触发，其他消息一律忽略 +COMMANDS = { + "status": "status", "状态": "status", + "power": "power", "功耗": "power", + "top": "top", "进程": "top", + "temp": "temp", "温度": "temp", + "net": "net", "网络": "net", + "help": "help", "帮助": "help", +} + + class Bot: def __init__(self, cfg: dict): self.cfg = cfg @@ -135,24 +147,28 @@ class Bot: cfg["server"]["host"], cfg["server"]["port"], cfg["server"].get("access_token", ""), ) + self._self_id: int | None = None # 机器人自己的 QQ 号，群聊 @ 校验用 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")): + msg_type = ev.get("message_type") + if msg_type == "group": + # 群聊必须 @ 机器人本人才响应 + if not await self._at_me(ev.get("message")): return - elif ev.get("message_type") == "private": + text = self._text_after_at(ev.get("message")) + elif msg_type == "private": text = self._extract_text(ev.get("message")) + if not text: # 纯图片/表情等无文本消息不响应 + return else: return reply = await self.handle_command(text) if reply is None: return - if ev.get("message_type") == "group": + if msg_type == "group": await self.ob.send_group_msg(ev["group_id"], reply) else: await self.ob.send_private_msg(ev["user_id"], reply) @@ -168,38 +184,69 @@ class Bot: ).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 "") + async def _get_self_id(self) -> int | None: + """获取并缓存机器人自己的 QQ 号。""" + if self._self_id is None: + try: + info = await self.ob.call_api("get_login_info") + self._self_id = int(info.get("user_id", 0)) or None + except Exception: + log.exception("获取机器人登录信息失败") + return self._self_id - @staticmethod - def _is_command(text: str) -> bool: - return text.startswith("/") or text.startswith("!") + async def _at_me(self, message) -> bool: + """校验消息中 @ 的是机器人本人（@其他人/全体不算）。""" + if self._self_id is None: + await self._get_self_id() + ats: list[str] = [] + if isinstance(message, list): + ats = [str(seg.get("data", {}).get("qq", "")) + for seg in message if isinstance(seg, dict) and seg.get("type") == "at"] + else: + ats = re.findall(r"\[CQ:at,qq=(\d+)", message or "") + return str(self._self_id or "") in ats + + def _text_after_at(self, message) -> str: + """取 @ 之后剩余的文本。""" + if isinstance(message, list): + after = False + parts = [] + for seg in message: + if not isinstance(seg, dict): + continue + if seg.get("type") == "at": + after = True + continue + if after and seg.get("type") == "text": + parts.append(seg.get("data", {}).get("text", "")) + return "".join(parts).strip() + if isinstance(message, str) and "[CQ:at" in message: + return message.split("]", 1)[-1].strip() + return self._extract_text(message) async def handle_command(self, text: str) -> str | None: cmd = text.lstrip("/!").strip().lower() cmd = cmd.split()[0] if cmd else "" + action = COMMANDS.get(cmd) + if action is None: # 非命令消息一律忽略 + return None try: - if cmd in ("status", "状态", ""): + if action == "status": return fmt_status(monitor.get_status()) - if cmd in ("power", "功耗"): + if action == "power": return fmt_power({"power": monitor.get_power()}) - if cmd in ("top", "进程"): + if action == "top": return fmt_top({"top": monitor.get_top_processes(10)}) - if cmd in ("temp", "温度"): + if action == "temp": return fmt_temps({"temps": monitor.get_temps()}) - if cmd in ("net", "网络"): + if action == "net": return fmt_net(monitor.get_net(interval=1.0)) - if cmd in ("help", "帮助"): + if action == "help": return templates.render("help") return None except Exception: log.exception("处理命令失败: %s", text) - return "❌ 命令执行出错，请查看服务端日志。" + return "[错误] 命令执行出错，请查看服务端日志。" async def run(self): await self.ob.start(self.on_event) diff --git a/tests/test_bot.py b/tests/test_bot.py index db6dfd8..7edfbbb 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -27,10 +27,15 @@ class FakeOneBotClient: async def _read(self): async for msg in self.ws: data = json.loads(msg.data) - if data.get("action"): # API 请求：记录消息内容并回执 - self.replies.append(data["params"].get("message", "")) + if data.get("action"): + if data["action"] == "get_login_info": + result = {"user_id": 88888, "nickname": "bot"} + else: + result = None + if data["action"].startswith("send_"): + self.replies.append(data["params"].get("message", "")) await self.ws.send_str(json.dumps( - {"status": "ok", "retcode": 0, "data": None, "echo": data.get("echo")})) + {"status": "ok", "retcode": 0, "data": result, "echo": data.get("echo")})) async def send_message(self, text: str): await self.ws.send_str(json.dumps({ @@ -38,6 +43,13 @@ class FakeOneBotClient: "user_id": 10000, "message": text, })) + async def send_group_message(self, message, group_id: int = 999): + """message 可为字符串或 CQ 段数组。""" + await self.ws.send_str(json.dumps({ + "post_type": "message", "message_type": "group", + "group_id": group_id, "user_id": 10000, "message": message, + })) + async def close(self): self._reader.cancel() await self.ws.close() @@ -90,3 +102,38 @@ async def test_status_cpu_display(client): reply = await wait_reply(client) # Linux 风格总占用 + 核心 assert "核心" in reply and "/" in reply + + +async def _expect_no_reply(client, send, timeout: float = 3.0): + """发送后确认机器人在超时内没有回复。""" + before = len(client.replies) + await send() + for _ in range(int(timeout * 10)): + if len(client.replies) > before: + raise AssertionError(f"不应回复但收到了: {client.replies[-1]}") + await asyncio.sleep(0.1) + + +@pytest.mark.asyncio +async def test_private_non_command_ignored(client): + # 私聊发普通消息/无文本消息不触发 + await _expect_no_reply(client, lambda: client.send_message("今天天气怎么样")) + await _expect_no_reply( + client, lambda: client.send_message( + [{"type": "image", "data": {"file": "abc.jpg"}}])) + + +@pytest.mark.asyncio +async def test_group_requires_at_me(client): + # 群聊不 @ 不响应 + await _expect_no_reply(client, lambda: client.send_group_message("/status")) + # @ 别人不响应 + await _expect_no_reply(client, lambda: client.send_group_message( + [{"type": "at", "data": {"qq": "66666"}}, + {"type": "text", "data": {"text": " /status"}}])) + # @ 本人才响应 + await client.send_group_message( + [{"type": "at", "data": {"qq": "88888"}}, + {"type": "text", "data": {"text": " /status"}}]) + reply = await wait_reply(client) + assert "服务器状态" in reply