Metadata-Version: 2.4
Name: qqbot-sdk
Version: 0.1.0
Summary: 现代化、人性化的腾讯 QQ 机器人 Python SDK (API v2)
Author: FlyCloud
License: MIT
Keywords: qq,bot,qqbot,tencent,sdk,async
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.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Framework :: AsyncIO
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: aiohttp>=3.9
Requires-Dist: cryptography>=41
Provides-Extra: dev
Requires-Dist: pytest>=7; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Dynamic: license-file

# qqbot-sdk

现代化、生产级的腾讯 QQ 机器人 Python **框架**,基于官方 [OpenAPI v2](https://bot.q.qq.com/wiki/develop/api-v2),面向 QQ 群机器人优化。

设计上博采众长:[botpy](https://github.com/tencent-connect/botpy)(官方 SDK,长期未维护)的接口对照、[NoneBot2](https://github.com/nonebot/nonebot2) 的规则/优先级/插件理念、[mirai](https://github.com/mamoe/mirai) 的事件与会话模型,并原生支持 **OneBot v12** 协议输出。

## 亮点一览

| 维度 | 能力 |
|---|---|
| 接口覆盖 | 官方 OpenAPI v2 全量:四场景消息、富媒体、频道/成员/身份组、禁言、公告、精华、表态、权限、日程、论坛、音频、API 权限申请 |
| 连接模式 | WebSocket(心跳/重连/Resume/多分片)+ Webhook(Ed25519 验签)可**同时多路运行**,自定义连接器可扩展 |
| 生态兼容 | 内置 **OneBot v12** 适配器(正向 WS + HTTP),NoneBot2 / Koishi 可直连 |
| 事件系统 | 洋葱模型**中间件**、声明式**规则**(`&`/`\|`/`~` 组合)、**优先级 + 阻断**、`on_any` 旁路监听 |
| 会话交互 | `msg.ask()` 一行实现多轮对话、`bot.wait_for()` 任意事件等待 |
| 插件系统 | 目录自动发现、运行时加载/卸载/**热重载**、独立存储命名空间、on_load/on_unload 钩子 |
| 指令系统 | 类型化参数(注解自动转换)、冷却、权限检查器、用法提示、自动 /帮助 |
| 数据管理 | 统一异步 KV 抽象:内存 / JSON / SQLite 三后端,会话状态 `bot.sessions` 按群+用户隔离 |
| 定时任务 | `@bot.schedule.every(3600)` 间隔任务 + 五段式 cron |
| 可观测性 | `/healthz` `/stats` `/metrics`(Prometheus)、事件/API/延迟指标、trace_id 透出 |
| 稳定性 | token 自动刷新、401 重试、出站令牌桶限流、网络错误退避重试、事件异常隔离、优雅停机(等在途任务) |

## 安装

```bash
pip install aiohttp cryptography
pip install -e .          # 本仓库根目录
```

Python ≥ 3.9。

## 三分钟上手

```python
import qqbot

bot = qqbot.Bot("你的AppID", "你的AppSecret")

@bot.on_message                  # QQ单聊 + 群@ + 频道@ 一网打尽
async def echo(msg: qqbot.Message):
    await msg.reply(f"你说了: {msg.text}")

bot.run()                        # WebSocket 模式,本地零配置可跑
```

`msg.reply()` 统一四种场景(单聊/群/频道/频道私信),自动携带 msg_id 走被动通道、自动递增 msg_seq;`reply(image=URL或路径或bytes)` 时群/单聊自动先走富媒体上传。

## 核心概念

### 连接模式(可多路并存)

```python
bot.add_websocket()                                   # 开发:零配置
bot.add_webhook(port=8443, max_timestamp_skew=300)    # 生产:官方推荐,自动验签+防重放
bot.add_connector(OneBotV12Adapter(port=6700))        # 生态:OneBot v12 输出
bot.run()
```

Webhook 的回调地址验证(op=13)由框架自动应答;WebSocket 的 intents 根据你注册的处理器**自动推导**。

### 规则 / 优先级 / 阻断(NoneBot2 风格)

```python
from qqbot import rules

@bot.on_group_message(rule=rules.keyword("天气") & rules.scene("group"))
async def weather(msg): ...

@bot.on_message(rule=rules.regex(r"^签到$"), priority=1, block=True)
async def sign(msg): ...          # 数字越小越先执行;block 阻断后续处理器
```

内置规则:`keyword / startswith / endswith / fullmatch / regex / scene / from_user / from_group / has_image`,支持 `&` `|` `~` 组合,自定义任何 `(data) -> bool` 皆可。

### 中间件(洋葱模型)

```python
@bot.middleware
async def audit(ctx: qqbot.Context, call_next):
    if ctx.message and is_spam(ctx.message.content):
        return                    # 不调 call_next => 拦截
    await call_next()
```

### 指令系统

```python
@bot.command("骰子", usage="/骰子 [次数] [面数]", cooldown=5)
async def roll(msg, times: int = 1, sides: int = 6):    # 注解自动转换类型
    ...

from qqbot import permissions as perm
@bot.command("重启", permission=perm.owner_only)         # 权限检查器
async def restart(msg): ...

bot.enable_help()                                        # 自动生成 /帮助
```

参数解析规则:无额外参数→只传 msg;单个无默认值 str→整段文本;多参数→按空格分词并按注解转换(int/float/bool);keyword-only str→吃掉剩余整段;解析失败自动回复用法。

### 会话式交互

```python
@bot.command("点餐")
async def order(msg):
    food = await msg.ask("想吃什么?")                    # 等同一用户下一条消息
    count = await food.ask(f"{food.text},要几份?")
    await count.reply("已下单 ✅")
```

### 插件

```python
# plugins/greet.py
from qqbot import Plugin
plugin = Plugin("greet", version="1.0")

@plugin.command("hello")
async def hello(msg): await msg.reply("你好!")

# main.py
bot.load_plugins("plugins")          # 自动发现
await bot.reload_plugin("greet")     # 热重载,改代码不重启
```

### 状态与数据

```python
bot = qqbot.Bot(appid, secret, storage=qqbot.SQLiteStore("bot.db"))

await bot.storage.set("global_key", {"any": "json"})
store = bot.storage.namespace("my_plugin")               # 插件隔离

async with bot.sessions.open(msg) as sess:               # 群+用户 维度会话状态
    sess["count"] = sess.get("count", 0) + 1             # 退出 with 自动保存
```

### 定时任务

```python
@bot.schedule.every(3600)            # 每小时
async def hourly(): ...

@bot.schedule.cron("0 8 * * *")      # 每天 8:00(分 时 日 月 周;周: 0=周一)
async def morning(): ...
```

### OneBot v12 桥接

```python
from qqbot.ext.onebot import OneBotV12Adapter
bot.add_websocket()
bot.add_connector(OneBotV12Adapter(port=6700, access_token="token"))
# NoneBot2 等以 ws://127.0.0.1:6700/onebot/v12 正向连接
```

选择 v12 而非 v11:v12 的 ID 为字符串,与官方 openid 无损兼容。已实现 `send_message / delete_message / get_self_info / get_status / get_version / get_supported_actions`,消息段 `text / image / mention / reply`,群管理事件以 `qq.*` 扩展 notice 下发。

### 可观测性 & 生产化

```python
bot = qqbot.Bot(appid, secret,
                rate_limit=20,        # 出站令牌桶限流(req/s)
                api_retries=2)        # 网络错误退避重试
bot.enable_observability(port=9100)   # /healthz /stats /metrics(Prometheus)

@bot.on_startup
async def up(b): ...
@bot.on_shutdown
async def down(b): ...                # SIGTERM/Ctrl+C 触发优雅停机
```

多账号:`qqbot.run_bots(bot1, bot2)`。

## 示例索引

| 文件 | 内容 |
|---|---|
| `examples/01_echo_websocket.py` | 最小复读机 |
| `examples/02_echo_webhook.py` | Webhook 生产部署 |
| `examples/03_commands.py` | 指令系统基础 |
| `examples/04_rich_content.py` | 图片/按钮/Markdown/Ark/按钮回调 |
| `examples/05_api_only.py` | 纯 API 主动推送 |
| `examples/06_middleware_rules.py` | 中间件 + 规则 + 优先级 |
| `examples/07_conversation.py` | 多轮对话 + 会话状态 |
| `examples/08_plugins/` | 插件化项目结构 + 热重载 |
| `examples/09_scheduler_production.py` | 定时任务 + 生产级配置 |
| `examples/10_onebot_bridge.py` | OneBot v12 桥接 NoneBot2 |

## 测试

```bash
pip install pytest pytest-asyncio
pytest tests            # 49 项:签名/构建器/模型/分发/规则/中间件/会话/存储/调度/插件/OneBot
```

## 项目结构

```
qqbot/
├── bot.py            # 总装:事件/中间件/指令/插件/会话/连接器/生命周期
├── api.py            # OpenAPI 全量封装
├── connection.py     # 连接器抽象(WebSocket / Webhook / 自定义)
├── gateway.py        # WebSocket 网关(心跳/重连/Resume/分片)
├── webhook.py        # Webhook 服务器(Ed25519 验签 + 防重放)
├── middleware.py     # 洋葱模型中间件
├── rules.py          # 声明式规则组合子
├── commands.py       # 增强指令(类型化参数/冷却/权限)
├── permissions.py    # 权限检查器
├── plugin.py         # 插件系统(热重载)
├── storage.py        # KV 存储抽象 + 会话管理(内存/JSON/SQLite)
├── scheduler.py      # 定时任务(interval + cron)
├── observability.py  # 指标 + 健康检查/Prometheus
├── limiter.py        # 令牌桶限流 / 冷却
├── models.py         # 统一消息模型 + 事件模型
├── event.py          # 事件定义、intents 映射、payload 解析
├── builders.py       # Markdown/Keyboard/Ark/Embed 构建器
├── http.py           # 鉴权 HTTP(401 重试/限流/重试/指标)
├── token.py          # access_token 自动管理
└── ext/
    └── onebot.py     # OneBot v12 适配器
```

## 已知边界

- 主动消息受平台配额限制(每目标每月 4 条),被动回复不受限,优先 `reply()` / `event.reply()`
- 私域专属能力(频道全量消息、论坛事件)公域机器人订阅会鉴权失败,框架已在自动推导时规避
- OneBot 适配器覆盖常用动作;未覆盖的平台能力可在同进程内用原生 API 补齐

## License

MIT
