diff --git a/CHANGELOG.md b/CHANGELOG.md index 1983f15..67eee3c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,15 +26,21 @@ - 组件构造签名瘦身：`WebsocketProtocol(config, session, token)` 移除未使用的 `queue`；`WebsocketConnecter` 改收 `(config, http, token, session, queue)`；`WebhookConnecter` 改收 `(config, queue, handle_event)`，op=13 应答经注入的回调取得，不再依赖具体 `EventEmitter`；`Session` 合并 `sequence_id`/`seq` 双属性为 `seq`（心跳序列号无历史时发 0 而非 null）；移除服务端不会推送的 HEARTBEAT 死代码 handler；`ApiError`/`TokenError` 的错误响应解析合并为 `model.error_parts` - **intents.toml 改为只按大类订阅**：配置项从"`[分组]` 表内逐事件开关"简化为"`分组名 = true/false`"（如 `GROUP_AND_C2C_EVENT = true`）——网关只接受分组位掩码，组内事件本就无法单独订阅，此前的逐事件开关只是"组内任一为 true 即订阅整组"的装饰。破坏性变更：旧格式（表内逐事件开关）不再兼容，与拼错的分组名、非布尔值一样在启动时直接抛 `ValueError`，而非静默少订阅 - 形参注入缓存化：handler 的签名与类型标注解析结果（此前每条事件都重新跑 `inspect.signature`/`get_type_hints`）合并缓存，`services` 命中改为每条事件实时判定——先注册 handler、后登记组件的既有装配顺序不受影响 +- **协议帧与业务事件分流，各接入方式配专有协议处理器**：`BaseProtocol` 从 `emitter.py` 移入新模块 `protocol.py`，接口由 `register(emitter)` 改为 `async on_frame(payload)`（返回需回送的响应体）；协议处理器不再注册进 emitter，而是由 `run_loop` 构造后**注入对应的 connecter**（`WebsocketConnecter(..., protocol)` / `WebhookConnecter(config, queue, protocol)`），由适配器逐帧调用。破坏性变更：只有 op=0 的业务事件进事件队列交给 `EventEmitter`，HELLO/READY/INVALID_SESSION/op=13 等协议帧在连接器内就地处理——用户 handler 不再收到协议帧；`EventEmitter.register_protocol()`、`EventEmitter.handle()`、`emitter._protocol`/`on_payload` 机制随之删除，协议类不再需要 `register()` 注册 handler。附带修掉一处时序耦合：`INVALID_SESSION` 的会话重置此前依赖分发泵在重连前消费掉该事件，现在在 connecter `break` 重连前同步完成，不会误 RESUME 已死会话 +- 事件分发热路径减负：`emit()` 先取事件名、无 listener 时不再构造 `Event`（每条未订阅事件/协议帧省一次对象与 dict 拷贝）；`Event.data` 改为惰性拷贝（只用 `.typed`/`.raw` 的 handler 不再触发 `dict(d)`） +- ws 重连收尾补 `await`：取消 `reply`/`heartbeat` 任务后 `gather(..., return_exceptions=True)` 取回结果，避免 "Task exception was never retrieved" 并确保它们在下一轮重连前真正停下 +- `EventQueue.put_event`/`put_reply` 去掉多余的 `return await` ### 移除 +- `EventEmitter.register_protocol()`、`EventEmitter.handle()` 与 `emitter` 持有协议层的机制（协议处理器改为由 connecter 持有并逐帧调用，见上文"变更"） - apscheduler 依赖（心跳迁入 connecter 后不再需要调度器） - 音乐插件示例（`plugins/`、根目录 `main.py`）已拆分为独立的点歌 bot 项目（含 Markdown 卡片、按钮回调与内置短网址多线程下载代理），不再随 SDK 仓库分发 ### 修复 -- **INVALID_SESSION 可续传标记失效**：op=9 的 `d` 是布尔值（true=可 RESUME），此前参数注入回退把 d 字典化成 `{}`（恒 falsy），任何 INVALID_SESSION 都会清空会话——RESUME 续传永远走不到，断线期间本可回放的消息丢失；注入回退改为直传原始 d，`async def on_invalid(resumable: bool)` 现在能拿到真实布尔 +- **INVALID_SESSION 可续传标记失效**：op=9 的 `d` 是布尔值（true=可 RESUME），此前参数注入回退把 d 字典化成 `{}`（恒 falsy），任何 INVALID_SESSION 都会清空会话——RESUME 续传永远走不到，断线期间本可回放的消息丢失。现协议帧不再经参数注入，`WebsocketProtocol` 直接读取 `d` 的真值，`resumable=true` 正确保留会话（协议帧处理测试覆盖） +- **会话重置时序**：`INVALID_SESSION` 的会话清空此前依赖分发泵在重连前消费掉该事件（慢 handler 会推迟它，导致带着已死 session_id 去 RESUME）；现在连接器在 `break` 重连前同步调用协议处理器完成重置 - token 请求失败时抛出 `TokenError`（携带 QQ 返回的 `code`/`message`），不再抛出掩盖真实原因的 `KeyError: 'access_token'` - 修正被动回复 `event_id` 的指引：官方要求取网关帧最外层的 id（`Event.event_id`），事件体里的裸 UUID（`Interaction.id` 等）不能作 `event_id`，此前文档与 docstring 的说法有误 - 响应 `Content-Type` 非 JSON 时不再抛 `ContentTypeError`（个别接口如回应互动返回 200 text/plain + `{}` body）；JSON 解析统一切换 ujson diff --git a/README.md b/README.md index edf3dd6..beff31a 100644 --- a/README.md +++ b/README.md @@ -100,8 +100,9 @@ async def handle(event: Event, api: BotApi): msg_id=event.message_id, ) # 失败抛 ApiError；富媒体先 upload_group_file 拿 file_info -# 业务事件（op=0）处理器的返回值不会回流； -# 协议层（op≠0）由 ws_protocol / webhook_protocol 的类处理，经 reply 队列发回。 +# 业务事件（op=0）才进分发，返回值不回流，回复请显式调 BotApi； +# 协议帧（HELLO/READY/INVALID_SESSION/op=13）由各接入方式的协议处理器 +# 在连接器内就地处理，不进队列、也不会触达 handler。 # 单个 handler 抛异常不会影响其他 handler 和主循环，异常会被记录。 if __name__ == "__main__": diff --git a/docs/API.md b/docs/API.md index ddb0400..ed6f13d 100644 --- a/docs/API.md +++ b/docs/API.md @@ -7,7 +7,8 @@ ```python from qqbotsdk import ( main, run_loop, # 入口（组件在 run_loop 显式装配） - EventEmitter, BaseProtocol, EventQueue, # 分发层 + EventEmitter, EventQueue, # 分发层 + BaseProtocol, # 协议处理器接口（自定义接入方式时继承） Config, Connecter, # 配置与连接器接口 ) ``` @@ -21,7 +22,7 @@ from qqbotsdk import ( | `EventQueue()` | 事件/应答队列；`EventEmitter` 默认自建，需多处共享时手动传入（`ee.queue` 取实例） | | `Config` | frozen dataclass，环境变量一次性读齐（见下文） | | `Connecter` | 连接适配器 Protocol（接口），按 `CONNECTER` 自动选择实现 | -| `BaseProtocol` | 协议层抽象基类，自定义协议处理器时继承并实现 `register(emitter)` | +| `BaseProtocol` | 协议处理器抽象基类（`protocol.py`），自定义接入方式时继承并实现 `on_frame(payload)` | 子包按需导入： @@ -36,18 +37,17 @@ from qqbotsdk.events import Event ```python ee = EventEmitter() -@ee.on("GROUP_AT_MESSAGE_CREATE") # DISPATCH 事件用 t 名 -@ee.on("INVALID_SESSION") # 协议事件用 Opcode 名（IDENTIFY/RESUME/HEARTBEAT/...） +@ee.on("GROUP_AT_MESSAGE_CREATE") # 业务事件用 t 名（仅 op=0 会分发到这里） async def handler(...): ... @ee.on("GROUP_AT_MESSAGE_CREATE", background=True) # 后台并发执行，不阻塞事件流 async def slow(...): ... ``` -- **事件命名**：op=0（DISPATCH）按 payload 的 `t` 名路由；其余按 `Opcode` 名路由。协议事件（op≠0）已由内置协议层处理，一般只需监听业务事件；handler 返回 dict 时协议事件的返回值会回流给 ws 发送（业务事件不会）。 +- **只分发业务事件**：`EventEmitter` 只处理 op=0（DISPATCH）事件，按 payload 的 `t` 名路由。协议帧（HELLO/READY/INVALID_SESSION/op=13 等）由各接入方式的协议处理器（`WebsocketProtocol`/`WebhookProtocol`）在连接器内就地处理，不进事件队列、也不会触达你的 handler。 - **异常隔离**：handler 异常就地记录（loguru），不影响同事件其他 handler 与主循环。 -- **后台并发**：`background=True` 的 handler 进后台任务，emit 不等它；同一事件内失去先后保证，返回值不回流，仅用于业务 handler。`run_loop` 退出时统一取消在飞后台任务（`ee.close()`）。 -- **`ee.handle(payload)`**：同步分发并返回首个 handler 的非空返回值，供协议层应答 op=13 等场景，业务代码不用。 +- **后台并发**：`background=True` 的 handler 进后台任务，emit 不等它；同一事件内失去先后保证，仅用于业务 handler。`run_loop` 退出时统一取消在飞后台任务（`ee.close()`）。 +- 业务 handler 的返回值不回流；回复消息请在 handler 里显式调 `BotApi`。 ### handler 参数注入约定 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 636981f..7376a5c 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -11,14 +11,14 @@ flowchart TB WH["WebhookConnecter\naiohttp server\n验签 + TTL去重"] end - subgraph 分发层["分发层 emitter.py"] - Q[("EventQueue\nmsg_queue / reply_queue")] - EE["EventEmitter\nop/t 路由 + 参数注入\nhandler 异常隔离\nbackground 后台并发"] + subgraph 协议层["协议层 protocol.py"] + WSP["WebsocketProtocol\nHELLO→IDENTIFY/RESUME\nREADY/INVALID_SESSION + s 序列号"] + WHP["WebhookProtocol\nop=13 验证应答"] end - subgraph 协议层["协议层 BaseProtocol"] - WSP["WebsocketProtocol\nIDENTIFY/RESUME\ns 序列号跟踪"] - WHP["WebhookProtocol\nop=13 验证应答"] + subgraph 分发层["分发层 emitter.py"] + Q[("EventQueue\nmsg_queue / reply_queue")] + EE["EventEmitter\nop=0/t 路由 + 参数注入\nhandler 异常隔离\nbackground 后台并发"] end subgraph 业务层["业务层"] @@ -35,12 +35,9 @@ flowchart TB QQ["QQ 开放平台"] - WS -- "事件 payload" --> Q - WH -- "事件 payload" --> Q + WS -- "业务事件 op=0" --> Q + WH -- "业务事件 op=0" --> Q Q -- "dispatch 泵" --> EE - EE -- "on_payload" --> WSP - WSP -- "IDENTIFY/RESUME 等\npayload 返回值" --> Q - WS -- "HEARTBEAT（心跳任务）" --> Q EE -- "Event 注入" --> USER USER -- "api: BotApi" --> API API --> QQ @@ -48,6 +45,9 @@ flowchart TB TK --> HTTP API --> TK WSP --> TK + WS -. "逐帧 on_frame\n（协议帧就地处理，不入队）" .-> WSP + WH -. "op=13 逐帧 on_frame" .-> WHP + WSP -. "IDENTIFY/RESUME 应答" .-> WS WHP --> CRY WH --> CRY CFG -. "构造传参" .-> WS & WH & API & TK & WSP & WHP @@ -59,30 +59,28 @@ flowchart TB sequenceDiagram participant QQ as QQ 开放平台 participant WS as WebsocketConnecter + participant P as WebsocketProtocol participant Q as EventQueue participant EE as EventEmitter - participant P as WebsocketProtocol participant H as 用户 handler QQ->>WS: HELLO (op=10) - WS->>Q: put_event - Q->>EE: dispatch 泵 - EE->>P: on_payload（记 s 序列号） - EE->>P: 路由到 HELLO handler - P-->>Q: IDENTIFY/RESUME payload → reply_queue + WS->>P: on_frame（握手判定） + P-->>WS: IDENTIFY / RESUME payload WS->>QQ: send（过滤 NOT_SET 字段） loop 每次心跳间隔（心跳任务随连接生灭） WS->>Q: HEARTBEAT payload WS->>QQ: send - QQ->>WS: HEARTBEAT_ACK（刷新看门狗时钟） + QQ->>WS: HEARTBEAT_ACK（刷新看门狗时钟，不进队列） end QQ->>WS: DISPATCH 事件 (op=0, t=GROUP_AT_MESSAGE_CREATE) - WS->>Q: put_event + WS->>P: on_frame（记 s 序列号） + WS->>Q: put_event（仅 op=0） Q->>EE: dispatch 泵 EE->>H: 按形参标注注入 Event/BotApi - H->>EE: 返回值不回流（op=0） + H->>EE: 返回值不回流 ``` ## 分层规则 @@ -91,33 +89,33 @@ sequenceDiagram | 层 | 模块 | 职责 | 不允许出现 | |---|---|---|---| -| 接入层 | `connecter.py` | 传输细节：握手、心跳、重连、验签、去重 | 事件业务语义、协议状态机 | -| 分发层 | `emitter.py` | 路由、参数注入、异常隔离、后台并发、应答回流 | 传输细节、具体协议状态机 | -| 协议层 | `ws_protocol.py` `webhook_protocol.py` | 鉴权/验证等协议状态机（会话语义） | 传输细节（只经队列收发） | +| 接入层 | `connecter.py` | 传输细节：握手、心跳、重连、验签、去重；逐帧驱动协议处理器 | 事件业务语义、协议状态机 | +| 协议层 | `protocol.py` `ws_protocol.py` `webhook_protocol.py` | 适配器专有的协议处理（`BaseProtocol.on_frame`）：鉴权/验证/会话语义 | 传输细节、业务语义 | +| 分发层 | `emitter.py` | 只分发业务事件（op=0）：路由、参数注入、异常隔离、后台并发 | 传输细节、协议帧、协议状态机 | | 业务层 | `api/` `payloads.py` + 用户代码 | OpenAPI 调用、payload 解析与机器人逻辑 | 协议细节 | -跨层共享的最小公共物：`model.py`（Opcode/Payload/Intent）、`queue.py`（事件通道）、`config.py`、`session.py`（ws 会话状态）、`payloads.py`（事件 dataclass 与解析，被 `events.py`/`emitter.py` 使用）。 +跨层共享的最小公共物：`model.py`（Opcode/Payload/Intent）、`protocol.py`（协议处理器接口，被接入层与协议层共同依赖）、`queue.py`（事件通道）、`config.py`、`session.py`（ws 会话状态）、`payloads.py`（事件 dataclass 与解析，被 `events.py`/`emitter.py` 使用）。 ## 生命周期与依赖装配 - 全部组件在 `run_loop`（`__init__.py`）显式构造装配，无 DI 框架，也无任何全局单例。 - 组件按类型登记进 `emitter.services`，handler 形参标注类型即得实例；用户自定义组件同样登记即可注入。 - 唯一的 `ClientSession` 在 `run_loop` 构造，`finally` 中先 `emitter.close()` 收尾后台 handler、再统一释放连接池（全 SDK 共享一个连接池）。 -- 接入方式的选择收敛在 `run_loop` 的 `match config.connecter` 装配分支，主体代码只依赖 `BaseProtocol`/`Connecter` 抽象；两个 connecter 的依赖面一致（config + queue + 各自所需的窄依赖），`WebhookConnecter` 经 `handle_event` 回调取得 op=13 应答，不感知 `EventEmitter`。 +- 接入方式的选择收敛在 `run_loop` 的 `match config.connecter` 装配分支：每种接入方式构造**自己的协议处理器**并注入对应的 connecter（主体不再持有协议层）；`WebhookConnecter` 经协议处理器的 `on_frame` 就地取得 op=13 应答，不感知 `EventEmitter`。两个 connecter 的依赖面一致（config + queue + 各自所需的窄依赖 + `BaseProtocol`）。 - 用户在 `main(ee)` 之前构造的 `EventEmitter` 会被复用（连同它的队列），保证 handler 注册在主体实际使用的实例上。 ## 关键设计决策 -1. **管线而非回调直调**：事件一律经 `EventQueue` 异步流转，接入与分发两侧解耦——Webhook handler 里不需要知道 ws 的存在，反之亦然。代价是多一跳队列延迟（可忽略）。 -2. **协议层只管会话语义，心跳归传输层**：`WebsocketProtocol` 负责 IDENTIFY/RESUME/序列号（说什么），`WebsocketConnecter` 负责连接与心跳（怎么保活）——心跳任务随连接生灭（HELLO 给出间隔后启动，断线即取消），断线期间不在 reply 队列堆积过期心跳，也免去了调度器的生命周期管理。协议类通过返回 payload dict 经 reply 队列外发。 +1. **业务事件走管线，协议帧走专有处理器**：只有 op=0 的业务事件经 `EventQueue` 异步流转给 `EventEmitter` 分发；HELLO/READY/INVALID_SESSION/op=13 等协议帧由 connecter 逐帧调用其专有协议处理器（`BaseProtocol.on_frame`）就地处理。这样用户 handler 不会看到协议帧，且"断线前重置会话"这类时序不再取决于分发泵的进度——`INVALID_SESSION` 的会话清空在 connecter `break` 重连前已同步完成，不会误 RESUME 一个已死会话。代价是多一跳队列延迟（可忽略）。 +2. **协议层只管会话语义，心跳归传输层**：`WebsocketProtocol` 负责 IDENTIFY/RESUME/序列号（说什么），`WebsocketConnecter` 负责连接与心跳（怎么保活）——心跳任务随连接生灭（HELLO 给出间隔后启动，断线即取消），断线期间不在 reply 队列堆积过期心跳，也免去了调度器的生命周期管理。协议处理器的应答经 `on_frame` 返回值由 connecter 写回（ws 走 reply 队列，webhook 走 HTTP 响应体）。 3. **看门狗两层互补**（见 `connecter.py` 模块注释）：静默超时抓连接彻底失联，ACK 期限检查抓"有事件流但心跳已死"。阈值 `_SILENCE_FACTOR=1.1`、`_ACK_FACTOR=1.2` 是按 ACK 自然节奏校准的，**不能压到 1× 整**，否则正常抖动会周期性误杀（误杀由 RESUME 兜底，不丢消息）。重连退避只在连接稳定存活（`_STABLE_SECONDS`）后才归一，抖动的服务端不会被 1s 间隔反复探测。 4. **handler 异常就地隔离**：`_invoke` 捕获记日志，`dispatch` 主循环另有兜底；单条事件的单个 handler 失败不影响进程。`background=True` 的 handler 进后台任务（emit 不等它），任务由 emitter 持强引用、`close()` 统一取消。 -5. **注入约定**：handler 参数按形参标注解析——`payloads.py` 里的 dataclass → 解析后的 payload 对象；`Event` → 事件封装；登记进 `emitter.services` 的组件类型（如 `BotApi`）→ 实例；未标注或未登记 → 回退传**原始 d**（协议事件的 d 可能是标量，如 INVALID_SESSION 的 resumable 布尔；dict 化会丢真值——这里曾出过 bug）。解析是纯同步查表（`get_type_hints` + dict），无框架、无运行时魔法。 +5. **注入约定**：handler 参数按形参标注解析——`payloads.py` 里的 dataclass → 解析后的 payload 对象；`Event` → 事件封装；登记进 `emitter.services` 的组件类型（如 `BotApi`）→ 实例；未标注或未登记 → 回退传**原始 d**。解析是纯同步查表（`get_type_hints` + 签名缓存），无框架、无运行时魔法。签名与类型标注每个 handler 只解析一次并缓存，但 `services` 命中在每条事件时实时判定，保证"先注册 handler、后登记组件"的装配顺序可用。 6. **OpenAPI 请求内核自治**：`api/core.py` 统一处理鉴权头、429 限流重试（优先 `Retry-After`，脏值容错，退避封顶）与 401 刷新 token 重试一次；v2/channel/guild 三个域方法集继承内核组合成 `BotApi`，新增域加模块即可，方法对用户始终平铺在 `BotApi` 上。 ## 扩展点 -- **新增接入方式**（如轮询/其他平台）：实现 `Connecter.run()` 与 `BaseProtocol.register()`，在 `run_loop` 的 `match` 装配分支加一个 `case`，`Config.load` 的白名单加一个名字。 +- **新增接入方式**（如轮询/其他平台）：实现 `Connecter.run()` 与 `BaseProtocol.on_frame()`，在 `run_loop` 的 `match` 装配分支加一个 `case`（构造该方式的协议处理器并注入 connecter），`Config.load` 的白名单加一个名字。 - **新增业务 API**：在 `api/` 对应域模块（`v2`/`channel`/`guild`）加方法即自动平铺到 `BotApi`，全部获得鉴权、限流重试与错误处理；新增域则新建模块并组合进 `api/__init__.py` 的 `BotApi`。群/单聊消息经 `_post_message` 统一处理 msg_id/msg_seq/media 语义；已有方法覆盖：单聊/群聊消息与富媒体上传、频道消息/撤回/表态、公告、精华消息、日程、禁言、频道信息查询（guild/channels/roles/member）。 - **新增事件类型 dataclass**：在 `payloads.py` 定义 dataclass 并登记 `EVENT_TYPES`，handler 即可按类型标注拿解析结果；未收录事件回退传 dict。当前已覆盖 intents.toml 的全部事件组（消息、成员变动、审核、表态、互动、论坛、音频）。 @@ -125,9 +123,9 @@ sequenceDiagram `tests/` 覆盖： -- 纯逻辑：crypto 往返、intents 解析、Config 校验、`_route` 路由边界、webhook 去重、序列号单调性、handler 异常隔离与 reply 回流、payload dataclass 解析、BotApi 各业务方法的拼参（Recording 子类截获 request）。 +- 纯逻辑：crypto 往返、intents 解析、Config 校验、协议帧处理（`test_route.py`：IDENTIFY/RESUME 选路、READY 捕获、s 序列号、INVALID_SESSION 重置、业务事件不产生应答）、webhook 去重、序列号单调性、handler 异常隔离、payload dataclass 解析、BotApi 各业务方法的拼参（Recording 子类截获 request）。 - **网络集成**（`test_ws_integration.py` / `test_webhook_integration.py`）： - - WebSocket：aiohttp 假网关（token 接口 + gateway + ws 协议行为）端到端跑通 HELLO → IDENTIFY → READY 会话捕获 → 心跳/ACK（心跳任务随连接生灭）→ 业务事件 dataclass 分发； + - WebSocket：aiohttp 假网关（token 接口 + gateway + ws 协议行为）端到端跑通 HELLO → IDENTIFY → READY 会话捕获 → 心跳/ACK（心跳任务随连接生灭）→ 业务事件 dataclass 分发，并断言协议帧（HELLO/HEARTBEAT_ACK）不会触达 handler； - Webhook：真实起 `WebhookConnecter` 的 http server（TestServer），覆盖 op=13 验证应答（签名内容正确性）、合法签名放行分发、坏签名 401、同 id 去重。 -另有 429/401 重试、token 并发锁、op=9 resumable 布尔注入、background 后台并发与 `close()` 收尾的专项单测。两套集成测试验证的都是真实连接路径，改接入层/协议层后跑 `uv run pytest` 即可回归。 +另有 429/401 重试、token 并发锁、background 后台并发与 `close()` 收尾的专项单测。两套集成测试验证的都是真实连接路径，改接入层/协议层后跑 `uv run pytest` 即可回归。 diff --git a/docs/GUIDE.md b/docs/GUIDE.md index e669965..f6f6272 100644 --- a/docs/GUIDE.md +++ b/docs/GUIDE.md @@ -116,11 +116,12 @@ handler 参数按标注解析的完整规则： | payload dataclass（如 `GroupAtMessage`） | 解析后的 dataclass 对象；未收录事件回退原始 dict | | `Event` | 事件封装对象（`.raw`/`.data`/`.typed`/`.user_id`/`.group_id`/`.content`/`.message_id`/`.event_id`） | | 组件类型（`BotApi`/`Session`/`Config` 等） | `emitter.services` 中按类型登记的实例；未登记回退传原始 d | -| 无标注 | 原始 d（业务事件即事件 dict；协议事件可能是标量，如 `INVALID_SESSION` 的 `resumable: bool`） | +| 无标注 | 原始 d（即事件 dict） | 注意： -- 业务事件（op=0）处理器的**返回值不会回流**，回复消息请在 handler 里显式调 API。 +- 只有**业务事件（op=0）**会分发给你的 handler；协议帧（HELLO/READY/INVALID_SESSION/op=13）由 SDK 的协议处理器就地处理，不触达 handler（`Ready` 是例外，它既是握手结果也是可监听的业务事件）。 +- 业务事件处理器的**返回值不会回流**，回复消息请在 handler 里显式调 API。 - 单个 handler 抛异常只记录日志，不影响同事件其他 handler 和主循环。 - 所在分组未订阅（intents.toml 中为 `false`）的事件不会到达 handler。 - 自定义组件也能注入：`ee.services[MyService] = MyService(...)` 登记后，handler 形参标注 `MyService` 即可拿到实例。 diff --git a/src/qqbotsdk/__init__.py b/src/qqbotsdk/__init__.py index 12aa69b..8455a25 100644 --- a/src/qqbotsdk/__init__.py +++ b/src/qqbotsdk/__init__.py @@ -15,8 +15,8 @@ from .api import BotApi from .config import Config from .connecter import Connecter as Connecter from .connecter import WebhookConnecter, WebsocketConnecter -from .emitter import BaseProtocol as BaseProtocol from .emitter import EventEmitter as EventEmitter +from .protocol import BaseProtocol as BaseProtocol from .queue import EventQueue as EventQueue from .session import Session from .token import AccessToken @@ -52,17 +52,20 @@ async def run_loop(emitter: EventEmitter | None = None) -> None: } ) + # 各接入方式配自己的协议处理器：协议帧由适配器就近处理，只有业务事件 + # （op=0）入队交给 emitter 分发（见 protocol.py / connecter.py） match config.connecter: case "websocket": protocol = WebsocketProtocol(config, session, token) - connecter = WebsocketConnecter(config, http, token, session, emitter.queue) + connecter = WebsocketConnecter( + config, http, token, session, emitter.queue, protocol + ) case "webhook": protocol = WebhookProtocol(config) - connecter = WebhookConnecter(config, emitter.queue, emitter.handle) + connecter = WebhookConnecter(config, emitter.queue, protocol) case unknown: raise ValueError(f"未知的 CONNECTER: {unknown}") - emitter.register_protocol(protocol) logger.info(f"qqbotsdk 启动，连接方式: {config.connecter}") try: await asyncio.gather(emitter.dispatch(), connecter.run()) diff --git a/src/qqbotsdk/connecter.py b/src/qqbotsdk/connecter.py index fc6aea3..9c2ec30 100644 --- a/src/qqbotsdk/connecter.py +++ b/src/qqbotsdk/connecter.py @@ -1,5 +1,9 @@ """连接适配器：把两种接入方式（WebSocket / Webhook）归一为事件流入、reply 队列流出。 +每条帧先交给本适配器注入的协议处理器（`BaseProtocol.on_frame`）处理协议语义， +只有业务事件（op=0）才入队交给 emitter 分发给用户 handler——协议帧与业务事件 +因此不共享分发路径。 + - WebsocketConnecter：gateway 握手 + ws 长连接，指数退避重连（连接稳定 存活才把退避归一，抖动的服务端不会被 1s 间隔反复探测）；心跳任务随 连接生灭——HELLO 给出间隔后周期投递，断线即取消，不在 reply 队列堆积 @@ -7,13 +11,12 @@ 收到消息时的 ACK 期限检查抓"有事件流但心跳已死"； - WebhookConnecter：aiohttp server 接收回调，验签失败 401， 按 payload id 做 60s TTL 去重，正常推送进事件队列；op=13 验证请求 - 经注入的 handle_event 同步取得应答（HTTP 响应不走 reply 队列）。 + 由协议处理器就地应答（HTTP 响应不走 reply 队列）。 """ import asyncio -from collections.abc import Awaitable, Callable from time import monotonic -from typing import Any, Protocol +from typing import Protocol, cast import aiohttp import ujson @@ -23,6 +26,7 @@ from loguru import logger from .config import Config from .crypto import verify_sig from .model import NOT_SET, Opcode, Payload, payload_of +from .protocol import BaseProtocol from .queue import EventQueue from .session import Session from .token import AccessToken @@ -51,12 +55,14 @@ class WebsocketConnecter: token: AccessToken, session: Session, queue: EventQueue, + protocol: BaseProtocol, ) -> None: self._config = config self._http = http self._token = token self._session = session self._queue = queue + self._protocol = protocol async def run(self) -> None: backoff: float = 1 @@ -105,6 +111,9 @@ class WebsocketConnecter: finally: reply.cancel() heartbeat.cancel() + # 取回被取消任务的结果，避免 "Task exception was never retrieved"， + # 也确保它们在下一轮重连前真正停下 + await asyncio.gather(reply, heartbeat, return_exceptions=True) logger.info("WebSocket 已断开") return monotonic() - started @@ -129,7 +138,17 @@ class WebsocketConnecter: interval_fut.set_result(interval) elif op == Opcode.HEARTBEAT_ACK: last_ack = monotonic() - await self._queue.put_event(data) + + # 每帧都交给本适配器的协议处理器（HELLO→IDENTIFY/RESUME、 + # READY→捕获会话、INVALID_SESSION→清会话、s→序列号）；只有业务 + # 事件（op=0）入队分发。协议帧不入队，所以 INVALID_SESSION 的 + # 会话重置在下面 break 之前已同步完成，重连不会误 RESUME 死会话。 + response = await self._protocol.on_frame(data) + if response is not None: + await self._queue.put_reply(cast(Payload, response)) + if op == Opcode.DISPATCH: + await self._queue.put_event(data) + if op in (Opcode.RECONNECT, Opcode.INVALID_SESSION): break if monotonic() - last_ack > interval * self._ACK_FACTOR: @@ -160,13 +179,13 @@ class WebhookConnecter: self, config: Config, queue: EventQueue, - handle_event: Callable[[Payload], Awaitable[Any]], + protocol: BaseProtocol, ) -> None: - """handle_event 供 op=13 验证请求同步取应答（通常传 emitter.handle）； - 与 ws 侧对齐，本类只依赖队列与该回调，不感知 EventEmitter。""" + """protocol 为本接入方式的协议处理器（WebhookProtocol），op=13 验证 + 请求由它就地应答；本类只依赖队列与该处理器，不感知 EventEmitter。""" self._config = config self._queue = queue - self._handle_event = handle_event + self._protocol = protocol self._seen: dict[str, float] = {} def _make_app(self) -> web.Application: @@ -206,7 +225,7 @@ class WebhookConnecter: if payload.get("op") == Opcode.VALIDATION: logger.info("收到 Webhook 验证请求") - ret = await self._handle_event(payload) + ret = await self._protocol.on_frame(payload) if ret is None: return web.json_response( {"error": "missing validation handler"}, status=500 @@ -224,5 +243,8 @@ class WebhookConnecter: logger.info(f"Webhook 重复推送已跳过: {event_id}") return web.json_response({"opcode": 12}) - await self._queue.put_event(payload) + if payload.get("op") == Opcode.DISPATCH: + await self._queue.put_event(payload) + else: + logger.warning(f"Webhook 收到非业务帧，已忽略: op={payload.get('op')}") return web.json_response({"opcode": 12}) diff --git a/src/qqbotsdk/emitter.py b/src/qqbotsdk/emitter.py index 03f0ed4..795555b 100644 --- a/src/qqbotsdk/emitter.py +++ b/src/qqbotsdk/emitter.py @@ -1,22 +1,21 @@ -"""事件分发主体：路由 payload → 触发 handler → 协议层应答回流。 +"""事件分发主体：路由业务事件 → 触发 handler → 异常隔离与后台并发。 -- `dispatch()` 泵循环消费 EventQueue.msg_queue，`emit()` 按 - op（DISPATCH 用 t 名、其余用 Opcode 名）路由到 `@on` 注册的 handler； +- `dispatch()` 泵循环消费 EventQueue.msg_queue，`emit()` 把 op=0 的业务事件 + 按 t 名路由到 `@on` 注册的 handler；协议帧（HELLO/INVALID_SESSION/op=13 等） + 不进队列，由连接适配器的协议处理器就地处理（见 protocol.py）； - handler 参数按形参标注注入：payload dataclass → 解析对象，`Event` → 事件封装，`services` 里登记的组件类型 → 实例，其余直传原始 d - （可能是 dict、标量甚至 None，协议事件的 d 不保证是 dict）； -- 业务 handler（op=0）返回值不回流；协议层（op≠0）返回的 payload - 进 reply 队列由 ws 发出。handler 异常就地隔离，不影响其他 handler。 + （可能是 dict、标量甚至 None）； +- 业务 handler 返回值不回流；handler 异常就地隔离，不影响其他 handler。 - `on(..., background=True)` 把 handler 放进后台任务并发执行： - 不阻塞事件流，但同事件 handler 间失去先后保证，返回值不回流； + 不阻塞事件流，但同事件 handler 间失去先后保证； 进程退出经 `close()` 统一取消在飞任务。 """ import asyncio import inspect -from abc import ABC, abstractmethod from collections.abc import Awaitable, Callable -from typing import Any, TypeVar, cast, get_type_hints, overload +from typing import Any, TypeVar, get_type_hints, overload from loguru import logger from pyee.asyncio import AsyncIOEventEmitter @@ -29,22 +28,11 @@ from .queue import EventQueue Handler = TypeVar("Handler", bound=Callable[..., Awaitable[Any]]) -class BaseProtocol(ABC): - """协议层处理器接口：具体接入方式（websocket/webhook）实现后注册到主体。""" - - @abstractmethod - def register(self, emitter: "EventEmitter") -> None: ... - - def on_payload(self, payload: Payload) -> None: # noqa: B027 - """路由后的每条 payload 回调，供协议类跟踪协议级状态（如 s 序列号）。""" - - class EventEmitter: def __init__(self, queue: EventQueue | None = None) -> None: self._ee = AsyncIOEventEmitter() self._queue = queue if queue is not None else EventQueue() self.services: dict[type, Any] = {} - self._protocol: BaseProtocol | None = None # handler → (类型标注, 形参名)：签名解析较贵，缓存后每条事件只查表 self._signatures: dict[ Callable[..., Any], tuple[dict[str, Any], tuple[str, ...]] @@ -57,10 +45,6 @@ class EventEmitter: """本 emitter 消费的事件队列；未显式传入时构造即自建，run_loop 复用同一实例。""" return self._queue - def register_protocol(self, protocol: BaseProtocol) -> None: - self._protocol = protocol - protocol.register(self) - @overload def on( self, event: Opcode | str, *, background: bool = False @@ -108,71 +92,34 @@ class EventEmitter: # dispatch 是主循环，单条事件路由失败不能带崩整个进程 logger.exception(f"事件分发异常，已跳过: {data} ({e})") - def _route(self, data: Payload) -> tuple[int, str, Event] | None: - op = data.get("op") - if not isinstance(op, int): - return None - - # 线上 payload 反序列化出来是 int，Opcode(9) 取名供非 DISPATCH 事件路由 - event_name: str | None - if op == Opcode.DISPATCH: - t = data.get("t") - event_name = t if isinstance(t, str) else None - else: - try: - event_name = Opcode(op).name - except ValueError: - return None - if not isinstance(event_name, str): - return None - - return op, event_name, Event(data) - async def emit(self, data: Payload) -> None: - routed = self._route(data) - if routed is None: - logger.warning(f"无法确定事件名，已跳过: {data}") + # 先取事件名，无 listener 就不必构造 Event（每条未订阅/协议帧都省一次拷贝） + op = data.get("op") + if op != Opcode.DISPATCH: + logger.warning(f"非业务事件不应进分发队列，已跳过: {data}") + return + t = data.get("t") + if not isinstance(t, str): + logger.warning(f"业务事件缺少 t 字段，已跳过: {data}") return - op, event_name, event = routed - if self._protocol is not None: - self._protocol.on_payload(data) - - listeners = self._ee.listeners(event_name) + listeners = self._ee.listeners(t) if not listeners: return - reply = op != Opcode.DISPATCH + event = Event(data) for fn in listeners: if fn in self._background: self._spawn(fn, event) else: - await self._invoke(fn, event, reply=reply) + await self._invoke(fn, event) - async def handle(self, data: Payload) -> Any: - """同步分发事件并返回首个处理器的非空返回值，用于 op=13 等需要应答的协议事件。""" - routed = self._route(data) - if routed is None: - return None - _, event_name, event = routed - - for fn in self._ee.listeners(event_name): - ret = await fn(*self._resolve_params(fn, event)) - if ret is not None: - return ret - return None - - async def _invoke( - self, fn: Callable[..., Awaitable[Any]], event: Event, reply: bool - ) -> None: + async def _invoke(self, fn: Callable[..., Awaitable[Any]], event: Event) -> None: # 用户 handler 的异常在这里就地隔离，不影响同一事件的其他 handler try: - ret: Any = await fn(*self._resolve_params(fn, event)) + await fn(*self._resolve_params(fn, event)) except Exception as e: logger.exception(f"事件处理器 {getattr(fn, '__qualname__', fn)} 异常: {e}") - return - if reply and isinstance(ret, dict): - await self._queue.put_reply(cast(Payload, ret)) def _spawn(self, fn: Callable[..., Awaitable[Any]], event: Event) -> None: """handler 进后台任务，emit 不等它；任务由 _bg_tasks 持强引用防 GC。""" diff --git a/src/qqbotsdk/events.py b/src/qqbotsdk/events.py index cbcfd21..c898b17 100644 --- a/src/qqbotsdk/events.py +++ b/src/qqbotsdk/events.py @@ -7,14 +7,13 @@ from .payloads import parse_event class Event: - """一条 DISPATCH 事件；属性取值兼容群（openid）与频道两套字段名。""" + """一条业务事件（op=0）；属性取值兼容群（openid）与频道两套字段名。""" def __init__(self, payload: Payload) -> None: self._payload = payload t = payload.get("t") self.type: str | None = t if isinstance(t, str) else None - d = payload.get("d") - self._d: dict[str, Any] = dict(d) if isinstance(d, dict) else {} + self._d: dict[str, Any] | None = None self._typed: object | None = None @property @@ -23,30 +22,34 @@ class Event: @property def data(self) -> dict[str, Any]: + """d 的只读副本；仅用到 `.typed`/`.raw` 的 handler 不会触发这次拷贝。""" + if self._d is None: + d = self._payload.get("d") + self._d = dict(d) if isinstance(d, dict) else {} return self._d @property def typed(self) -> object: """d 的 dataclass 解析结果（见 payloads.py）；未知事件类型为 dict。""" if self._typed is None: - self._typed = parse_event(self.type, self._d) + self._typed = parse_event(self.type, self.data) return self._typed @property def user_id(self) -> str | None: - return self._d.get("user_id") or self._d.get("from_user_id") + return self.data.get("user_id") or self.data.get("from_user_id") @property def group_id(self) -> str | None: - return self._d.get("group_openid") + return self.data.get("group_openid") @property def content(self) -> str | None: - return self._d.get("content") + return self.data.get("content") @property def message_id(self) -> str | None: - return self._d.get("id") + return self.data.get("id") @property def event_id(self) -> str | None: diff --git a/src/qqbotsdk/protocol.py b/src/qqbotsdk/protocol.py new file mode 100644 index 0000000..eb7be0d --- /dev/null +++ b/src/qqbotsdk/protocol.py @@ -0,0 +1,25 @@ +"""协议帧处理器接口：由连接适配器逐帧调用，不走事件队列。 + +只有业务事件（op=0）进入事件队列交给 `EventEmitter` 分发给用户 handler； +HELLO/READY/INVALID_SESSION/op=13 这类协议帧在适配器内就地处理，因为它们是 +握手续传语义——混进业务分发既让用户 handler 看到不该看的协议帧，也让"断线 +前重置会话"这类时序取决于分发泵的进度。各接入方式的实现见 `ws_protocol.py` +与 `webhook_protocol.py`。 +""" + +from abc import ABC, abstractmethod +from typing import Any + +from .model import Payload + + +class BaseProtocol(ABC): + """适配器专有的协议处理器：实现方返回需要回送的响应体。""" + + @abstractmethod + async def on_frame(self, payload: Payload) -> Payload | dict[str, Any] | None: + """处理一条帧，返回需回送的响应体（无需回送则 None）。 + + IDENTIFY/RESUME 等握手应答、op=13 验证应答都经返回值回送；业务事件 + （op=0）不必处理，连接器会照常入队分发。 + """ diff --git a/src/qqbotsdk/webhook_protocol.py b/src/qqbotsdk/webhook_protocol.py index 6953755..a7ead99 100644 --- a/src/qqbotsdk/webhook_protocol.py +++ b/src/qqbotsdk/webhook_protocol.py @@ -1,11 +1,15 @@ -"""Webhook 接入的协议层：op=13 验证请求应答（签名 event_ts + plain_token）。""" +"""Webhook 接入的协议层：op=13 验证请求应答（签名 event_ts + plain_token）。 -from typing import override +由 `WebhookConnecter` 在验签前就地调用；返回的应答体由连接器直接写回 HTTP +响应，不走 reply 队列。 +""" + +from typing import Any, cast, override from .config import Config from .crypto import sign_msg -from .emitter import BaseProtocol, EventEmitter -from .model import Opcode, ValidationData +from .model import Opcode, Payload, ValidationData +from .protocol import BaseProtocol class WebhookProtocol(BaseProtocol): @@ -13,10 +17,13 @@ class WebhookProtocol(BaseProtocol): self._config = config @override - def register(self, emitter: EventEmitter) -> None: - emitter.on(Opcode.VALIDATION)(self.validation) - - async def validation(self, data: ValidationData) -> dict[str, str]: + async def on_frame(self, payload: Payload) -> dict[str, Any] | None: + if payload.get("op") != Opcode.VALIDATION: + return None + d = payload.get("d") + if not isinstance(d, dict): + return None + data = cast(ValidationData, d) return { "plain_token": data["plain_token"], "signature": sign_msg( diff --git a/src/qqbotsdk/ws_protocol.py b/src/qqbotsdk/ws_protocol.py index 0d92384..7446381 100644 --- a/src/qqbotsdk/ws_protocol.py +++ b/src/qqbotsdk/ws_protocol.py @@ -1,8 +1,10 @@ -"""WebSocket 接入的协议层：IDENTIFY/RESUME 鉴权、序列号跟踪与会话保持。 +"""WebSocket 接入的协议层：握手鉴权、会话捕获与失效重置。 -心跳与连接生命周期归 WebsocketConnecter；本层只负责会话语义—— -识别 HELLO 后按有无 session_id 决定 RESUME 续传或 IDENTIFY 新建， -收到不可恢复的 INVALID_SESSION（d=false）才清空会话重新 IDENTIFY。 +由 `WebsocketConnecter` 逐帧调用，不经过事件队列：HELLO → 按有无 +session_id 决定 RESUME 续传或 IDENTIFY 新建并返回应答；READY → 捕获 +session_id（该帧同时是 op=0，仍会入队分发给用户的 READY handler）； +INVALID_SESSION → d=false 时清空会话，使重连走 IDENTIFY；每条帧的 s +在此更新，供心跳与 RESUME 取用。 """ from tomllib import load @@ -11,16 +13,8 @@ from typing import override from loguru import logger from .config import Config -from .emitter import BaseProtocol, EventEmitter -from .model import ( - IdentifyData, - Intent, - Opcode, - Payload, - ResumeData, - payload_of, -) -from .payloads import Ready +from .model import IdentifyData, Intent, Opcode, Payload, ResumeData, payload_of +from .protocol import BaseProtocol from .session import Session from .token import AccessToken @@ -67,18 +61,21 @@ class WebsocketProtocol(BaseProtocol): self._token = token @override - def register(self, emitter: EventEmitter) -> None: - emitter.on(Opcode.HELLO)(self.identify) - emitter.on("READY")(self.ready) - emitter.on(Opcode.INVALID_SESSION)(self.invalid_session) - - @override - def on_payload(self, payload: Payload) -> None: + async def on_frame(self, payload: Payload) -> Payload | None: seq = payload.get("s") if isinstance(seq, int): self._session.update_sequence(seq) - async def identify(self) -> Payload: + op = payload.get("op") + if op == Opcode.HELLO: + return await self._handshake() + if op == Opcode.INVALID_SESSION: + self._invalid_session(payload.get("d")) + elif op == Opcode.DISPATCH and payload.get("t") == "READY": + self._ready(payload.get("d")) + return None + + async def _handshake(self) -> Payload: access_token = await self._token.get_access_token() d: ResumeData | IdentifyData @@ -101,11 +98,15 @@ class WebsocketProtocol(BaseProtocol): return payload_of(op, d) - async def ready(self, ready: Ready) -> None: - self._session.session_id = ready.session_id - logger.info("会话已建立（READY）") + def _ready(self, d: object) -> None: + if isinstance(d, dict): + session_id = d.get("session_id") + if isinstance(session_id, str): + self._session.session_id = session_id + logger.info("会话已建立（READY）") - async def invalid_session(self, resumable: bool) -> None: - if not resumable: + def _invalid_session(self, d: object) -> None: + # 官方语义：d 为布尔，true=可 RESUME 续传 + if not d: self._session.session_id = None logger.warning("会话已失效，下次连接将重新 IDENTIFY") diff --git a/tests/test_emitter.py b/tests/test_emitter.py index 9a1e74f..102edc0 100644 --- a/tests/test_emitter.py +++ b/tests/test_emitter.py @@ -2,14 +2,12 @@ import asyncio import pytest -from qqbotsdk.config import Config from qqbotsdk.emitter import EventEmitter +from qqbotsdk.events import Event from qqbotsdk.model import Opcode, payload_of from qqbotsdk.payloads import GroupAtMessage from qqbotsdk.queue import EventQueue from qqbotsdk.session import Session -from qqbotsdk.token import AccessToken -from qqbotsdk.ws_protocol import WebsocketProtocol def test_sequence_only_moves_forward(): @@ -41,7 +39,8 @@ async def test_handler_exception_is_isolated(): @pytest.mark.asyncio -async def test_reply_payload_is_enqueued(): +async def test_protocol_frame_never_reaches_business_handler(): + """协议帧改由适配器的协议处理器就近处理，不再经 emitter 回流应答。""" queue = EventQueue() emitter = EventEmitter(queue) @@ -50,9 +49,7 @@ async def test_reply_payload_is_enqueued(): return payload_of(Opcode.HEARTBEAT, 42) await emitter.emit({"op": Opcode.HEARTBEAT, "d": 0}) - reply = await queue.get_reply() - assert reply["op"] == Opcode.HEARTBEAT - assert reply["d"] == 42 + assert queue.reply_queue.empty() @pytest.mark.asyncio @@ -126,40 +123,23 @@ async def test_service_registered_after_handler_still_injects(): @pytest.mark.asyncio -async def test_scalar_d_falls_back_to_raw_value(): - """回归：op=9 INVALID_SESSION 的 d 是布尔值，DI 回退必须传原始 d。 - 此前实现把 d dict 化成 {}（恒 falsy），resumable=true 也被当成不可续传。 - 走真实协议层 handler 验证路由 → 注入 → 会话状态的全链路。""" - config = Config( - app_id="a", - app_secret="s", - base_url="https://x", - timeout=None, # type: ignore[arg-type] - connecter="websocket", - intents_file="no-such.toml", - webhook_host="0.0.0.0", - webhook_port=8080, - webhook_path="/x", - ) +async def test_typed_and_data_see_same_content(): + """`.typed` 与 `.data` 均可用且一致；只取 `.typed` 的 handler 不触发 data 拷贝。""" emitter = EventEmitter() - session = Session() - session.session_id = "S1" - emitter.register_protocol( - WebsocketProtocol(config, session, AccessToken(config, None)) # type: ignore[arg-type] + seen: list[tuple[str | None, str | None]] = [] + + @emitter.on("GROUP_AT_MESSAGE_CREATE") + async def handler(msg: GroupAtMessage, event: Event): + seen.append((msg.content, event.data["content"])) + + await emitter.emit( + { + "op": Opcode.DISPATCH, + "t": "GROUP_AT_MESSAGE_CREATE", + "d": {"content": "hi", "group_openid": "G1"}, + } ) - seen: list[bool] = [] - - @emitter.on(Opcode.INVALID_SESSION) - async def spy(resumable: bool): - seen.append(resumable) - - await emitter.emit({"op": Opcode.INVALID_SESSION, "d": True}) - assert seen == [True] - assert session.session_id == "S1" # 可续传，会话保留 - - await emitter.emit({"op": Opcode.INVALID_SESSION, "d": False}) - assert seen == [True, False] - assert session.session_id is None # 不可续传，会话清空 + assert seen == [("hi", "hi")] @pytest.mark.asyncio diff --git a/tests/test_route.py b/tests/test_route.py index 0a60da5..2dce242 100644 --- a/tests/test_route.py +++ b/tests/test_route.py @@ -1,36 +1,128 @@ +"""协议帧处理测试：协议处理器（WebsocketProtocol）就近处理握手续传语义， +不进事件分发；只有业务事件（op=0）由 emitter 分发给用户 handler。""" + import pytest +from qqbotsdk.config import Config from qqbotsdk.emitter import EventEmitter from qqbotsdk.model import Opcode +from qqbotsdk.session import Session +from qqbotsdk.ws_protocol import WebsocketProtocol -@pytest.fixture -def emitter() -> EventEmitter: - return EventEmitter() # 无参构造，覆盖 queue 默认自建路径 +class FakeToken: + """手搓 token 源，避免构造真实 AccessToken 去发网络请求。""" + + async def get_access_token(self) -> str: + return "tok" -def test_route_dispatch_uses_t(emitter: EventEmitter): - routed = emitter._route({"op": Opcode.DISPATCH, "t": "READY", "d": {}}) - assert routed is not None - op, name, event = routed - assert op == Opcode.DISPATCH - assert name == "READY" - assert event.type == "READY" +def make_protocol(session: Session | None = None) -> tuple[WebsocketProtocol, Session]: + session = session or Session() + config = Config( + app_id="app", + app_secret="secret", + base_url="https://api.bot.qq.com", + timeout=None, # type: ignore[arg-type] + connecter="websocket", + intents_file="no-such.toml", + webhook_host="0.0.0.0", + webhook_port=8080, + webhook_path="/x", + ) + return WebsocketProtocol(config, session, FakeToken()), session # type: ignore[arg-type] -def test_route_non_dispatch_uses_opcode_name(emitter: EventEmitter): - routed = emitter._route({"op": Opcode.HELLO, "d": {"heartbeat_interval": 1}}) - assert routed is not None - assert routed[1] == "HELLO" +@pytest.mark.asyncio +async def test_hello_without_session_identifies(): + protocol, _ = make_protocol() + reply = await protocol.on_frame( + {"op": Opcode.HELLO, "d": {"heartbeat_interval": 1}} + ) + assert reply is not None + assert reply["op"] == Opcode.IDENTIFY + assert reply["d"]["token"] == "QQBot tok" + assert reply["d"]["shard"] == (0, 1) -def test_route_rejects_missing_op(emitter: EventEmitter): - assert emitter._route({"d": {}}) is None +@pytest.mark.asyncio +async def test_hello_with_session_resumes(): + protocol, session = make_protocol() + session.session_id = "S1" + session.update_sequence(7) + reply = await protocol.on_frame({"op": Opcode.HELLO, "d": {}}) + assert reply is not None + assert reply["op"] == Opcode.RESUME + assert reply["d"]["session_id"] == "S1" + assert reply["d"]["seq"] == 7 -def test_route_rejects_unknown_opcode(emitter: EventEmitter): - assert emitter._route({"op": 999}) is None +@pytest.mark.asyncio +async def test_ready_captures_session_without_reply(): + protocol, session = make_protocol() + reply = await protocol.on_frame( + {"op": Opcode.DISPATCH, "t": "READY", "d": {"session_id": "S9"}} + ) + assert reply is None + assert session.session_id == "S9" -def test_route_rejects_dispatch_without_t(emitter: EventEmitter): - assert emitter._route({"op": Opcode.DISPATCH, "d": {}}) is None +@pytest.mark.asyncio +async def test_sequence_is_tracked_from_frames(): + protocol, session = make_protocol() + await protocol.on_frame({"op": Opcode.DISPATCH, "t": "X", "s": 5, "d": {}}) + await protocol.on_frame( + {"op": Opcode.DISPATCH, "t": "X", "s": 3, "d": {}} + ) # 回退被忽略 + assert session.seq == 5 + + +@pytest.mark.asyncio +async def test_invalid_session_resets_only_when_not_resumable(): + protocol, session = make_protocol() + session.session_id = "S1" + await protocol.on_frame({"op": Opcode.INVALID_SESSION, "d": True}) + assert session.session_id == "S1" # 可续传，会话保留 + await protocol.on_frame({"op": Opcode.INVALID_SESSION, "d": False}) + assert session.session_id is None # 不可续传，下次重连走 IDENTIFY + + +@pytest.mark.asyncio +async def test_business_event_returns_no_reply(): + """业务事件（op=0 非 READY）协议层不产生应答，只留给 emitter 分发。""" + protocol, _ = make_protocol() + reply = await protocol.on_frame( + {"op": Opcode.DISPATCH, "t": "GROUP_AT_MESSAGE_CREATE", "d": {"content": "hi"}} + ) + assert reply is None + + +@pytest.mark.asyncio +async def test_emit_drops_protocol_frames(): + """协议帧不进分发队列；万一进来也不得触达业务 handler。""" + emitter = EventEmitter() + called: list[str] = [] + + @emitter.on(Opcode.HELLO) + async def on_hello(_): + called.append("hello") + + await emitter.emit({"op": Opcode.HELLO, "d": {"heartbeat_interval": 1}}) + assert called == [] + + +@pytest.mark.asyncio +async def test_emit_drops_dispatch_without_t(): + emitter = EventEmitter() + + @emitter.on("READY") + async def on_ready(_): + raise AssertionError("不应被调用") + + await emitter.emit({"op": Opcode.DISPATCH, "d": {}}) + + +@pytest.mark.asyncio +async def test_emit_ignores_missing_op(): + emitter = EventEmitter() + await emitter.emit({"d": {}}) # 无 op，静默跳过 diff --git a/tests/test_webhook_integration.py b/tests/test_webhook_integration.py index deb33da..87d702d 100644 --- a/tests/test_webhook_integration.py +++ b/tests/test_webhook_integration.py @@ -51,8 +51,7 @@ def event_body(event_id: str, content: str) -> dict[str, Any]: async def test_webhook_end_to_end(): config = make_config(0) emitter = EventEmitter() - emitter.register_protocol(WebhookProtocol(config)) - connecter = WebhookConnecter(config, emitter.queue, emitter.handle) + connecter = WebhookConnecter(config, emitter.queue, WebhookProtocol(config)) received: list[GroupAtMessage] = [] diff --git a/tests/test_ws_integration.py b/tests/test_ws_integration.py index 9cf2d68..d92c228 100644 --- a/tests/test_ws_integration.py +++ b/tests/test_ws_integration.py @@ -11,6 +11,7 @@ from aiohttp import ClientSession, WSMsgType, web from qqbotsdk.config import Config from qqbotsdk.connecter import WebsocketConnecter from qqbotsdk.emitter import EventEmitter +from qqbotsdk.model import Opcode from qqbotsdk.payloads import GroupAtMessage from qqbotsdk.queue import EventQueue from qqbotsdk.session import Session @@ -115,16 +116,25 @@ async def test_websocket_end_to_end(): session = Session() emitter = EventEmitter(queue) protocol = WebsocketProtocol(config, session, token) - emitter.register_protocol(protocol) - connecter = WebsocketConnecter(config, http, token, session, queue) + connecter = WebsocketConnecter(config, http, token, session, queue, protocol) received: list[GroupAtMessage] = [] + # 协议帧（HELLO/HEARTBEAT_ACK）绝不该触达 handler——只有 op=0 业务事件入队 + protocol_frames: list[object] = [] @emitter.on("GROUP_AT_MESSAGE_CREATE") async def on_msg(msg: GroupAtMessage) -> None: received.append(msg) gw.got_dispatch.set() + @emitter.on(Opcode.HELLO) + async def on_hello(_) -> None: + protocol_frames.append("HELLO") + + @emitter.on(Opcode.HEARTBEAT_ACK) + async def on_ack(_) -> None: + protocol_frames.append("HEARTBEAT_ACK") + dispatch_task = asyncio.create_task(emitter.dispatch()) connect_task = asyncio.create_task(connecter.run()) try: @@ -146,6 +156,8 @@ async def test_websocket_end_to_end(): assert len(received) == 1 assert received[0].content == "ping" assert received[0].group_openid == "G1" + # 协议帧（HELLO/HEARTBEAT_ACK）没进分发路径 + assert protocol_frames == [] finally: dispatch_task.cancel() connect_task.cancel()