简体中文 English 繁體中文 日本語 Русский
本文为静态镜像，内容以交互版为准 在交互式文档中心打开 →

ErisPulse.Core.Event.command 模块

模块概述

ErisPulse 命令处理模块

提供基于装饰器的命令注册和处理功能

命令是特殊的消息事件处理器（ErisPulse 扩展类型），其用户侧配置 由统一覆写系统持有（ErisPulse.event.overrides，见 :mod:ErisPulse.Core.Event.overrides）：用户黑白名单（ACL） 在 overrides.acl 类别（命令名支持 glob，acl_default_allow 兜底严格模式）； 实现参数覆写（master / hidden / aliases 等）在 overrides.command 类别 （用户优先语义）。本模块的命令判定链消费覆写系统的生效结果。

提示

  1. 支持命令别名和命令组
  2. 支持命令权限控制（master / permission 函数 / 覆写系统 ACL）
  3. 支持命令帮助系统
  4. 支持等待用户回复交互
  5. 支持声明式参数与选项（args= / options=）：自动类型转换、本地化错误提示与 usage 生成

类列表

class CommandHandler

命令处理器

提供命令注册、处理和管理功能

方法列表

_cooldowns()

内部方法 命令冷却状态表（cooldown= 声明）

_rate_limits()

内部方法 命令限流状态表（rate_limit= 滑动窗口）

_usage_counts()

内部方法 配额内存计数表（usage= 持久化读缓存）

_refresh_command_config()

从配置读取命令解析相关参数

支持配置热更新：config.updated 事件触发后再次调用即可刷新 前缀 / 大小写 / 空格前缀 / 是否须 @机器人 等解析参数。

_on_config_updated(_data: dict)

配置变更回调：刷新命令解析参数、实现参数覆盖与用户 ACL

_recompute_max_name_tokens()

内部方法 重算已注册命令名 / 别名中的最大 token 数

命令名支持空格分隔的子命令形式（如 "admin add"），匹配阶段按 最长前缀尝试；此缓存在注册与注销后重算，消息分发期只做字典查找。

_resolve_command_tokens(parts: list[str])

内部方法 按"最长前缀匹配"从已切分的命令 token 中解析命令名

依次尝试 parts[:n]（n 从最大注册 token 数降到 1）组成的候选名， 先查别名映射再查命令表；命中即返回，未命中继续降级尝试。

_inherited_permission(name: str)

内部方法 沿命令名父链向上查找最近声明了权限函数的祖先命令

子命令（空格分隔的多 token 命令名，如 "admin add"）自身未声明 permission 时调用：逐级去掉末尾 token 查找已注册祖先，返回第一个 声明了权限的祖先的权限函数——保护父命令即保护其下全部子命令 （声明了权限的祖先会跳过未声明权限的中间祖先继续上溯）。 仅读取注册值，不递归用户覆写。

parse_rate_limit(spec: str)

解析限流声明（如 "5/minute"、"10/s"）为 (次数, 窗口秒)

滑动窗口语义：窗口内至多放行 次数 次，超出静默丢弃。单位支持 second / minute / hour / day（含单字母缩写与可选数值前缀，大小写不敏感）。

parse_usage(spec: str)

解析配额声明（如 "3/day"）为 (次数, 周期单位)

自然周期语义：周期边界对齐本地时区的自然分钟 / 小时 / 日（如 day 为 当日 00:00 起，次日自动重置），与 :meth:parse_rate_limit 的滑动窗口 相区分（rate_limit 防瞬时刷屏，usage_limit 管业务配额）。

usage_period_key(unit: str)

计算当前自然周期的标识键（本地时区）

内部方法 供分发期配额判定使用；周期切换键随之变化即自动重置

_scope()

内部方法 延迟获取作用域单例（避免模块初始化阶段的循环依赖）

用于模块维度作用域检查与事件上下文提取。

返回值 (scope): 单例（ScopeManager）

__call__(name: str | list[str] | None = None, aliases: list[str] | None = None, group: str | None = None, priority: int = 0, permission: Callable | None = None, help: str | None = None, usage: str | None = None, hidden: bool = False, master: bool = False, args: str | None = None, options: dict | None = None, cooldown: str | None = None, cooldown_key: str = 'user', cooldown_reply: str | None = None, rate_limit: str | None = None, rate_limit_key: str = 'user', rate_limit_reply: str | None = None, usage_limit: str | None = None, usage_limit_key: str = 'user', usage_limit_reply: str | None = None, deprecated: str | None = None, deprecated_reject: bool = False)

命令装饰器

>>> @command("roll", args="<count:int> [sides:int=6]",
...          options={"verbose": "-v/--verbose", "label": "--label"})
... async def roll(event, count: int, sides: int = 6, verbose: bool = False, label: str = ""):
...     await event.reply(f"掷了 {count} 次 {sides} 面骰")
>>> @command("daily", cooldown="1d", cooldown_key="user", cooldown_reply="今天已签到")
... async def daily(event):
...     await event.reply("签到成功！")
>>> @command("search", rate_limit="5/minute", rate_limit_key="user")
... async def search(event): ...
>>> @command("oldcmd", deprecated="请用 /newcmd", deprecated_reject=True)
... async def old(event): ...
unregister(handler: Callable)

注销命令处理器

unregister_by_owner(owner: str)

内部方法 按归属者精确移除命令

