Merge pull request '命令监听配置重构：SQLite → YAML + 黑白名单过滤优化' (#2) from feat/command-scope into main

Reviewed-on: #2

.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=

.gitignore

@@ -15,3 +15,9 @@ uploads/
 
 # Environment
 .env
+
+# Dynamic config
+settings.yaml
+
+# Legacy data
+data/

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 号（默认回复到原发送者） |
 

config.py

@@ -1,40 +1,284 @@
-"""项目配置：所有值从环境变量读取，未配置时使用安全默认值。"""
+"""项目配置：静态配置从环境变量读取，命令监听配置通过 settings.yaml 动态管理。"""
 
+import logging
 import os
 import uuid
+from dataclasses import dataclass
 from pathlib import Path
 
+import yaml
 from dotenv import load_dotenv
 
 # 加载 .env 文件（优先从插件目录查找）
 load_dotenv(Path(__file__).parent / ".env")
 
-# ── 鉴权 ────────────────────────────────────────────────────
-WEBHOOK_API_KEY: str = os.environ.get("WEBHOOK_API_KEY", "") or uuid.uuid4().hex
+logger = logging.getLogger("webhook-plugin.config")
 
-# ── 网络 ─────────────────────────────────────────────────────
+# ── 静态配置（环境变量，运行时不变）────────────────────────────
+WEBHOOK_API_KEY: str = os.environ.get("WEBHOOK_API_KEY", "") or uuid.uuid4().hex
 HOST: str = os.environ.get("WEBHOOK_HOST", "0.0.0.0")
 try:
     PORT: int = int(os.environ.get("WEBHOOK_PORT", "8081"))
 except ValueError:
     PORT = 8081
 
-# ── 上传 ─────────────────────────────────────────────────────
 UPLOAD_DIR: Path = Path(os.environ.get("UPLOAD_DIR", str(Path(__file__).parent / "uploads")))
-# 单个文件最大 20 MB
 MAX_UPLOAD_SIZE: int = int(os.environ.get("MAX_UPLOAD_SIZE", str(20 * 1024 * 1024)))
-# 允许的文件扩展名（小写，不含点），为空则不限制
 ALLOWED_EXTENSIONS: set[str] = set(
     filter(None, os.environ.get("ALLOWED_EXTENSIONS", "").lower().split(","))
 )
 
-# ── QQ API ───────────────────────────────────────────────────
 QQ_API_TIMEOUT: float = float(os.environ.get("QQ_API_TIMEOUT", "10"))
 QQ_API_MAX_RETRIES: int = int(os.environ.get("QQ_API_MAX_RETRIES", "2"))
 
-# ── 命令监听 ────────────────────────────────────────────────
-COMMAND_PREFIX: str = os.environ.get("COMMAND_PREFIX", "#")
-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"))
+# ── YAML 文件路径 ────────────────────────────────────────────
+SETTINGS_YAML_PATH: Path = Path(__file__).parent / "settings.yaml"
+
+# ── 动态配置 dataclass ──────────────────────────────────────
+@dataclass
+class CommandConfig:
+    """命令监听动态配置，通过 settings.yaml 热重载。"""
+
+    prefix: str = "#"
+    length_min: int = 2
+    length_max: int = 4
+    scope: str = "all"           # all / group / private
+    list_enabled: bool = False   # 是否启用黑白名单过滤
+    list_mode: str = "allow"     # allow / deny
+    allowed_groups: frozenset[str] = frozenset()
+    denied_groups: frozenset[str] = frozenset()
+    allowed_users: frozenset[str] = frozenset()
+    denied_users: frozenset[str] = frozenset()
+    at_sender: bool = True
+    callback_url: str = ""
+    callback_timeout: int = 180
+
+
+# 全局单例 —— 所有消费者通过 config.command.xxx 访问，避免 stale import 问题
+command = CommandConfig()
+
+# ── 环境变量默认值（YAML 缺失 key 时的回退）────────────────────
+_ENV_DEFAULTS: dict = {
+    "prefix": os.environ.get("COMMAND_PREFIX", "#"),
+    "length_min": int(os.environ.get("COMMAND_LENGTH_MIN", "2")),
+    "length_max": int(os.environ.get("COMMAND_LENGTH_MAX", "4")),
+    "scope": os.environ.get("COMMAND_SCOPE", "all"),
+    "list_enabled": os.environ.get("COMMAND_LIST_ENABLED", "false").lower() in ("true", "1", "yes"),
+    "list_mode": os.environ.get("COMMAND_LIST_MODE", "allow"),
+    "allowed_groups": os.environ.get("COMMAND_ALLOWED_GROUPS", ""),
+    "denied_groups": os.environ.get("COMMAND_DENIED_GROUPS", ""),
+    "allowed_users": os.environ.get("COMMAND_ALLOWED_USERS", ""),
+    "denied_users": os.environ.get("COMMAND_DENIED_USERS", ""),
+    "at_sender": os.environ.get("COMMAND_AT_SENDER", "true").lower() in ("true", "1", "yes"),
+    "callback_url": os.environ.get("COMMAND_CALLBACK_URL", ""),
+    "callback_timeout": int(os.environ.get("COMMAND_CALLBACK_TIMEOUT", "180")),
+}
+
+# ── YAML I/O ─────────────────────────────────────────────────
+_YAML_HEADER = (
+    "# settings.yaml - 命令监听动态配置\n"
+    "# 可通过后台管理界面修改，也可手动编辑（自动热重载）\n\n"
+)
+
+
+def _yaml_defaults() -> dict:
+    """返回默认 YAML 结构（首次启动时写入）。"""
+    return {
+        "command": {
+            "prefix": "#",
+            "length_min": 2,
+            "length_max": 4,
+            "scope": "all",
+            "list_enabled": False,
+            "list_mode": "allow",
+            "allowed_groups": [],
+            "denied_groups": [],
+            "allowed_users": [],
+            "denied_users": [],
+            "at_sender": True,
+            "callback_url": "",
+            "callback_timeout": 180,
+        }
+    }
+
+
+def load_yaml() -> dict:
+    """读取 settings.yaml，文件不存在时返回空字典。"""
+    if not SETTINGS_YAML_PATH.exists():
+        return {}
+    with open(SETTINGS_YAML_PATH, "r", encoding="utf-8") as f:
+        data = yaml.safe_load(f)
+    return data if isinstance(data, dict) else {}
+
+
+def save_yaml(data: dict) -> None:
+    """将配置写入 settings.yaml。"""
+    with open(SETTINGS_YAML_PATH, "w", encoding="utf-8") as f:
+        f.write(_YAML_HEADER)
+        yaml.dump(data, f, default_flow_style=False, allow_unicode=True, sort_keys=False)
+
+
+def ensure_settings_yaml() -> None:
+    """首次启动时创建 settings.yaml，合并环境变量覆盖。"""
+    if SETTINGS_YAML_PATH.exists():
+        return
+    data = _yaml_defaults()
+    # 用环境变量覆盖默认值
+    env_map = {
+        "prefix": ("COMMAND_PREFIX", str),
+        "length_min": ("COMMAND_LENGTH_MIN", int),
+        "length_max": ("COMMAND_LENGTH_MAX", int),
+        "scope": ("COMMAND_SCOPE", str),
+        "list_mode": ("COMMAND_LIST_MODE", str),
+        "list_enabled": ("COMMAND_LIST_ENABLED", "bool"),
+        "allowed_groups": ("COMMAND_ALLOWED_GROUPS", "csv_list"),
+        "denied_groups": ("COMMAND_DENIED_GROUPS", "csv_list"),
+        "allowed_users": ("COMMAND_ALLOWED_USERS", "csv_list"),
+        "denied_users": ("COMMAND_DENIED_USERS", "csv_list"),
+        "at_sender": ("COMMAND_AT_SENDER", "bool"),
+        "callback_url": ("COMMAND_CALLBACK_URL", str),
+        "callback_timeout": ("COMMAND_CALLBACK_TIMEOUT", int),
+    }
+    for key, (env_name, conv) in env_map.items():
+        env_val = os.environ.get(env_name)
+        if env_val is not None:
+            if conv == "csv_list":
+                data["command"][key] = [v.strip() for v in env_val.split(",") if v.strip()]
+            elif conv == "bool":
+                data["command"][key] = env_val.lower() in ("true", "1", "yes")
+            else:
+                data["command"][key] = conv(env_val)
+    save_yaml(data)
+    logger.info("Created settings.yaml with defaults (env-var overrides applied)")
+
+
+# ── 将 YAML 数据应用到全局 command 对象 ───────────────────────
+def _apply_yaml_to_command(data: dict) -> None:
+    """将 YAML command 段合并到全局 command 对象，缺失 key 回退到环境变量默认值。"""
+    cmd_data = data.get("command", {})
+
+    command.prefix = str(cmd_data.get("prefix", _ENV_DEFAULTS["prefix"]))
+    command.length_min = int(cmd_data.get("length_min", _ENV_DEFAULTS["length_min"]))
+    command.length_max = int(cmd_data.get("length_max", _ENV_DEFAULTS["length_max"]))
+    command.scope = str(cmd_data.get("scope", _ENV_DEFAULTS["scope"]))
+    command.list_mode = str(cmd_data.get("list_mode", _ENV_DEFAULTS["list_mode"]))
+
+    # list_enabled: YAML bool → Python bool
+    list_enabled_val = cmd_data.get("list_enabled", _ENV_DEFAULTS["list_enabled"])
+    if isinstance(list_enabled_val, str):
+        command.list_enabled = list_enabled_val.lower() in ("true", "1", "yes")
+    else:
+        command.list_enabled = bool(list_enabled_val)
+
+    command.callback_url = str(cmd_data.get("callback_url", _ENV_DEFAULTS["callback_url"]))
+    command.callback_timeout = int(cmd_data.get("callback_timeout", _ENV_DEFAULTS["callback_timeout"]))
+
+    # at_sender: YAML bool → Python bool
+    at_sender_val = cmd_data.get("at_sender", _ENV_DEFAULTS["at_sender"])
+    if isinstance(at_sender_val, str):
+        command.at_sender = at_sender_val.lower() in ("true", "1", "yes")
+    else:
+        command.at_sender = bool(at_sender_val)
+
+    # 列表字段: YAML list → frozenset
+    for field_name in ("allowed_groups", "denied_groups", "allowed_users", "denied_users"):
+        raw = cmd_data.get(field_name)
+        if raw is None:
+            # 回退到环境变量默认值
+            env_default = _ENV_DEFAULTS[field_name]
+            if isinstance(env_default, str):
+                setattr(command, field_name, frozenset(
+                    v.strip() for v in env_default.split(",") if v.strip()
+                ))
+            else:
+                setattr(command, field_name, frozenset(env_default) if env_default else frozenset())
+        elif isinstance(raw, str):
+            setattr(command, field_name, frozenset(
+                v.strip() for v in raw.split(",") if v.strip()
+            ))
+        elif isinstance(raw, list):
+            setattr(command, field_name, frozenset(str(v).strip() for v in raw if str(v).strip()))
+        else:
+            setattr(command, field_name, frozenset())
+
+    # 重建命令匹配正则
+    from .handlers.command import rebuild_pattern
+    rebuild_pattern()
+
+
+# ── 公共 API ─────────────────────────────────────────────────
+def reload_settings() -> None:
+    """重新读取 settings.yaml 并应用到全局 command 对象。"""
+    data = load_yaml()
+    _apply_yaml_to_command(data)
+    logger.info("Config reloaded from settings.yaml")
+
+
+def get_settings_flat() -> dict[str, str]:
+    """返回所有动态配置的扁平字典 {key: str_value}，供管理 API 使用。"""
+    return {
+        "command_prefix": command.prefix,
+        "command_length_min": str(command.length_min),
+        "command_length_max": str(command.length_max),
+        "command_scope": command.scope,
+        "command_list_mode": command.list_mode,
+        "command_list_enabled": str(command.list_enabled).lower(),
+        "command_allowed_groups": ",".join(sorted(command.allowed_groups)),
+        "command_denied_groups": ",".join(sorted(command.denied_groups)),
+        "command_allowed_users": ",".join(sorted(command.allowed_users)),
+        "command_denied_users": ",".join(sorted(command.denied_users)),
+        "command_at_sender": str(command.at_sender).lower(),
+        "command_callback_url": command.callback_url,
+        "command_callback_timeout": str(command.callback_timeout),
+    }
+
+
+# API 平坦 key → (YAML key, 类型转换) 映射
+_API_KEY_MAP: dict[str, tuple[str, ...]] = {
+    "command_prefix": ("prefix", str),
+    "command_length_min": ("length_min", int),
+    "command_length_max": ("length_max", int),
+    "command_scope": ("scope", str),
+    "command_list_mode": ("list_mode", str),
+    "command_list_enabled": ("list_enabled", "bool"),
+    "command_at_sender": ("at_sender", "bool"),
+    "command_callback_url": ("callback_url", str),
+    "command_callback_timeout": ("callback_timeout", int),
+    "command_allowed_groups": ("allowed_groups", "csv_list"),
+    "command_denied_groups": ("denied_groups", "csv_list"),
+    "command_allowed_users": ("allowed_users", "csv_list"),
+    "command_denied_users": ("denied_users", "csv_list"),
+}
+
+
+def update_settings_from_api(data: dict[str, str]) -> dict[str, str]:
+    """从管理 API 接收扁平字典，合并到 settings.yaml 并热重载。返回已应用的字段。"""
+    allowed_keys = set(_API_KEY_MAP.keys())
+    filtered = {k: str(v) for k, v in data.items() if k in allowed_keys}
+    if not filtered:
+        return {}
+
+    # 读取当前 YAML，合并修改
+    yaml_data = load_yaml()
+    if "command" not in yaml_data:
+        yaml_data["command"] = {}
+    cmd = yaml_data["command"]
+
+    for flat_key, value in filtered.items():
+        yaml_key, conv = _API_KEY_MAP[flat_key]
+        if conv == "csv_list":
+            cmd[yaml_key] = [v.strip() for v in value.split(",") if v.strip()]
+        elif conv == "bool":
+            cmd[yaml_key] = value.lower() in ("true", "1", "yes")
+        elif conv is int:
+            try:
+                cmd[yaml_key] = int(value)
+            except ValueError:
+                pass
+        else:
+            cmd[yaml_key] = value
+
+    save_yaml(yaml_data)
+    # 写入后立即应用（不等文件 watcher）
+    _apply_yaml_to_command(yaml_data)
+    return filtered

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 UPLOAD_DIR, command
 from ..handlers.message import _resolve_url
 
 
@@ -18,7 +18,7 @@ def build_command_pattern() -> re.Pattern:
     - 其他非空白字符 = 1
     """
     return re.compile(
-        rf"^{re.escape(COMMAND_PREFIX)}(\S{{{COMMAND_LENGTH_MIN},{COMMAND_LENGTH_MAX}}})(?:\s+(.+))?$",
+        rf"^{re.escape(command.prefix)}(\S{{{command.length_min},{command.length_max}}})(?:\s+(.+))?$",
         re.DOTALL,
     )
 
@@ -26,6 +26,12 @@ def build_command_pattern() -> re.Pattern:
 COMMAND_PATTERN = build_command_pattern()
 
 
+def rebuild_pattern() -> None:
+    """动态配置变更后重新编译正则。"""
+    global COMMAND_PATTERN
+    COMMAND_PATTERN = build_command_pattern()
+
+
 def parse_command(raw_message: str) -> dict | None:
     """解析消息，匹配命令模式。返回 {command, content, raw_message} 或 None。"""
     match = COMMAND_PATTERN.match(raw_message.strip())
@@ -50,28 +56,28 @@ async def send_command_callback(data: dict, event, api, logger) -> None:
             {"type": "file", "url": "..."},
             {"type": "video", "url": "..."}
         ],
-        "at_sender": true                      // 是否 @发送者（默认 true，仅群聊）
+        "at_sender": true                      // 是否 @发送者（默认取配置，仅群聊）
     }
 
     所有字段均为可选，无回复内容时返回空 JSON 即可。
     回复会引用触发命令的原消息。
     """
-    if not COMMAND_CALLBACK_URL:
-        logger.warning("COMMAND_CALLBACK_URL 未配置，跳过命令回调")
+    if not command.callback_url:
+        logger.warning("callback_url 未配置，跳过命令回调")
         return
 
     try:
         async with aiohttp.ClientSession() as session:
             async with session.post(
-                COMMAND_CALLBACK_URL,
+                command.callback_url,
                 json=data,
-                timeout=aiohttp.ClientTimeout(total=COMMAND_CALLBACK_TIMEOUT),
+                timeout=aiohttp.ClientTimeout(total=command.callback_timeout),
             ) as resp:
                 if resp.status >= 400:
                     body = await resp.text()
                     logger.error(
                         "命令回调失败: status=%d url=%s body=%s",
-                        resp.status, COMMAND_CALLBACK_URL, body[:200],
+                        resp.status, command.callback_url, body[:200],
                     )
                     return
 
@@ -87,12 +93,13 @@ async def send_command_callback(data: dict, event, api, logger) -> None:
                 await _handle_reply(result, event.data, api, logger)
 
     except Exception as exc:
-        logger.error("命令回调异常: url=%s error=%s", COMMAND_CALLBACK_URL, exc)
+        logger.error("命令回调异常: url=%s error=%s", command.callback_url, exc)
 
 
 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)

