Metadata-Version: 2.3
Name: qq-botpy-sdk
Version: 2.1.1
Summary: Community-maintained async Python SDK for the QQ Bot Open Platform
Author: Teahouse Studios
License: MIT License
         
         Copyright (c) 2021 Tencent
         
         Permission is hereby granted, free of charge, to any person obtaining a copy
         of this software and associated documentation files (the "Software"), to deal
         in the Software without restriction, including without limitation the rights
         to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
         copies of the Software, and to permit persons to whom the Software is
         furnished to do so, subject to the following conditions:
         
         The above copyright notice and this permission notice shall be included in all
         copies or substantial portions of the Software.
         
         THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
         IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
         FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
         AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
         LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
         OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
         SOFTWARE.
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Dist: httpx>=0.28.1,<1
Requires-Dist: httpx-ws>=0.7.2,<1
Requires-Dist: apscheduler>=3.11.3,<4
Requires-Dist: cryptography>=50.0.0,<51
Requires-Dist: pyyaml>=6.0.3,<7
Requires-Python: >=3.10, <4.0
Project-URL: Homepage, https://github.com/Teahouse-Studios/qq-botpy-sdk
Project-URL: Repository, https://github.com/Teahouse-Studios/qq-botpy-sdk
Project-URL: Issues, https://github.com/Teahouse-Studios/qq-botpy-sdk/issues
Description-Content-Type: text/markdown

# qq-botpy-sdk

基于 QQ 机器人开放平台的异步 Python SDK 与机器人开发框架。

