✨ feat(command): 添加监听范围过滤和回复 @控制

- 新增 COMMAND_SCOPE 配置，支持 all/group/private 过滤消息来源 - 新增 COMMAND_ALLOWED_GROUPS 群号白名单，逗号分隔，留空不限制 - 新增 COMMAND_ALLOWED_USERS QQ 号白名单，逗号分隔，留空不限制 - 新增 COMMAND_AT_SENDER 配置，控制回复时是否 @发送者（默认 true） - 回调响应中 at_sender 字段可覆盖全局配置 - 更新 .env.example 和 README.md 文档
2026-05-03 00:01:10 +08:00
parent 84f671741b
commit ed6e27f162
5 changed files with 98 additions and 17 deletions
--- a/.env.example
+++ b/.env.example
@@ -19,7 +19,19 @@ QQ_API_MAX_RETRIES=2
 # ── 命令监听 ──
 # 命令前缀，默认 #
 COMMAND_PREFIX=#
-# 命令名长度（中文字数），默认 4
-COMMAND_LENGTH=4
+# 命令名最小字符数，默认 2
+COMMAND_LENGTH_MIN=2
+# 命令名最大字符数，默认 4
+COMMAND_LENGTH_MAX=4
+# 监听范围：all（群+私）、group（仅群）、private（仅私），默认 all
+COMMAND_SCOPE=all
+# 允许的群号，逗号分隔，留空不限制，例：123456,789012
+COMMAND_ALLOWED_GROUPS=
+# 允许的 QQ 号，逗号分隔，留空不限制，例：111111,222222
+COMMAND_ALLOWED_USERS=
+# 回复时是否 @发送者，默认 true
+COMMAND_AT_SENDER=true
+# 回调超时秒数，默认 180（生图等耗时命令需要较长超时）
+COMMAND_CALLBACK_TIMEOUT=180
 # 匹配到命令后的回调 URL，留空则不监听
 COMMAND_CALLBACK_URL=
--- a/README.md
+++ b/README.md
@@ -42,7 +42,13 @@ uv run python -m ncatbot
 | `QQ_API_TIMEOUT` | 否 | `10` | QQ API 超时秒数 |
 | `QQ_API_MAX_RETRIES` | 否 | `2` | QQ API 失败重试次数 |
 | `COMMAND_PREFIX` | 否 | `#` | 命令前缀 |
-| `COMMAND_LENGTH` | 否 | `4` | 命令名字符数（中文字数） |
+| `COMMAND_LENGTH_MIN` | 否 | `2` | 命令名最小字符数 |
+| `COMMAND_LENGTH_MAX` | 否 | `4` | 命令名最大字符数 |
+| `COMMAND_SCOPE` | 否 | `all` | 监听范围：`all`（群+私）、`group`（仅群）、`private`（仅私） |
+| `COMMAND_ALLOWED_GROUPS` | 否 | 空（不限） | 允许的群号，逗号分隔，如 `123456,789012` |
+| `COMMAND_ALLOWED_USERS` | 否 | 空（不限） | 允许的 QQ 号，逗号分隔，如 `111111,222222` |
+| `COMMAND_AT_SENDER` | 否 | `true` | 回复时是否 @发送者 |
+| `COMMAND_CALLBACK_TIMEOUT` | 否 | `180` | 回调超时秒数 |
 | `COMMAND_CALLBACK_URL` | 否 | 空（不监听） | 命令匹配后的回调 URL |

 ## 接口说明
@@ -187,22 +193,34 @@ curl -X POST http://localhost:8081/webhook \

 ### 命令监听

-插件会自动监听 QQ 消息，当消息以 `#四个中文字+空格` 开头时，将命令内容 POST 到 `COMMAND_CALLBACK_URL`。
+插件会自动监听 QQ 消息，当消息以 `#命令名` 开头时，将命令内容 POST 到 `COMMAND_CALLBACK_URL`。

 **匹配规则：**

 ```
 #测试命令 你好世界
 │      │ │       │
-│      │ │       └── 命令内容（content）
-│      └── 空格分隔
-└── 命令名（4个中文字）
+│      │ │       └── 命令内容（content，可选）
+│      └── 空格分隔（可选）
+└── 命令名（2~4个字符）
 └── 前缀（默认 #）
 ```