middleware.py

@@ -10,9 +10,9 @@ from .response import error
 
 @web.middleware
 async def auth_middleware(request: web.Request, handler):
-    """对 /upload 和 /webhook 路径强制校验 API Key。"""
-    # 健康检查不需要鉴权
-    if request.path == "/healthz":
+    """对需要鉴权的路径校验 API Key。/healthz 和 /admin/ 及 /api/ 开头的路径不需要鉴权。"""
+    # 不需要鉴权的路径
+    if request.path == "/healthz" or request.path.startswith("/admin") or request.path.startswith("/api/"):
         return await handler(request)
 
     auth_header = request.headers.get("Authorization", "")

plugin.py

@@ -7,7 +7,17 @@ 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 (
+    HOST,
+    PORT,
+    UPLOAD_DIR,
+    WEBHOOK_API_KEY,
+    SETTINGS_YAML_PATH,
+    command,
+    ensure_settings_yaml,
+    reload_settings,
+)
+from .handlers.admin import admin_page_handler, api_get_settings, api_reload_settings, api_update_settings
 from .handlers.command import parse_command, send_command_callback
 from .handlers.health import health_handler
 from .handlers.message import webhook_handler
@@ -28,31 +38,43 @@ class WebHookPlugin(NcatBotPlugin):
         self._webhook_runner: web.AppRunner | None = None
         self._cleanup_task: asyncio.Task | None = None
         self._listener_task: asyncio.Task | None = None
