diff --git a/CHANGELOG.md b/CHANGELOG.md index 4f17965..c528f8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,18 +10,28 @@ - Markdown 与内嵌按钮一等支持：`post_group_message`/`post_c2c_message` 新增 `markdown`/`keyboard` 参数（自动置 msg_type=2，与 content 互斥）；新增 `respond_interaction()` 回应按钮回调；`Interaction` payload 补齐 `chat_type`/`scene`/`group_openid` 等官方字段 - `Event` 新增 `event_id` 属性（推送信封顶层的事件 id，与 op/t/d 平级）：非消息事件（GROUP_ADD_ROBOT 等）的被动回复不再需要手动 `event.raw["id"]` - 按钮快捷构造与点击处理：`button()`/`keyboard()` 生成内嵌按钮（免手写嵌套 dict）；`Interaction` 新增 `button_data` 属性（被点按钮的 action.data）；examples/group_bot 补 `/btn` 卡片与 `INTERACTION_CREATE` 回调示例 +- handler 后台并发：`@ee.on(..., background=True)` 把 handler 放进后台任务，不阻塞事件流（同事件内失去先后保证，返回值不回流）；`run_loop` 退出经 `emitter.close()` 统一取消在飞任务 +- OpenAPI 限流与鉴权自愈：`BotApi.request` 对 429 自动重试（最多 3 次，优先 `Retry-After`，脏值容错解析，退避封顶 30s）；401 强制刷新 token 后重试一次，token 被服务端提前作废不再直接抛错 - 文档站点：基于 docsify（侧边栏、全文搜索、代码复制、mermaid 图渲染、分页），`npx docsify-cli serve` 本地预览；新增 `examples/README.md` 示例索引 ### 变更 - **移除 dishka 依赖，改为内置参数注入**：handler 形参直接标注组件类型（`api: BotApi`）即注入，不再需要 `Inject[T]` 包装；组件改在 `run_loop` 显式装配并按类型登记到 `emitter.services`（自定义组件同样登记即可注入）。破坏性变更：`Inject`、`make_container`、`use_container` 随之移除，原 `api: Inject[BotApi]` 写法改为 `api: BotApi` +- `EventEmitter` 的 `queue` 参数改为可选：不传自动创建队列，`EventEmitter()` 即可直接注册 handler；需多处共享队列时仍可显式传 `EventQueue`（`ee.queue` 取实例）。文档与示例统一改为无参构造 +- **`api.py` 拆分为 `api/` 包**：`core.py` 请求内核（鉴权/限流重试/错误归一）+ `v2.py`（群/单聊与按钮构造）、`channel.py`（频道消息）、`guild.py`（频道管理与查询）三个域方法集组合成 `BotApi`；方法名与导入路径不变（`from qqbotsdk.api import BotApi, button, keyboard` 照旧），新增域加模块即可 +- **心跳归传输层**：心跳从 `WebsocketProtocol`（定时调度任务）迁入 `WebsocketConnecter`（asyncio 任务随连接生灭），断线期间不再往 reply 队列堆积过期心跳，重连也不会先冲出陈旧心跳 +- 重连退避只在连接稳定存活（≥60s）后才重置归一，抖动的服务端不再被 1s 间隔反复探测 +- `AccessToken` 并发安全：换取过程持 `asyncio.Lock` 双重检查（并发首调只发一次请求）；到期判断改用 monotonic 时钟并预留 30s 提前刷新余量；新增 `invalidate()` 供 401 时强制作废 +- 组件构造签名瘦身：`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` ### 移除 +- 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)` 现在能拿到真实布尔 - 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 d4763bf..7f4f230 100644 --- a/README.md +++ b/README.md @@ -69,9 +69,9 @@ WEBHOOK_PATH=/qqbot/webhook 编写入口（如 `main.py`），把自己构造的 `EventEmitter` 传给 `main()`： ```python -from qqbotsdk import EventEmitter, EventQueue, main +from qqbotsdk import EventEmitter, main -ee = EventEmitter(EventQueue()) +ee = EventEmitter() # 方式一：裸参数（d 为原始事件数据） @ee.on("GROUP_AT_MESSAGE_CREATE") diff --git a/docs/API.md b/docs/API.md index 87334dc..2b4993e 100644 --- a/docs/API.md +++ b/docs/API.md @@ -16,9 +16,9 @@ from qqbotsdk import ( |---|---| | `main(emitter=None)` | 同步入口：装配组件并阻塞运行，Ctrl+C 优雅退出 | | `run_loop(emitter=None)` | 异步入口：`asyncio.run(run_loop(ee))` 的内部实现，可自行 await | -| `EventEmitter(queue)` | 分发主体，`@ee.on(...)` 注册 handler | +| `EventEmitter(queue=None)` | 分发主体，`@ee.on(..., background=True)` 注册 handler（可选后台并发）；`queue` 不传自动创建 | | `ee.services` | `dict[type, Any]`，按类型登记可注入组件；`run_loop` 装配 SDK 内置组件，也可登记自定义类型 | -| `EventQueue()` | 事件/应答队列；自建 emitter 时传入，`run_loop` 复用同一实例 | +| `EventQueue()` | 事件/应答队列；`EventEmitter` 默认自建，需多处共享时手动传入（`ee.queue` 取实例） | | `Config` | frozen dataclass，环境变量一次性读齐（见下文） | | `Connecter` | 连接适配器 Protocol（接口），按 `CONNECTER` 自动选择实现 | | `BaseProtocol` | 协议层抽象基类，自定义协议处理器时继承并实现 `register(emitter)` | @@ -34,15 +34,19 @@ from qqbotsdk.events import Event ## EventEmitter ```python -ee = EventEmitter(EventQueue()) +ee = EventEmitter() @ee.on("GROUP_AT_MESSAGE_CREATE") # DISPATCH 事件用 t 名 @ee.on("INVALID_SESSION") # 协议事件用 Opcode 名（IDENTIFY/RESUME/HEARTBEAT/...） 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 发送（业务事件不会）。 - **异常隔离**：handler 异常就地记录（loguru），不影响同事件其他 handler 与主循环。 +- **后台并发**：`background=True` 的 handler 进后台任务，emit 不等它；同一事件内失去先后保证，返回值不回流，仅用于业务 handler。`run_loop` 退出时统一取消在飞后台任务（`ee.close()`）。 - **`ee.handle(payload)`**：同步分发并返回首个 handler 的非空返回值，供协议层应答 op=13 等场景，业务代码不用。 ### handler 参数注入约定 @@ -51,10 +55,10 @@ async def handler(...): ... | 标注 | 得到 | 未命中时 | |---|---|---| -| payload dataclass（`payloads` 中的类型） | `Event.typed` 解析对象 | 回退传原始 dict | +| payload dataclass（`payloads` 中的类型） | `Event.typed` 解析对象 | 回退传原始 d | | `Event` | 事件封装对象 | — | -| 组件类型（`BotApi`/`Session`/`Config` 等已登记进 `ee.services` 的类型） | 登记的实例 | 回退传原始 dict | -| 无标注 | 原始事件 dict | — | +| 组件类型（`BotApi`/`Session`/`Config` 等已登记进 `ee.services` 的类型） | 登记的实例 | 回退传原始 d | +| 无标注 | 原始 d（业务事件即事件 dict；协议事件可能是标量，如 `INVALID_SESSION` 的 `resumable: bool`） | — | ## Event @@ -92,6 +96,8 @@ async def handler(...): ... `from qqbotsdk.api import BotApi`。经参数注入（`api: BotApi`）。自动附带 `Authorization: QQBot ` 鉴权头并刷新 token；HTTP ≥400 抛 `ApiError`。 +实现按域拆在 `qqbotsdk/api/` 包内（`core` 请求内核 + `v2`/`channel`/`guild` 方法集组合成 `BotApi`），方法名与下表一致；429 限流自动重试（最多 3 次，优先 `Retry-After`，退避封顶 30s），401 强制刷新 token 后重试一次。 + ### 通用 | 方法 | 说明 | @@ -159,7 +165,7 @@ async def handler(...): ... ### ApiError -`ApiError(RuntimeError)`，HTTP ≥400 时抛出： +`ApiError(RuntimeError)`，重试后仍失败（HTTP ≥400）时抛出： | 属性 | 说明 | |---|---| diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 76d229f..636981f 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -7,29 +7,29 @@ ```mermaid flowchart TB subgraph 接入层["接入层 connecter.py"] - WS["WebsocketConnecter\ngateway握手 + ws长连接\n指数退避重连 + 心跳看门狗"] + WS["WebsocketConnecter\ngateway握手 + ws长连接\n心跳随连接生灭 + 心跳看门狗\n指数退避重连（稳定连接才归一）"] WH["WebhookConnecter\naiohttp server\n验签 + TTL去重"] end subgraph 分发层["分发层 emitter.py"] Q[("EventQueue\nmsg_queue / reply_queue")] - EE["EventEmitter\nop/t 路由 + 参数注入\nhandler 异常隔离"] + EE["EventEmitter\nop/t 路由 + 参数注入\nhandler 异常隔离\nbackground 后台并发"] end subgraph 协议层["协议层 BaseProtocol"] - WSP["WebsocketProtocol\nIDENTIFY/RESUME\napscheduler 心跳\ns 序列号跟踪"] + WSP["WebsocketProtocol\nIDENTIFY/RESUME\ns 序列号跟踪"] WHP["WebhookProtocol\nop=13 验证应答"] end subgraph 业务层["业务层"] - API["BotApi\nREST 封装 + ApiError"] + API["BotApi\nREST 封装（core + v2/channel/guild）\n429/401 自动重试"] USER["用户 handler\n@ee.on(...)"] end subgraph 基础设施["基础设施"] CFG["Config\n.env 一次性读齐"] HTTP[("ClientSession\n共享连接池")] - TK["AccessToken\n缓存 + 自动续期"] + TK["AccessToken\n缓存 + 自动续期\n并发锁 + 401 作废"] CRY["crypto\nEd25519 验签/签名"] end @@ -39,7 +39,8 @@ flowchart TB WH -- "事件 payload" --> Q Q -- "dispatch 泵" --> EE EE -- "on_payload" --> WSP - WSP -- "IDENTIFY/HEARTBEAT 等\npayload 返回值" --> Q + WSP -- "IDENTIFY/RESUME 等\npayload 返回值" --> Q + WS -- "HEARTBEAT（心跳任务）" --> Q EE -- "Event 注入" --> USER USER -- "api: BotApi" --> API API --> QQ @@ -71,8 +72,8 @@ sequenceDiagram P-->>Q: IDENTIFY/RESUME payload → reply_queue WS->>QQ: send（过滤 NOT_SET 字段） - loop 每次心跳间隔 - P->>Q: HEARTBEAT payload（apscheduler 定时） + loop 每次心跳间隔（心跳任务随连接生灭） + WS->>Q: HEARTBEAT payload WS->>QQ: send QQ->>WS: HEARTBEAT_ACK（刷新看门狗时钟） end @@ -90,10 +91,10 @@ sequenceDiagram | 层 | 模块 | 职责 | 不允许出现 | |---|---|---|---| -| 接入层 | `connecter.py` | 传输细节：握手、重连、验签、去重 | 事件业务语义、handler 调用 | -| 分发层 | `emitter.py` | 路由、参数注入、异常隔离、应答回流 | 传输细节、具体协议状态机 | -| 协议层 | `ws_protocol.py` `webhook_protocol.py` | 鉴权/心跳/验证等协议状态机 | 传输细节（只经队列收发） | -| 业务层 | `api.py` `payloads.py` + 用户代码 | OpenAPI 调用、payload 解析与机器人逻辑 | 协议细节 | +| 接入层 | `connecter.py` | 传输细节：握手、心跳、重连、验签、去重 | 事件业务语义、协议状态机 | +| 分发层 | `emitter.py` | 路由、参数注入、异常隔离、后台并发、应答回流 | 传输细节、具体协议状态机 | +| 协议层 | `ws_protocol.py` `webhook_protocol.py` | 鉴权/验证等协议状态机（会话语义） | 传输细节（只经队列收发） | +| 业务层 | `api/` `payloads.py` + 用户代码 | OpenAPI 调用、payload 解析与机器人逻辑 | 协议细节 | 跨层共享的最小公共物：`model.py`（Opcode/Payload/Intent）、`queue.py`（事件通道）、`config.py`、`session.py`（ws 会话状态）、`payloads.py`（事件 dataclass 与解析，被 `events.py`/`emitter.py` 使用）。 @@ -101,22 +102,23 @@ sequenceDiagram - 全部组件在 `run_loop`（`__init__.py`）显式构造装配，无 DI 框架，也无任何全局单例。 - 组件按类型登记进 `emitter.services`，handler 形参标注类型即得实例；用户自定义组件同样登记即可注入。 -- 唯一的 `ClientSession` 在 `run_loop` 构造，`finally` 中统一释放（全 SDK 共享一个连接池）。 -- 接入方式的选择收敛在 `run_loop` 的 `match config.connecter` 装配分支，主体代码只依赖 `BaseProtocol`/`Connecter` 抽象。 +- 唯一的 `ClientSession` 在 `run_loop` 构造，`finally` 中先 `emitter.close()` 收尾后台 handler、再统一释放连接池（全 SDK 共享一个连接池）。 +- 接入方式的选择收敛在 `run_loop` 的 `match config.connecter` 装配分支，主体代码只依赖 `BaseProtocol`/`Connecter` 抽象；两个 connecter 的依赖面一致（config + queue + 各自所需的窄依赖），`WebhookConnecter` 经 `handle_event` 回调取得 op=13 应答，不感知 `EventEmitter`。 - 用户在 `main(ee)` 之前构造的 `EventEmitter` 会被复用（连同它的队列），保证 handler 注册在主体实际使用的实例上。 ## 关键设计决策 1. **管线而非回调直调**：事件一律经 `EventQueue` 异步流转，接入与分发两侧解耦——Webhook handler 里不需要知道 ws 的存在，反之亦然。代价是多一跳队列延迟（可忽略）。 -2. **协议层状态机独立**（`BaseProtocol.register/on_payload`）：emitter 不认识"心跳""序列号"这些 ws 概念；协议类通过返回 payload dict 经 reply 队列外发。 -3. **看门狗两层互补**（见 `connecter.py` 模块注释）：静默超时抓连接彻底失联，ACK 期限检查抓"有事件流但心跳已死"。阈值 `_SILENCE_FACTOR=1.1`、`_ACK_FACTOR=1.2` 是按 ACK 自然节奏校准的，**不能压到 1× 整**，否则正常抖动会周期性误杀（误杀由 RESUME 兜底，不丢消息）。 -4. **handler 异常就地隔离**：`_invoke` 捕获记日志，`dispatch` 主循环另有兜底；单条事件的单个 handler 失败不影响进程。 -5. **注入约定**：handler 参数按形参标注解析——`payloads.py` 里的 dataclass → 解析后的 payload 对象；`Event` → 事件封装；登记进 `emitter.services` 的组件类型（如 `BotApi`）→ 实例；未标注或未登记 → 回退传原始 `d`，保持旧处理器兼容。解析是纯同步查表（`get_type_hints` + dict），无框架、无运行时魔法。 +2. **协议层只管会话语义，心跳归传输层**：`WebsocketProtocol` 负责 IDENTIFY/RESUME/序列号（说什么），`WebsocketConnecter` 负责连接与心跳（怎么保活）——心跳任务随连接生灭（HELLO 给出间隔后启动，断线即取消），断线期间不在 reply 队列堆积过期心跳，也免去了调度器的生命周期管理。协议类通过返回 payload dict 经 reply 队列外发。 +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），无框架、无运行时魔法。 +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` 的白名单加一个名字。 -- **新增业务 API**：在 `api.py` 的 `BotApi` 上加方法（或独立模块组合 `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 的全部事件组（消息、成员变动、审核、表态、互动、论坛、音频）。 ## 测试 @@ -125,7 +127,7 @@ sequenceDiagram - 纯逻辑：crypto 往返、intents 解析、Config 校验、`_route` 路由边界、webhook 去重、序列号单调性、handler 异常隔离与 reply 回流、payload dataclass 解析、BotApi 各业务方法的拼参（Recording 子类截获 request）。 - **网络集成**（`test_ws_integration.py` / `test_webhook_integration.py`）： - - WebSocket：aiohttp 假网关（token 接口 + gateway + ws 协议行为）端到端跑通 HELLO → IDENTIFY → READY 会话捕获 → apscheduler 心跳/ACK → 业务事件 dataclass 分发； + - WebSocket：aiohttp 假网关（token 接口 + gateway + ws 协议行为）端到端跑通 HELLO → IDENTIFY → READY 会话捕获 → 心跳/ACK（心跳任务随连接生灭）→ 业务事件 dataclass 分发； - Webhook：真实起 `WebhookConnecter` 的 http server（TestServer），覆盖 op=13 验证应答（签名内容正确性）、合法签名放行分发、坏签名 401、同 id 去重。 -两套集成测试验证的都是真实连接路径，改接入层/协议层后跑 `uv run pytest` 即可回归。 +另有 429/401 重试、token 并发锁、op=9 resumable 布尔注入、background 后台并发与 `close()` 收尾的专项单测。两套集成测试验证的都是真实连接路径，改接入层/协议层后跑 `uv run pytest` 即可回归。 diff --git a/docs/GUIDE.md b/docs/GUIDE.md index 572f08b..268e40a 100644 --- a/docs/GUIDE.md +++ b/docs/GUIDE.md @@ -76,9 +76,9 @@ SDK 启动时读取 `intents.toml` 计算订阅掩码，把想收的事件改成 用装饰器把 handler 注册到 `EventEmitter`，事件名即线上 payload 的 `t` 名（如 `GROUP_AT_MESSAGE_CREATE`）。三种写法可混用： ```python -from qqbotsdk import EventEmitter, EventQueue +from qqbotsdk import EventEmitter -ee = EventEmitter(EventQueue()) +ee = EventEmitter() # 方式一：裸参数，d 为原始事件 dict @ee.on("GROUP_AT_MESSAGE_CREATE") @@ -107,8 +107,8 @@ handler 参数按标注解析的完整规则： |---|---| | payload dataclass（如 `GroupAtMessage`） | 解析后的 dataclass 对象；未收录事件回退原始 dict | | `Event` | 事件封装对象（`.raw`/`.data`/`.typed`/`.user_id`/`.group_id`/`.content`/`.message_id`/`.event_id`） | -| 组件类型（`BotApi`/`Session`/`Config` 等） | `emitter.services` 中按类型登记的实例；未登记回退传原始 dict | -| 无标注 | 原始事件 dict | +| 组件类型（`BotApi`/`Session`/`Config` 等） | `emitter.services` 中按类型登记的实例；未登记回退传原始 d | +| 无标注 | 原始 d（业务事件即事件 dict；协议事件可能是标量，如 `INVALID_SESSION` 的 `resumable: bool`） | 注意： @@ -117,6 +117,18 @@ handler 参数按标注解析的完整规则： - 未订阅（intents.toml 为 `false`）的事件不会到达 handler。 - 自定义组件也能注入：`ee.services[MyService] = MyService(...)` 登记后，handler 形参标注 `MyService` 即可拿到实例。 +### 后台并发 handler + +默认同一事件的所有 handler 串行执行，某个慢 handler 会延迟后续事件。调用密集、不关心先后顺序的 handler 可加 `background=True` 丢进后台任务，事件流不再等它： + +```python +@ee.on("GROUP_AT_MESSAGE_CREATE", background=True) +async def slow_work(msg: GroupAtMessage, api: BotApi): + ... # 耗时处理，不阻塞其他事件的分发 +``` + +代价是同一事件内 handler 间失去先后保证，且返回值不回流（本就只对协议事件回流），因此仅用于业务 handler。进程退出时 `run_loop` 会统一取消在飞的后台任务。 + ## 发消息 ### 被动回复（最常用） diff --git a/examples/group_bot/main.py b/examples/group_bot/main.py index b966f91..518312e 100644 --- a/examples/group_bot/main.py +++ b/examples/group_bot/main.py @@ -10,12 +10,12 @@ from loguru import logger -from qqbotsdk import EventEmitter, EventQueue, main +from qqbotsdk import EventEmitter, main from qqbotsdk.api import BotApi, button, keyboard from qqbotsdk.events import Event from qqbotsdk.payloads import C2CMessage, GroupAtMessage, Interaction -ee = EventEmitter(EventQueue()) +ee = EventEmitter() @ee.on("GROUP_AT_MESSAGE_CREATE") diff --git a/examples/webhook_bot/main.py b/examples/webhook_bot/main.py index ef0cfe8..305784d 100644 --- a/examples/webhook_bot/main.py +++ b/examples/webhook_bot/main.py @@ -11,11 +11,11 @@ 运行：uv run python examples/webhook_bot/main.py """ -from qqbotsdk import EventEmitter, EventQueue, main +from qqbotsdk import EventEmitter, main from qqbotsdk.api import BotApi from qqbotsdk.payloads import GroupAtMessage -ee = EventEmitter(EventQueue()) +ee = EventEmitter() @ee.on("GROUP_AT_MESSAGE_CREATE") diff --git a/pyproject.toml b/pyproject.toml index 7f7ce72..180c13a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,6 @@ authors = [ requires-python = ">=3.13" dependencies = [ "aiohttp>=3.14.3", - "apscheduler>=3.11.3", "cryptography>=50.0.1", "loguru>=0.7.3", "pyee>=14.0.0", diff --git a/src/qqbotsdk/__init__.py b/src/qqbotsdk/__init__.py index c8aa652..12aa69b 100644 --- a/src/qqbotsdk/__init__.py +++ b/src/qqbotsdk/__init__.py @@ -1,6 +1,6 @@ """qqbotsdk：QQ 官方机器人 SDK 入口。 -用法：构造 `EventEmitter(EventQueue())` 注册 handler，交给 `main(ee)` +用法：构造 `EventEmitter()` 注册 handler，交给 `main(ee)` 启动；连接方式由 .env 的 CONNECTER 决定（websocket / webhook）。 架构详情见 docs/ARCHITECTURE.md。 """ @@ -33,30 +33,32 @@ async def run_loop(emitter: EventEmitter | None = None) -> None: 参数注入；共享的 HTTP 连接池随 run_loop 结束释放。 """ if emitter is None: - emitter = EventEmitter(EventQueue()) + emitter = EventEmitter() config = Config.load() http = ClientSession(timeout=config.timeout) session = Session() token = AccessToken(config, http) api = BotApi(config, http, token) - emitter.services.update({ - EventEmitter: emitter, - EventQueue: emitter.queue, - Config: config, - Session: session, - ClientSession: http, - AccessToken: token, - BotApi: api, - }) + emitter.services.update( + { + EventEmitter: emitter, + EventQueue: emitter.queue, + Config: config, + Session: session, + ClientSession: http, + AccessToken: token, + BotApi: api, + } + ) match config.connecter: case "websocket": - protocol = WebsocketProtocol(config, emitter.queue, session, token) - connecter = WebsocketConnecter(config, http, token, emitter.queue) + protocol = WebsocketProtocol(config, session, token) + connecter = WebsocketConnecter(config, http, token, session, emitter.queue) case "webhook": - protocol = WebhookProtocol(config, emitter.queue, session) - connecter = WebhookConnecter(config, emitter, emitter.queue) + protocol = WebhookProtocol(config) + connecter = WebhookConnecter(config, emitter.queue, emitter.handle) case unknown: raise ValueError(f"未知的 CONNECTER: {unknown}") @@ -65,6 +67,8 @@ async def run_loop(emitter: EventEmitter | None = None) -> None: try: await asyncio.gather(emitter.dispatch(), connecter.run()) finally: + # 在飞的后台 handler 先于连接池收尾，避免用到已关闭的 ClientSession + await emitter.close() # 全 SDK 共享一个连接池，统一在此释放 await http.close() diff --git a/src/qqbotsdk/api.py b/src/qqbotsdk/api.py deleted file mode 100644 index 5401fd2..0000000 --- a/src/qqbotsdk/api.py +++ /dev/null @@ -1,477 +0,0 @@ -"""OpenAPI 客户端：自动附带鉴权头，HTTP ≥400 抛 ApiError。 - -消息接口按接入面分组：v2 群/单聊（openid）、频道（channel_id）。 -被动回复规则（官方文档 2026-09 核对）：携带 msg_id（或 event_id， -二选一）为被动回复——单聊 60 分钟内最多 4 次，群聊 5 分钟内最多 5 次； -同一条消息多次回复需递增 msg_seq（相同 msg_id+msg_seq 会发送失败）； -不携带则为主动消息（另有频控，群/单聊每日每对象 1000 条）。 -""" - -from typing import Any -from uuid import uuid4 - -import ujson -from aiohttp import ClientSession -from loguru import logger - -from .config import Config -from .token import AccessToken - -# msg_type 取值 -MSG_TYPE_TEXT = 0 -MSG_TYPE_MARKDOWN = 2 -MSG_TYPE_MEDIA = 7 -# file_type 取值（富媒体上传） -FILE_TYPE_IMAGE = 1 -FILE_TYPE_VIDEO = 2 -FILE_TYPE_VOICE = 3 -FILE_TYPE_FILE = 4 - - -def button( - label: str, - data: str = "", - *, - type: int = 1, - permission: int = 2, - style: int = 1, - id: str | None = None, -) -> dict: - """构造单个内嵌按钮 dict，供 keyboard() 组装后传入 keyboard= 参数。 - - type: 1 回调（点击推 INTERACTION_CREATE）/ 0 跳转链接（data 为 url）/ - 2 指令（data 插入输入框）。permission: 2 所有人 / 1 管理员 / 0 指定 - 用户。label 官方限 10 字以内；要补嵌套字段（如 visited_label、 - prompt）直接修改返回的 dict。 - """ - return { - "id": id or uuid4().hex[:8], - "render_data": {"label": label, "style": style}, - "action": {"type": type, "permission": {"type": permission}, "data": data}, - } - - -def keyboard(*rows: dict | list[dict]) -> dict: - """组装内嵌键盘（keyboard= 参数）：每个位置参数为一行， - 单个按钮独占一行，按钮列表同行并排。""" - return { - "content": { - "rows": [ - {"buttons": row if isinstance(row, list) else [row]} for row in rows - ] - } - } - - -class ApiError(RuntimeError): - """OpenAPI 调用失败：HTTP 非 2xx 或业务 code 非 0。""" - - def __init__(self, status: int, code: Any, message: Any) -> None: - self.status = status - self.code = code - self.message = message - super().__init__(f"API 调用失败: HTTP {status}, code={code}, message={message}") - - -class BotApi: - """通用 REST 封装；具体的业务接口在此之上做薄封装。""" - - def __init__(self, config: Config, http: ClientSession, token: AccessToken) -> None: - self._config = config - self._http = http - self._token = token - - async def request(self, method: str, path: str, **kwargs) -> Any: - access_token = await self._token.get_access_token() - headers = {"Authorization": f"QQBot {access_token}"} - async with self._http.request( - method, - f"{self._config.base_url}{path}", - headers=headers, - **kwargs, - ) as resp: - # 个别接口 200 响应的 Content-Type 不是 JSON（body 仍是 JSON， - # 如回应互动返回 200 text/plain + {}），跳过标头校验 - data = await resp.json(loads=ujson.loads, content_type=None) - if resp.status >= 400: - logger.warning(f"{method} {path} 失败: {data}") - code = data.get("code") if isinstance(data, dict) else None - message = data.get("message") if isinstance(data, dict) else data - raise ApiError(resp.status, code, message) - return data - - async def get(self, path: str, **kwargs) -> Any: - return await self.request("GET", path, **kwargs) - - async def post(self, path: str, **kwargs) -> Any: - return await self.request("POST", path, **kwargs) - - async def delete(self, path: str, **kwargs) -> Any: - return await self.request("DELETE", path, **kwargs) - - async def put(self, path: str, **kwargs) -> Any: - return await self.request("PUT", path, **kwargs) - - async def patch(self, path: str, **kwargs) -> Any: - return await self.request("PATCH", path, **kwargs) - - # ---------- 通用 ---------- - - async def me(self) -> dict: - """机器人自身信息（GET /users/@me）。""" - return await self.get("/users/@me") - - async def respond_interaction(self, interaction_id: str, code: int = 0) -> dict: - """回应互动事件（PUT /interactions/{interaction_id}）。 - - 按钮点击（INTERACTION_CREATE type=11）等互动收到后必须回应， - 否则客户端一直 loading 直到超时；同一 id 只能回应一次。 - """ - return await self.put(f"/interactions/{interaction_id}", json={"code": code}) - - # ---------- v2 单聊（C2C） ---------- - - async def post_c2c_message( - self, - openid: str, - content: str = "", - msg_id: str | None = None, - msg_seq: int = 1, - media: dict | None = None, - event_id: str | None = None, - markdown: str | dict | None = None, - keyboard: dict | None = None, - **extra: Any, - ) -> dict: - """发单聊消息（POST /v2/users/{openid}/messages）。 - - media 为富媒体上传接口返回的 {"file_info": ...} 结构， - 此时 content 作媒体下方文本。msg_id（回复消息）与 event_id - （响应事件）二选一，均为被动回复。markdown 为字符串或完整 - dict（msg_type=2，与 content 互斥）；keyboard 为内嵌键盘 - 结构（{"content": {"rows": [...]}} 或 {"id": ...}）。 - """ - return await self._post_message( - f"/v2/users/{openid}/messages", - content=content, - msg_id=msg_id, - msg_seq=msg_seq, - media=media, - event_id=event_id, - markdown=markdown, - keyboard=keyboard, - **extra, - ) - - async def upload_c2c_file( - self, openid: str, file_type: int, url: str, srv_send_msg: bool = False - ) -> dict: - """上传单聊富媒体（POST /v2/users/{openid}/files）。 - - 返回含 file_info，可传入 post_c2c_message 的 media 直接发送， - 或 srv_send_msg=True 直接作为消息发出。 - """ - return await self.post( - f"/v2/users/{openid}/files", - json={"file_type": file_type, "url": url, "srv_send_msg": srv_send_msg}, - ) - - # ---------- v2 群聊 ---------- - - async def post_group_message( - self, - group_openid: str, - content: str = "", - msg_id: str | None = None, - msg_seq: int = 1, - media: dict | None = None, - event_id: str | None = None, - markdown: str | dict | None = None, - keyboard: dict | None = None, - **extra: Any, - ) -> dict: - """发群消息（POST /v2/groups/{group_openid}/messages）。 - - msg_id（回复消息）与 event_id（响应 GROUP_ADD_ROBOT / - INTERACTION_CREATE 等事件）二选一。markdown 为字符串或完整 - dict（msg_type=2，与 content 互斥）；keyboard 为内嵌键盘 - 结构（{"content": {"rows": [...]}} 或 {"id": ...}）。 - """ - return await self._post_message( - f"/v2/groups/{group_openid}/messages", - content=content, - msg_id=msg_id, - msg_seq=msg_seq, - media=media, - event_id=event_id, - markdown=markdown, - keyboard=keyboard, - **extra, - ) - - async def upload_group_file( - self, group_openid: str, file_type: int, url: str, srv_send_msg: bool = False - ) -> dict: - """上传群富媒体（POST /v2/groups/{group_openid}/files）。""" - return await self.post( - f"/v2/groups/{group_openid}/files", - json={"file_type": file_type, "url": url, "srv_send_msg": srv_send_msg}, - ) - - async def _post_message( - self, - path: str, - content: str, - msg_id: str | None, - msg_seq: int, - media: dict | None, - event_id: str | None, - markdown: str | dict | None = None, - keyboard: dict | None = None, - **extra: Any, - ) -> dict: - # media > markdown > 纯文本；markdown 与 content 互斥（官方约定） - if media is not None: - body: dict[str, Any] = {"msg_type": MSG_TYPE_MEDIA, "media": media} - if content: - body["content"] = content - elif markdown is not None: - body = { - "msg_type": MSG_TYPE_MARKDOWN, - "markdown": markdown if isinstance(markdown, dict) else {"content": markdown}, - } - else: - body = {"msg_type": MSG_TYPE_TEXT} - if content: - body["content"] = content - if msg_id: - body["msg_id"] = msg_id - body["msg_seq"] = msg_seq - elif event_id: - body["event_id"] = event_id - body["msg_seq"] = msg_seq - if keyboard is not None: - body["keyboard"] = keyboard - body.update(extra) - return await self.post(path, json=body) - - async def _withdraw_v2_message(self, path: str, message_id: str) -> dict: - # 发送超过 2 分钟的消息不可撤回 - return await self.delete(f"{path}/messages/{message_id}") - - async def withdraw_group_message(self, group_openid: str, message_id: str) -> dict: - """撤回群消息（DELETE /v2/groups/{group_openid}/messages/{message_id}）。 - - 群管理员可撤回自己与普通成员的消息；普通成员仅能撤回自己的。 - """ - return await self._withdraw_v2_message(f"/v2/groups/{group_openid}", message_id) - - async def withdraw_c2c_message(self, openid: str, message_id: str) -> dict: - """撤回单聊消息（DELETE /v2/users/{openid}/messages/{message_id}）。""" - return await self._withdraw_v2_message(f"/v2/users/{openid}", message_id) - - async def mute_group_member(self, group_openid: str, mute_expire_at: str) -> dict: - """群成员禁言（POST /v2/groups/{group_openid}/restrict_chat_setting）。 - - mute_expire_at 为禁言到期时间（RFC3339 格式，如 - "2026-01-01T12:00:00+08:00"）；传当前时间之前即解除禁言。 - """ - return await self.post( - f"/v2/groups/{group_openid}/restrict_chat_setting", - json={"mute_expire_at": mute_expire_at}, - ) - - # ---------- 频道 ---------- - - async def post_channel_message( - self, - channel_id: str, - content: str = "", - msg_id: str | None = None, - msg_seq: int = 1, - image: str | None = None, - **extra: Any, - ) -> dict: - """发频道消息（POST /channels/{channel_id}/messages）。 - - image 为图片 URL；其余字段（embed/ark/markdown 等）经 extra 透传。 - """ - body: dict[str, Any] = {"msg_seq": msg_seq} - if content: - body["content"] = content - if msg_id: - body["msg_id"] = msg_id - if image: - body["image"] = image - body.update(extra) - return await self.post(f"/channels/{channel_id}/messages", json=body) - - async def withdraw_channel_message( - self, channel_id: str, message_id: str, hide_tip: bool = False - ) -> dict: - """撤回频道消息（DELETE /channels/{channel_id}/messages/{message_id}）。 - - hide_tip 控制是否隐藏撤回提示小灰条（官方参数名为 hidetip）； - 仅私域机器人可用。 - """ - return await self.delete( - f"/channels/{channel_id}/messages/{message_id}", - params={"hidetip": str(hide_tip).lower()}, - ) - - async def put_channel_reaction( - self, channel_id: str, message_id: str, emoji_type: int, emoji_id: str - ) -> dict: - """对频道消息表态（PUT .../reactions/{type}/{emoji_id}）。 - - emoji_type=1 为系统表情，emoji_id 为系统表情序号（字符串）。 - """ - return await self.put( - f"/channels/{channel_id}/messages/{message_id}" - f"/reactions/{emoji_type}/{emoji_id}" - ) - - async def delete_channel_reaction( - self, channel_id: str, message_id: str, emoji_type: int, emoji_id: str - ) -> dict: - """删除自己对该消息的表态（DELETE .../reactions/{type}/{emoji_id}）。""" - return await self.delete( - f"/channels/{channel_id}/messages/{message_id}" - f"/reactions/{emoji_type}/{emoji_id}" - ) - - # ---------- 频道管理 ---------- - - async def create_guild_announce( - self, - guild_id: str, - channel_id: str, - message_id: str, - announces_type: int = 0, - **extra: Any, - ) -> dict: - """创建频道公告（POST /guilds/{guild_id}/announces）。 - - announces_type: 0 成员公告（默认）、1 欢迎公告； - recommend_channels 等经 extra 透传（与消息类公告互斥）。 - """ - body = { - "channel_id": channel_id, - "message_id": message_id, - "announces_type": announces_type, - } - body.update(extra) - return await self.post(f"/guilds/{guild_id}/announces", json=body) - - async def delete_guild_announce(self, guild_id: str, message_id: str) -> dict: - """删除频道公告（DELETE /guilds/{guild_id}/announces/{message_id}）。 - - 删除推荐子频道类型的公告需传 message_id="all"。 - """ - return await self.delete(f"/guilds/{guild_id}/announces/{message_id}") - - async def pin_channel_message(self, channel_id: str, message_id: str) -> dict: - """添加精华消息（PUT /channels/{channel_id}/pins/{message_id}）。""" - return await self.put(f"/channels/{channel_id}/pins/{message_id}") - - async def unpin_channel_message(self, channel_id: str, message_id: str) -> dict: - """移出精华消息（DELETE /channels/{channel_id}/pins/{message_id}）。""" - return await self.delete(f"/channels/{channel_id}/pins/{message_id}") - - async def get_channel_pins(self, channel_id: str) -> dict: - """获取精华消息列表（GET /channels/{channel_id}/pins）。""" - return await self.get(f"/channels/{channel_id}/pins") - - async def get_channel_schedules( - self, channel_id: str, since: int | None = None - ) -> list | dict: - """获取子频道日程列表（GET /channels/{channel_id}/schedules）。 - - 默认返回当天日程；传 since（毫秒级时间戳）返回结束时间在其后的日程。 - """ - params = {"since": str(since)} if since is not None else None - return await self.get(f"/channels/{channel_id}/schedules", params=params) - - async def get_channel_schedule(self, channel_id: str, schedule_id: str) -> dict: - """获取单个日程（GET /channels/{channel_id}/schedules/{schedule_id}）。""" - return await self.get(f"/channels/{channel_id}/schedules/{schedule_id}") - - async def create_channel_schedule( - self, - channel_id: str, - name: str, - start_timestamp: str, - end_timestamp: str, - jump_channel_id: str, - remind_type: str = "0", - description: str = "", - ) -> dict: - """创建日程（POST /channels/{channel_id}/schedules）。 - - 时间为毫秒级时间戳字符串；remind_type: "0" 不提醒 / "1" 开始时提醒 - （官方字段为字符串）。请求体按官方约定包在 "schedule" 字段里。 - 需管理频道权限；单管理员每天限 10 次。 - """ - return await self.post( - f"/channels/{channel_id}/schedules", - json={ - "schedule": { - "name": name, - "description": description, - "start_timestamp": start_timestamp, - "end_timestamp": end_timestamp, - "jump_channel_id": jump_channel_id, - "remind_type": remind_type, - } - }, - ) - - async def update_channel_schedule( - self, channel_id: str, schedule_id: str, **fields: Any - ) -> dict: - """修改日程（PATCH /channels/{channel_id}/schedules/{schedule_id}）。 - - schedule 内的字段（name/start_timestamp/...）经 fields 透传。 - """ - return await self.patch( - f"/channels/{channel_id}/schedules/{schedule_id}", - json={"schedule": {"id": schedule_id, **fields}}, - ) - - async def delete_channel_schedule(self, channel_id: str, schedule_id: str) -> dict: - """删除日程（DELETE /channels/{channel_id}/schedules/{schedule_id}）。""" - return await self.delete(f"/channels/{channel_id}/schedules/{schedule_id}") - - async def mute_guild_member( - self, guild_id: str, user_id: str, mute_seconds: int = 0 - ) -> dict: - """禁言/解禁频道成员（PATCH /guilds/{guild_id}/members/{user_id}/mute）。 - - mute_seconds 为禁言时长（秒），传 0 解除禁言；上限 28 天。 - """ - return await self.patch( - f"/guilds/{guild_id}/members/{user_id}/mute", - json={"mute_seconds": str(mute_seconds)}, - ) - - # ---------- 频道信息查询 ---------- - - async def get_guild(self, guild_id: str) -> dict: - """获取频道信息（GET /guilds/{guild_id}）。""" - return await self.get(f"/guilds/{guild_id}") - - async def get_guild_channels(self, guild_id: str) -> list | dict: - """获取频道下的子频道列表（GET /channels/{guild_id}/channels）。""" - return await self.get(f"/channels/{guild_id}/channels") - - async def get_channel(self, channel_id: str) -> dict: - """获取子频道信息（GET /channels/{channel_id}）。""" - return await self.get(f"/channels/{channel_id}") - - async def get_guild_roles(self, guild_id: str) -> dict: - """获取频道身份组列表（GET /guilds/{guild_id}/roles）。""" - return await self.get(f"/guilds/{guild_id}/roles") - - async def get_guild_member(self, guild_id: str, user_id: str) -> dict: - """获取成员详情（GET /guilds/{guild_id}/members/{user_id}）。""" - return await self.get(f"/guilds/{guild_id}/members/{user_id}") diff --git a/src/qqbotsdk/api/__init__.py b/src/qqbotsdk/api/__init__.py new file mode 100644 index 0000000..903bf97 --- /dev/null +++ b/src/qqbotsdk/api/__init__.py @@ -0,0 +1,30 @@ +"""OpenAPI 客户端：BotApi 按域组合——v2 群/单聊、频道消息、频道管理。 + +各域模块是方法集（继承 BaseApi 共享请求内核：鉴权头注入、429 自动 +重试、401 刷新 token 重试、统一 ApiError），方法名与拆分前完全一致。 +""" + +from .channel import ChannelApi +from .core import ApiError +from .core import BaseApi as BaseApi +from .guild import GuildApi +from .v2 import V2Api, button, keyboard + + +class BotApi(V2Api, ChannelApi, GuildApi): + """通用 REST 封装；具体业务接口见各域方法集。 + + BaseApi（请求内核）经域方法集继承自然落在 MRO 末尾， + 构造签名 (config, http, token) 由它提供。 + """ + + +__all__ = [ + "ApiError", + "BotApi", + "ChannelApi", + "GuildApi", + "V2Api", + "button", + "keyboard", +] diff --git a/src/qqbotsdk/api/channel.py b/src/qqbotsdk/api/channel.py new file mode 100644 index 0000000..fa76bf4 --- /dev/null +++ b/src/qqbotsdk/api/channel.py @@ -0,0 +1,66 @@ +"""频道消息方法集（channel_id 接入面）：发送、撤回与表情表态。""" + +from typing import Any + +from .core import BaseApi + + +class ChannelApi(BaseApi): + """频道消息方法集；请求内核（鉴权/重试/错误）由 BaseApi 提供。""" + + async def post_channel_message( + self, + channel_id: str, + content: str = "", + msg_id: str | None = None, + msg_seq: int = 1, + image: str | None = None, + **extra: Any, + ) -> dict: + """发频道消息（POST /channels/{channel_id}/messages）。 + + image 为图片 URL；其余字段（embed/ark/markdown 等）经 extra 透传。 + """ + body: dict[str, Any] = {"msg_seq": msg_seq} + if content: + body["content"] = content + if msg_id: + body["msg_id"] = msg_id + if image: + body["image"] = image + body.update(extra) + return await self.post(f"/channels/{channel_id}/messages", json=body) + + async def withdraw_channel_message( + self, channel_id: str, message_id: str, hide_tip: bool = False + ) -> dict: + """撤回频道消息（DELETE /channels/{channel_id}/messages/{message_id}）。 + + hide_tip 控制是否隐藏撤回提示小灰条（官方参数名为 hidetip）； + 仅私域机器人可用。 + """ + return await self.delete( + f"/channels/{channel_id}/messages/{message_id}", + params={"hidetip": str(hide_tip).lower()}, + ) + + async def put_channel_reaction( + self, channel_id: str, message_id: str, emoji_type: int, emoji_id: str + ) -> dict: + """对频道消息表态（PUT .../reactions/{type}/{emoji_id}）。 + + emoji_type=1 为系统表情，emoji_id 为系统表情序号（字符串）。 + """ + return await self.put( + f"/channels/{channel_id}/messages/{message_id}" + f"/reactions/{emoji_type}/{emoji_id}" + ) + + async def delete_channel_reaction( + self, channel_id: str, message_id: str, emoji_type: int, emoji_id: str + ) -> dict: + """删除自己对该消息的表态（DELETE .../reactions/{type}/{emoji_id}）。""" + return await self.delete( + f"/channels/{channel_id}/messages/{message_id}" + f"/reactions/{emoji_type}/{emoji_id}" + ) diff --git a/src/qqbotsdk/api/core.py b/src/qqbotsdk/api/core.py new file mode 100644 index 0000000..f8bdbab --- /dev/null +++ b/src/qqbotsdk/api/core.py @@ -0,0 +1,112 @@ +"""OpenAPI 请求内核：鉴权头注入、限流/鉴权失败重试与统一错误。 + +失败语义（官方文档 2026-07 核对）： +- 429 触发令牌桶限流，请求未被服务端处理，重试安全——优先按 + Retry-After 等待（线上见过 "30,120" 这类脏值，解析容错），缺失时 + 指数退避，最多重试 `_RETRY_LIMIT` 次，等待时长封顶 `_RETRY_CAP`； +- 401 多为 token 被服务端提前作废，强制刷新 token 后重试一次； +- 其余 HTTP ≥400 抛 ApiError，code/message 取自响应体。 +""" + +import asyncio +from typing import Any + +import ujson +from aiohttp import ClientResponse, ClientSession +from loguru import logger + +from ..config import Config +from ..model import error_parts +from ..token import AccessToken + + +class ApiError(RuntimeError): + """OpenAPI 调用失败：HTTP 非 2xx 或业务 code 非 0。""" + + def __init__(self, status: int, code: Any, message: Any) -> None: + self.status = status + self.code = code + self.message = message + super().__init__(f"API 调用失败: HTTP {status}, code={code}, message={message}") + + +class BaseApi: + """REST 请求内核；域方法集（v2/channel/guild）继承它做薄封装。""" + + # 429 重试次数上限；Retry-After 缺失时的退避基数（秒）与等待封顶 + _RETRY_LIMIT = 3 + _RETRY_BASE = 1.0 + _RETRY_CAP = 30.0 + + def __init__(self, config: Config, http: ClientSession, token: AccessToken) -> None: + self._config = config + self._http = http + self._token = token + + async def request(self, method: str, path: str, **kwargs) -> Any: + auth_retried = False + for attempt in range(self._RETRY_LIMIT + 1): + access_token = await self._token.get_access_token() + headers = {"Authorization": f"QQBot {access_token}"} + async with self._http.request( + method, + f"{self._config.base_url}{path}", + headers=headers, + **kwargs, + ) as resp: + # 个别接口 200 响应的 Content-Type 不是 JSON（body 仍是 JSON， + # 如回应互动返回 200 text/plain + {}），跳过标头校验 + data = await resp.json(loads=ujson.loads, content_type=None) + if resp.status == 429 and attempt < self._RETRY_LIMIT: + delay = self._retry_delay(resp, attempt) + logger.warning(f"{method} {path} 触发频率限制，{delay:.0f}s 后重试") + await asyncio.sleep(delay) + continue + if resp.status == 401 and not auth_retried: + auth_retried = True + self._token.invalidate() + logger.warning(f"{method} {path} 鉴权失效，刷新 token 后重试") + continue + if resp.status >= 400: + logger.warning(f"{method} {path} 失败: {data}") + raise ApiError(resp.status, *error_parts(data)) + return data + raise ApiError(429, None, "频率限制重试次数用尽") # 兜底，理论不可达 + + def _retry_delay(self, resp: ClientResponse, attempt: int) -> float: + raw = resp.headers.get("Retry-After") + if raw: + try: + return min(float(raw.split(",")[0].strip()), self._RETRY_CAP) + except ValueError: + pass + return min(self._RETRY_BASE * 2**attempt, self._RETRY_CAP) + + # ---------- 通用 ---------- + + async def me(self) -> dict: + """机器人自身信息（GET /users/@me）。""" + return await self.get("/users/@me") + + async def respond_interaction(self, interaction_id: str, code: int = 0) -> dict: + """回应互动事件（PUT /interactions/{interaction_id}）。 + + 按钮点击（INTERACTION_CREATE type=11）等互动收到后必须回应， + 否则客户端一直 loading 直到超时；同一 id 只能回应一次。 + """ + return await self.put(f"/interactions/{interaction_id}", json={"code": code}) + + async def get(self, path: str, **kwargs) -> Any: + return await self.request("GET", path, **kwargs) + + async def post(self, path: str, **kwargs) -> Any: + return await self.request("POST", path, **kwargs) + + async def delete(self, path: str, **kwargs) -> Any: + return await self.request("DELETE", path, **kwargs) + + async def put(self, path: str, **kwargs) -> Any: + return await self.request("PUT", path, **kwargs) + + async def patch(self, path: str, **kwargs) -> Any: + return await self.request("PATCH", path, **kwargs) diff --git a/src/qqbotsdk/api/guild.py b/src/qqbotsdk/api/guild.py new file mode 100644 index 0000000..5568313 --- /dev/null +++ b/src/qqbotsdk/api/guild.py @@ -0,0 +1,147 @@ +"""频道（guild）管理方法集：公告、精华消息、日程、成员禁言与信息查询。""" + +from typing import Any + +from .core import BaseApi + + +class GuildApi(BaseApi): + """频道管理与查询方法集；请求内核（鉴权/重试/错误）由 BaseApi 提供。""" + + # ---------- 公告与精华 ---------- + + async def create_guild_announce( + self, + guild_id: str, + channel_id: str, + message_id: str, + announces_type: int = 0, + **extra: Any, + ) -> dict: + """创建频道公告（POST /guilds/{guild_id}/announces）。 + + announces_type: 0 成员公告（默认）、1 欢迎公告； + recommend_channels 等经 extra 透传（与消息类公告互斥）。 + """ + body = { + "channel_id": channel_id, + "message_id": message_id, + "announces_type": announces_type, + } + body.update(extra) + return await self.post(f"/guilds/{guild_id}/announces", json=body) + + async def delete_guild_announce(self, guild_id: str, message_id: str) -> dict: + """删除频道公告（DELETE /guilds/{guild_id}/announces/{message_id}）。 + + 删除推荐子频道类型的公告需传 message_id="all"。 + """ + return await self.delete(f"/guilds/{guild_id}/announces/{message_id}") + + async def pin_channel_message(self, channel_id: str, message_id: str) -> dict: + """添加精华消息（PUT /channels/{channel_id}/pins/{message_id}）。""" + return await self.put(f"/channels/{channel_id}/pins/{message_id}") + + async def unpin_channel_message(self, channel_id: str, message_id: str) -> dict: + """移出精华消息（DELETE /channels/{channel_id}/pins/{message_id}）。""" + return await self.delete(f"/channels/{channel_id}/pins/{message_id}") + + async def get_channel_pins(self, channel_id: str) -> dict: + """获取精华消息列表（GET /channels/{channel_id}/pins）。""" + return await self.get(f"/channels/{channel_id}/pins") + + # ---------- 日程 ---------- + + async def get_channel_schedules( + self, channel_id: str, since: int | None = None + ) -> list | dict: + """获取子频道日程列表（GET /channels/{channel_id}/schedules）。 + + 默认返回当天日程；传 since（毫秒级时间戳）返回结束时间在其后的日程。 + """ + params = {"since": str(since)} if since is not None else None + return await self.get(f"/channels/{channel_id}/schedules", params=params) + + async def get_channel_schedule(self, channel_id: str, schedule_id: str) -> dict: + """获取单个日程（GET /channels/{channel_id}/schedules/{schedule_id}）。""" + return await self.get(f"/channels/{channel_id}/schedules/{schedule_id}") + + async def create_channel_schedule( + self, + channel_id: str, + name: str, + start_timestamp: str, + end_timestamp: str, + jump_channel_id: str, + remind_type: str = "0", + description: str = "", + ) -> dict: + """创建日程（POST /channels/{channel_id}/schedules）。 + + 时间为毫秒级时间戳字符串；remind_type: "0" 不提醒 / "1" 开始时提醒 + （官方字段为字符串）。请求体按官方约定包在 "schedule" 字段里。 + 需管理频道权限；单管理员每天限 10 次。 + """ + return await self.post( + f"/channels/{channel_id}/schedules", + json={ + "schedule": { + "name": name, + "description": description, + "start_timestamp": start_timestamp, + "end_timestamp": end_timestamp, + "jump_channel_id": jump_channel_id, + "remind_type": remind_type, + } + }, + ) + + async def update_channel_schedule( + self, channel_id: str, schedule_id: str, **fields: Any + ) -> dict: + """修改日程（PATCH /channels/{channel_id}/schedules/{schedule_id}）。 + + schedule 内的字段（name/start_timestamp/...）经 fields 透传。 + """ + return await self.patch( + f"/channels/{channel_id}/schedules/{schedule_id}", + json={"schedule": {"id": schedule_id, **fields}}, + ) + + async def delete_channel_schedule(self, channel_id: str, schedule_id: str) -> dict: + """删除日程（DELETE /channels/{channel_id}/schedules/{schedule_id}）。""" + return await self.delete(f"/channels/{channel_id}/schedules/{schedule_id}") + + # ---------- 成员管理与信息查询 ---------- + + async def mute_guild_member( + self, guild_id: str, user_id: str, mute_seconds: int = 0 + ) -> dict: + """禁言/解禁频道成员（PATCH /guilds/{guild_id}/members/{user_id}/mute）。 + + mute_seconds 为禁言时长（秒），传 0 解除禁言；上限 28 天。 + """ + return await self.patch( + f"/guilds/{guild_id}/members/{user_id}/mute", + json={"mute_seconds": str(mute_seconds)}, + ) + + async def get_guild(self, guild_id: str) -> dict: + """获取频道信息（GET /guilds/{guild_id}）。""" + return await self.get(f"/guilds/{guild_id}") + + async def get_guild_channels(self, guild_id: str) -> list | dict: + """获取频道下的子频道列表（GET /channels/{guild_id}/channels）。""" + return await self.get(f"/channels/{guild_id}/channels") + + async def get_channel(self, channel_id: str) -> dict: + """获取子频道信息（GET /channels/{channel_id}）。""" + return await self.get(f"/channels/{channel_id}") + + async def get_guild_roles(self, guild_id: str) -> dict: + """获取频道身份组列表（GET /guilds/{guild_id}/roles）。""" + return await self.get(f"/guilds/{guild_id}/roles") + + async def get_guild_member(self, guild_id: str, user_id: str) -> dict: + """获取成员详情（GET /guilds/{guild_id}/members/{user_id}）。""" + return await self.get(f"/guilds/{guild_id}/members/{user_id}") diff --git a/src/qqbotsdk/api/v2.py b/src/qqbotsdk/api/v2.py new file mode 100644 index 0000000..7a76f10 --- /dev/null +++ b/src/qqbotsdk/api/v2.py @@ -0,0 +1,218 @@ +"""v2 群聊/单聊方法集（openid 接入面）：消息、富媒体上传、撤回、禁言， +以及内嵌按钮/键盘的快捷构造。 + +被动回复规则（官方文档 2026-09 核对）：携带 msg_id（或 event_id， +二选一）为被动回复——单聊 60 分钟内最多 4 次，群聊 5 分钟内最多 5 次； +同一条消息多次回复需递增 msg_seq（相同 msg_id+msg_seq 会发送失败）； +不携带则为主动消息（另有频控，群/单聊每日每对象 1000 条）。 +""" + +from typing import Any +from uuid import uuid4 + +from .core import BaseApi + +# msg_type 取值 +MSG_TYPE_TEXT = 0 +MSG_TYPE_MARKDOWN = 2 +MSG_TYPE_MEDIA = 7 +# file_type 取值（富媒体上传） +FILE_TYPE_IMAGE = 1 +FILE_TYPE_VIDEO = 2 +FILE_TYPE_VOICE = 3 +FILE_TYPE_FILE = 4 + + +def button( + label: str, + data: str = "", + *, + type: int = 1, + permission: int = 2, + style: int = 1, + id: str | None = None, +) -> dict: + """构造单个内嵌按钮 dict，供 keyboard() 组装后传入 keyboard= 参数。 + + type: 1 回调（点击推 INTERACTION_CREATE）/ 0 跳转链接（data 为 url）/ + 2 指令（data 插入输入框）。permission: 2 所有人 / 1 管理员 / 0 指定 + 用户。label 官方限 10 字以内；要补嵌套字段（如 visited_label、 + prompt）直接修改返回的 dict。 + """ + return { + "id": id or uuid4().hex[:8], + "render_data": {"label": label, "style": style}, + "action": {"type": type, "permission": {"type": permission}, "data": data}, + } + + +def keyboard(*rows: dict | list[dict]) -> dict: + """组装内嵌键盘（keyboard= 参数）：每个位置参数为一行， + 单个按钮独占一行，按钮列表同行并排。""" + return { + "content": { + "rows": [ + {"buttons": row if isinstance(row, list) else [row]} for row in rows + ] + } + } + + +class V2Api(BaseApi): + """v2 群聊/单聊方法集；请求内核（鉴权/重试/错误）由 BaseApi 提供。""" + + async def post_c2c_message( + self, + openid: str, + content: str = "", + msg_id: str | None = None, + msg_seq: int = 1, + media: dict | None = None, + event_id: str | None = None, + markdown: str | dict | None = None, + keyboard: dict | None = None, + **extra: Any, + ) -> dict: + """发单聊消息（POST /v2/users/{openid}/messages）。 + + media 为富媒体上传接口返回的 {"file_info": ...} 结构， + 此时 content 作媒体下方文本。msg_id（回复消息）与 event_id + （响应事件）二选一，均为被动回复。markdown 为字符串或完整 + dict（msg_type=2，与 content 互斥）；keyboard 为内嵌键盘 + 结构（{"content": {"rows": [...]}} 或 {"id": ...}）。 + """ + return await self._post_message( + f"/v2/users/{openid}/messages", + content=content, + msg_id=msg_id, + msg_seq=msg_seq, + media=media, + event_id=event_id, + markdown=markdown, + keyboard=keyboard, + **extra, + ) + + async def upload_c2c_file( + self, openid: str, file_type: int, url: str, srv_send_msg: bool = False + ) -> dict: + """上传单聊富媒体（POST /v2/users/{openid}/files）。 + + 返回含 file_info，可传入 post_c2c_message 的 media 直接发送， + 或 srv_send_msg=True 直接作为消息发出。 + """ + return await self._upload_v2_file( + f"/v2/users/{openid}", file_type, url, srv_send_msg + ) + + async def post_group_message( + self, + group_openid: str, + content: str = "", + msg_id: str | None = None, + msg_seq: int = 1, + media: dict | None = None, + event_id: str | None = None, + markdown: str | dict | None = None, + keyboard: dict | None = None, + **extra: Any, + ) -> dict: + """发群消息（POST /v2/groups/{group_openid}/messages）。 + + msg_id（回复消息）与 event_id（响应 GROUP_ADD_ROBOT / + INTERACTION_CREATE 等事件）二选一。markdown 为字符串或完整 + dict（msg_type=2，与 content 互斥）；keyboard 为内嵌键盘 + 结构（{"content": {"rows": [...]}} 或 {"id": ...}）。 + """ + return await self._post_message( + f"/v2/groups/{group_openid}/messages", + content=content, + msg_id=msg_id, + msg_seq=msg_seq, + media=media, + event_id=event_id, + markdown=markdown, + keyboard=keyboard, + **extra, + ) + + async def upload_group_file( + self, group_openid: str, file_type: int, url: str, srv_send_msg: bool = False + ) -> dict: + """上传群富媒体（POST /v2/groups/{group_openid}/files）。""" + return await self._upload_v2_file( + f"/v2/groups/{group_openid}", file_type, url, srv_send_msg + ) + + async def _upload_v2_file( + self, base: str, file_type: int, url: str, srv_send_msg: bool + ) -> dict: + return await self.post( + f"{base}/files", + json={"file_type": file_type, "url": url, "srv_send_msg": srv_send_msg}, + ) + + async def _post_message( + self, + path: str, + content: str, + msg_id: str | None, + msg_seq: int, + media: dict | None, + event_id: str | None, + markdown: str | dict | None = None, + keyboard: dict | None = None, + **extra: Any, + ) -> dict: + # media > markdown > 纯文本；markdown 与 content 互斥（官方约定） + if media is not None: + body: dict[str, Any] = {"msg_type": MSG_TYPE_MEDIA, "media": media} + if content: + body["content"] = content + elif markdown is not None: + body = { + "msg_type": MSG_TYPE_MARKDOWN, + "markdown": markdown + if isinstance(markdown, dict) + else {"content": markdown}, + } + else: + body = {"msg_type": MSG_TYPE_TEXT} + if content: + body["content"] = content + if msg_id: + body["msg_id"] = msg_id + body["msg_seq"] = msg_seq + elif event_id: + body["event_id"] = event_id + body["msg_seq"] = msg_seq + if keyboard is not None: + body["keyboard"] = keyboard + body.update(extra) + return await self.post(path, json=body) + + async def _withdraw_v2_message(self, path: str, message_id: str) -> dict: + # 发送超过 2 分钟的消息不可撤回 + return await self.delete(f"{path}/messages/{message_id}") + + async def withdraw_group_message(self, group_openid: str, message_id: str) -> dict: + """撤回群消息（DELETE /v2/groups/{group_openid}/messages/{message_id}）。 + + 群管理员可撤回自己与普通成员的消息；普通成员仅能撤回自己的。 + """ + return await self._withdraw_v2_message(f"/v2/groups/{group_openid}", message_id) + + async def withdraw_c2c_message(self, openid: str, message_id: str) -> dict: + """撤回单聊消息（DELETE /v2/users/{openid}/messages/{message_id}）。""" + return await self._withdraw_v2_message(f"/v2/users/{openid}", message_id) + + async def mute_group_member(self, group_openid: str, mute_expire_at: str) -> dict: + """群成员禁言（POST /v2/groups/{group_openid}/restrict_chat_setting）。 + + mute_expire_at 为禁言到期时间（RFC3339 格式，如 + "2026-01-01T12:00:00+08:00"）；传当前时间之前即解除禁言。 + """ + return await self.post( + f"/v2/groups/{group_openid}/restrict_chat_setting", + json={"mute_expire_at": mute_expire_at}, + ) diff --git a/src/qqbotsdk/connecter.py b/src/qqbotsdk/connecter.py index 3c0e5ca..fc6aea3 100644 --- a/src/qqbotsdk/connecter.py +++ b/src/qqbotsdk/connecter.py @@ -1,15 +1,19 @@ """连接适配器：把两种接入方式（WebSocket / Webhook）归一为事件流入、reply 队列流出。 -- WebsocketConnecter：gateway 握手 + ws 长连接，指数退避重连； - 看门狗两层互补——receive 的静默超时抓"连接彻底静默"， +- WebsocketConnecter：gateway 握手 + ws 长连接，指数退避重连（连接稳定 + 存活才把退避归一，抖动的服务端不会被 1s 间隔反复探测）；心跳任务随 + 连接生灭——HELLO 给出间隔后周期投递，断线即取消，不在 reply 队列堆积 + 过期心跳；看门狗两层互补——receive 的静默超时抓"连接彻底静默"， 收到消息时的 ACK 期限检查抓"有事件流但心跳已死"； - WebhookConnecter：aiohttp server 接收回调，验签失败 401， - 按 payload id 做 60s TTL 去重，正常推送进事件队列。 + 按 payload id 做 60s TTL 去重，正常推送进事件队列；op=13 验证请求 + 经注入的 handle_event 同步取得应答（HTTP 响应不走 reply 队列）。 """ import asyncio +from collections.abc import Awaitable, Callable from time import monotonic -from typing import Protocol +from typing import Any, Protocol import aiohttp import ujson @@ -18,9 +22,9 @@ from loguru import logger from .config import Config from .crypto import verify_sig -from .emitter import EventEmitter -from .model import NOT_SET, Opcode, Payload +from .model import NOT_SET, Opcode, Payload, payload_of from .queue import EventQueue +from .session import Session from .token import AccessToken @@ -37,25 +41,30 @@ class WebsocketConnecter: # 不能压到 1× 整——ACK 的自然间隔就是 1×，正常抖动会贴边误杀；TCP 下丢包表现为延迟而非静默 _SILENCE_FACTOR = 1.1 _ACK_FACTOR = 1.2 + # 连接存活超过该时长才算稳定，才把退避归一 + _STABLE_SECONDS = 60.0 def __init__( self, config: Config, http: aiohttp.ClientSession, token: AccessToken, + session: Session, queue: EventQueue, ) -> None: self._config = config self._http = http self._token = token + self._session = session self._queue = queue async def run(self) -> None: backoff: float = 1 while True: try: - await self._connect() - backoff = 1 + lived = await self._connect() + if lived >= self._STABLE_SECONDS: + backoff = 1 except RateLimitError: logger.warning("gateway 触发频率限制") backoff = 60 @@ -67,7 +76,9 @@ class WebsocketConnecter: await asyncio.sleep(backoff) backoff = min(backoff * 2, 60) - async def _connect(self) -> None: + async def _connect(self) -> float: + """建立一次连接直到断开，返回存活秒数（供 run 判断是否重置退避）。""" + started = monotonic() access_token = await self._token.get_access_token() headers = {"Authorization": f"QQBot {access_token}"} @@ -84,15 +95,22 @@ class WebsocketConnecter: async with self._http.ws_connect(gateway, headers=headers) as ws: logger.info("WebSocket 已连接") - receive = asyncio.create_task(self.receive_helper(ws)) + # 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)) + heartbeat = asyncio.create_task(self.heartbeat_helper(interval)) try: await receive finally: reply.cancel() + heartbeat.cancel() logger.info("WebSocket 已断开") + return monotonic() - started - async def receive_helper(self, ws: aiohttp.ClientWebSocketResponse) -> None: + async def receive_helper( + self, ws: aiohttp.ClientWebSocketResponse, interval_fut: asyncio.Future[float] + ) -> None: interval: float = 0 last_ack = monotonic() while not ws.closed: @@ -107,6 +125,8 @@ class WebsocketConnecter: d = data.get("d") if isinstance(d, dict): interval = d.get("heartbeat_interval", 41_250) / 1000 + if not interval_fut.done(): + interval_fut.set_result(interval) elif op == Opcode.HEARTBEAT_ACK: last_ack = monotonic() await self._queue.put_event(data) @@ -115,6 +135,14 @@ class WebsocketConnecter: if monotonic() - last_ack > interval * self._ACK_FACTOR: raise TimeoutError + async def heartbeat_helper(self, interval_fut: asyncio.Future[float]) -> None: + """HELLO 给出间隔后周期投递心跳；断线随任务取消，不堆积过期心跳。""" + interval = await interval_fut + while True: + await asyncio.sleep(interval) + # 服务端不会推送 op=1，心跳只由本任务经 reply 队列发出 + await self._queue.put_reply(payload_of(Opcode.HEARTBEAT, self._session.seq)) + async def reply_helper(self, ws: aiohttp.ClientWebSocketResponse) -> None: while not ws.closed: payload: Payload = await self._queue.get_reply() @@ -129,11 +157,16 @@ class WebhookConnecter: _DEDUP_TTL = 60 def __init__( - self, config: Config, emitter: EventEmitter, queue: EventQueue + self, + config: Config, + queue: EventQueue, + handle_event: Callable[[Payload], Awaitable[Any]], ) -> None: + """handle_event 供 op=13 验证请求同步取应答（通常传 emitter.handle）； + 与 ws 侧对齐，本类只依赖队列与该回调，不感知 EventEmitter。""" self._config = config - self._emitter = emitter self._queue = queue + self._handle_event = handle_event self._seen: dict[str, float] = {} def _make_app(self) -> web.Application: @@ -173,7 +206,7 @@ class WebhookConnecter: if payload.get("op") == Opcode.VALIDATION: logger.info("收到 Webhook 验证请求") - ret = await self._emitter.handle(payload) + ret = await self._handle_event(payload) if ret is None: return web.json_response( {"error": "missing validation handler"}, status=500 diff --git a/src/qqbotsdk/emitter.py b/src/qqbotsdk/emitter.py index aaa8a34..9acfaa9 100644 --- a/src/qqbotsdk/emitter.py +++ b/src/qqbotsdk/emitter.py @@ -3,11 +3,16 @@ - `dispatch()` 泵循环消费 EventQueue.msg_queue，`emit()` 按 op（DISPATCH 用 t 名、其余用 Opcode 名）路由到 `@on` 注册的 handler； - handler 参数按形参标注注入：payload dataclass → 解析对象，`Event` → - 事件封装，`services` 里登记的组件类型 → 实例，其余直传原始 d； + 事件封装，`services` 里登记的组件类型 → 实例，其余直传原始 d + （可能是 dict、标量甚至 None，协议事件的 d 不保证是 dict）； - 业务 handler（op=0）返回值不回流；协议层（op≠0）返回的 payload 进 reply 队列由 ws 发出。handler 异常就地隔离，不影响其他 handler。 +- `on(..., background=True)` 把 handler 放进后台任务并发执行： + 不阻塞事件流，但同事件 handler 间失去先后保证，返回值不回流； + 进程退出经 `close()` 统一取消在飞任务。 """ +import asyncio import inspect from abc import ABC, abstractmethod from collections.abc import Awaitable, Callable @@ -35,16 +40,18 @@ class BaseProtocol(ABC): class EventEmitter: - def __init__(self, queue: EventQueue) -> None: + def __init__(self, queue: EventQueue | None = None) -> None: self._ee = AsyncIOEventEmitter() - self._queue = queue + self._queue = queue if queue is not None else EventQueue() self.services: dict[type, Any] = {} self._protocol: BaseProtocol | None = None self._hints: dict[Callable[..., Any], dict[str, Any]] = {} + self._background: set[Callable[..., Any]] = set() + self._bg_tasks: set[asyncio.Task[None]] = set() @property def queue(self) -> EventQueue: - """本 emitter 消费的事件队列；外部构造 emitter 时由 run_loop 复用同一实例。""" + """本 emitter 消费的事件队列；未显式传入时构造即自建，run_loop 复用同一实例。""" return self._queue def register_protocol(self, protocol: BaseProtocol) -> None: @@ -52,25 +59,42 @@ class EventEmitter: protocol.register(self) @overload - def on(self, event: Opcode | str) -> Callable[[Handler], Handler]: ... + def on( + self, event: Opcode | str, *, background: bool = False + ) -> Callable[[Handler], Handler]: ... @overload - def on(self, event: Opcode | str, fn: Handler) -> Handler: ... + def on( + self, event: Opcode | str, fn: Handler, *, background: bool = False + ) -> Handler: ... def on( - self, event: Opcode | str, fn: Handler | None = None + self, + event: Opcode | str, + fn: Handler | None = None, + *, + background: bool = False, ) -> Handler | Callable[[Handler], Handler]: name = event.name if isinstance(event, Opcode) else event def decorator(fn: Handler) -> Handler: self._ee.on(name, fn) + if background: + self._background.add(fn) return fn if fn is None: return decorator - self._ee.on(name, fn) - return fn + return decorator(fn) + + async def close(self) -> None: + """取消全部在飞后台任务并等待收尾；run_loop 退出时先于连接池调用。""" + if not self._bg_tasks: + return + for task in self._bg_tasks: + task.cancel() + await asyncio.gather(*self._bg_tasks, return_exceptions=True) async def dispatch(self) -> None: while True: @@ -115,8 +139,12 @@ class EventEmitter: if not listeners: return + reply = op != Opcode.DISPATCH for fn in listeners: - await self._invoke(fn, event, reply=op != Opcode.DISPATCH) + if fn in self._background: + self._spawn(fn, event) + else: + await self._invoke(fn, event, reply=reply) async def handle(self, data: Payload) -> Any: """同步分发事件并返回首个处理器的非空返回值，用于 op=13 等需要应答的协议事件。""" @@ -143,11 +171,29 @@ class EventEmitter: 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。""" + task = asyncio.create_task(self._run_background(fn, event)) + self._bg_tasks.add(task) + task.add_done_callback(self._bg_tasks.discard) + + async def _run_background( + self, fn: Callable[..., Awaitable[Any]], event: Event + ) -> None: + try: + await fn(*self._resolve_params(fn, event)) + except Exception as e: + logger.exception( + f"后台事件处理器 {getattr(fn, '__qualname__', fn)} 异常: {e}" + ) + def _resolve_params( self, fn: Callable[..., Awaitable[Any]], event: Event ) -> list[Any]: """按 handler 形参标注解析：payload dataclass → 解析对象，`Event` → 事件 - 封装，`services` 登记的组件类型 → 实例，其余（含无标注）→ 原始 d。""" + 封装，`services` 登记的组件类型 → 实例，其余（含无标注）→ 原始 d + （Event.data 是 dict 化视图，标量 d 如 INVALID_SESSION 的 true/false + 必须走这里才能拿到原值）。""" hints = self._hints.get(fn) if hints is None: hints = get_type_hints(fn) @@ -163,5 +209,5 @@ class EventEmitter: elif hint in self.services: args.append(self.services[hint]) else: - args.append(event.data) + args.append(event.raw.get("d")) return args diff --git a/src/qqbotsdk/model.py b/src/qqbotsdk/model.py index 04550df..0bc2d5b 100644 --- a/src/qqbotsdk/model.py +++ b/src/qqbotsdk/model.py @@ -89,3 +89,14 @@ class Intent(IntFlag): def payload_of(op: Opcode, d: Any = NOT_SET) -> Payload: return {"id": NOT_SET, "op": op, "d": d, "s": NOT_SET, "t": NOT_SET} + + +def error_parts(data: Any) -> tuple[Any, Any]: + """从错误响应体取 (code, message)；非 dict 响应整体作 message。 + + QQ 的失败响应 HTTP 可能仍是 200 且 Content-Type 不规范， + ApiError/TokenError 统一用它暴露真实原因。 + """ + if isinstance(data, dict): + return data.get("code"), data.get("message") + return None, data diff --git a/src/qqbotsdk/session.py b/src/qqbotsdk/session.py index 11bdf1b..8b8b5ab 100644 --- a/src/qqbotsdk/session.py +++ b/src/qqbotsdk/session.py @@ -6,19 +6,14 @@ class Session: def __init__(self) -> None: self.session_id: str | None = None - self._sequence_id: int | None = None - - @property - def sequence_id(self) -> int | None: - return self._sequence_id - - def update_sequence(self, value: int) -> None: - """只接受不回退的序列号，乱序/重放的旧值会被忽略。""" - current = self._sequence_id or 0 - if value >= current: - self._sequence_id = value + self._seq: int | None = None @property def seq(self) -> int: - """RESUME 时使用的序列号，无历史序列号时为 0。""" - return self._sequence_id or 0 + """已收到的最大 s 序列号（心跳/RESUME 用），无历史时为 0。""" + return self._seq or 0 + + def update_sequence(self, value: int) -> None: + """只接受不回退的序列号，乱序/重放的旧值会被忽略。""" + if value >= self.seq: + self._seq = value diff --git a/src/qqbotsdk/token.py b/src/qqbotsdk/token.py index 37e7169..5f05e0b 100644 --- a/src/qqbotsdk/token.py +++ b/src/qqbotsdk/token.py @@ -1,12 +1,19 @@ -"""access_token 获取与缓存：到期前直接复用，过期自动重新换取。""" +"""access_token 获取与缓存：到期前直接复用，过期自动重新换取。 -import time +并发安全：换取过程持 asyncio.Lock 双重检查，多个协程同时首调只发一次请求； +到期判断基于 monotonic 时钟并预留提前刷新余量，不受系统改表时间影响， +也避免卡着到期点刷新在边界上 401。 +""" + +import asyncio +from time import monotonic from typing import Any import ujson from aiohttp import ClientSession from .config import Config +from .model import error_parts class TokenError(RuntimeError): @@ -26,34 +33,48 @@ class TokenError(RuntimeError): class AccessToken: + # 提前刷新余量（秒）：expires_in 末段视为不可信 + _EXPIRY_MARGIN = 30.0 + def __init__(self, config: Config, http: ClientSession) -> None: self._config = config self._http = http self._access_token = "" self._expires_at = 0.0 + self._lock = asyncio.Lock() async def get_access_token(self) -> str: - if time.time() < self._expires_at: + if not self._expired(): return self._access_token + async with self._lock: + if not self._expired(): + return self._access_token - payload = { - "appId": self._config.app_id, - "clientSecret": self._config.app_secret, - } + payload = { + "appId": self._config.app_id, + "clientSecret": self._config.app_secret, + } - async with self._http.post( - f"{self._config.base_url}/app/getAppAccessToken", - json=payload, - ) as resp: - # 与 request() 同理：个别接口以非 JSON Content-Type 返回 JSON body - data = await resp.json(loads=ujson.loads, content_type=None) - # 失败时 QQ 返回 code/message 且 HTTP 可能仍是 200， - # 以 access_token 是否在为准判断成败，抛出真实原因 - if not isinstance(data, dict) or "access_token" not in data: - code = data.get("code") if isinstance(data, dict) else None - message = data.get("message") if isinstance(data, dict) else data - raise TokenError(resp.status, code, message) - self._access_token = data["access_token"] - self._expires_at = time.time() + int(data["expires_in"]) + async with self._http.post( + f"{self._config.base_url}/app/getAppAccessToken", + json=payload, + ) as resp: + # 与 ApiCore.request 同理：个别接口以非 JSON Content-Type 返回 JSON body + data = await resp.json(loads=ujson.loads, content_type=None) + # 失败时 QQ 返回 code/message 且 HTTP 可能仍是 200， + # 以 access_token 是否在为准判断成败，抛出真实原因 + if not isinstance(data, dict) or "access_token" not in data: + raise TokenError(resp.status, *error_parts(data)) + self._access_token = data["access_token"] + self._expires_at = monotonic() + max( + int(data["expires_in"]) - self._EXPIRY_MARGIN, 1 + ) return self._access_token + + def invalidate(self) -> None: + """作废缓存 token（如 OpenAPI 侧 401），下次获取强制重新换取。""" + self._expires_at = 0.0 + + def _expired(self) -> bool: + return monotonic() >= self._expires_at diff --git a/src/qqbotsdk/webhook_protocol.py b/src/qqbotsdk/webhook_protocol.py index 1038d54..6953755 100644 --- a/src/qqbotsdk/webhook_protocol.py +++ b/src/qqbotsdk/webhook_protocol.py @@ -6,12 +6,10 @@ from .config import Config from .crypto import sign_msg from .emitter import BaseProtocol, EventEmitter from .model import Opcode, ValidationData -from .queue import EventQueue -from .session import Session class WebhookProtocol(BaseProtocol): - def __init__(self, config: Config, queue: EventQueue, session: Session) -> None: + def __init__(self, config: Config) -> None: self._config = config @override diff --git a/src/qqbotsdk/ws_protocol.py b/src/qqbotsdk/ws_protocol.py index b3251fa..7b3a331 100644 --- a/src/qqbotsdk/ws_protocol.py +++ b/src/qqbotsdk/ws_protocol.py @@ -1,19 +1,18 @@ -"""WebSocket 接入的协议层：IDENTIFY/RESUME 鉴权、定时心跳、序列号跟踪。 +"""WebSocket 接入的协议层：IDENTIFY/RESUME 鉴权、序列号跟踪与会话保持。 -识别 HELLO 后启动 apscheduler 心跳；断线重连时若已有 session_id -则发 RESUME 续传，收到不可恢复的 INVALID_SESSION 才清空会话重新 IDENTIFY。 +心跳与连接生命周期归 WebsocketConnecter；本层只负责会话语义—— +识别 HELLO 后按有无 session_id 决定 RESUME 续传或 IDENTIFY 新建， +收到不可恢复的 INVALID_SESSION（d=false）才清空会话重新 IDENTIFY。 """ from tomllib import load from typing import override -from apscheduler.schedulers.asyncio import AsyncIOScheduler from loguru import logger from .config import Config from .emitter import BaseProtocol, EventEmitter from .model import ( - HelloData, IdentifyData, Intent, Opcode, @@ -22,7 +21,6 @@ from .model import ( payload_of, ) from .payloads import Ready -from .queue import EventQueue from .session import Session from .token import AccessToken @@ -49,24 +47,14 @@ def get_intents(path: str) -> int: class WebsocketProtocol(BaseProtocol): - def __init__( - self, - config: Config, - queue: EventQueue, - session: Session, - token: AccessToken, - ) -> None: + def __init__(self, config: Config, session: Session, token: AccessToken) -> None: self._config = config - self._queue = queue self._session = session self._token = token - self._scheduler = AsyncIOScheduler() - self._heartbeat_job_id = "heartbeat" @override def register(self, emitter: EventEmitter) -> None: emitter.on(Opcode.HELLO)(self.identify) - emitter.on(Opcode.HEARTBEAT)(self.heartbeat) emitter.on("READY")(self.ready) emitter.on(Opcode.INVALID_SESSION)(self.invalid_session) @@ -76,17 +64,7 @@ class WebsocketProtocol(BaseProtocol): if isinstance(seq, int): self._session.update_sequence(seq) - async def identify(self, data: HelloData) -> Payload: - if not self._scheduler.running: - self._scheduler.add_job( - self.send_heartbeat, - "interval", - seconds=data["heartbeat_interval"] / 1000, - id=self._heartbeat_job_id, - ) - self._scheduler.start() - logger.info(f"心跳任务已启动，间隔 {data['heartbeat_interval']}ms") - + async def identify(self) -> Payload: access_token = await self._token.get_access_token() d: ResumeData | IdentifyData @@ -109,12 +87,6 @@ class WebsocketProtocol(BaseProtocol): return payload_of(op, d) - async def heartbeat(self, _) -> Payload: - return payload_of(Opcode.HEARTBEAT, self._session.sequence_id) - - async def send_heartbeat(self) -> None: - await self._queue.put_reply(await self.heartbeat(None)) - async def ready(self, ready: Ready) -> None: self._session.session_id = ready.session_id logger.info("会话已建立（READY）") diff --git a/tests/test_api.py b/tests/test_api.py index d38c346..8a6abc9 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,7 +1,8 @@ +import pytest from aiohttp import ClientSession, ClientTimeout, web from aiohttp.test_utils import TestServer -from qqbotsdk.api import BotApi, button, keyboard +from qqbotsdk.api import ApiError, BotApi, button, keyboard from qqbotsdk.config import Config from qqbotsdk.token import AccessToken @@ -155,8 +156,9 @@ async def test_post_message_markdown_dict_and_event_id(): async def test_media_takes_precedence_over_markdown(): api, cap = make_api() - await api.post_group_message("G1", content="看图", msg_id="m1", - media={"file_info": "f"}, markdown="# t") + await api.post_group_message( + "G1", content="看图", msg_id="m1", media={"file_info": "f"}, markdown="# t" + ) body = cap["call"][2]["json"] assert body["msg_type"] == 7 assert "markdown" not in body @@ -228,3 +230,142 @@ async def test_request_tolerates_non_json_content_type(): await http.close() finally: await server.close() + + +# ---------- 请求内核：429 限流重试 / 401 刷新 token 重试 ---------- + + +class ScriptedResponse: + def __init__(self, status: int, payload, headers: dict | None = None) -> None: + self.status = status + self.headers = headers or {} + self._payload = payload + + async def json(self, **kwargs): + return self._payload + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + +class ScriptedSession: + """按序弹出响应，记录 request 调用。""" + + def __init__(self, *responses: ScriptedResponse) -> None: + self._responses = list(responses) + self.calls: list[tuple[str, str]] = [] + + def request(self, method: str, url: str, **kwargs): + self.calls.append((method, url)) + return self._responses.pop(0) + + +class StubToken: + def __init__(self) -> None: + self.fetches = 0 + self.invalidated = 0 + + async def get_access_token(self) -> str: + self.fetches += 1 + return "t1" + + def invalidate(self) -> None: + self.invalidated += 1 + + +def make_core_api(http: ScriptedSession, token: StubToken | None = None) -> BotApi: + config = Config( + app_id="id", + app_secret="s", + base_url="https://x", + timeout=ClientTimeout(total=5), + connecter="websocket", + intents_file="intents.toml", + webhook_host="0.0.0.0", + webhook_port=8080, + webhook_path="/qqbot/webhook", + ) + return BotApi(config, http, token or StubToken()) # type: ignore[arg-type] + + +async def test_request_retries_429_honoring_retry_after(monkeypatch): + delays: list[float] = [] + + async def fake_sleep(delay): + delays.append(delay) + + monkeypatch.setattr("qqbotsdk.api.core.asyncio.sleep", fake_sleep) + http = ScriptedSession( + ScriptedResponse( + 429, {"code": 11244, "message": "limit"}, {"Retry-After": "2"} + ), + ScriptedResponse(200, {"ok": 1}), + ) + api = make_core_api(http) + assert await api.request("POST", "/v1/x") == {"ok": 1} + assert len(http.calls) == 2 + assert delays == [2.0] + + +async def test_request_429_dirty_header_falls_back_to_backoff(monkeypatch): + delays: list[float] = [] + + async def fake_sleep(delay): + delays.append(delay) + + monkeypatch.setattr("qqbotsdk.api.core.asyncio.sleep", fake_sleep) + http = ScriptedSession( + # 线上见过的脏值 "30,120"：取首段 30 + ScriptedResponse(429, {}, {"Retry-After": "30,120"}), + # 无 Retry-After：指数退避 1 * 2**1 = 2 + ScriptedResponse(429, {}), + ScriptedResponse(200, {}), + ) + api = make_core_api(http) + assert await api.get("/v1/x") == {} + assert delays == [30.0, 2.0] + + +async def test_request_429_retry_after_capped(monkeypatch): + delays: list[float] = [] + + async def fake_sleep(delay): + delays.append(delay) + + monkeypatch.setattr("qqbotsdk.api.core.asyncio.sleep", fake_sleep) + http = ScriptedSession( + ScriptedResponse(429, {}, {"Retry-After": "3600"}), + ScriptedResponse(200, {}), + ) + api = make_core_api(http) + await api.get("/v1/x") + assert delays == [30.0] # 封顶，不真睡一小时 + + +async def test_request_429_exhausted_raises_api_error(monkeypatch): + async def fake_sleep(delay): + pass + + monkeypatch.setattr("qqbotsdk.api.core.asyncio.sleep", fake_sleep) + http = ScriptedSession(*(ScriptedResponse(429, {}) for _ in range(4))) + api = make_core_api(http) + with pytest.raises(ApiError) as exc_info: + await api.get("/v1/x") + assert exc_info.value.status == 429 + assert len(http.calls) == 4 # 首次 + 3 次重试 + + +async def test_request_401_invalidates_token_and_retries_once(): + http = ScriptedSession( + ScriptedResponse(401, {"code": -1, "message": "unauthorized"}), + ScriptedResponse(200, {"ok": 1}), + ) + token = StubToken() + api = make_core_api(http, token) + assert await api.get("/v1/x") == {"ok": 1} + assert token.invalidated == 1 + assert token.fetches == 2 + assert len(http.calls) == 2 diff --git a/tests/test_dedup.py b/tests/test_dedup.py index f566ab0..97ddeb7 100644 --- a/tests/test_dedup.py +++ b/tests/test_dedup.py @@ -1,6 +1,5 @@ from qqbotsdk.config import Config from qqbotsdk.connecter import WebhookConnecter -from qqbotsdk.emitter import EventEmitter from qqbotsdk.queue import EventQueue @@ -16,7 +15,11 @@ def make_connecter() -> WebhookConnecter: webhook_port=8080, webhook_path="/qqbot/webhook", ) - return WebhookConnecter(config, EventEmitter(EventQueue()), EventQueue()) + + async def noop_handle(payload): + return None + + return WebhookConnecter(config, EventQueue(), noop_handle) def test_first_push_is_not_duplicate(): diff --git a/tests/test_emitter.py b/tests/test_emitter.py index 308a097..160422f 100644 --- a/tests/test_emitter.py +++ b/tests/test_emitter.py @@ -1,28 +1,27 @@ +import asyncio + import pytest +from qqbotsdk.config import Config from qqbotsdk.emitter import EventEmitter 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(): s = Session() - assert s.sequence_id is None + assert s.seq == 0 s.update_sequence(5) s.update_sequence(3) # 回退被忽略 - assert s.sequence_id == 5 + assert s.seq == 5 s.update_sequence(7) - assert s.sequence_id == 7 assert s.seq == 7 -def test_seq_defaults_to_zero(): - s = Session() - assert s.seq == 0 - - @pytest.mark.asyncio async def test_handler_exception_is_isolated(): queue = EventQueue() @@ -96,7 +95,93 @@ async def test_unregistered_annotation_falls_back_to_raw_dict(): async def handler(dep: Session): # Session 未登记进 services received.append(dep) - await emitter.emit( - {"op": Opcode.DISPATCH, "t": "READY", "d": {"session_id": "s1"}} - ) + await emitter.emit({"op": Opcode.DISPATCH, "t": "READY", "d": {"session_id": "s1"}}) assert received == [{"session_id": "s1"}] + + +@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", + ) + emitter = EventEmitter() + session = Session() + session.session_id = "S1" + emitter.register_protocol( + WebsocketProtocol(config, session, AccessToken(config, None)) # type: ignore[arg-type] + ) + 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 # 不可续传，会话清空 + + +@pytest.mark.asyncio +async def test_background_handler_does_not_block_dispatch(): + """background=True：慢 handler 进后台任务，不阻塞同一事件的后续 handler。""" + emitter = EventEmitter() + slow_done = asyncio.Event() + quick_done = asyncio.Event() + + @emitter.on("GROUP_AT_MESSAGE_CREATE", background=True) + async def slow(msg: GroupAtMessage): + await asyncio.sleep(0.05) + slow_done.set() + + @emitter.on("GROUP_AT_MESSAGE_CREATE") + async def quick(msg: GroupAtMessage): + quick_done.set() + + await emitter.emit( + { + "op": Opcode.DISPATCH, + "t": "GROUP_AT_MESSAGE_CREATE", + "d": {"content": "hi"}, + } + ) + # emit 返回时快 handler 已跑完，慢 handler 仍在后台 + assert quick_done.is_set() + assert not slow_done.is_set() + await asyncio.wait_for(slow_done.wait(), timeout=2) + await asyncio.sleep(0) # 让已完成任务的 done_callback 清理 _bg_tasks + await emitter.close() + assert not emitter._bg_tasks # 收尾后在飞任务清空 + + +@pytest.mark.asyncio +async def test_close_cancels_inflight_background_tasks(): + emitter = EventEmitter() + cancelled = asyncio.Event() + + @emitter.on("READY", background=True) + async def forever(): + try: + await asyncio.sleep(60) + except asyncio.CancelledError: + cancelled.set() + raise + + await emitter.emit({"op": Opcode.DISPATCH, "t": "READY", "d": {}}) + await asyncio.sleep(0) # 让后台任务实际启动（未启动的任务被 cancel 不会执行函数体） + await emitter.close() + assert cancelled.is_set() diff --git a/tests/test_route.py b/tests/test_route.py index dd27c5e..0a60da5 100644 --- a/tests/test_route.py +++ b/tests/test_route.py @@ -2,12 +2,11 @@ import pytest from qqbotsdk.emitter import EventEmitter from qqbotsdk.model import Opcode -from qqbotsdk.queue import EventQueue @pytest.fixture def emitter() -> EventEmitter: - return EventEmitter(EventQueue()) + return EventEmitter() # 无参构造，覆盖 queue 默认自建路径 def test_route_dispatch_uses_t(emitter: EventEmitter): diff --git a/tests/test_token.py b/tests/test_token.py index 59bdd6b..646c686 100644 --- a/tests/test_token.py +++ b/tests/test_token.py @@ -1,4 +1,6 @@ -"""AccessToken：成功换取与失败时抛 TokenError（带 QQ 的 code/message）。""" +"""AccessToken：缓存复用、失败抛 TokenError、并发只换取一次、invalidate 强制刷新。""" + +import asyncio import pytest from aiohttp import ClientTimeout @@ -7,27 +9,8 @@ from qqbotsdk.config import Config from qqbotsdk.token import AccessToken, TokenError -def make_token(payload) -> AccessToken: - class FakeResponse: - status = 200 - - def __init__(self) -> None: - self._payload = payload - - async def json(self, **kwargs): - return self._payload - - async def __aenter__(self): - return self - - async def __aexit__(self, *exc): - return False - - class FakeSession: - def post(self, *args, **kwargs): - return FakeResponse() - - config = Config( +def make_config() -> Config: + return Config( app_id="id", app_secret="s", base_url="https://x", @@ -38,14 +21,46 @@ def make_token(payload) -> AccessToken: webhook_port=8080, webhook_path="/qqbot/webhook", ) - return AccessToken(config, FakeSession()) # type: ignore[arg-type] + + +class FakeResponse: + def __init__(self, payload, status: int = 200) -> None: + self.status = status + self._payload = payload + + async def json(self, **kwargs): + return self._payload + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + +class FakeSession: + """按序弹出响应体，记录 post 次数。""" + + def __init__(self, *payloads) -> None: + self._payloads = list(payloads) + self.posts = 0 + + def post(self, *args, **kwargs): + self.posts += 1 + return FakeResponse(self._payloads.pop(0)) + + +def make_token(*payloads) -> AccessToken: + return AccessToken(make_config(), FakeSession(*payloads)) # type: ignore[arg-type] async def test_success_caches_token(): - token = make_token({"access_token": "t1", "expires_in": 7200}) + session = FakeSession({"access_token": "t1", "expires_in": 7200}) + token = AccessToken(make_config(), session) # type: ignore[arg-type] assert await token.get_access_token() == "t1" - # 缓存期内直接复用，不再发请求（FakeSession 也没有第二次响应的意义） + # 缓存期内直接复用，不再发请求 assert await token.get_access_token() == "t1" + assert session.posts == 1 async def test_error_payload_raises_token_error(): @@ -62,3 +77,24 @@ async def test_non_dict_payload_raises_token_error(): with pytest.raises(TokenError) as exc_info: await token.get_access_token() assert exc_info.value.message == "gateway busy" + + +async def test_concurrent_first_fetch_only_requests_once(): + session = FakeSession({"access_token": "t1", "expires_in": 7200}) + token = AccessToken(make_config(), session) # type: ignore[arg-type] + + results = await asyncio.gather(*(token.get_access_token() for _ in range(5))) + assert results == ["t1"] * 5 + assert session.posts == 1 # 双重检查锁：并发首调只换取一次 + + +async def test_invalidate_forces_refetch(): + session = FakeSession( + {"access_token": "t1", "expires_in": 7200}, + {"access_token": "t2", "expires_in": 7200}, + ) + token = AccessToken(make_config(), session) # type: ignore[arg-type] + assert await token.get_access_token() == "t1" + token.invalidate() # 模拟服务端提前作废 + assert await token.get_access_token() == "t2" + assert session.posts == 2 diff --git a/tests/test_webhook_integration.py b/tests/test_webhook_integration.py index fc93761..deb33da 100644 --- a/tests/test_webhook_integration.py +++ b/tests/test_webhook_integration.py @@ -14,8 +14,6 @@ from qqbotsdk.connecter import WebhookConnecter from qqbotsdk.crypto import build_seed from qqbotsdk.emitter import EventEmitter from qqbotsdk.payloads import GroupAtMessage -from qqbotsdk.queue import EventQueue -from qqbotsdk.session import Session from qqbotsdk.webhook_protocol import WebhookProtocol SECRET = "webhook-test-secret" @@ -52,10 +50,9 @@ def event_body(event_id: str, content: str) -> dict[str, Any]: @pytest.mark.asyncio async def test_webhook_end_to_end(): config = make_config(0) - queue = EventQueue() - emitter = EventEmitter(queue) - emitter.register_protocol(WebhookProtocol(config, queue, Session())) - connecter = WebhookConnecter(config, emitter, queue) + emitter = EventEmitter() + emitter.register_protocol(WebhookProtocol(config)) + connecter = WebhookConnecter(config, emitter.queue, emitter.handle) received: list[GroupAtMessage] = [] diff --git a/tests/test_ws_integration.py b/tests/test_ws_integration.py index 2475bbc..9cf2d68 100644 --- a/tests/test_ws_integration.py +++ b/tests/test_ws_integration.py @@ -114,9 +114,9 @@ async def test_websocket_end_to_end(): queue = EventQueue() session = Session() emitter = EventEmitter(queue) - protocol = WebsocketProtocol(config, queue, session, token) + protocol = WebsocketProtocol(config, session, token) emitter.register_protocol(protocol) - connecter = WebsocketConnecter(config, http, token, queue) + connecter = WebsocketConnecter(config, http, token, session, queue) received: list[GroupAtMessage] = [] diff --git a/uv.lock b/uv.lock index b53bc2a..e76a46a 100644 --- a/uv.lock +++ b/uv.lock @@ -104,18 +104,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fb/76/641ae371508676492379f16e2fa48f4e2c11741bd63c48be4b12a6b09cba/aiosignal-1.4.0-py3-none-any.whl", hash = "sha256:053243f8b92b990551949e63930a839ff0cf0b0ebbe0597b0f3fb19e1a0fe82e", size = 7490, upload-time = "2025-07-03T22:54:42.156Z" }, ] -[[package]] -name = "apscheduler" -version = "3.11.3" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "tzlocal" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/8c/6b/eeff360196bb20b312c9e762a820fd1b2c6d809466c755ef57863478e454/apscheduler-3.11.3.tar.gz", hash = "sha256:cd2fcc9330039a81a5893472ad49facf23a6d5604cbe1d918c835c6de7834d5a", size = 110312, upload-time = "2026-06-28T19:39:22.493Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/42/c9/8638db32514dbb9157b3d82680c6faea89283523edf9ed2415ea3884f2ae/apscheduler-3.11.3-py3-none-any.whl", hash = "sha256:bbeb2ec02d23d3c06a6c07ed7f0f3939ada6680eb121fae809a69bb42c537a30", size = 66024, upload-time = "2026-06-28T19:39:20.982Z" }, -] - [[package]] name = "attrs" version = "26.1.0" @@ -632,7 +620,6 @@ version = "0.1.0" source = { editable = "." } dependencies = [ { name = "aiohttp" }, - { name = "apscheduler" }, { name = "cryptography" }, { name = "loguru" }, { name = "pyee" }, @@ -651,7 +638,6 @@ dev = [ [package.metadata] requires-dist = [ { name = "aiohttp", specifier = ">=3.14.3" }, - { name = "apscheduler", specifier = ">=3.11.3" }, { name = "cryptography", specifier = ">=50.0.1" }, { name = "loguru", specifier = ">=0.7.3" }, { name = "pyee", specifier = ">=14.0.0" }, @@ -701,27 +687,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/49/d3/b8441a820a491ddfc024b0b0cf0393375b75ea13866d9c66727e54c2fc80/typing_extensions-4.16.0-py3-none-any.whl", hash = "sha256:481caa481374e813c1b176ada14e97f1f67a4539ce9cfeb3f350d78d6370c2e8", size = 45571, upload-time = "2026-07-02T08:40:04.659Z" }, ] -[[package]] -name = "tzdata" -version = "2026.3" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/92/ff/5a28bdfd8c3ebec42564ac7d0e54ca3db65044a9314a97f9564fa7a1e926/tzdata-2026.3.tar.gz", hash = "sha256:4a1518b8993086a7982523e071643f3c0e5f213e75b21318e78bcabfff9d1415", size = 198674, upload-time = "2026-07-10T08:50:37.887Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/e5/6d/b53b99a9f2766d095985947a5782f1702cabb129a34f7a802d7197af832f/tzdata-2026.3-py2.py3-none-any.whl", hash = "sha256:dc096730c87af6cab1b171c9d532be840741ff5d459015e7f6947bd7d7e54931", size = 348168, upload-time = "2026-07-10T08:50:36.46Z" }, -] - -[[package]] -name = "tzlocal" -version = "5.4.4" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "tzdata", marker = "sys_platform == 'win32'" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/81/5b/879b2f932adfa7a053c360d50bc896c977fa6426109185f7c12ebdd0cb9d/tzlocal-5.4.4.tar.gz", hash = "sha256:8dbb8660838688a7b6ba4fed31d18dedf842afb4d47ca050d6d891c2c15f3be4", size = 31170, upload-time = "2026-06-29T08:03:40.026Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/9e/a4/017a7a6cbe387d961a688ec31364ae60a5c4e22c96ae9921b79a947c855d/tzlocal-5.4.4-py3-none-any.whl", hash = "sha256:aae09f0126a8a86fa736be266eb4a471380d26a0de3bc14844e7821fee3e2a15", size = 18115, upload-time = "2026-06-29T08:03:38.666Z" }, -] - [[package]] name = "ujson" version = "6.0.0"