async wait_reply(event: dict[str, Any], prompt: str | None = None, timeout: float = DEFAULT_WAIT_TIMEOUT_SECS, callback: Callable[[dict[str, Any]], Awaitable[Any]] | None = None, validator: Callable[[dict[str, Any]], bool] | None = None, method: str = DEFAULT_SEND_METHOD, pattern: str | None = None, regex: str | None = None, session: bool = False, cmdpass: bool | None = None)

等待用户回复

async _handle_message(event: dict[str, Any])

处理消息事件中的命令

内部方法 内部使用的方法，用于从消息中解析并执行命令

_is_command_text(text: str)

判定文本是否形如一条已注册命令（前缀 + 命令名/别名命中），不执行命令

供交互等待（wait_reply）的命令穿透判定复用——等待期间命中命令的 消息放行给命令分发器执行。判定口径与 :meth:_try_execute_command 的匹配逻辑一致（前缀 / 大小写归一 / 子命令最长前缀匹配）。

async _try_execute_command(event: 'Event', original_text: str, prefix: str)

尝试执行命令

内部方法 内部使用的方法，用于尝试解析和执行命令

async _check_pending_reply(event: 'Event')

检查是否是等待回复的消息

判定链（会话键命中 → pattern/regex 过滤 → validator 校验 → 权限复查 → 唤醒等待方并认领事件）委托交互会话管理器 :meth:~ErisPulse.Core.Event.interaction.InteractionManager.resolve。

async _send_event_text(event: dict[str, Any], text: str)

内部方法 向事件来源会话发送文本（各 send* 提示的公共发送通道）

async _send_permission_denied(event: dict[str, Any])

发送权限拒绝消息

内部方法 内部使用的方法

async _send_command_error(event: dict[str, Any], error: str)

发送命令错误消息

内部方法 内部使用的方法

async _send_args_error(event: dict[str, Any], text: str)

发送命令参数错误消息（args= / options= 解析失败时的本地化提示 + 用法）

内部方法 内部使用的方法

_cooldown_scope_key(kind: str, event: 'Event')

内部方法 计算冷却作用域键（复用 platform:bot:目标 会话键体系）

_usage_line(cmd_name: str, effective: dict, display_prefix: str | None = None)

内部方法 计算命令的生效 usage 行（帮助展示与参数错误提示共用）

优先取开发者声明的 usage=（含覆写）；未声明且注册了 args= / options= 时按声明自动生成；否则回退 {前缀}{命令名}。

bind_message_handler(handler: BaseEventHandler)

内部方法 绑定到共享的消息事件处理器

将命令分发器 _handle_message 注册到共享的 BaseEventHandler 中， 使命令处理和通用消息处理共享同一个优先级队列。

_register_dispatcher()

内部方法 将命令分发器注册到共享 handler（如尚未注册）

_clear_commands()

内部方法 清除所有已注册的命令，并从共享 handler 中注销命令分发器

返回值: 被清除的命令数量

get_command(name: str)

获取命令信息（返回合并覆写系统命令参数后的生效参数）

传入作用域上下文（event 或 platform / bot_id / session_id 任一）时，命令归属模块在当前会话不可用则返回 None（与分发静默语义一致）。

示例:

>>> command.get_command("admin")
>>> command.get_command("admin", event=event)   # 会话不可用时返回 None
get_commands()

获取所有命令

传入作用域上下文时，过滤掉当前会话不可用模块的命令（值为原始注册信息， 需要覆盖合并后的生效参数请用 :meth:get_command / :meth:get_visible_commands）； 不传上下文时返回完整注册表（与原行为一致）。

get_group_commands(group: str)

获取命令组中的命令

传入作用域上下文时，过滤掉当前会话不可用模块的命令。

get_visible_commands()

获取所有可见命令（非隐藏命令）

可见性判定读取覆写系统命令参数（event.overrides.command.<module>.<command>.hidden）： 用户显式覆盖 hidden 后，帮助列表随之变化（用户优先）。 传入作用域上下文（event 或 platform / bot_id / session_id 任一）时，额外按模块维度过滤该会话不可用模块的命令（与分发静默语义一致）。

_context_from_event(event: Any)

内部方法 从事件提取作用域查询上下文（platform / bot / session）

_resolve_query_context(event: Any = None, platform: str | None = None, bot_id: str | None = None, session_id: str | None = None)

内部方法 归一查询上下文：event 与显式关键字参数合并（显式参数优先）

返回值 ({"platform":): str, "bot_id": str|None, "session_id": str|None}； 完全未提供任何上下文时返回 None（不做会话过滤）

_effective_info(name: str, info: dict)

内部方法 合并实现参数覆盖后的命令生效参数（帮助渲染与可见性判定用）

help(command_name: str | None = None, show_hidden: bool = False, event: Any = None)

生成帮助信息

传入 event 时按作用域对输出做会话感知调整：① 模块维度—— 该会话（platform / bot / session）下被作用域禁用的模块，其命令不再列出 （与分发静默语义一致）；② 覆盖——帮助文本 / usage / 可见性读取 event.overrides.command 覆写值（用户优先）。 子命令（空格分隔多 token 命令名）在其可见父命令下缩进展示。

返回值 (帮助信息字符串): 示例:

>>> # 全量帮助（不感知会话）
>>> command.help()
>>> # 会话感知帮助：只列出当前会话可用的命令
>>> command.help(event=event)
_owner_blocked(info: dict, ctx: dict[str, str | None])

内部方法 判断命令归属模块在给定作用域上下文下是否被模块维度禁用