+        self._watcher_task: asyncio.Task | None = None
 
     async def on_load(self):
+        # 初始化 settings.yaml 并加载配置
+        ensure_settings_yaml()
+        reload_settings()
+
         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(%s) 回调=%s",
+            command.prefix,
+            command.length_min,
+            command.length_max,
+            command.scope,
+            "开" if command.list_enabled else "关",
+            command.list_mode,
+            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())
+        self._watcher_task = asyncio.create_task(self._config_watcher())
 
     async def on_close(self):
-        if self._listener_task is not None:
-            self._listener_task.cancel()
-            try:
-                await self._listener_task
-            except asyncio.CancelledError:
-                pass
-            self._listener_task = None
-        if self._cleanup_task is not None:
-            self._cleanup_task.cancel()
-            try:
-                await self._cleanup_task
-            except asyncio.CancelledError:
-                pass
-            self._cleanup_task = None
+        for task_attr in ("_watcher_task", "_listener_task", "_cleanup_task"):
+            task = getattr(self, task_attr)
+            if task is not None:
+                task.cancel()
+                try:
+                    await task
+                except asyncio.CancelledError:
+                    pass
+                setattr(self, task_attr, None)
         await self._stop_webhook()
         self.logger.info("Webhook 插件已卸载")
 
@@ -65,6 +87,21 @@ class WebHookPlugin(NcatBotPlugin):
             except Exception as exc:
                 self.logger.error("清理过期文件失败: %s", exc)
 