- 前缀、命令名长度可通过 `COMMAND_PREFIX`、`COMMAND_LENGTH` 配置
+- 命令名支持中文、数字、字母等任意非空白字符，每个字符计 1
+- 前缀通过 `COMMAND_PREFIX` 配置，长度范围通过 `COMMAND_LENGTH_MIN` / `COMMAND_LENGTH_MAX` 配置
+- `#测试命令`、`#1a`、`#abc` 均可触发
 - 不配置 `COMMAND_CALLBACK_URL` 则不监听

+**监听范围过滤：**
+
+- `COMMAND_SCOPE`：控制监听范围
+  - `all`（默认）：群聊 + 私聊都监听
+  - `group`：仅监听群聊
+  - `private`：仅监听私聊
+- `COMMAND_ALLOWED_GROUPS`：群号白名单，逗号分隔，留空不限制
+- `COMMAND_ALLOWED_USERS`：QQ 号白名单，逗号分隔，留空不限制
+- 三个条件同时生效，必须全部满足才触发回调
+
 **回调请求体（POST JSON）：**

 ```json
@@ -218,7 +236,7 @@ curl -X POST http://localhost:8081/webhook \

 | 字段 | 说明 |
 |---|---|
-| `command` | 命令名（4个中文字） |
+| `command` | 命令名（2~4个字符） |
 | `content` | 命令后的内容 |
 | `raw_message` | 原始消息文本 |
 | `user_id` | 发送者 QQ 号 |
@@ -249,6 +267,7 @@ curl -X POST http://localhost:8081/webhook \
 |---|---|
 | `reply` | 纯文本回复（与 `messages` 二选一，`messages` 优先） |
 | `messages` | 批量回复数组，格式同 `/webhook` 的 `messages` 字段 |
+| `at_sender` | 是否 @发送者（默认取 `COMMAND_AT_SENDER` 配置，仅群聊生效） |
 | `group_id` | 可选，覆盖回复目标群号（默认回复到原群） |
 | `user_id` | 可选，覆盖回复目标 QQ 号（默认回复到原发送者） |

--- a/config.py
+++ b/config.py
@@ -38,3 +38,11 @@ COMMAND_LENGTH_MIN: int = int(os.environ.get("COMMAND_LENGTH_MIN", "2"))
 COMMAND_LENGTH_MAX: int = int(os.environ.get("COMMAND_LENGTH_MAX", "4"))
 COMMAND_CALLBACK_URL: str = os.environ.get("COMMAND_CALLBACK_URL", "")
 COMMAND_CALLBACK_TIMEOUT: int = int(os.environ.get("COMMAND_CALLBACK_TIMEOUT", "180"))
+COMMAND_SCOPE: str = os.environ.get("COMMAND_SCOPE", "all")  # all / group / private
+COMMAND_ALLOWED_GROUPS: frozenset[str] = frozenset(
+    filter(None, os.environ.get("COMMAND_ALLOWED_GROUPS", "").split(","))
+)
+COMMAND_ALLOWED_USERS: frozenset[str] = frozenset(
+    filter(None, os.environ.get("COMMAND_ALLOWED_USERS", "").split(","))
+)
+COMMAND_AT_SENDER: bool = os.environ.get("COMMAND_AT_SENDER", "true").lower() in ("true", "1", "yes")
--- a/handlers/command.py
+++ b/handlers/command.py
@@ -4,7 +4,7 @@ import re

 import aiohttp

-from ..config import COMMAND_CALLBACK_TIMEOUT, COMMAND_CALLBACK_URL, COMMAND_LENGTH_MAX, COMMAND_LENGTH_MIN, COMMAND_PREFIX, UPLOAD_DIR
+from ..config import COMMAND_AT_SENDER, COMMAND_CALLBACK_TIMEOUT, COMMAND_CALLBACK_URL, COMMAND_LENGTH_MAX, COMMAND_LENGTH_MIN, COMMAND_PREFIX, UPLOAD_DIR
 from ..handlers.message import _resolve_url


@@ -50,7 +50,7 @@ async def send_command_callback(data: dict, event, api, logger) -> None:
            {"type": "file", "url": "..."},
            {"type": "video", "url": "..."}
        ],
-        "at_sender": true                      // 是否 @发送者（默认 true，仅群聊）
+        "at_sender": true                      // 是否 @发送者（默认取 COMMAND_AT_SENDER 配置，仅群聊）
    }

    所有字段均为可选，无回复内容时返回空 JSON 即可。
@@ -92,7 +92,8 @@ async def send_command_callback(data: dict, event, api, logger) -> None:

 async def _handle_reply(result: dict, msg_event, api, logger) -> None:
    """处理回调响应，引用原消息自动回复。msg_event 是 GroupMessageEvent / PrivateMessageEvent。"""
-    at_sender = result.get("at_sender", True)
+    # at_sender: 回调响应中的值优先，未指定则使用全局配置
+    at_sender = result.get("at_sender", COMMAND_AT_SENDER)
    messages = result.get("messages")
    reply = result.get("reply")
    group_id = getattr(msg_event, "group_id", None)
--- a/plugin.py
+++ b/plugin.py
@@ -7,7 +7,20 @@ import os
 from aiohttp import web
 from ncatbot.plugin import NcatBotPlugin

-from .config import COMMAND_CALLBACK_URL, COMMAND_LENGTH_MAX, COMMAND_LENGTH_MIN, COMMAND_PREFIX, HOST, PORT, UPLOAD_DIR, WEBHOOK_API_KEY
+from .config import (
+    COMMAND_ALLOWED_GROUPS,
+    COMMAND_ALLOWED_USERS,
+    COMMAND_AT_SENDER,
+    COMMAND_CALLBACK_URL,
+    COMMAND_LENGTH_MAX,
+    COMMAND_LENGTH_MIN,
+    COMMAND_PREFIX,
+    COMMAND_SCOPE,
+    HOST,
+    PORT,
+    UPLOAD_DIR,
+    WEBHOOK_API_KEY,
+)
 from .handlers.command import parse_command, send_command_callback
 from .handlers.health import health_handler
 from .handlers.message import webhook_handler
@@ -31,9 +44,18 @@ class WebHookPlugin(NcatBotPlugin):

    async def on_load(self):
        self.logger.info("Webhook 插件已加载")
-        self.logger.info("WEBHOOK_API_KEY: %s", "已配置" if os.environ.get("WEBHOOK_API_KEY") else "自动生成")
-        self.logger.info("命令监听: 前缀=%s 长度=%d~%d 回调=%s", COMMAND_PREFIX, COMMAND_LENGTH_MIN, COMMAND_LENGTH_MAX,
-                         COMMAND_CALLBACK_URL or "未配置")
+        self.logger.info(
+            "WEBHOOK_API_KEY: %s",
+            "已配置" if os.environ.get("WEBHOOK_API_KEY") else "自动生成",
+        )
+        self.logger.info(
+            "命令监听: 前缀=%s 长度=%d~%d 范围=%s 回调=%s",
+            COMMAND_PREFIX,
+            COMMAND_LENGTH_MIN,
+            COMMAND_LENGTH_MAX,
+            COMMAND_SCOPE,
+            COMMAND_CALLBACK_URL or "未配置",
+        )
        asyncio.create_task(self._start_webhook())
        self._cleanup_task = asyncio.create_task(self._cleanup_loop())
        self._listener_task = asyncio.create_task(self._message_listener())
@@ -77,6 +99,23 @@ class WebHookPlugin(NcatBotPlugin):
                        parsed = parse_command(raw_message)
                        if not parsed:
                            continue
+
+                        # 范围过滤：group / private / all
+                        is_group = hasattr(event.data, "group_id")
+                        if COMMAND_SCOPE == "group" and not is_group:
+                            continue
+                        if COMMAND_SCOPE == "private" and is_group:
+                            continue
+
+                        # 群白名单过滤
+                        if COMMAND_ALLOWED_GROUPS and is_group:
+                            if event.data.group_id not in COMMAND_ALLOWED_GROUPS:
+                                continue
+
+                        # 用户白名单过滤
+                        if COMMAND_ALLOWED_USERS and event.data.user_id not in COMMAND_ALLOWED_USERS:
+                            continue
+
                        # 构建回调数据
                        data = {
                            "command": parsed["command"],
@@ -89,7 +128,9 @@ class WebHookPlugin(NcatBotPlugin):
                            data["group_id"] = event.data.group_id
                        self.logger.info(
                            "命令监听匹配: command=%s user=%s group=%s",
-                            parsed["command"], data["user_id"], data.get("group_id", "-"),
+                            parsed["command"],
+                            data["user_id"],
+                            data.get("group_id", "-"),
                        )
                        asyncio.create_task(
                            send_command_callback(data, event, self.api, self.logger)