refactor: 工程化收尾——去 pyee、预计算解析计划与派生密钥、命名与接口一致化

以「语义可读、逻辑清晰、减少心智负担」为目标,不追求极致速度:

- **移除 pyee 依赖**:EventEmitter 改为内置的 `事件名 → handler 列表` 注册表
  (on() 追加、emit() 顺序遍历)。SDK 用不到 pyee 的 once/new_listener/错误
  事件机制,去掉后少一个依赖、分发语义一眼可读。注册语义不变(同函数重复
  注册只生效一次),并补测试锁定「重复注册只生效一次」「按注册顺序执行」。
  破坏性:once/remove_listener 等装饰性 API 不再提供。

- **去掉热路径上的重复计算**(纯缓存,不改语义):
  · payloads 按类缓存解析计划(字段名 + 是否嵌套 Author),不再每次事件重跑
    dataclasses.fields 与联合类型解包;未收录字段照旧忽略(新增回归测试,
    线上 payload 会带 author.union_openid 这类字段)。实测 7.5µs → 2.5µs。
  · crypto 按 AppSecret 缓存派生私钥(每个 webhook 请求都要验签,派生一次
    约 40µs);公钥派生只差 3µs,不再多缓存一层。

- **命名与接口一致性**:
  · WebsocketConnecter.reply_helper → send_helper(它从出站缓冲取帧发送,
    原名会让人误以为还有 reply 队列;model.py 的对应注释同步)
  · Connecter 由 typing.Protocol 改为 abc.ABC,与 BaseProtocol 风格统一

- **webhook 去重改摊还清理**:过期判定按当前 id 做(O(1),且 TTL 外重发的
  同一 id 仍按新事件处理——原先的写法在阈值清理下会误判为重复,已由测试
  抓出并修正),整表清过期项只在表长到阈值时执行,不再每请求 O(n) 扫描。

测试 85 全绿(新增未知字段/解析计划缓存/注册语义/去重摊还与 TTL 过期),
ruff check/format + pyright 干净。文档 ARCHITECTURE/API/CHANGELOG 同步。
This commit is contained in:
snowy-yang committed 2026-09-20 22:45:34 +08:00
1 parent 1a7111973b
commit 1552c5207b
13 files changed
+174 -52

No files matched your search