+    async def _config_watcher(self) -> None:
+        """轮询 settings.yaml 的 mtime，变更时热重载配置。"""
+        last_mtime: float = 0.0
+        while True:
+            await asyncio.sleep(2)
+            try:
+                if SETTINGS_YAML_PATH.exists():
+                    current_mtime = SETTINGS_YAML_PATH.stat().st_mtime
+                    if current_mtime != last_mtime:
+                        last_mtime = current_mtime
+                        reload_settings()
+                        self.logger.info("settings.yaml changed, config reloaded")
+            except Exception as exc:
+                self.logger.error("Config watcher error: %s", exc)
+
     async def _message_listener(self) -> None:
         """监听 QQ 消息，匹配命令模式后转发到外部回调。"""
         try:
@@ -77,6 +114,31 @@ 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.list_enabled:
+                            if command.list_mode == "allow":
+                                # 白名单模式（OR 逻辑）：用户在名单 OR 群在名单 → 放行
+                                # 名单为空时视为不限制
+                                if command.allowed_users or command.allowed_groups:
+                                    user_ok = event.data.user_id in command.allowed_users
+                                    group_ok = is_group and event.data.group_id in command.allowed_groups
+                                    if not (user_ok or group_ok):
+                                        continue
+                            elif command.list_mode == "deny":
+                                # 黑名单模式：用户在名单 OR 群在名单 → 拒绝
+                                user_blocked = event.data.user_id in command.denied_users
+                                group_blocked = is_group and event.data.group_id in command.denied_groups
+                                if user_blocked or group_blocked:
+                                    continue
+
                         # 构建回调数据
                         data = {
                             "command": parsed["command"],
@@ -85,11 +147,13 @@ class WebHookPlugin(NcatBotPlugin):
                             "user_id": event.data.user_id,
                             "message_id": event.data.message_id,
                         }
