diff --git a/CHANGELOG.md b/CHANGELOG.md index 5d2f372..32d8e2b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,7 +30,11 @@ - 事件分发热路径减负：`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` -- `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)） ### 移除 diff --git a/docs/API.md b/docs/API.md index 64417db..8f86e38 100644 --- a/docs/API.md +++ b/docs/API.md @@ -21,7 +21,7 @@ from qqbotsdk import ( | `ee.services` | `dict[type, Any]`，按类型登记可注入组件；`run_loop` 装配 SDK 内置组件，也可登记自定义类型 | | `EventQueue()` | 业务事件通道（连接层 → 分发层，单向）；`EventEmitter` 默认自建，需多处共享时手动传入（`ee.queue` 取实例） | | `Config` | frozen dataclass，环境变量一次性读齐（见下文） | -| `Connecter` | 连接适配器 Protocol（接口），按 `CONNECTER` 自动选择实现 | +| `Connecter` | 连接适配器抽象基类（ABC），按 `CONNECTER` 自动选择实现 | | `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。 +- **注册与顺序**：同一函数对同一事件重复注册只生效一次；handler 按注册顺序串行执行（`background=True` 的除外）。 - **异常隔离**：handler 异常就地记录（loguru），不影响同事件其他 handler 与主循环。 - **后台并发**：`background=True` 的 handler 进后台任务，emit 不等它；同一事件内失去先后保证，仅用于业务 handler。`run_loop` 退出时统一取消在飞后台任务（`ee.close()`）。 - 业务 handler 的返回值不回流；回复消息请在 handler 里显式调 `BotApi`。 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 0d8c191..37699c0 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -110,12 +110,14 @@ sequenceDiagram 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 间隔反复探测。 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、后登记组件"的装配顺序可用。 -6. **OpenAPI 请求内核自治**：`api/core.py` 统一处理鉴权头、429 限流重试（优先 `Retry-After`，脏值容错，退避封顶）与 401 刷新 token 重试一次；v2/channel/guild 三个域方法集继承内核组合成 `BotApi`，新增域加模块即可，方法对用户始终平铺在 `BotApi` 上。 +5. **注入约定**：handler 参数按形参标注解析——`payloads.py` 里的 dataclass → 解析后的 payload 对象；`Event` → 事件封装；登记进 `emitter.services` 的组件类型（如 `BotApi`）→ 实例；未标注或未登记 → 回退传**原始 d**。解析是纯同步查表，无框架、无运行时魔法：每个 handler 的 `get_type_hints` + 形参名只解析一次并缓存，但 `services` 命中在每条事件时实时判定，保证"先注册 handler、后登记组件"的装配顺序可用。 +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）。 - **新增事件类型 dataclass**：在 `payloads.py` 定义 dataclass 并登记 `EVENT_TYPES`，handler 即可按类型标注拿解析结果；未收录事件回退传 dict。当前已覆盖 intents.toml 的全部事件组（消息、成员变动、审核、表态、互动、论坛、音频）。 @@ -123,7 +125,7 @@ sequenceDiagram `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`）： - WebSocket：aiohttp 假网关（token 接口 + gateway + ws 协议行为）端到端跑通 HELLO → IDENTIFY → READY 会话捕获 → 心跳/ACK（心跳任务随连接生灭）→ 业务事件 dataclass 分发，并断言协议帧（HELLO/HEARTBEAT_ACK）不会触达 handler； - Webhook：真实起 `WebhookConnecter` 的 http server（TestServer），覆盖 op=13 验证应答（签名内容正确性）、合法签名放行分发、坏签名 401、同 id 去重。 diff --git a/pyproject.toml b/pyproject.toml index 180c13a..816d9c4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,7 +13,6 @@ dependencies = [ "aiohttp>=3.14.3", "cryptography>=50.0.1", "loguru>=0.7.3", - "pyee>=14.0.0", "python-dotenv>=1.2.3", "ujson>=6.0.0", ] diff --git a/src/qqbotsdk/connecter.py b/src/qqbotsdk/connecter.py index f238575..1b71853 100644 --- a/src/qqbotsdk/connecter.py +++ b/src/qqbotsdk/connecter.py @@ -15,8 +15,9 @@ """ import asyncio +from abc import ABC, abstractmethod from time import monotonic -from typing import Protocol, cast +from typing import cast import aiohttp import ujson @@ -32,7 +33,10 @@ from .session import Session from .token import AccessToken -class Connecter(Protocol): +class Connecter(ABC): + """连接适配器接口：两种接入方式都实现 `run()` 常驻运行。""" + + @abstractmethod async def run(self) -> None: ... @@ -40,7 +44,7 @@ class RateLimitError(RuntimeError): """gateway 接口触发频率限制（code=100017）。""" -class WebsocketConnecter: +class WebsocketConnecter(Connecter): # 看门狗阈值（× 心跳间隔）：贴着 ACK 的自然节奏（1×）跑，误杀由 RESUME 兜底不丢消息； # 不能压到 1× 整——ACK 的自然间隔就是 1×，正常抖动会贴边误杀；TCP 下丢包表现为延迟而非静默 _SILENCE_FACTOR = 1.1 @@ -63,7 +67,7 @@ class WebsocketConnecter: self._session = session self._queue = queue self._protocol = protocol - # 出站帧缓冲（心跳 + 协议应答）：由 reply_helper 单任务发送， + # 出站帧缓冲（心跳 + 协议应答）：由 send_helper 单任务发送， # 维持 aiohttp ws 单写者不变量，生产者（心跳）也不会被发送阻塞 self._outbound: asyncio.Queue[Payload] = asyncio.Queue() @@ -107,16 +111,16 @@ class WebsocketConnecter: # HELLO 才给出心跳间隔；receive 解析到后经 Future 交给心跳任务 interval: asyncio.Future[float] = asyncio.get_running_loop().create_future() 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)) try: await receive finally: - reply.cancel() + send.cancel() heartbeat.cancel() # 取回被取消任务的结果，避免 "Task exception was never retrieved"， # 也确保它们在下一轮重连前真正停下 - await asyncio.gather(reply, heartbeat, return_exceptions=True) + await asyncio.gather(send, heartbeat, return_exceptions=True) logger.info("WebSocket 已断开") return monotonic() - started @@ -165,7 +169,8 @@ class WebsocketConnecter: # 服务端不会推送 op=1，心跳只由本任务投进出站缓冲 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: payload = await self._outbound.get() 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})") -class WebhookConnecter: +class WebhookConnecter(Connecter): _DEDUP_TTL = 60 + # 去重表长度达到该值才做一次过期清理（摊还 O(1)，避免每请求全表扫描） + _DEDUP_SWEEP_AT = 1024 def __init__( self, @@ -212,13 +219,21 @@ class WebhookConnecter: await runner.cleanup() def _duplicated(self, event_id: str | None) -> bool: + """TTL 内见过的同 id 视为重复推送。 + + 过期判断按当前 id 单独做（O(1)），保证 TTL 外重发的同一 id 会被当作 + 新事件处理；整表清过期项只在表长到阈值时做一次，避免每请求 O(n) 扫描。 + """ if not event_id: return False now = monotonic() - for key in [k for k, ts in self._seen.items() if now - ts > self._DEDUP_TTL]: - del self._seen[key] - if event_id in self._seen: + seen_at = self._seen.get(event_id) + if seen_at is not None and now - seen_at <= self._DEDUP_TTL: 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 return False diff --git a/src/qqbotsdk/crypto.py b/src/qqbotsdk/crypto.py index b1dc398..28c199d 100644 --- a/src/qqbotsdk/crypto.py +++ b/src/qqbotsdk/crypto.py @@ -3,6 +3,8 @@ seed 派生规则（官方约定）：AppSecret 的 utf8 字节反复倍增补足 32 字节。 """ +from functools import cache + from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric import ed25519 @@ -16,16 +18,24 @@ def build_seed(secret: str) -> bytes: 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: - private_key = ed25519.Ed25519PrivateKey.from_private_bytes(build_seed(secret)) - return private_key.sign(msg.encode("utf-8")).hex() + return _private_key(secret).sign(msg.encode("utf-8")).hex() def verify_sig(secret: str, timestamp: str, signature: str, body: bytes) -> bool: try: sig = bytes.fromhex(signature) - private_key = ed25519.Ed25519PrivateKey.from_private_bytes(build_seed(secret)) - private_key.public_key().verify(sig, timestamp.encode("utf-8") + body) + _private_key(secret).public_key().verify(sig, timestamp.encode("utf-8") + body) return True except (InvalidSignature, ValueError): return False diff --git a/src/qqbotsdk/emitter.py b/src/qqbotsdk/emitter.py index 795555b..a7936b2 100644 --- a/src/qqbotsdk/emitter.py +++ b/src/qqbotsdk/emitter.py @@ -18,7 +18,6 @@ from collections.abc import Awaitable, Callable from typing import Any, TypeVar, get_type_hints, overload from loguru import logger -from pyee.asyncio import AsyncIOEventEmitter from .events import Event from .model import Opcode, Payload @@ -30,8 +29,9 @@ Handler = TypeVar("Handler", bound=Callable[..., Awaitable[Any]]) class EventEmitter: def __init__(self, queue: EventQueue | None = None) -> None: - self._ee = AsyncIOEventEmitter() self._queue = queue if queue is not None else EventQueue() + # 事件名 → handler（按注册顺序）；同名事件重复注册同一函数只生效一次 + self._handlers: dict[str, list[Callable[..., Awaitable[Any]]]] = {} self.services: dict[type, Any] = {} # handler → (类型标注, 形参名)：签名解析较贵，缓存后每条事件只查表 self._signatures: dict[ @@ -65,7 +65,9 @@ class EventEmitter: name = event.name if isinstance(event, Opcode) else event 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: self._background.add(fn) return fn @@ -93,7 +95,7 @@ class EventEmitter: logger.exception(f"事件分发异常，已跳过: {data} ({e})") async def emit(self, data: Payload) -> None: - # 先取事件名，无 listener 就不必构造 Event（每条未订阅/协议帧都省一次拷贝） + # 先取事件名，无 handler 就不必构造 Event（每条未订阅事件都省一次拷贝） op = data.get("op") if op != Opcode.DISPATCH: logger.warning(f"非业务事件不应进分发队列，已跳过: {data}") @@ -103,12 +105,13 @@ class EventEmitter: logger.warning(f"业务事件缺少 t 字段，已跳过: {data}") return - listeners = self._ee.listeners(t) - if not listeners: + handlers = self._handlers.get(t) + if not handlers: return event = Event(data) - for fn in listeners: + # 遍历快照：handler 在分发过程中注册新 handler，不影响本次分发 + for fn in list(handlers): if fn in self._background: self._spawn(fn, event) else: diff --git a/src/qqbotsdk/model.py b/src/qqbotsdk/model.py index 1fb28de..de5d05a 100644 --- a/src/qqbotsdk/model.py +++ b/src/qqbotsdk/model.py @@ -1,7 +1,7 @@ """协议模型：Opcode、payload 及各事件 d 字段的 TypedDict、Intent 位掩码。 Payload 里 `NOT_SET` 哨兵标记"外发时应省略"的字段，由 -WebsocketConnecter.reply_helper 统一过滤。 +WebsocketConnecter.send_helper 统一过滤。 """ from enum import IntEnum, IntFlag diff --git a/src/qqbotsdk/payloads.py b/src/qqbotsdk/payloads.py index a69de80..a54ca4c 100644 --- a/src/qqbotsdk/payloads.py +++ b/src/qqbotsdk/payloads.py @@ -8,7 +8,7 @@ handler 直接标注具体类型即可拿到解析后的对象（如 from dataclasses import dataclass, fields 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) @@ -339,15 +339,34 @@ def _is_author_type(annotation: object) -> bool: 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: + """按解析计划填充 dataclass。 + + d 里未收录的字段一律忽略——线上 payload 会带新增或未声明的字段 + （如 author.union_openid），直接透传会让 dataclass 构造报错。 + """ kwargs = {} - for f in fields(cls): - if f.name not in d: + for name, nested_author in _parse_plan(cls): + if name not in d: continue - value = d[f.name] - if isinstance(value, dict) and _is_author_type(f.type): + value = d[name] + if nested_author and isinstance(value, dict): value = _build(Author, value) - kwargs[f.name] = value + kwargs[name] = value return cls(**kwargs) diff --git a/tests/test_dedup.py b/tests/test_dedup.py index 97ddeb7..c437480 100644 --- a/tests/test_dedup.py +++ b/tests/test_dedup.py @@ -1,3 +1,5 @@ +from time import monotonic + from qqbotsdk.config import Config from qqbotsdk.connecter import WebhookConnecter from qqbotsdk.queue import EventQueue @@ -48,3 +50,25 @@ def test_expired_entries_are_evicted(): c._seen[key] -= c._DEDUP_TTL + 1 assert not c._duplicated("evt-1") 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 diff --git a/tests/test_emitter.py b/tests/test_emitter.py index ca879c6..53ac6e6 100644 --- a/tests/test_emitter.py +++ b/tests/test_emitter.py @@ -20,6 +20,38 @@ def test_sequence_only_moves_forward(): 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 async def test_handler_exception_is_isolated(): queue = EventQueue() diff --git a/tests/test_payloads.py b/tests/test_payloads.py index 4afd215..38c291e 100644 --- a/tests/test_payloads.py +++ b/tests/test_payloads.py @@ -1,6 +1,7 @@ from qqbotsdk.emitter import EventEmitter from qqbotsdk.model import Opcode from qqbotsdk.payloads import ( + _PARSE_PLANS, C2CMessage, GroupAtMessage, Interaction, @@ -10,7 +11,33 @@ from qqbotsdk.payloads import ( 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( "GROUP_AT_MESSAGE_CREATE", { diff --git a/uv.lock b/uv.lock index e76a46a..d663072 100644 --- a/uv.lock +++ b/uv.lock @@ -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" }, ] -[[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]] name = "pygments" version = "2.21.0" @@ -622,7 +610,6 @@ dependencies = [ { name = "aiohttp" }, { name = "cryptography" }, { name = "loguru" }, - { name = "pyee" }, { name = "python-dotenv" }, { name = "ujson" }, ] @@ -640,7 +627,6 @@ requires-dist = [ { name = "aiohttp", specifier = ">=3.14.3" }, { name = "cryptography", specifier = ">=50.0.1" }, { name = "loguru", specifier = ">=0.7.3" }, - { name = "pyee", specifier = ">=14.0.0" }, { name = "python-dotenv", specifier = ">=1.2.3" }, { name = "ujson", specifier = ">=6.0.0" }, ]