fix: require @bot in groups, verify at target, ignore non-command and textless messages

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)