+5 -1
View File
@@ -30,7 +30,11 @@
- 事件分发热路径减负:`emit()` 先取事件名、无 listener 时不再构造 `Event`(每条未订阅事件/协议帧省一次对象与 dict 拷贝);`Event.data` 改为惰性拷贝(只用 `.typed`/`.raw` 的 handler 不再触发 `dict(d)`) - 事件分发热路径减负:`emit()` 先取事件名、无 listener 时不再构造 `Event`(每条未订阅事件/协议帧省一次对象与 dict 拷贝);`Event.data` 改为惰性拷贝(只用 `.typed`/`.raw` 的 handler 不再触发 `dict(d)`)
- ws 重连收尾补 `await`:取消 `reply`/`heartbeat` 任务后 `gather(..., return_exceptions=True)` 取回结果,避免 "Task exception was never retrieved" 并确保它们在下一轮重连前真正停下 - ws 重连收尾补 `await`:取消 `reply`/`heartbeat` 任务后 `gather(..., return_exceptions=True)` 取回结果,避免 "Task exception was never retrieved" 并确保它们在下一轮重连前真正停下
- `EventQueue.put_event`/`put_reply` 去掉多余的 `return await` - `EventQueue.put_event`/`put_reply` 去掉多余的 `return await`
- `EventQueue` 收敛为**单向业务事件通道**:删除 `reply_queue` 与 `put_reply`/`get_reply`(协议分流后引用它的只剩 `WebsocketConnecter`)。ws 的出站帧(心跳 + 协议应答)改由 `WebsocketConnecter` 私有出站缓冲承载、`reply_helper` 单任务发送(维持 aiohttp ws 单写者不变量);webhook 的应答直接写 HTTP 响应体。破坏性变更:`EventQueue.reply_queue` 与 `put_reply`/`get_reply` 不再存在 - `EventQueue` 收敛为**单向业务事件通道**:删除 `reply_queue` 与 `put_reply`/`get_reply`(协议分流后引用它的只剩 `WebsocketConnecter`)。ws 的出站帧(心跳 + 协议应答)改由 `WebsocketConnecter` 私有出站缓冲承载、`send_helper` 单任务发送(维持 aiohttp ws 单写者不变量);webhook 的应答直接写 HTTP 响应体。破坏性变更:`EventQueue.reply_queue` 与 `put_reply`/`get_reply` 不再存在
- **移除 pyee 依赖**:`EventEmitter` 改为内置的 `事件名 → handler 列表` 注册表(`on()` 追加、`emit()` 顺序遍历),不再需要 pyee 的 `once`/`new_listener`/错误事件机制。注册语义保持不变(同函数重复注册只生效一次、按注册顺序串行执行)。破坏性变更:`once`/`remove_listener` 等 pyee 装饰性 API 不再提供(SDK 与公开文档本就不涉及)
- 热路径去掉重复计算(不改语义):`payloads` 按类缓存解析计划(不再每次事件重跑 `dataclasses.fields` 与联合类型解包,实测 7.5µs→2.5µs;未收录字段照旧忽略);`crypto` 按 AppSecret 缓存派生私钥(验签每请求省约 40µs 的密钥派生)
- `WebsocketConnecter` 的 `reply_helper` 更名为 `send_helper`(它从出站缓冲取帧发送,原名会让人以为还有 reply 队列);`Connecter` 由 `typing.Protocol` 改为 `abc.ABC`,与 `BaseProtocol` 的接口风格一致
- webhook 去重改为摊还清理:过期判定按当前 id 做(O(1),TTL 外重发的同一 id 仍按新事件处理),整表清过期项只在表长到阈值时执行(原先每请求全表扫描,O(n))
### 移除 ### 移除
+2 -1
View File
@@ -21,7 +21,7 @@ from qqbotsdk import (
| `ee.services` | `dict[type, Any]`,按类型登记可注入组件;`run_loop` 装配 SDK 内置组件,也可登记自定义类型 | | `ee.services` | `dict[type, Any]`,按类型登记可注入组件;`run_loop` 装配 SDK 内置组件,也可登记自定义类型 |
| `EventQueue()` | 业务事件通道(连接层 → 分发层,单向);`EventEmitter` 默认自建,需多处共享时手动传入(`ee.queue` 取实例) | | `EventQueue()` | 业务事件通道(连接层 → 分发层,单向);`EventEmitter` 默认自建,需多处共享时手动传入(`ee.queue` 取实例) |
| `Config` | frozen dataclass,环境变量一次性读齐(见下文) | | `Config` | frozen dataclass,环境变量一次性读齐(见下文) |
| `Connecter` | 连接适配器 Protocol(接口),按 `CONNECTER` 自动选择实现 | | `Connecter` | 连接适配器抽象基类(ABC),按 `CONNECTER` 自动选择实现 |
| `BaseProtocol` | 协议处理器抽象基类(`protocol.py`),自定义接入方式时继承并实现 `on_frame(payload)` | | `BaseProtocol` | 协议处理器抽象基类(`protocol.py`),自定义接入方式时继承并实现 `on_frame(payload)` |
子包按需导入: 子包按需导入:
@@ -45,6 +45,7 @@ async def slow(...): ...
``` ```
- **只分发业务事件**:`EventEmitter` 只处理 op=0(DISPATCH)事件,按 payload 的 `t` 名路由。协议帧(HELLO/READY/INVALID_SESSION/op=13 等)由各接入方式的协议处理器(`WebsocketProtocol`/`WebhookProtocol`)在连接器内就地处理,不进事件队列、也不会触达你的 handler。 - **只分发业务事件**:`EventEmitter` 只处理 op=0(DISPATCH)事件,按 payload 的 `t` 名路由。协议帧(HELLO/READY/INVALID_SESSION/op=13 等)由各接入方式的协议处理器(`WebsocketProtocol`/`WebhookProtocol`)在连接器内就地处理,不进事件队列、也不会触达你的 handler。
- **注册与顺序**:同一函数对同一事件重复注册只生效一次;handler 按注册顺序串行执行(`background=True` 的除外)。
- **异常隔离**:handler 异常就地记录(loguru),不影响同事件其他 handler 与主循环。 - **异常隔离**:handler 异常就地记录(loguru),不影响同事件其他 handler 与主循环。
- **后台并发**:`background=True` 的 handler 进后台任务,emit 不等它;同一事件内失去先后保证,仅用于业务 handler。`run_loop` 退出时统一取消在飞后台任务(`ee.close()`)。 - **后台并发**:`background=True` 的 handler 进后台任务,emit 不等它;同一事件内失去先后保证,仅用于业务 handler。`run_loop` 退出时统一取消在飞后台任务(`ee.close()`)。
- 业务 handler 的返回值不回流;回复消息请在 handler 里显式调 `BotApi`。 - 业务 handler 的返回值不回流;回复消息请在 handler 里显式调 `BotApi`。
+6 -4
View File
@@ -110,12 +110,14 @@ sequenceDiagram
2. **协议层只管会话语义,心跳归传输层**:`WebsocketProtocol` 负责 IDENTIFY/RESUME/序列号(说什么),`WebsocketConnecter` 负责连接与心跳(怎么保活)——心跳任务随连接生灭(HELLO 给出间隔后启动,断线即取消),断线期间不在出站缓冲堆积过期心跳,也免去了调度器的生命周期管理。协议处理器的应答经 `on_frame` 返回值交给 connecter:ws 与心跳一起进 connecter 私有的出站缓冲、由单任务发送(维持 ws 单写者不变量),webhook 则直接写进 HTTP 响应体。 2. **协议层只管会话语义,心跳归传输层**:`WebsocketProtocol` 负责 IDENTIFY/RESUME/序列号(说什么),`WebsocketConnecter` 负责连接与心跳(怎么保活)——心跳任务随连接生灭(HELLO 给出间隔后启动,断线即取消),断线期间不在出站缓冲堆积过期心跳,也免去了调度器的生命周期管理。协议处理器的应答经 `on_frame` 返回值交给 connecter:ws 与心跳一起进 connecter 私有的出站缓冲、由单任务发送(维持 ws 单写者不变量),webhook 则直接写进 HTTP 响应体。
3. **看门狗两层互补**(见 `connecter.py` 模块注释):静默超时抓连接彻底失联,ACK 期限检查抓"有事件流但心跳已死"。阈值 `_SILENCE_FACTOR=1.1`、`_ACK_FACTOR=1.2` 是按 ACK 自然节奏校准的,**不能压到 1× 整**,否则正常抖动会周期性误杀(误杀由 RESUME 兜底,不丢消息)。重连退避只在连接稳定存活(`_STABLE_SECONDS`)后才归一,抖动的服务端不会被 1s 间隔反复探测。 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()` 统一取消。 4. **handler 异常就地隔离**:`_invoke` 捕获记日志,`dispatch` 主循环另有兜底;单条事件的单个 handler 失败不影响进程。`background=True` 的 handler 进后台任务(emit 不等它),任务由 emitter 持强引用、`close()` 统一取消。
5. **注入约定**:handler 参数按形参标注解析——`payloads.py` 里的 dataclass → 解析后的 payload 对象;`Event` → 事件封装;登记进 `emitter.services` 的组件类型(如 `BotApi`)→ 实例;未标注或未登记 → 回退传**原始 d**。解析是纯同步查表(`get_type_hints` + 签名缓存),无框架、无运行时魔法。签名与类型标注每个 handler 只解析一次并缓存,但 `services` 命中在每条事件时实时判定,保证"先注册 handler、后登记组件"的装配顺序可用。 5. **注入约定**:handler 参数按形参标注解析——`payloads.py` 里的 dataclass → 解析后的 payload 对象;`Event` → 事件封装;登记进 `emitter.services` 的组件类型(如 `BotApi`)→ 实例;未标注或未登记 → 回退传**原始 d**。解析是纯同步查表,无框架、无运行时魔法:每个 handler 的 `get_type_hints` + 形参名只解析一次并缓存,但 `services` 命中在每条事件时实时判定,保证"先注册 handler、后登记组件"的装配顺序可用。
6. **OpenAPI 请求内核自治**:`api/core.py` 统一处理鉴权头、429 限流重试(优先 `Retry-After`,脏值容错,退避封顶)与 401 刷新 token 重试一次;v2/channel/guild 三个域方法集继承内核组合成 `BotApi`,新增域加模块即可,方法对用户始终平铺在 `BotApi` 上。 6. **事件注册不依赖第三方库**:`EventEmitter` 内部就是一个 `事件名 → handler 列表` 的字典,`on()` 追加、`emit()` 顺序遍历。不需要 pyee 那套 `once`/`new_listener`/错误事件机制(SDK 用不到),少一个依赖、语义也一眼可读。注册语义:同一函数对同一事件重复注册只生效一次(沿用原先 pyee 的行为),分发按注册顺序串行执行(`background=True` 的除外)。
7. **解析计划与派生密钥都预计算**:`payloads._parse_plan` 把 dataclass 的"字段名 + 是否需解析成 Author"按类缓存一次(不再每次事件重跑 `dataclasses.fields` 与联合类型解包),解析时未收录字段一律忽略——线上 payload 会带新增/未声明字段,透传会让构造报错。`crypto` 按 AppSecret 缓存派生出的 Ed25519 私钥(派生约 40µs,而每个 webhook 请求都要验签)。都是"值为输入的纯函数、结果与输入一一对应",缓存安全且收益明确;不追求极致速度,只去掉明显的重复计算。
8. **OpenAPI 请求内核自治**:`api/core.py` 统一处理鉴权头、429 限流重试(优先 `Retry-After`,脏值容错,退避封顶)与 401 刷新 token 重试一次;v2/channel/guild 三个域方法集继承内核组合成 `BotApi`,新增域加模块即可,方法对用户始终平铺在 `BotApi` 上。
## 扩展点 ## 扩展点
- **新增接入方式**(如轮询/其他平台):实现 `Connecter.run()` 与 `BaseProtocol.on_frame()`,在 `run_loop` 的 `match` 装配分支加一个 `case`(构造该方式的协议处理器并注入 connecter),`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)。 - **新增业务 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 的全部事件组(消息、成员变动、审核、表态、互动、论坛、音频)。 - **新增事件类型 dataclass**:在 `payloads.py` 定义 dataclass 并登记 `EVENT_TYPES`,handler 即可按类型标注拿解析结果;未收录事件回退传 dict。当前已覆盖 intents.toml 的全部事件组(消息、成员变动、审核、表态、互动、论坛、音频)。
@@ -123,7 +125,7 @@ sequenceDiagram
`tests/` 覆盖: `tests/` 覆盖:
- 纯逻辑:crypto 往返、intents 解析、Config 校验、协议帧处理(`test_route.py`:IDENTIFY/RESUME 选路、READY 捕获、s 序列号、INVALID_SESSION 重置、业务事件不产生应答)、webhook 去重、序列号单调性、handler 异常隔离、payload dataclass 解析、BotApi 各业务方法的拼参(Recording 子类截获 request)。 - 纯逻辑:crypto 往返、intents 解析、Config 校验、协议帧处理(`test_route.py`:IDENTIFY/RESUME 选路、READY 捕获、s 序列号、INVALID_SESSION 重置、业务事件不产生应答)、webhook 去重(含 TTL 过期与摊还清理)、序列号单调性、handler 异常隔离、handler 注册语义(同函数重复注册只生效一次、按注册顺序执行)、payload dataclass 解析(含未收录字段忽略、解析计划缓存)、BotApi 各业务方法的拼参(Recording 子类截获 request)。
- **网络集成**(`test_ws_integration.py` / `test_webhook_integration.py`): - **网络集成**(`test_ws_integration.py` / `test_webhook_integration.py`):
- WebSocket:aiohttp 假网关(token 接口 + gateway + ws 协议行为)端到端跑通 HELLO → IDENTIFY → READY 会话捕获 → 心跳/ACK(心跳任务随连接生灭)→ 业务事件 dataclass 分发,并断言协议帧(HELLO/HEARTBEAT_ACK)不会触达 handler; - WebSocket:aiohttp 假网关(token 接口 + gateway + ws 协议行为)端到端跑通 HELLO → IDENTIFY → READY 会话捕获 → 心跳/ACK(心跳任务随连接生灭)→ 业务事件 dataclass 分发,并断言协议帧(HELLO/HEARTBEAT_ACK)不会触达 handler;
- Webhook:真实起 `WebhookConnecter` 的 http server(TestServer),覆盖 op=13 验证应答(签名内容正确性)、合法签名放行分发、坏签名 401、同 id 去重。 - Webhook:真实起 `WebhookConnecter` 的 http server(TestServer),覆盖 op=13 验证应答(签名内容正确性)、合法签名放行分发、坏签名 401、同 id 去重。
-1
View File
@@ -13,7 +13,6 @@ dependencies = [
"aiohttp>=3.14.3", "aiohttp>=3.14.3",
"cryptography>=50.0.1", "cryptography>=50.0.1",
"loguru>=0.7.3", "loguru>=0.7.3",
"pyee>=14.0.0",
"python-dotenv>=1.2.3", "python-dotenv>=1.2.3",
"ujson>=6.0.0", "ujson>=6.0.0",
] ]
+27 -12
View File
@@ -15,8 +15,9 @@
""" """
import asyncio import asyncio
from abc import ABC, abstractmethod
from time import monotonic from time import monotonic
from typing import Protocol, cast from typing import cast
import aiohttp import aiohttp
import ujson import ujson
@@ -32,7 +33,10 @@ from .session import Session
from .token import AccessToken from .token import AccessToken
class Connecter(Protocol): class Connecter(ABC):
"""连接适配器接口:两种接入方式都实现 `run()` 常驻运行。"""
@abstractmethod
async def run(self) -> None: ... async def run(self) -> None: ...
@@ -40,7 +44,7 @@ class RateLimitError(RuntimeError):
"""gateway 接口触发频率限制(code=100017)。""" """gateway 接口触发频率限制(code=100017)。"""
class WebsocketConnecter: class WebsocketConnecter(Connecter):
# 看门狗阈值(× 心跳间隔):贴着 ACK 的自然节奏(1×)跑,误杀由 RESUME 兜底不丢消息; # 看门狗阈值(× 心跳间隔):贴着 ACK 的自然节奏(1×)跑,误杀由 RESUME 兜底不丢消息;
# 不能压到 1× 整——ACK 的自然间隔就是 1×,正常抖动会贴边误杀;TCP 下丢包表现为延迟而非静默 # 不能压到 1× 整——ACK 的自然间隔就是 1×,正常抖动会贴边误杀;TCP 下丢包表现为延迟而非静默
_SILENCE_FACTOR = 1.1 _SILENCE_FACTOR = 1.1
@@ -63,7 +67,7 @@ class WebsocketConnecter:
self._session = session self._session = session
self._queue = queue self._queue = queue
self._protocol = protocol self._protocol = protocol
# 出站帧缓冲(心跳 + 协议应答):由 reply_helper 单任务发送, # 出站帧缓冲(心跳 + 协议应答):由 send_helper 单任务发送,
# 维持 aiohttp ws 单写者不变量,生产者(心跳)也不会被发送阻塞 # 维持 aiohttp ws 单写者不变量,生产者(心跳)也不会被发送阻塞
self._outbound: asyncio.Queue[Payload] = asyncio.Queue() self._outbound: asyncio.Queue[Payload] = asyncio.Queue()
@@ -107,16 +111,16 @@ class WebsocketConnecter:
# HELLO 才给出心跳间隔;receive 解析到后经 Future 交给心跳任务 # HELLO 才给出心跳间隔;receive 解析到后经 Future 交给心跳任务
interval: asyncio.Future[float] = asyncio.get_running_loop().create_future() interval: asyncio.Future[float] = asyncio.get_running_loop().create_future()
receive = asyncio.create_task(self.receive_helper(ws, interval)) receive = asyncio.create_task(self.receive_helper(ws, interval))
reply = asyncio.create_task(self.reply_helper(ws)) send = asyncio.create_task(self.send_helper(ws))
heartbeat = asyncio.create_task(self.heartbeat_helper(interval)) heartbeat = asyncio.create_task(self.heartbeat_helper(interval))
try: try:
await receive await receive
finally: finally:
reply.cancel() send.cancel()
heartbeat.cancel() heartbeat.cancel()
# 取回被取消任务的结果,避免 "Task exception was never retrieved", # 取回被取消任务的结果,避免 "Task exception was never retrieved",
# 也确保它们在下一轮重连前真正停下 # 也确保它们在下一轮重连前真正停下
await asyncio.gather(reply, heartbeat, return_exceptions=True) await asyncio.gather(send, heartbeat, return_exceptions=True)
logger.info("WebSocket 已断开") logger.info("WebSocket 已断开")
return monotonic() - started return monotonic() - started
@@ -165,7 +169,8 @@ class WebsocketConnecter:
# 服务端不会推送 op=1,心跳只由本任务投进出站缓冲 # 服务端不会推送 op=1,心跳只由本任务投进出站缓冲
await self._outbound.put(payload_of(Opcode.HEARTBEAT, self._session.seq)) await self._outbound.put(payload_of(Opcode.HEARTBEAT, self._session.seq))
async def reply_helper(self, ws: aiohttp.ClientWebSocketResponse) -> None: async def send_helper(self, ws: aiohttp.ClientWebSocketResponse) -> None:
"""唯一的 ws 写者:从出站缓冲取帧发送(保持 aiohttp 单写者约束)。"""
while not ws.closed: while not ws.closed:
payload = await self._outbound.get() payload = await self._outbound.get()
body = {k: v for k, v in payload.items() if v != NOT_SET} body = {k: v for k, v in payload.items() if v != NOT_SET}
@@ -175,8 +180,10 @@ class WebsocketConnecter:
logger.warning(f"WebSocket 发送失败,丢弃该帧: {body} ({e})") logger.warning(f"WebSocket 发送失败,丢弃该帧: {body} ({e})")
class WebhookConnecter: class WebhookConnecter(Connecter):
_DEDUP_TTL = 60 _DEDUP_TTL = 60
# 去重表长度达到该值才做一次过期清理(摊还 O(1),避免每请求全表扫描)
_DEDUP_SWEEP_AT = 1024
def __init__( def __init__(
self, self,
@@ -212,13 +219,21 @@ class WebhookConnecter:
await runner.cleanup() await runner.cleanup()
def _duplicated(self, event_id: str | None) -> bool: def _duplicated(self, event_id: str | None) -> bool:
"""TTL 内见过的同 id 视为重复推送。
过期判断按当前 id 单独做(O(1)),保证 TTL 外重发的同一 id 会被当作
新事件处理;整表清过期项只在表长到阈值时做一次,避免每请求 O(n) 扫描。
"""
if not event_id: if not event_id:
return False return False
now = monotonic() now = monotonic()
for key in [k for k, ts in self._seen.items() if now - ts > self._DEDUP_TTL]: seen_at = self._seen.get(event_id)
del self._seen[key] if seen_at is not None and now - seen_at <= self._DEDUP_TTL:
if event_id in self._seen:
return True return True
if len(self._seen) >= self._DEDUP_SWEEP_AT:
self._seen = {
k: ts for k, ts in self._seen.items() if now - ts <= self._DEDUP_TTL
}
self._seen[event_id] = now self._seen[event_id] = now
return False return False
+14 -4
View File
@@ -3,6 +3,8 @@
seed 派生规则(官方约定):AppSecret 的 utf8 字节反复倍增补足 32 字节。 seed 派生规则(官方约定):AppSecret 的 utf8 字节反复倍增补足 32 字节。
""" """
from functools import cache
from cryptography.exceptions import InvalidSignature from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric import ed25519 from cryptography.hazmat.primitives.asymmetric import ed25519
@@ -16,16 +18,24 @@ def build_seed(secret: str) -> bytes:
return seed[:_ED25519_SEED_SIZE] return seed[:_ED25519_SEED_SIZE]
@cache
def _private_key(secret: str) -> ed25519.Ed25519PrivateKey:
"""AppSecret → Ed25519 私钥。
派生一次约 40µs(每个 webhook 请求都要验签),而进程内 secret 是固定的,
按 secret 缓存即可把这一步省掉。公钥派生只差 3µs,不值得再缓存一层。
"""
return ed25519.Ed25519PrivateKey.from_private_bytes(build_seed(secret))
def sign_msg(secret: str, msg: str) -> str: def sign_msg(secret: str, msg: str) -> str:
private_key = ed25519.Ed25519PrivateKey.from_private_bytes(build_seed(secret)) return _private_key(secret).sign(msg.encode("utf-8")).hex()
return private_key.sign(msg.encode("utf-8")).hex()
def verify_sig(secret: str, timestamp: str, signature: str, body: bytes) -> bool: def verify_sig(secret: str, timestamp: str, signature: str, body: bytes) -> bool:
try: try:
sig = bytes.fromhex(signature) sig = bytes.fromhex(signature)
private_key = ed25519.Ed25519PrivateKey.from_private_bytes(build_seed(secret)) _private_key(secret).public_key().verify(sig, timestamp.encode("utf-8") + body)
private_key.public_key().verify(sig, timestamp.encode("utf-8") + body)
return True return True
except (InvalidSignature, ValueError): except (InvalidSignature, ValueError):
return False return False
+10 -7
View File
@@ -18,7 +18,6 @@ from collections.abc import Awaitable, Callable
from typing import Any, TypeVar, get_type_hints, overload from typing import Any, TypeVar, get_type_hints, overload
from loguru import logger from loguru import logger
from pyee.asyncio import AsyncIOEventEmitter
from .events import Event from .events import Event
from .model import Opcode, Payload from .model import Opcode, Payload
@@ -30,8 +29,9 @@ Handler = TypeVar("Handler", bound=Callable[..., Awaitable[Any]])
class EventEmitter: class EventEmitter:
def __init__(self, queue: EventQueue | None = None) -> None: def __init__(self, queue: EventQueue | None = None) -> None:
self._ee = AsyncIOEventEmitter()
self._queue = queue if queue is not None else EventQueue() self._queue = queue if queue is not None else EventQueue()
# 事件名 → handler(按注册顺序);同名事件重复注册同一函数只生效一次
self._handlers: dict[str, list[Callable[..., Awaitable[Any]]]] = {}
self.services: dict[type, Any] = {} self.services: dict[type, Any] = {}
# handler → (类型标注, 形参名):签名解析较贵,缓存后每条事件只查表 # handler → (类型标注, 形参名):签名解析较贵,缓存后每条事件只查表
self._signatures: dict[ self._signatures: dict[
@@ -65,7 +65,9 @@ class EventEmitter:
name = event.name if isinstance(event, Opcode) else event name = event.name if isinstance(event, Opcode) else event
def decorator(fn: Handler) -> Handler: def decorator(fn: Handler) -> Handler:
self._ee.on(name, fn) handlers = self._handlers.setdefault(name, [])
if fn not in handlers:
handlers.append(fn)
if background: if background:
self._background.add(fn) self._background.add(fn)
return fn return fn
@@ -93,7 +95,7 @@ class EventEmitter:
logger.exception(f"事件分发异常,已跳过: {data} ({e})") logger.exception(f"事件分发异常,已跳过: {data} ({e})")
async def emit(self, data: Payload) -> None: async def emit(self, data: Payload) -> None:
# 先取事件名,无 listener 就不必构造 Event(每条未订阅/协议帧都省一次拷贝) # 先取事件名,无 handler 就不必构造 Event(每条未订阅事件都省一次拷贝)
op = data.get("op") op = data.get("op")
if op != Opcode.DISPATCH: if op != Opcode.DISPATCH:
logger.warning(f"非业务事件不应进分发队列,已跳过: {data}") logger.warning(f"非业务事件不应进分发队列,已跳过: {data}")
@@ -103,12 +105,13 @@ class EventEmitter:
logger.warning(f"业务事件缺少 t 字段,已跳过: {data}") logger.warning(f"业务事件缺少 t 字段,已跳过: {data}")
return return
listeners = self._ee.listeners(t) handlers = self._handlers.get(t)
if not listeners: if not handlers:
return return
event = Event(data) event = Event(data)
for fn in listeners: # 遍历快照:handler 在分发过程中注册新 handler,不影响本次分发
for fn in list(handlers):
if fn in self._background: if fn in self._background:
self._spawn(fn, event) self._spawn(fn, event)
else: else:
+1 -1
View File
@@ -1,7 +1,7 @@
"""协议模型:Opcode、payload 及各事件 d 字段的 TypedDict、Intent 位掩码。 """协议模型:Opcode、payload 及各事件 d 字段的 TypedDict、Intent 位掩码。
Payload 里 `NOT_SET` 哨兵标记"外发时应省略"的字段,由 Payload 里 `NOT_SET` 哨兵标记"外发时应省略"的字段,由
WebsocketConnecter.reply_helper 统一过滤。 WebsocketConnecter.send_helper 统一过滤。
""" """
from enum import IntEnum, IntFlag from enum import IntEnum, IntFlag
+25 -6
View File
@@ -8,7 +8,7 @@ handler 直接标注具体类型即可拿到解析后的对象(如
from dataclasses import dataclass, fields from dataclasses import dataclass, fields
from types import UnionType from types import UnionType
from typing import Union, get_args, get_origin from typing import Union, get_args, get_origin, get_type_hints
@dataclass(slots=True) @dataclass(slots=True)
@@ -339,15 +339,34 @@ def _is_author_type(annotation: object) -> bool:
return False return False
# 字段名与标注都是类的静态属性,解析计划只需构建一次。每次事件都跑一遍
# dataclasses.fields 与联合类型解包,是热路径上最贵的一步。
_PARSE_PLANS: dict[type, tuple[tuple[str, bool], ...]] = {}
def _parse_plan(cls: type) -> tuple[tuple[str, bool], ...]:
"""(字段名, 该字段的 dict 值是否需解析成 Author),按定义顺序。"""
if (plan := _PARSE_PLANS.get(cls)) is None:
hints = get_type_hints(cls)
plan = tuple((f.name, _is_author_type(hints[f.name])) for f in fields(cls))
_PARSE_PLANS[cls] = plan
return plan
def _build(cls: type, d: dict) -> object: def _build(cls: type, d: dict) -> object:
"""按解析计划填充 dataclass。
d 里未收录的字段一律忽略——线上 payload 会带新增或未声明的字段
(如 author.union_openid),直接透传会让 dataclass 构造报错。
"""
kwargs = {} kwargs = {}
for f in fields(cls): for name, nested_author in _parse_plan(cls):
if f.name not in d: if name not in d:
continue continue
value = d[f.name] value = d[name]
if isinstance(value, dict) and _is_author_type(f.type): if nested_author and isinstance(value, dict):
value = _build(Author, value) value = _build(Author, value)
kwargs[f.name] = value kwargs[name] = value
return cls(**kwargs) return cls(**kwargs)
+24
View File
@@ -1,3 +1,5 @@
from time import monotonic
from qqbotsdk.config import Config from qqbotsdk.config import Config
from qqbotsdk.connecter import WebhookConnecter from qqbotsdk.connecter import WebhookConnecter
from qqbotsdk.queue import EventQueue from qqbotsdk.queue import EventQueue
@@ -48,3 +50,25 @@ def test_expired_entries_are_evicted():
c._seen[key] -= c._DEDUP_TTL + 1 c._seen[key] -= c._DEDUP_TTL + 1
assert not c._duplicated("evt-1") assert not c._duplicated("evt-1")
assert "evt-1" in c._seen # 以新时间戳重新记录 assert "evt-1" in c._seen # 以新时间戳重新记录
def test_expired_id_is_treated_as_new():
"""TTL 之外重发的同一 id 是新事件,不能被当成重复丢掉。"""
c = make_connecter()
assert not c._duplicated("evt-x")
c._seen["evt-x"] -= c._DEDUP_TTL + 1 # 只把这一个改成过期
assert not c._duplicated("evt-x")
def test_table_is_swept_only_at_threshold():
"""表长未到阈值时不做全表清理(摊还 O(1)),过期项留在表里也不影响判定。"""
c = make_connecter()
assert not c._duplicated("old")
c._seen["old"] -= c._DEDUP_TTL + 1
assert not c._duplicated("new") # 未到阈值:不清理
assert "old" in c._seen
# 灌到阈值,下一次判定触发清理,过期项被移除
for i in range(c._DEDUP_SWEEP_AT):
c._seen[f"filler-{i}"] = monotonic()
assert not c._duplicated("another")
assert "old" not in c._seen
+32
View File
@@ -20,6 +20,38 @@ def test_sequence_only_moves_forward():
assert s.seq == 7 assert s.seq == 7
@pytest.mark.asyncio
async def test_same_handler_registered_twice_runs_once():
"""同名事件重复注册同一函数只生效一次(沿用 pyee 的语义)。"""
emitter = EventEmitter()
calls: list[str] = []
async def handler(_):
calls.append("hit")
emitter.on("READY", handler)
emitter.on("READY", handler)
await emitter.emit({"op": Opcode.DISPATCH, "t": "READY", "d": {}})
assert calls == ["hit"]
@pytest.mark.asyncio
async def test_handlers_run_in_registration_order():
emitter = EventEmitter()
order: list[int] = []
@emitter.on("READY")
async def first(_):
order.append(1)
@emitter.on("READY")
async def second(_):
order.append(2)
await emitter.emit({"op": Opcode.DISPATCH, "t": "READY", "d": {}})
assert order == [1, 2]
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_handler_exception_is_isolated(): async def test_handler_exception_is_isolated():
queue = EventQueue() queue = EventQueue()
+28 -1
View File
@@ -1,6 +1,7 @@
from qqbotsdk.emitter import EventEmitter from qqbotsdk.emitter import EventEmitter
from qqbotsdk.model import Opcode from qqbotsdk.model import Opcode
from qqbotsdk.payloads import ( from qqbotsdk.payloads import (
_PARSE_PLANS,
C2CMessage, C2CMessage,
GroupAtMessage, GroupAtMessage,
Interaction, Interaction,
@@ -10,7 +11,33 @@ from qqbotsdk.payloads import (
from qqbotsdk.queue import EventQueue from qqbotsdk.queue import EventQueue
def test_parse_group_at_message(): def test_parse_ignores_unknown_fields():
"""回归:线上 payload 会带未收录字段(如 author.union_openid),
解析必须忽略而非透传给 dataclass 构造(那会直接报错)。"""
msg = parse_event(
"GROUP_AT_MESSAGE_CREATE",
{
"id": "msgid-9",
"content": "hi",
"group_openid": "G1",
"unknown_top_level": {"whatever": 1},
"author": {"member_openid": "M9", "union_openid": "U-should-be-dropped"},
},
)
assert isinstance(msg, GroupAtMessage)
assert msg.content == "hi"
assert msg.author is not None and msg.author.member_openid == "M9"
def test_parse_plan_is_cached_per_class():
"""解析计划按类缓存,且二次解析结果与首次一致。"""
d = {"content": "x", "group_openid": "G"}
first = parse_event("GROUP_AT_MESSAGE_CREATE", d)
second = parse_event("GROUP_AT_MESSAGE_CREATE", d)
assert isinstance(first, GroupAtMessage) and isinstance(second, GroupAtMessage)
assert first.content == second.content == "x"
assert GroupAtMessage in _PARSE_PLANS
msg = parse_event( msg = parse_event(
"GROUP_AT_MESSAGE_CREATE", "GROUP_AT_MESSAGE_CREATE",
{ {
Generated
-14
View File
@@ -543,18 +543,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/0c/c3/44f3fbbfa403ea2a7c779186dc20772604442dde72947e7d01069cbe98e3/pycparser-3.0-py3-none-any.whl", hash = "sha256:b727414169a36b7d524c1c3e31839a521725078d7b2ff038656844266160a992", size = 48172, upload-time = "2026-01-21T14:26:50.693Z" }, { url = "https://files.pythonhosted.org/packages/0c/c3/44f3fbbfa403ea2a7c779186dc20772604442dde72947e7d01069cbe98e3/pycparser-3.0-py3-none-any.whl", hash = "sha256:b727414169a36b7d524c1c3e31839a521725078d7b2ff038656844266160a992", size = 48172, upload-time = "2026-01-21T14:26:50.693Z" },
] ]
[[package]]
name = "pyee"
version = "14.0.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/1d/f1/fdedc2c75c3e31a330659c85e5793bb18b3397981fbf0844c6dee5b18926/pyee-14.0.0.tar.gz", hash = "sha256:76dd0f4314ecd27f02dc73589dea7fd3853f9b6176d8ef9b122860657e3602de", size = 98760, upload-time = "2026-08-13T04:26:11.021Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/81/12/5347938b1f9a6453f0dbdfcc3e2388a1320ef9b9ec17fbefbc4ab647ea98/pyee-14.0.0-py3-none-any.whl", hash = "sha256:3ac2d3229a9677f7de2c33d7f52fe25b638a46b19c413fea2edc8c6d0a644e4d", size = 15553, upload-time = "2026-08-13T04:26:09.916Z" },
]
[[package]] [[package]]
name = "pygments" name = "pygments"
version = "2.21.0" version = "2.21.0"
@@ -622,7 +610,6 @@ dependencies = [
{ name = "aiohttp" }, { name = "aiohttp" },
{ name = "cryptography" }, { name = "cryptography" },
{ name = "loguru" }, { name = "loguru" },
{ name = "pyee" },
{ name = "python-dotenv" }, { name = "python-dotenv" },
{ name = "ujson" }, { name = "ujson" },
] ]
@@ -640,7 +627,6 @@ requires-dist = [
{ name = "aiohttp", specifier = ">=3.14.3" }, { name = "aiohttp", specifier = ">=3.14.3" },
{ name = "cryptography", specifier = ">=50.0.1" }, { name = "cryptography", specifier = ">=50.0.1" },
{ name = "loguru", specifier = ">=0.7.3" }, { name = "loguru", specifier = ">=0.7.3" },
{ name = "pyee", specifier = ">=14.0.0" },
{ name = "python-dotenv", specifier = ">=1.2.3" }, { name = "python-dotenv", specifier = ">=1.2.3" },
{ name = "ujson", specifier = ">=6.0.0" }, { name = "ujson", specifier = ">=6.0.0" },
] ]