-                        if hasattr(event.data, "group_id"):
+                        if is_group:
                             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)
@@ -106,6 +170,11 @@ class WebHookPlugin(NcatBotPlugin):
         app.router.add_get("/healthz", health_handler)
         app.router.add_post("/webhook", webhook_handler)
         app.router.add_post("/upload", upload_handler)
+        # 后台管理
+        app.router.add_get("/admin/", admin_page_handler)
+        app.router.add_get("/api/settings", api_get_settings)
+        app.router.add_put("/api/settings", api_update_settings)
+        app.router.add_post("/api/settings/reload", api_reload_settings)
         return app
 
     async def _start_webhook(self):
@@ -117,9 +186,10 @@ class WebHookPlugin(NcatBotPlugin):
         await site.start()
         self.logger.info("Webhook 已启动: %s:%d", HOST, PORT)
         self.logger.info("上传目录: %s", UPLOAD_DIR)
+        self.logger.info("后台管理: http://%s:%d/admin/", HOST, PORT)
 
     async def _stop_webhook(self):
         if self._webhook_runner is not None:
             await self._webhook_runner.cleanup()
             self._webhook_runner = None
-            self.logger.info("Webhook 已停止")
+            self.logger.info("Webhook 已停止")

pyproject.toml

@@ -8,4 +8,5 @@ dependencies = [
     "ncatbot5>=5.5.2.post3",
     "aiohttp>=3.9",
     "python-dotenv>=1.0",
+    "pyyaml>=6.0",
 ]

uv.lock

@@ -911,6 +911,7 @@ dependencies = [
     { name = "aiohttp" },
     { name = "ncatbot5" },
     { name = "python-dotenv" },
+    { name = "pyyaml" },
 ]
 
 [package.metadata]
@@ -918,6 +919,7 @@ requires-dist = [
     { name = "aiohttp", specifier = ">=3.9" },
     { name = "ncatbot5", specifier = ">=5.5.2.post3" },
     { name = "python-dotenv", specifier = ">=1.0" },
+    { name = "pyyaml", specifier = ">=6.0" },
 ]
 
 [[package]]