[![Python](https://img.shields.io/badge/Python-3.10%2B-blue)](https://www.python.org/)
[![License](https://img.shields.io/badge/License-MIT-green)](./LICENSE)
[![PyPI](https://img.shields.io/pypi/v/qq-botpy-sdk)](https://pypi.org/project/qq-botpy-sdk/)
[![Repository](https://img.shields.io/badge/GitHub-Teahouse--Studios%2Fqq--botpy--sdk-181717?logo=github)](https://github.com/Teahouse-Studios/qq-botpy-sdk)
[![Status](https://img.shields.io/badge/status-community%20maintained-orange)](#项目状态)

## 项目状态

本仓库最初 fork 自腾讯维护的
[tencent-connect/botpy](https://github.com/tencent-connect/botpy)，现已转为由 Teahouse Studios 独立维护的
社区项目。PyPI 发布包名为 `qq-botpy-sdk`，Python 导入名继续保持为 `botpy`。

请特别注意：

- 本项目不是腾讯官方 SDK，也不代表腾讯或 QQ 机器人开放平台。
- 问题反馈、功能请求和代码贡献请提交到
  [Teahouse-Studios/qq-botpy-sdk](https://github.com/Teahouse-Studios/qq-botpy-sdk)，不要提交给历史上游。
- 独立维护期间会继续整理版本策略和公开 API，升级前请阅读 [迁移指南](./MIGRATION.md)。
- QQ、QQ 机器人及相关名称和商标归其权利人所有。

## 为什么维护这个分支

维护这个项目的直接原因是历史上游已经长期缺乏维护，同时原框架在连接稳定性、协议完整性、资源生命周期、
错误处理和现代群聊/C2C 场景中存在一系列问题，无法仅靠零散补丁解决。因此，本项目选择独立维护并对协议层、
高层 API 和工程结构进行系统改造，重点解决：

- Gateway 心跳 ACK、Resume、关闭码分类和稳定重连。
- 长时间运行时的 Token 生命周期和网络资源释放。
- C2C、群聊、频道、频道私信的统一事件和消息发送模型。
- WebSocket 与 Webhook 双事件传输。
- 中间件、持久化、媒体分片上传、流式消息等生产能力。
- 结构化错误、非幂等请求安全和可测试的协议组件。

## 主要能力

### Gateway 与传输

- 按服务端 `heartbeat_interval` 发送心跳并携带最新序列号。
- 检测 Opcode 11 Heartbeat ACK，超时后主动恢复连接。
- 支持 Identify、Resume、服务端重连指令和 Invalid Session。
- 按关闭码选择刷新 Token、清理 Session、退避重连或停止。
- 重连期间暂停消息发送，待所有分片 READY/RESUMED 后继续。
- 支持 JSON 文件或自定义 Store 持久化 Gateway Session。
- 支持 WebSocket、内置 Webhook 服务和自定义事件传输适配器。
- Webhook 支持回调地址验证、Ed25519 验签和事件 ACK。

### HTTP 与鉴权

- 并发安全的 access token 缓存、提前刷新和后台刷新。
- 401 强制刷新一次 Token 后重试。
- 结构化 API、认证、限流和传输异常。
- 解析 `Retry-After`，安全方法支持指数退避。
- 按平台 `err_code` 处理消息发送失败：回退主动消息、换 `msg_seq` 重发、指数退避重发，
  见[消息发送失败处理](#消息发送失败处理)。
- 出站请求整流：按官方文档核对后的接口频率限制排队限速，见[请求整流与限速](#请求整流与限速)。
- POST/PATCH 默认不自动重试，避免非幂等消息重复发送。
- C2C/群聊仅带 `msg_id` 的被动回复会对不确定的传输失败重试一次。
- 可配置 API 地址、Token 地址、User-Agent 和 SSLContext/私有 CA。
- 通过 `client.api.request/get/post/put/patch/delete()` 调用尚未封装的 REST API。

REST API 默认调用 `https://api.bot.qq.com`；沙盒兼容参数不会切换到其他域名。可通过
`Client(base_url="https://your-api.example.com")` 配置自定义调用域名；如果 token 服务使用独立网关，
可同时配置 `token_base_url`。

### 消息、媒体与 Interaction

- 统一的 `RawEvent`、`InboundMessage`、`ReplyTarget` 和 `InteractionContext`。
- 统一发送文本、Markdown、Ark、Embed、Keyboard 和媒体消息。
- 长文本超过 5000 字符时自动切分。
- 被动回复次数或时间窗口超限后自动转主动消息。
- 支持 Typing、Wakeup、消息撤回和出站 `ref_idx` hook。
- 支持图片、视频、语音、普通文件的 URL、Base64、bytes 和本地文件来源。
- 5 MiB 以上自动切换分片上传，支持哈希、并发、进度回调和业务错误重试。
- 按内容、目标和媒体类型缓存服务端 `file_info`。
- 支持 C2C replace-mode 流式消息。
- 支持群信息、入群申请审批、成员禁言与自动审批策略管理。
- 支持声明式全局菜单和多指令面板，并在启动时同步平台配置。
- 提供高层 Interaction ACK。

### 中间件与存储

- 消息过滤、内容清洗、访问策略、限流、并发保护和 Mention Gate。
- Slash Command、Typing Indicator、错误处理和恢复中间件。
- History Buffer、Quote Ref 和 LLM Envelope。
- 通用异步 KV Store，以及 Session、History、Ref Index 适配器。

### 工程集成

- 保留历史 botpy 事件回调，便于渐进迁移。
- 支持标准库 logging 和 Loguru。
- 支持自定义 Webhook Server、Session Store、KV Store 和中间件。
- 核心协议组件可独立测试和使用。

## 环境要求

- Python 3.10 或更高版本。
- QQ 机器人开放平台提供的 AppID 和 AppSecret。
- 使用 WebSocket 时，运行环境需要能够访问 QQ Gateway 和 REST API。
- 使用 Webhook 时，需要平台可访问的公网 HTTPS 回调地址。

## 安装

### 从 PyPI 安装

```bash
pip install --upgrade qq-botpy-sdk
```

安装包名是 `qq-botpy-sdk`，Python 导入名仍然是：

```python
import botpy
```

### 从当前仓库安装

需要使用尚未发布的最新代码时，可以直接从仓库安装：

```bash
pip install "git+https://github.com/Teahouse-Studios/qq-botpy-sdk.git"
```

uv 项目可以使用：

```bash
uv add "qq-botpy-sdk @ git+https://github.com/Teahouse-Studios/qq-botpy-sdk.git"
```

> `qq-botpy-sdk` 是当前独立维护版的发布包名。历史 `qq-botpy` 包属于不同的发布来源，不代表本仓库。

### 本地开发安装

```bash
git clone https://github.com/Teahouse-Studios/qq-botpy-sdk.git
cd qq-botpy-sdk
uv sync
```

## 快速开始

下面的示例使用统一消息回调，可同时处理标准化后的 C2C、群聊、频道和频道私信消息：

```python
import os

import botpy


class MyClient(botpy.Client):
    async def on_message_context(self, context):
        message = context.message
        print(message.event_type, message.author_id, message.content)
        await self.send_text(
            message.reply_target,
            f"收到：{message.content}",
        )

    async def on_ready(self):
        print(f"机器人 {self.robot.name} 已连接")


intents = botpy.Intents(
    public_messages=True,
    public_guild_messages=True,
    direct_message=True,
    interaction=True,
)

client = MyClient(intents=intents)
client.run(
    appid=os.environ["QQBOT_APP_ID"],
    secret=os.environ["QQBOT_APP_SECRET"],
)
```

历史事件回调仍然可用，例如 `on_at_message_create`、`on_c2c_message_create`、
`on_group_at_message_create` 和 `on_interaction_create`。新项目建议优先使用统一消息与原始事件回调，
旧项目可以渐进迁移。

## 常用配置

### 持久化 Gateway Session

```python
from botpy.protocol import JsonFileSessionStore

client = MyClient(
    intents=intents,
    session_store=JsonFileSessionStore("./.botpy-sessions"),
)
```

进程重启后，客户端会优先尝试 Resume，并从保存的序列号之后补发事件。失效、过期或分片数量不匹配的
Session 会被自动清理。

### 重连期间的消息发送

```python
client = MyClient(
    intents=intents,
    gateway_send_timeout=30.0,
)
```

WebSocket 模式下，Gateway 正在连接或重连时，消息会等待所有已注册分片恢复，不会立即发出 HTTP
请求。`gateway_send_timeout` 默认为 30 秒；设为 `None` 可一直等待，设为 `0` 则立即失败。首次请求前
等待超时时，`TransportError.attempts` 为 `0`；若重试等待超时，则保留此前实际请求次数。

### 请求整流与限速

QQ 开放平台为每个 REST 接口声明了 `接口频率限制`，消息类接口还会叠加「主动消息」的 Bot 维度、
单关系维度和每日额度限制。SDK 默认启用**队列化令牌桶**：同一通道上的请求先到先发，按配额均匀
发出，收到 `429` 时按 `Retry-After` 让整个通道一起退避，避免瞬时并发撞限流。

默认限制表来自官方文档逐条核对（详见 [docs/RATE_LIMITS.md](docs/RATE_LIMITS.md)，核对脚本
`generated_docs/fetch_docs.py` + `generated_docs/extract_rate_limits.py`）：

| 通道 | 说明 | 示例 |
| --- | --- | --- |
| `route` | 接口级 `接口频率限制` | `POST /v2/groups/{}/messages` → 100 QPS |
| `bot` | 主动消息的 Bot 维度配额，全 Bot 共享 | 已认证群聊 60 QPM、单聊 10 QPS |
| `relationship` | 主动消息的单关系配额，按群/用户/子频道分别计数 | 20 QPM + 1000 条/天 |
| `proactive` / `relationship`（子频道） | 子频道每 1s 最多 5 条，主动推送每天 20 条 | 5 QPS |

被动回复（带 `msg_id`/`event_id`）只受接口级与「始终生效」配额约束，不会消耗主动消息额度。
媒体预签名地址等非 `base_url` 主机上的传输不计入接口配额。

**对话消息优先**：同一通道内的等待队列按优先级排序，被动回复（用户正在等响应）插到队头，
主动/批量推送排在其后，避免一批推送把队列占满导致对话消息饿死。优先级只决定谁先拿到下一个
令牌，不改变整体吞吐；连续放行 4 个高优先级请求后会强制放行一个等待最久的批量请求，防止低
优先级被反向饿死（`rate_limit={"low_priority_interval": ...}`，`None` 关闭）。

并发槽位（`max_concurrency`）在取令牌**之前**获取，令牌紧挨着真正发送才被消耗，否则等待
槽位期间堆积的令牌会在槽位释放后成串发出、绕过速率约束。`max_buckets` 超出后只回收「令牌桶
已补满且惩罚已结束」的空闲通道；所有通道都还有未恢复状态时允许超过该上限，以免回收把已消耗
的平台配额静默清零。

```python
# 默认：启用整流，按文档限制排队
client = MyClient(intents=intents)

# 主动消息配额默认按「已认证」档位整流（平台没有查询认证等级的接口）：
#   已认证（默认）：单聊 10 QPS，群聊 60 QPM
#   未认证：单聊 5 QPS + 30 QPM，群聊 30 QPM
# 机器人未通过企业/个人认证时必须显式降档，否则会超出平台配额被限流。
client = MyClient(intents=intents, rate_limit={"certification": "unverified"})

# 追加全局在途请求上限，进一步压制突发
client = MyClient(intents=intents, rate_limit={"max_concurrency": 8})

# 完全关闭整流（自行保证速率）
client = MyClient(intents=intents, rate_limit=False)

# 完全自定义：注入自己的 RateLimiter
from botpy.protocol import RateLimitBudget, RateLimiter, RequestPriority, RouteRule

limiter = RateLimiter(
    route_rules={"POST /v2/groups/{}/messages": RouteRule((RateLimitBudget(20, 60.0),))},
)
client = MyClient(intents=intents, rate_limit=limiter)
```

`RateLimiter` 可调参数：`certification`（`"certified"` 默认 / `"unverified"`）、`max_concurrency`、
`route_rules`、`default_budgets`（未匹配接口的兜底配额，默认 50 QPS）、`max_buckets`、
`default_penalty`、`max_pacing_window`、`low_priority_interval`。

窗口长于 `max_pacing_window`（默认 3600 秒）的配额属于「每日额度」：用尽时 SDK 记录一条 warning
并仍然发送，由平台返回结构化错误，而不会让消息挂起数小时。可用 `limiter.snapshot()` 查看各通道
的排队深度与剩余令牌。

### 消息发送失败处理

平台把消息发送的业务错误放在响应体的 `err_code` 里，`Client.send()` 会据此决定下一步动作
（完整错误码表见 [docs/SEND_ERRORS.md](docs/SEND_ERRORS.md)）：

| 情况 | 处理 |
| --- | --- |
| `msg_id`/`event_id` 失效（`40034005`、`40034128`、`304103` 等） | 去掉被动回复上下文，回退为主动消息重发一次 |
| 消息被去重（`40054005`） | 首次发送换一个新 `msg_seq` 重发一次；仍失败直接抛错 |
| 网络超时 / 连接中断 | 按 `3 × 2ⁿ` 秒（3、6、12、24……）指数退避重发；主动消息只重发"确定没发出"的失败 |
| 已知不可达（`40034101`、`40054002`、`40054003` 等） | 立即抛出 `ApiError`，不重试 |
| 其它未收录错误码 | 同样立即抛出，绝不重复投递 |

单次逻辑发送的总时长上限为 60 秒，超时抛出 `MessageSendTimeoutError`（`TransportError`
子类，带 `attempts` / `elapsed` / `timeout`）。重发过程中再次遇到 `40054005` 会立即停止：
此时消息可能已经发出但 id 不可知。

网络失败区分「请求确定没有发出」和「已发出但结果未知」：前者任何消息都重发（不可能重复
投递），后者只有被动回复重发——主动消息没有去重保护，默认不重发以免用户收到两条。
需要旧行为时用 `MessageSendPolicy(replay_ambiguous_proactive=True)`。

```python
from botpy.protocol import MessageSendPolicy

client = MyClient(
    intents=intents,
    send_policy=MessageSendPolicy(total_timeout=60.0, backoff_base=3.0),
)
```

### Webhook 模式

```python
client = MyClient(
    intents=intents,
    transport="webhook",
    webhook_host="0.0.0.0",
    webhook_port=8080,
    webhook_path="/callback",
)
```

生产环境建议在 SDK 前使用反向代理终止 TLS，并将平台回调地址配置为公网 HTTPS URL。
Webhook 事件默认只接受时间戳误差不超过 300 秒的签名，并在进程内拒绝相同签名的重复投递；
如需兼容自定义时钟或测试环境，可直接构造 `WebhookTransport` 并设置 `signature_max_age`。

### 中间件

```python
from botpy.middleware import (
    RateLimitTier,
    ScopePolicy,
    access_policy,
    concurrency_guard,
    content_sanitizer,
    rate_limiter,
)

client.use(
    content_sanitizer(collapse_whitespace=True, parse_face_tags=True),
    access_policy(group=ScopePolicy(mode="allowlist", allow=("group-openid",))),
    rate_limiter(per_sender=RateLimitTier(max_requests=5, window_seconds=60)),
    concurrency_guard(strategy="queue", max_queue=3),
)
```

### 发送 Markdown 和媒体

```python
await client.send_markdown(
    context.reply_target,
    "# Markdown 消息",
)

result = await client.send_image(
    context.reply_target,
    local_path="./image.png",
    content="图片说明",
)

print(result.upload["file_info"], result.message["id"])

# Markdown 图片等场景需要 URL：upload_media_url 强制分片上传并返回临时直链
uploaded = await client.upload_media_url(
    context.reply_target,
    MediaFileType.IMAGE,
    local_path="./image.png",
)

print(uploaded.raw_url, uploaded.ttl)
```

### C2C 流式消息

```python
stream = client.open_stream(context.reply_target, throttle_ms=500)
full_text = ""

try:
    async for token in model_stream():
        full_text += token
        await stream.update(full_text)
    await stream.complete()
except Exception:
    stream.cancel()
    raise
```

流式消息只支持携带入站 `message_id` 的 C2C 目标。

### 调用未封装的 REST API

```python
guilds = await client.api.get(
    "/users/@me/guilds",
    params={"limit": 100},
)

token = await client.api.get_token()
```

### 自定义菜单与指令面板

```python
from botpy.configuration import Menu, Panel

client = MyClient(
    intents=intents,
    menu=Menu(
        items=[
            Menu.message("帮助", "/help"),
            Menu.link("官网", "https://example.com"),
        ]
    ),
    panels=[
        Panel(
            "main",
            scope="c2c",
            items=[Panel.command("查询天气", desc="查询当前天气")],
        )
    ],
    config_sync_strict=False,
)
```

SDK 在客户端启动后同步声明的单个全局菜单和多个受管面板。多副本部署、strict 模式和非破坏性同步边界见
[自定义菜单与指令面板](./docs/MENU_PANEL.md)。

### 使用 Loguru

```python
import sys

import botpy
from loguru import logger

logger.remove()
logger.configure(extra={"stdlib_logger": "app"})
logger.add(
    sys.stderr,
    enqueue=True,
    format="{time} | {level} | {extra[stdlib_logger]} | {message}",
)

client = MyClient(
    intents=intents,
    loguru_logger=logger,
)
```

完整说明见 [Loguru 配置指南](./docs/LOGURU.md)。

## 文档

| 文档 | 内容 |
| --- | --- |
| [API 参考](./docs/API.md) | 新高层接口、统一发送、媒体、REST 和 Interaction |
| [自定义菜单与指令面板](./docs/MENU_PANEL.md) | 声明式 Menu/Panel、启动同步和多副本部署风险 |
| [群管理 API 与事件](./docs/GROUP_MANAGEMENT.md) | 群信息、入群审批、成员禁言、自动审批策略和群成员事件 |
| [请求整流与接口限速](./docs/RATE_LIMITS.md) | 通道划分、文档核对后的接口配额表和整流配置 |
| [消息发送失败处理](./docs/SEND_ERRORS.md) | `err_code` 分类、回退主动消息、换 seq 与指数退避规则 |
| [迁移指南](./MIGRATION.md) | 协议层改造后的行为变化与不兼容项 |
| [Loguru 配置指南](./docs/LOGURU.md) | 日志桥接、轮转、结构化字段和根 logger 接管 |
| [发布指南](./docs/RELEASING.md) | GitHub Release、版本校验和 PyPI Trusted Publishing |
| [事件监听](./docs/事件监听.md) | 历史事件回调列表 |
| [示例目录](./examples/README.md) | 频道、群聊、C2C、媒体和管理事件示例 |

## 示例

[`examples`](./examples/) 目录保留了历史 API 示例，并增加了现代协议层组合示例：

- `demo_modern_client.py`：统一消息、Session、Interaction、缓存和限流。
- `demo_group_reply_text.py` / `demo_group_reply_file.py`：群聊消息与媒体。
- `demo_c2c_reply_text.py` / `demo_c2c_reply_file.py`：C2C 消息与媒体。
- `demo_at_reply.py`：频道 AT 消息。
- `demo_dms_reply.py`：频道私信。
- `demo_recall.py`：消息撤回。

部分历史示例直接读取 `examples/config.yaml`。不要提交包含真实 AppSecret 的配置文件。

## 开发与测试

安装依赖：

```bash
uv sync
```

运行纯本地测试：

```bash
uv run python -m unittest discover -s tests -p "test_[!a]*.py"
```

`tests/test_api.py` 使用真实平台凭证，并包含创建、修改或删除线上资源的 API 测试。除非你明确准备了隔离的
测试机器人和资源，否则不要运行该文件。

编译和项目元数据检查：

```bash
uv run python -m compileall -q botpy examples
uv lock --check
uv build
```

## 当前限制

- Python 版没有捆绑 Node 参考实现中的 `silk-wasm` 或 MP3 WASM 解码器；提供原生格式直传、
  PCM/WAV 工具和可选系统 FFmpeg 支持。
- 内置 JSON Session/KV Store 面向单进程轻量部署；多进程部署应接入 Redis、SQL 或其他共享存储。
- QQ 开放平台仍可能增加事件和字段；未知事件可通过 `on_raw_event` 获取，未知发送字段可通过
  `Client.send(..., extra={...})` 透传。

## 来源与许可证

本项目派生自 [tencent-connect/botpy](https://github.com/tencent-connect/botpy)。感谢原项目维护者和历史
贡献者提供的基础实现。

本仓库继续采用 [MIT License](./LICENSE)。原始版权声明保留在许可证文件和相关源文件中。
