* 滚木

* feat: 重构核心架构,增强类型安全与插件管理

本次提交对核心模块进行了深度重构,引入 Pydantic 增强配置管理的类型安全性,并全面优化了插件管理系统。

主要变更详情:

1. 核心架构与配置
   - 重构配置加载模块:引入 Pydantic 模型 (`core/config_models.py`),提供严格的配置项类型检查、验证及默认值管理。
   - 统一模块结构:规范化模块导入路径,移除冗余的 `__init__.py` 文件,提升项目结构的清晰度。
   - 性能优化:集成 Redis 缓存支持 (`RedisManager`),有效降低高频 API 调用开销,提升响应速度。

2. 插件系统升级
   - 实现热重载机制:新增插件文件变更监听功能,支持开发过程中自动重载插件,提升开发效率。
   - 优化生命周期管理:改进插件加载与卸载逻辑,支持精确卸载指定插件及其关联的命令、事件处理器和定时任务。

3. 功能特性增强
   - 新增媒体 API:引入 `MediaAPI` 模块,封装图片、语音等富媒体资源的获取与处理接口。
   - 完善权限体系:重构权限管理系统,实现管理员与操作员的分级控制,支持更细粒度的命令权限校验。

4. 代码质量与稳定性
   - 全面类型修复:解决 `mypy` 静态类型检查发现的大量类型错误(包括 `CommandManager`、`EventFactory` 及 `Bot` API 签名不匹配问题)。
   - 增强错误处理:优化消息处理管道的异常捕获机制,完善关键路径的日志记录,提升系统运行稳定性。

* feat: 添加测试用例并优化代码结构

refactor(permission_manager): 调整初始化顺序和逻辑
fix(admin_manager): 修复初始化逻辑和目录创建问题
feat(ws): 优化Bot实例初始化条件
feat(message): 增强MessageSegment功能并添加测试
feat(events): 支持字符串格式的消息解析
test: 添加核心功能测试用例
refactor(plugin_manager): 改进插件路径处理
style: 清理无用导入和代码
chore: 更新依赖项

* refactor(handler): 移除TYPE_CHECKING并直接导入Bot类

简化类型注解,直接导入Bot类而非使用TYPE_CHECKING条件导入,提高代码可读性和维护性

* fix(command_manager): 修复插件卸载时元信息移除不精确的问题

修复 CommandManager 中 unload_plugin 方法移除插件元信息时使用 startswith 导致可能误删其他插件的问题,改为精确匹配
同时调整相关测试用例验证精确匹配行为

* refactor: 清理未使用的导入和更新文档结构

docs: 添加config_models.py到项目结构文档
docs: 调整数据目录位置到core/data下
docs: 更新权限管理器文档描述

* 文档更新

* 更新thpic插件 支持一次返回多张图

* feat: 添加测试覆盖率并修复相关问题

refactor(redis_manager): 移除冗余的ConnectionError处理
refactor(event_handler): 优化Bot类型注解
refactor(factory): 移除未使用的GroupCardNoticeEvent

test: 添加全面的单元测试覆盖
- 添加test_import.py测试模块导入
- 添加test_debug.py测试插件加载调试
- 添加test_plugin_error.py测试错误处理
- 添加test_config_loader.py测试配置加载
- 添加test_redis_manager.py测试Redis管理
- 添加test_bot.py测试Bot功能
- 扩展test_models.py测试消息模型
- 添加test_plugin_manager_coverage.py测试插件管理
- 添加test_executor.py测试代码执行器
- 添加test_ws.py测试WebSocket
- 添加test_api.py测试API接口
- 添加test_core_managers.py测试核心管理模块

fix(plugin_manager): 修复插件加载日志变量问题

覆盖率已到达86%(忽略插件)

* 更新/help指令,现在会发送图片

* feat(help): 重构帮助系统为图片渲染模式

添加浏览器管理器和图片管理器,用于通过 Playwright 渲染帮助菜单为图片
重构命令管理器以支持图片缓存和同步功能
添加 HTML 模板用于帮助菜单渲染

* build: 更新依赖文件 requirements.txt

* build: 更新依赖文件

* feat: 添加性能优化和架构文档,更新依赖和核心模块

refactor(browser_manager): 实现页面池机制以提升性能
refactor(image_manager): 添加模板缓存并集成页面池
refactor(bili_parser): 迁移到异步HTTP请求并实现会话复用
docs: 新增性能优化、架构设计和最佳实践文档
chore: 更新requirements.txt添加新依赖

* docs: 更新文档内容并优化语言风格

重构所有文档内容,使用更简洁直接的语言风格
更新架构、插件开发、部署等核心文档
优化代码示例和图表说明
统一术语和格式规范

* docs: 更新文档内容,简化语言并修正格式

- 简化插件开发指南中的描述,移除冗余内容
- 调整部署文档中的Python版本说明
- 优化最佳实践文档的措辞和格式
- 更新性能优化文档,删除不准确的数据
- 重构核心概念文档,使用更简洁的语言
- 修正README中的项目描述和技术栈说明
- 更新快速上手文档,简化安装步骤
- 调整事件流转文档的描述方式
- 简化架构文档内容
- 更新指令处理文档,添加参数注入示例
- 优化单例管理器文档的表述

* refactor(core): 优化权限管理和事件模型

- 重构 AdminManager 和 PermissionManager 以 Redis 为主要数据源
- 为所有事件模型添加 slots=True 提升性能
- 更新文档说明 Mypyc 编译注意事项
- 清理测试和调试文件
- 移动静态资源到 web_static 目录

* feat: 添加模块编译脚本和导出依赖功能

refactor(events): 移除数据类的slots参数以提升兼容性
build: 更新requirements.txt依赖列表

* docs: 更新性能优化文档并修复命令管理器帮助输出

更新性能优化相关文档,详细说明 Python 3.14 JIT 编译器的使用方法和原理,补充与 Mypyc 的互补策略。同时修复命令管理器中帮助信息的输出方式,移除图片发送仅保留文本输出。

调整部署文档结构,明确两种性能优化方案(AOT 和 JIT)的配置方法和适用场景。完善架构文档中关于 JIT 的原理和启用方式说明。

* feat(help): 重构帮助菜单界面并优化样式

refactor(bili_parser): 修复 API 响应 content-type 问题
fix(command_manager): 添加帮助图片获取的错误处理
docs(deployment): 简化部署文档并移除 JIT 相关内容

* feat: 新增自动同意请求插件和API文档

docs: 更新文档结构和内容

* refactor(scripts): 重构并优化脚本文件结构

feat(scripts): 添加Python环境检查脚本
feat(scripts): 增强依赖导出脚本功能
perf(plugins/bili_parser): 优化B站解析器性能和代码结构
style(plugins/bili_parser): 统一代码风格和常量命名

---------

Co-authored-by: baby20162016 <2185823427@qq.com>
This commit is contained in:
镀铬酸钾
2026-01-18 21:07:01 +08:00
committed by GitHub
parent 0bb339c5be
commit ad8f7e761f
21 changed files with 2903 additions and 155 deletions

398
docs/api/account.md Normal file
View File

@@ -0,0 +1,398 @@
# 账号 API
这一页讲的是怎么管理机器人自己的账号:查看登录信息、设置在线状态、修改资料、退出登录等等。这些都是跟机器人自身相关的操作。
## 账号信息
### `get_login_info` - 获取登录信息
```python
async def get_login_info(self, no_cache: bool = False) -> LoginInfo
```
获取当前登录的机器人账号信息。默认会缓存 1 小时。
**参数:**
- `no_cache`: 是否跳过缓存,直接从服务器获取
**返回值:**
- `LoginInfo`: 登录信息对象
**示例:**
```python
info = await bot.get_login_info()
print(f"机器人QQ号: {info.user_id}")
print(f"机器人昵称: {info.nickname}")
```
`LoginInfo` 对象包含:
- `user_id`: 机器人 QQ 号
- `nickname`: 机器人昵称
### `get_version_info` - 获取版本信息
```python
async def get_version_info(self) -> VersionInfo
```
获取 OneBot v11 实现的版本信息(比如 NapCatQQ 的版本)。
**返回值:**
- `VersionInfo`: 版本信息对象
**示例:**
```python
version = await bot.get_version_info()
print(f"客户端: {version.app_name}")
print(f"版本: {version.app_version}")
print(f"OneBot 协议版本: {version.protocol_version}")
```
`VersionInfo` 对象包含:
- `app_name`: 客户端名称(如 "NapCatQQ"
- `app_version`: 客户端版本
- `protocol_version`: 支持的 OneBot 协议版本
### `get_status` - 获取运行状态
```python
async def get_status(self) -> Status
```
获取 OneBot 实现的运行状态信息。
**返回值:**
- `Status`: 状态信息对象
**示例:**
```python
status = await bot.get_status()
print(f"在线: {status.online}")
print(f"状态: {status.status}")
print(f"正常: {status.good}")
```
`Status` 对象包含:
- `online`: 是否在线
- `status`: 状态描述
- `good`: 运行是否正常
## 状态设置
### `set_self_longnick` - 设置个性签名
```python
async def set_self_longnick(self, long_nick: str) -> Dict[str, Any]
```
设置机器人账号的个性签名QQ 资料里的那个长签名)。
**参数:**
- `long_nick`: 要设置的个性签名内容
**示例:**
```python
@matcher.command("setsign")
async def handle_setsign(event: MessageEvent, args: str):
if not args:
await event.reply("需要签名内容")
return
await event.bot.set_self_longnick(args)
await event.reply("个性签名已更新")
```
### `set_online_status` - 设置在线状态
```python
async def set_online_status(self, status_code: int) -> Dict[str, Any]
```
设置机器人的在线状态(在线、离开、忙碌等)。
**参数:**
- `status_code`: 状态码
- `1`: 在线
- `2`: 离开
- `3`: 忙碌
- `4`: 请勿打扰
- `5`: 隐身
- 其他值取决于客户端支持
**示例:**
```python
# 设置为隐身
await bot.set_online_status(5)
```
### `set_diy_online_status` - 设置自定义在线状态
```python
async def set_diy_online_status(
self,
face_id: int,
face_type: int,
wording: str
) -> Dict[str, Any]
```
设置自定义的在线状态(需要客户端支持)。
**参数:**
- `face_id`: 状态表情 ID
- `face_type`: 状态表情类型
- `wording`: 状态描述文本
**示例:**
```python
# 设置为"摸鱼中"
await bot.set_diy_online_status(
face_id=100,
face_type=1,
wording="摸鱼中"
)
```
### `set_input_status` - 设置"正在输入"状态
```python
async def set_input_status(
self,
user_id: int,
event_type: int
) -> Dict[str, Any]
```
向指定用户显示"对方正在输入..."的状态提示。
**参数:**
- `user_id`: 目标用户的 QQ 号
- `event_type`: 事件类型(具体含义取决于客户端)
**示例:**
```python
# 向某个用户显示"正在输入"
await bot.set_input_status(123456, 1)
```
## 资料修改
### `set_qq_profile` - 设置个人资料
```python
async def set_qq_profile(self, **kwargs) -> Dict[str, Any]
```
设置机器人账号的个人资料。
**参数:**
- `**kwargs`: 个人资料的相关参数,具体字段请参考 OneBot v11 规范
**示例:**
```python
# 修改昵称
await bot.set_qq_profile(nickname="新的昵称")
# 修改多个字段
await bot.set_qq_profile(
nickname="新昵称",
sex="female",
age=18,
level=50
)
```
### `set_qq_avatar` - 设置头像
```python
async def set_qq_avatar(self, **kwargs) -> Dict[str, Any]
```
设置机器人账号的头像。
**参数:**
- `**kwargs`: 头像的相关参数,具体字段请参考 OneBot v11 规范
**示例:**
```python
# 设置头像(具体参数格式取决于客户端)
await bot.set_qq_avatar(file="path/to/avatar.jpg")
```
## 系统操作
### `bot_exit` - 退出登录
```python
async def bot_exit(self) -> Dict[str, Any]
```
让机器人进程退出(需要客户端支持)。谨慎使用!
**示例:**
```python
@matcher.command("shutdown", permission="admin")
async def handle_shutdown(event: MessageEvent):
await event.reply("机器人正在退出...")
await event.bot.bot_exit()
```
### `clean_cache` - 清理缓存
```python
async def clean_cache(self) -> Dict[str, Any]
```
清理 OneBot 客户端的缓存。
**示例:**
```python
@matcher.command("clearcache", permission="admin")
async def handle_clearcache(event: MessageEvent):
await event.bot.clean_cache()
await event.reply("缓存已清理")
```
### `get_clientkey` - 获取客户端密钥
```python
async def get_clientkey(self) -> Dict[str, Any]
```
获取客户端密钥(通常用于 QQ 登录相关操作)。
**返回值:**
- 包含客户端密钥的字典
## 实用示例
### 机器人状态查询插件
```python
@matcher.command("status")
async def handle_status(event: MessageEvent):
# 获取各种信息
login_info = await event.bot.get_login_info()
version_info = await event.bot.get_version_info()
status_info = await event.bot.get_status()
# 构建状态消息
msg = "🤖 机器人状态\n"
msg += f"QQ号: {login_info.user_id}\n"
msg += f"昵称: {login_info.nickname}\n"
msg += f"客户端: {version_info.app_name} v{version_info.app_version}\n"
msg += f"协议: OneBot v{version_info.protocol_version}\n"
msg += f"状态: {'在线' if status_info.online else '离线'}\n"
msg += f"运行: {'正常' if status_info.good else '异常'}"
await event.reply(msg)
```
### 自动切换状态
```python
import asyncio
from datetime import datetime
async def auto_status_scheduler(bot):
"""
定时自动切换状态
"""
while True:
now = datetime.now().hour
if 9 <= now < 18:
# 工作时间:在线
await bot.set_online_status(1)
status_text = "工作中"
elif 18 <= now < 22:
# 晚上:离开
await bot.set_online_status(2)
status_text = "休息中"
else:
# 深夜:隐身
await bot.set_online_status(5)
status_text = "睡眠模式"
# 设置个性签名
await bot.set_self_longnick(f"当前状态: {status_text} | 最后更新: {datetime.now():%H:%M}")
# 每小时更新一次
await asyncio.sleep(3600)
# 在初始化插件时启动
# (注意:这只是一个示例,实际使用需要考虑插件生命周期)
```
### 资料备份与恢复
```python
import json
@matcher.command("backupprofile", permission="admin")
async def handle_backup_profile(event: MessageEvent):
"""
备份当前资料到文件
"""
# 获取当前登录信息
login_info = await event.bot.get_login_info()
# 构建备份数据
backup_data = {
"user_id": login_info.user_id,
"nickname": login_info.nickname,
"backup_time": datetime.now().isoformat()
}
# 保存到文件
filename = f"profile_backup_{login_info.user_id}.json"
with open(filename, "w", encoding="utf-8") as f:
json.dump(backup_data, f, ensure_ascii=False, indent=2)
await event.reply(f"资料已备份到 {filename}")
@matcher.command("restoreprofile", permission="admin")
async def handle_restore_profile(event: MessageEvent, args: str):
"""
从备份恢复资料
"""
if not args:
await event.reply("需要备份文件名")
return
try:
with open(args, "r", encoding="utf-8") as f:
backup_data = json.load(f)
# 恢复资料(这里只是示例,实际可能需要更多字段)
await event.bot.set_qq_profile(
nickname=backup_data.get("nickname", "")
)
await event.reply("资料已恢复")
except Exception as e:
await event.reply(f"恢复失败: {e}")
```
## 注意事项
1. **权限**: 修改资料、退出登录等操作通常需要机器人有相应权限。
2. **频率限制**: 不要频繁修改资料或状态,可能被限制。
3. **客户端支持**: 不是所有 OneBot 客户端都支持全部 API使用前最好测试一下。
4. **谨慎操作**: `bot_exit` 会让机器人下线,谨慎使用。
## 重复的方法
`AccountAPI` 中还包含了一些与好友、群组相关的方法,这些方法在其他模块中也有定义:
- `get_stranger_info()`: 同 [好友 API](./friend.md#get_stranger_info---获取陌生人信息)
- `get_friend_list()`: 同 [好友 API](./friend.md#get_friend_list---获取好友列表)
- `get_group_list()`: 同 [群组 API](./group.md#get_group_list---获取群列表)
这些方法在 `AccountAPI` 中的实现可能略有不同(比如缓存逻辑),但功能相同。建议使用对应模块中的版本,因为那些是专门为该功能设计的。
## 下一步
- [好友 API](./friend.md): 管理好友相关功能
- [群组 API](./group.md): 管理群聊相关功能
- [消息 API](./message.md): 怎么发消息、撤回消息

130
docs/api/base.md Normal file
View File

@@ -0,0 +1,130 @@
# API 基础
这一页讲的是 NEO Bot 里 API 调用的底层原理。如果你只是写插件发消息,可以直接跳过这页,去看 [消息 API](./message.md)。
但如果你想了解背后发生了什么,或者想自己封装一些高级功能,那这里的信息会帮到你。
## API 调用流程
简单来说,当你调用 `bot.send_group_msg()` 时:
1. **你的插件**`bot.send_group_msg(123456, "hello")`
2. **Bot 类** → 把它打包成 OneBot 标准的 JSON
3. **WebSocket** → 通过 `ws.py` 发给 NapCatQQ或其他 OneBot 实现)
4. **OneBot 实现** → 收到请求,真的把消息发到 QQ 群里
5. **响应返回** → 原路返回,告诉 Bot “消息发送成功”
整个过程是异步的,所以你要用 `await`
## call_api 方法
所有 API 最终都会调用 `BaseAPI.call_api()` 方法。这是最底层的接口:
```python
async def call_api(self, action: str, params: Optional[Dict[str, Any]] = None) -> Any:
```
- `action`: API 动作名,比如 `"send_group_msg"``"get_login_info"`
- `params`: 参数字典,比如 `{"group_id": 123456, "message": "hello"}`
### 返回值
`call_api` 返回的是 OneBot 响应中的 `data` 字段。如果 API 调用失败(返回 `{"status": "failed", ...}`),它会记录一条警告日志,但**不会抛出异常**(除非网络错误)。
这样设计是为了让插件能更灵活地处理失败情况。比如:
```python
try:
result = await bot.call_api("send_group_msg", {"group_id": 123456, "message": "test"})
if result is None:
print("API 调用失败,但没抛异常")
except Exception as e:
print(f"网络或底层错误: {e}")
```
## 响应格式
OneBot v11 的标准响应格式是:
```json
{
"status": "ok",
"retcode": 0,
"data": { ... },
"message": "",
"echo": "请求时的 echo 值(如果有)"
}
```
- `status`: `"ok"``"failed"`
- `retcode`: 状态码0 表示成功
- `data`: 真正的返回数据
- `message`: 错误信息(失败时)
- `echo`: 用来匹配请求和响应的标识WebSocket 用)
NEO Bot 的 `call_api` 方法会自动提取 `data` 字段返回给你。如果 `status``"failed"`,它会在日志里记录警告,但依然返回 `data`(通常是 `None` 或空字典)。
## 错误处理
API 调用可能因为各种原因失败:
1. **网络问题**: WebSocket 断开、超时
2. **权限不足**: 机器人不是管理员却想踢人
3. **参数错误**: 群号不存在、消息太长
4. **客户端不支持**: 某些 OneBot 实现可能没实现某些 API
建议在插件里做好错误处理:
```python
@matcher.command("kick")
async def handle_kick(event: MessageEvent, args: str):
target_id = int(args) if args.isdigit() else 0
if not target_id:
await event.reply("参数错误,需要 QQ 号")
return
try:
result = await event.bot.set_group_kick(event.group_id, target_id)
if result.get("status") == "failed":
await event.reply(f"踢人失败: {result.get('message', '未知错误')}")
else:
await event.reply("踢人成功")
except Exception as e:
await event.reply(f"网络错误: {e}")
```
## 直接调用 vs 高级封装
NEO Bot 提供了两种调用 API 的方式:
### 1. 直接调用 `call_api`
```python
await bot.call_api("send_group_msg", {"group_id": 123456, "message": "hello"})
```
**什么时候用?**
- 你想调用的 API 没有被封装成独立方法(很少见)
- 你在调试,想看看原始请求和响应
- 你在写框架代码,需要动态生成 action 名
### 2. 使用封装好的方法
```python
await bot.send_group_msg(123456, "hello")
```
**这是推荐的方式**,因为:
- 有类型提示,编辑器能帮你补全
- 参数有文档,不用去查 OneBot 标准
- 有些方法有额外逻辑(比如缓存、参数转换)
## 下一步
现在你了解了 API 调用的基础。接下来可以去看看具体的 API 类别:
- [消息 API](./message.md): 最常用,先看这个
- [群组 API](./group.md): 管理群聊
- [好友 API](./friend.md): 好友相关操作
- [账号 API](./account.md): 机器人自身状态
- [媒体 API](./media.md): 图片、语音

273
docs/api/friend.md Normal file
View File

@@ -0,0 +1,273 @@
# 好友 API
这一页讲的是怎么管理好友:获取好友列表、给好友点赞、处理加好友请求,还有获取陌生人信息。
## 好友列表
### `get_friend_list` - 获取好友列表
```python
async def get_friend_list(self, no_cache: bool = False) -> List[FriendInfo]
```
获取机器人账号的所有好友列表。默认会缓存 1 小时。
**参数:**
- `no_cache`: 是否跳过缓存,直接从服务器获取最新列表
**返回值:**
- `List[FriendInfo]`: 好友信息对象列表
**示例:**
```python
friends = await bot.get_friend_list()
print(f"我有 {len(friends)} 个好友")
for friend in friends:
print(f"{friend.user_id}: {friend.nickname} (备注: {friend.remark})")
```
`FriendInfo` 对象包含以下字段:
- `user_id`: QQ 号
- `nickname`: 昵称
- `remark`: 备注(你给好友设置的备注名)
- 其他可能的信息字段
## 陌生人信息
### `get_stranger_info` - 获取陌生人信息
```python
async def get_stranger_info(
self,
user_id: int,
no_cache: bool = False
) -> StrangerInfo
```
获取非好友的 QQ 用户信息。默认会缓存 1 小时。
**参数:**
- `user_id`: 目标用户的 QQ 号
- `no_cache`: 是否跳过缓存
**返回值:**
- `StrangerInfo`: 陌生人信息对象
**示例:**
```python
@matcher.command("who")
async def handle_who(event: MessageEvent, args: str):
if not args.isdigit():
await event.reply("参数错误,需要 QQ 号")
return
target_id = int(args)
info = await event.bot.get_stranger_info(target_id)
msg = f"用户 {target_id} 的信息:\n"
msg += f"昵称: {info.nickname}\n"
msg += f"性别: {info.sex}\n"
msg += f"年龄: {info.age}\n"
msg += f"等级: {info.level}"
await event.reply(msg)
```
`StrangerInfo` 对象包含以下字段:
- `user_id`: QQ 号
- `nickname`: 昵称
- `sex`: 性别(`male`/`female`/`unknown`
- `age`: 年龄
- `level`: QQ 等级
- 其他可能的信息字段
## 互动功能
### `send_like` - 发送点赞(戳一戳)
```python
async def send_like(
self,
user_id: int,
times: int = 1
) -> Dict[str, Any]
```
给指定用户发送"戳一戳"(点赞)。每天有次数限制,建议不要超过 10 次。
**参数:**
- `user_id`: 目标用户的 QQ 号
- `times`: 点赞次数,建议 1-10 次
**示例:**
```python
@matcher.command("like")
async def handle_like(event: MessageEvent, args: str):
# 给发送者点赞
await event.bot.send_like(event.user_id, times=1)
await event.reply("给你点了个赞!")
# 如果提供了参数,给指定用户点赞
if args.isdigit():
target_id = int(args)
await event.bot.send_like(target_id, times=1)
await event.reply(f"{target_id} 点了个赞!")
```
**注意:**
- 不是所有 OneBot 实现都支持这个 API
- 有每日次数限制,不要滥用
- 对方可能关闭了"戳一戳"功能,这时会失败
## 加好友请求处理
### `set_friend_add_request` - 处理加好友请求
```python
async def set_friend_add_request(
self,
flag: str,
approve: bool = True,
remark: str = ""
) -> Dict[str, Any]
```
处理收到的加好友请求。需要在 `request` 事件中调用。
**参数:**
- `flag`: 请求标识,从 `request` 事件的 `flag` 字段获取
- `approve`: 是否同意,`True` 同意,`False` 拒绝
- `remark`: 同意请求时,为该好友设置的备注(可选)
**示例:**
```python
from models.events.request import RequestEvent
from core.managers.command_manager import matcher
# 处理所有加好友请求
@matcher.on_event(RequestEvent)
async def handle_friend_request(event: RequestEvent):
if event.request_type == "friend":
# 自动同意并设置备注
await event.bot.set_friend_add_request(
flag=event.flag,
approve=True,
remark=f"自动添加-{event.user_id}"
)
# 给新好友发个欢迎消息
await event.bot.send_private_msg(
event.user_id,
"你好!我是机器人,已自动通过你的好友请求。"
)
```
## 实用示例
### 好友信息查询插件
```python
@matcher.command("friendinfo")
async def handle_friendinfo(event: MessageEvent):
# 获取好友列表
friends = await event.bot.get_friend_list()
# 按备注名排序
sorted_friends = sorted(friends, key=lambda f: f.remark or f.nickname)
# 生成好友列表消息
if len(sorted_friends) > 50:
msg = f"好友太多啦只显示前50个{len(sorted_friends)}个)\n"
sorted_friends = sorted_friends[:50]
else:
msg = f"我的好友列表(共{len(sorted_friends)}个):\n"
for i, friend in enumerate(sorted_friends, 1):
remark_display = friend.remark if friend.remark else "(无备注)"
msg += f"{i}. {friend.nickname} ({friend.user_id}) - 备注: {remark_display}\n"
await event.reply(msg)
```
### 自动通过特定用户的好友请求
```python
@matcher.on_event(RequestEvent)
async def handle_specific_friend_request(event: RequestEvent):
if event.request_type != "friend":
return
# 允许列表
allowed_users = [123456, 789012, 345678]
if event.user_id in allowed_users:
# 自动同意
await event.bot.set_friend_add_request(
flag=event.flag,
approve=True,
remark="重要联系人"
)
# 发送欢迎消息
await event.bot.send_private_msg(
event.user_id,
"你好!已通过你的好友请求。\n"
"发送 /help 查看可用指令。"
)
else:
# 拒绝其他人
await event.bot.set_friend_add_request(
flag=event.flag,
approve=False,
reason="仅限授权用户添加"
)
```
### 批量给好友发送消息(谨慎使用!)
```python
@matcher.command("broadcast", permission="admin")
async def handle_broadcast(event: MessageEvent, args: str):
if not args:
await event.reply("需要广播内容")
return
# 获取好友列表
friends = await event.bot.get_friend_list()
success_count = 0
fail_count = 0
await event.reply(f"开始向 {len(friends)} 个好友发送广播...")
for friend in friends:
try:
await event.bot.send_private_msg(friend.user_id, args)
success_count += 1
# 避免发送太快被限制
await asyncio.sleep(0.5)
except Exception as e:
print(f"发送给 {friend.user_id} 失败: {e}")
fail_count += 1
await event.reply(
f"广播完成!\n"
f"成功: {success_count}\n"
f"失败: {fail_count}"
)
```
**注意**:批量发送消息容易被腾讯限制,谨慎使用!
## 注意事项
1. **频率限制**: 获取好友列表、查询陌生人信息等操作有频率限制。
2. **缓存**: 好友列表和陌生人信息默认缓存 1 小时,如果需要实时数据,设 `no_cache=True`
3. **权限**: 有些 API 需要特定的权限或客户端支持。
4. **隐私**: 处理好友请求时,注意保护用户隐私。
## 下一步
- [账号 API](./account.md): 管理机器人自己的信息
- [群组 API](./group.md): 管理群聊相关功能
- [消息 API](./message.md): 怎么发消息、撤回消息

506
docs/api/group.md Normal file
View File

@@ -0,0 +1,506 @@
# 群组 API
管群是个技术活。这一页讲的是怎么管理群聊:踢人、禁言、改名片、设管理员……所有跟群相关的操作都在这里。
## 权限说明
**重要提醒**:很多群管理 API 需要机器人有相应的权限:
- **管理员权限**:禁言、踢人、改群名片等
- **群主权限**:解散群、设置管理员等
如果机器人权限不足API 调用会失败。建议先检查机器人的权限,或者做好错误处理。
## 成员管理
### `set_group_kick` - 踢出群聊
```python
async def set_group_kick(
self,
group_id: int,
user_id: int,
reject_add_request: bool = False
) -> Dict[str, Any]
```
把指定成员踢出群聊。
**参数:**
- `group_id`: 群号
- `user_id`: 要踢出的成员的 QQ 号
- `reject_add_request`: 是否同时拒绝该用户此后的加群请求(默认 `False`
**示例:**
```python
@matcher.command("kick")
async def handle_kick(event: MessageEvent, args: str):
if not args.isdigit():
await event.reply("参数错误,需要 QQ 号")
return
target_id = int(args)
await event.bot.set_group_kick(event.group_id, target_id)
await event.reply(f"已踢出 {target_id}")
```
### `set_group_ban` - 禁言/解除禁言
```python
async def set_group_ban(
self,
group_id: int,
user_id: int,
duration: int = 1800
) -> Dict[str, Any]
```
禁言群成员。设置 `duration=0` 可以解除禁言。
**参数:**
- `group_id`: 群号
- `user_id`: 要禁言的成员的 QQ 号
- `duration`: 禁言时长,单位秒。默认 1800 秒30 分钟0 表示解除禁言
**示例:**
```python
# 禁言 10 分钟
await bot.set_group_ban(123456, 789012, duration=600)
# 解除禁言
await bot.set_group_ban(123456, 789012, duration=0)
```
### `set_group_anonymous_ban` - 禁言匿名用户
```python
async def set_group_anonymous_ban(
self,
group_id: int,
anonymous: Optional[Dict[str, Any]] = None,
duration: int = 1800,
flag: Optional[str] = None
) -> Dict[str, Any]
```
禁言发送匿名消息的用户。需要从消息事件的 `anonymous` 字段获取匿名用户信息。
**参数:**
- `group_id`: 群号
- `anonymous`: 匿名用户对象(从事件中获取)
- `duration`: 禁言时长,单位秒
- `flag`: 匿名用户的 flag 标识(从事件中获取)
**示例:**
```python
@matcher.command("ban_anonymous")
async def handle_ban_anonymous(event: GroupMessageEvent):
if not event.anonymous:
await event.reply("这不是匿名消息")
return
# 方法 1: 使用 anonymous 对象
await event.bot.set_group_anonymous_ban(
event.group_id,
anonymous=event.anonymous,
duration=3600 # 禁言 1 小时
)
# 方法 2: 使用 flag如果事件中有的话
# await event.bot.set_group_anonymous_ban(
# event.group_id,
# flag=event.anonymous.get("flag"),
# duration=3600
# )
```
### `set_group_whole_ban` - 全员禁言
```python
async def set_group_whole_ban(
self,
group_id: int,
enable: bool = True
) -> Dict[str, Any]
```
开启或关闭全员禁言。
**参数:**
- `group_id`: 群号
- `enable`: `True` 开启全员禁言,`False` 关闭
**示例:**
```python
# 开启全员禁言
await bot.set_group_whole_ban(123456, enable=True)
# 关闭全员禁言
await bot.set_group_whole_ban(123456, enable=False)
```
## 权限设置
### `set_group_admin` - 设置/取消管理员
```python
async def set_group_admin(
self,
group_id: int,
user_id: int,
enable: bool = True
) -> Dict[str, Any]
```
设置或取消群管理员。**需要机器人是群主**。
**参数:**
- `group_id`: 群号
- `user_id`: 目标成员的 QQ 号
- `enable`: `True` 设为管理员,`False` 取消管理员
**示例:**
```python
# 设某人为管理员
await bot.set_group_admin(123456, 789012, enable=True)
# 取消某人的管理员
await bot.set_group_admin(123456, 789012, enable=False)
```
### `set_group_anonymous` - 匿名聊天设置
```python
async def set_group_anonymous(
self,
group_id: int,
enable: bool = True
) -> Dict[str, Any]
```
开启或关闭群匿名聊天功能。**需要机器人是管理员**。
**参数:**
- `group_id`: 群号
- `enable`: `True` 开启匿名,`False` 关闭
## 成员信息
### `set_group_card` - 设置群名片
```python
async def set_group_card(
self,
group_id: int,
user_id: int,
card: str = ""
) -> Dict[str, Any]
```
设置群成员的群名片(群内显示的名称)。传空字符串可以删除群名片,恢复为昵称。
**参数:**
- `group_id`: 群号
- `user_id`: 目标成员的 QQ 号
- `card`: 要设置的群名片内容,空字符串表示删除
**示例:**
```python
# 设置群名片
await bot.set_group_card(123456, 789012, "技术大佬")
# 删除群名片(恢复为昵称)
await bot.set_group_card(123456, 789012, "")
```
### `set_group_special_title` - 设置专属头衔
```python
async def set_group_special_title(
self,
group_id: int,
user_id: int,
special_title: str = "",
duration: int = -1
) -> Dict[str, Any]
```
为群成员设置专属头衔(群主/管理员才有权限设置)。**需要机器人是群主**。
**参数:**
- `group_id`: 群号
- `user_id`: 目标成员的 QQ 号
- `special_title`: 专属头衔内容,空字符串表示删除
- `duration`: 头衔有效期,单位秒。-1 表示永久
**示例:**
```python
# 设置永久头衔
await bot.set_group_special_title(123456, 789012, "御用摄影师", duration=-1)
# 设置 7 天有效的头衔
await bot.set_group_special_title(123456, 789012, "本周活跃之星", duration=7*24*3600)
# 删除头衔
await bot.set_group_special_title(123456, 789012, "")
```
## 群信息管理
### `set_group_name` - 修改群名
```python
async def set_group_name(
self,
group_id: int,
group_name: str
) -> Dict[str, Any]
```
修改群名称。**需要机器人是群主或管理员**。
**参数:**
- `group_id`: 群号
- `group_name`: 新的群名称
**示例:**
```python
await bot.set_group_name(123456, "技术交流群")
```
### `set_group_leave` - 退出/解散群聊
```python
async def set_group_leave(
self,
group_id: int,
is_dismiss: bool = False
) -> Dict[str, Any]
```
退出群聊,如果是群主还可以解散群。
**参数:**
- `group_id`: 群号
- `is_dismiss`: 是否解散群(仅群主有效)
**示例:**
```python
# 普通退群
await bot.set_group_leave(123456)
# 解散群(需要是群主)
await bot.set_group_leave(123456, is_dismiss=True)
```
## 获取信息
### `get_group_info` - 获取群信息
```python
async def get_group_info(
self,
group_id: int,
no_cache: bool = False
) -> GroupInfo
```
获取群的详细信息,包括群名、成员数、创建时间等。默认会缓存 1 小时。
**参数:**
- `group_id`: 群号
- `no_cache`: 是否跳过缓存,直接从服务器获取最新信息
**返回值:**
- `GroupInfo` 对象,包含群信息
**示例:**
```python
info = await bot.get_group_info(123456)
print(f"群名: {info.group_name}")
print(f"成员数: {info.member_count}")
print(f"创建时间: {info.create_time}")
```
### `get_group_list` - 获取群列表
```python
async def get_group_list(self) -> List[GroupInfo]
```
获取机器人加入的所有群列表。
**示例:**
```python
groups = await bot.get_group_list()
for group in groups:
print(f"{group.group_id}: {group.group_name}")
```
### `get_group_member_info` - 获取群成员信息
```python
async def get_group_member_info(
self,
group_id: int,
user_id: int,
no_cache: bool = False
) -> GroupMemberInfo
```
获取指定群成员的详细信息,包括昵称、群名片、加群时间、最后发言时间等。
**参数:**
- `group_id`: 群号
- `user_id`: 成员 QQ 号
- `no_cache`: 是否跳过缓存
**返回值:**
- `GroupMemberInfo` 对象
**示例:**
```python
member = await bot.get_group_member_info(123456, 789012)
print(f"昵称: {member.nickname}")
print(f"群名片: {member.card}")
print(f"权限: {member.role}") # owner, admin, member
```
### `get_group_member_list` - 获取群成员列表
```python
async def get_group_member_list(self, group_id: int) -> List[GroupMemberInfo]
```
获取群的所有成员列表。
**示例:**
```python
members = await bot.get_group_member_list(123456)
print(f"群里有 {len(members)} 个成员")
for member in members:
print(f"{member.user_id}: {member.nickname}")
```
### `get_group_honor_info` - 获取群荣誉信息
```python
async def get_group_honor_info(
self,
group_id: int,
type: str
) -> GroupHonorInfo
```
获取群的荣誉信息,比如龙王、群聊之火、快乐源泉等。
**参数:**
- `group_id`: 群号
- `type`: 荣誉类型,可选值:
- `"talkative`:" 龙王(发言最多)
- `"performer"`: 群聊之火(发言最活跃)
- `"legend"`: 群传奇(连续多天发言最多)
- `"strong_newbie"`: 冒尖小萌新(新人中发言最多)
- `"emotion"`: 快乐源泉(发送表情包最多)
**示例:**
```python
honor = await bot.get_group_honor_info(123456, "talkative")
print(f"本周龙王: {honor.current_talkative.user_id}")
```
## 加群请求处理
### `set_group_add_request` - 处理加群请求/邀请
```python
async def set_group_add_request(
self,
flag: str,
sub_type: str,
approve: bool = True,
reason: str = ""
) -> Dict[str, Any]
```
处理加群请求或邀请。需要在 `request` 事件中调用。
**参数:**
- `flag`: 请求标识,从 `request` 事件的 `flag` 字段获取
- `sub_type`: 请求类型,`"add"`(加群请求)或 `"invite"`(群邀请)
- `approve`: 是否同意,`True` 同意,`False` 拒绝
- `reason`: 拒绝理由(仅在 `approve=False` 时有效)
**示例:**
```python
from models.events.request import RequestEvent
# 在请求事件处理函数中
async def handle_group_request(event: RequestEvent):
if event.request_type == "group":
# 自动同意所有加群请求
await event.bot.set_group_add_request(
flag=event.flag,
sub_type=event.sub_type,
approve=True
)
```
## 实用示例
### 自动同意加群请求
```python
from models.events.request import RequestEvent
from core.managers.command_manager import matcher
@matcher.on_event(RequestEvent)
async def handle_all_requests(event: RequestEvent):
if event.request_type == "group":
# 检查是否来自特定用户
if event.user_id in [123456, 789012]:
await event.bot.set_group_add_request(
flag=event.flag,
sub_type=event.sub_type,
approve=True
)
await event.bot.send_private_msg(
event.user_id,
f"已同意你的加群请求,欢迎加入!"
)
```
### 群活跃度统计
```python
@matcher.command("active")
async def handle_active(event: MessageEvent):
# 获取群成员列表
members = await event.bot.get_group_member_list(event.group_id)
# 找出最后发言时间最近的一批成员
active_members = sorted(
members,
key=lambda m: m.last_sent_time or 0,
reverse=True
)[:10]
# 生成统计消息
msg = "本群最近活跃成员TOP10:\n"
for i, member in enumerate(active_members, 1):
msg += f"{i}. {member.nickname} (最后发言: {member.last_sent_time})\n"
await event.reply(msg)
```
## 注意事项
1. **权限检查**: 调用管理 API 前,最好先检查机器人的权限。
2. **频率限制**: 不要频繁调用 API尤其是获取群成员列表这种大数据量的操作。
3. **缓存**: 获取信息的 API 默认有缓存,如果需要实时数据,记得设 `no_cache=True`
4. **错误处理**: 管理操作可能失败(权限不足、参数错误等),要做好错误处理。
## 下一步
- [好友 API](./friend.md): 处理好友相关操作
- [账号 API](./account.md): 管理机器人自身状态
- [消息 API](./message.md): 怎么发消息、撤回消息

61
docs/api/index.md Normal file
View File

@@ -0,0 +1,61 @@
# API 参考
嘿,这里是 NEO Bot 的 API 参考文档。
如果你在写插件,那这里就是你的工具库。所有能和 OneBot 交互的方法都在这了。
## 快速导航
### 1. 基础概念
- [API 调用方式](./base.md): 怎么调用 API、参数格式、返回格式
- [消息段 (MessageSegment)](./message.md#消息段): 除了文字,还能发图片、表情、@人……
### 2. 分类 API
- [消息 API](./message.md): 发消息、撤回、转发
- [群组 API](./group.md): 管群、禁言、踢人、改名片
- [好友 API](./friend.md): 好友列表、点赞、加好友请求
- [账号 API](./account.md): 机器人自己的信息、状态设置
- [媒体 API](./media.md): 图片、语音相关
### 3. 高级功能
- [合并转发](./message.md#合并转发): 怎么发那种一条消息展开好多条的“聊天记录”
- [智能回复](./message.md#智能回复): `event.reply()``bot.send()` 怎么选
## 怎么用这些 API
在插件里,你拿到的 `event` 对象自带一个 `bot` 属性,那就是你的机器人实例:
```python
from core.managers.command_manager import matcher
from models.events.message import MessageEvent
@matcher.command("test")
async def handle_test(event: MessageEvent):
# 方法 1: 快捷回复(推荐)
await event.reply("你好!")
# 方法 2: 直接调用 bot 上的 API
bot = event.bot
await bot.send_group_msg(123456, "这是一条群消息")
# 方法 3: 如果你只有 bot 实例,没有 event
# (这种情况比较少见,一般只在初始化时用到)
await bot.get_login_info()
```
大部分时候,用 `event.reply()` 就够了。它帮你判断是群聊还是私聊,自动调用正确的 API。
## 兼容性说明
NEO Bot 基于 **OneBot v11** 标准实现,兼容:
- [NapCatQQ](https://github.com/NapNeko/NapCatQQ) (推荐)
- go-cqhttp
- 以及其他实现了 OneBot v11 标准的客户端
但要注意:不同客户端的实现细节可能有差异。比如某些 API 可能不支持,或者参数格式稍有不同。
如果你发现某个 API 调用失败,先看看日志里的错误信息,或者去对应客户端的文档里查查。
## 接下来?
挑一个你感兴趣的类别开始看吧。建议从 [消息 API](./message.md) 开始,因为发消息是最常用的功能。

259
docs/api/media.md Normal file
View File

@@ -0,0 +1,259 @@
# 媒体 API
这一页讲的是怎么处理图片、语音等媒体文件。虽然方法不多,但都很实用。
## 能力检查
### `can_send_image` - 检查是否可以发送图片
```python
async def can_send_image(self) -> Dict[str, Any]
```
检查当前上下文是否允许发送图片。
**返回值:**
- 包含检查结果的字典,通常有 `yes``no` 字段
**示例:**
```python
@matcher.command("sendpic")
async def handle_sendpic(event: MessageEvent, args: str):
# 先检查能不能发图片
result = await event.bot.can_send_image()
if result.get("yes"):
# 可以发图片
await event.reply(MessageSegment.image("https://example.com/image.jpg"))
else:
# 不能发图片
await event.reply("当前环境不支持发送图片")
```
### `can_send_record` - 检查是否可以发送语音
```python
async def can_send_record(self) -> Dict[str, Any]
```
检查当前上下文是否允许发送语音消息。
**示例:**
```python
result = await bot.can_send_record()
if result.get("yes"):
print("可以发语音")
else:
print("不能发语音")
```
## 图片信息
### `get_image` - 获取图片信息
```python
async def get_image(self, file: str) -> Dict[str, Any]
```
获取图片的详细信息比如大小、尺寸、MD5 等。
**参数:**
- `file`: 图片文件名、路径或 URL
**返回值:**
- 包含图片信息的字典
**示例:**
```python
@matcher.command("imageinfo")
async def handle_imageinfo(event: MessageEvent):
# 检查消息中是否有图片
for segment in event.message:
if segment.type == "image":
file = segment.data.get("file", "")
if file:
# 获取图片信息
info = await event.bot.get_image(file)
await event.reply(
f"图片信息:\n"
f"大小: {info.get('size', '未知')} 字节\n"
f"尺寸: {info.get('width', '?')}x{info.get('height', '?')}\n"
f"MD5: {info.get('md5', '未知')}"
)
return
await event.reply("消息中没有图片")
```
## 实际应用示例
### 图片转发器
```python
@matcher.command("forwardimage")
async def handle_forwardimage(event: MessageEvent, args: str):
"""
将收到的图片转发到指定群
用法: /forwardimage 群号
"""
if not args.isdigit():
await event.reply("参数错误,需要群号")
return
target_group = int(args)
# 查找消息中的图片
images = []
for segment in event.message:
if segment.type == "image":
images.append(segment)
if not images:
await event.reply("消息中没有图片")
return
# 检查是否能发图片到目标群
can_send = await event.bot.can_send_image()
if not can_send.get("yes"):
await event.reply("当前环境不支持发送图片")
return
# 转发所有图片
for image in images:
await event.bot.send_group_msg(target_group, image)
await asyncio.sleep(0.5) # 避免发送太快
await event.reply(f"已转发 {lenimages()} 张图片到群 {target_group}")
```
### 图片信息查询插件
```python
@matcher.on_event(GroupMessageEvent)
async def handle_image_autoinfo(event: GroupMessageEvent):
"""
自动回复图片信息(当有人发图片时)
"""
# 只处理包含图片的消息
images = [seg for seg in event.message if seg.type == "image"]
if not images:
return
# 只处理第一张图片(避免消息太长)
image_seg = images[0]
file = image_seg.data.get("file", "")
if not file:
return
try:
# 获取图片信息
info = await event.bot.get_image(file)
# 构建回复消息
msg = "📷 图片信息n\"
if "size" in info:
size_kb = info["size"] / 1024
msg += f"大小: {size_kb:.1f} KB\n"
if "width" in info and "height" in info:
msg += f"尺寸: {info['width']}×{info['height']}\n"
if "md5" in info:
msg += f"MD5: {info['md5'][:8]}...\n"
await event.reply(msg)
except Exception as e:
# 获取图片信息失败,静默处理
pass
```
### 图片发送安全检查
```python
async def safe_send_image(bot, target_id, image_url, is_group=True):
"""
安全发送图片:先检查是否能发,再发送
"""
# 检查发送能力
can_send = await bot.can_send_image()
if not can_send.get("yes"):
return False, "当前环境不支持发送图片"
# 检查图片是否存在(简单检查)
if not image_url:
return False, "图片URL为空"
try:
# 发送图片
if is_group:
await bot.send_group_msg(target_id, MessageSegment.image(image_url))
else:
await bot.send_private_msg(target_id, MessageSegment.image(image_url))
return True, "图片发送成功"
except Exception as e:
return False, f"发送失败: {e}"
@matcher.command("safepic")
async def handle_safepic(event: MessageEvent, args: str):
"""
安全发送图片示例
"""
if not args:
await event.reply("需要图片URL")
return
# 是判断群聊还是私聊
is_group = hasattr(event, "group_id") and event.group_id
if is_group:
target_id = event.group_id
else:
target_id = event.user_id
# 安全发送
success, message = await safe_send_image(
event.bot, target_id, args, is_group
)
if not success:
await event.reply(message)
```
## 注意事项
1. **客户端支持**: 不是所有 OneBot 客户端都完全支持媒体 API。
2. **网络限制**: 发送图片和语音可能受网络环境限制。
3. **文件大小**: 图片和语音文件有大小限制,太大的文件可能发送失败。
4. **缓存**: 图片默认会缓存,重复发送同一图片会更快。
5. **安全性**: 不要发送可疑或非法内容。
## 常见问题
### Q: 为什么 `can_send_image` 总是返回可以?
A: 这取决于 OneBot 客户端的实现。有些客户端可能不检查实际能力,总是返回可以。
### Q: 怎么发送本地图片?
A: 使用 `file://` 协议或直接使用本地路径:
```python
# 本地文件路径
image = MessageSegment.image("file:///path/to/image.jpg")
# 或者(取决于客户端)
image = MessageSegment.image("/path/to/image.jpg")
```
### Q: 怎么发送语音消息?
A: NEO Bot 目前没有封装发送语音的 API但你可以通过 `call_api` 直接调用:
```python
await bot.call_api("send_group_msg", {
"group_id": 123456,
"message": [{
"type": "record",
"data": {"file": "http://example.com/voice.amr"}
}]
})
```
## 下一步
- [消息 API](./message.md): 怎么发消息、撤回消息,包含消息段的使用
- [群组 API](./group.md): 管理群聊相关功能
- [好友 API](./friend.md): 管理好友相关功能

309
docs/api/message.md Normal file
View File

@@ -0,0 +1,309 @@
# 消息 API
发消息是机器人最基础的功能。这一页讲的是怎么发消息、撤回消息、转发消息,以及怎么用消息段(图片、@人、表情等等)。
## 快速开始
### 发一条简单的消息
```python
from core.managers.command_manager import matcher
from models.events.message import MessageEvent
@matcher.command("hello")
async def handle_hello(event: MessageEvent):
# 方法 1: 直接回复(最常用)
await event.reply("你好呀!")
# 方法 2: 通过 bot 实例发消息
await event.bot.send_group_msg(event.group_id, "这是一条群消息")
# 如果是私聊,可以用 send_private_msg
# await event.bot.send_private_msg(event.user_id, "这是一条私聊消息")
```
`event.reply()` 是最简单的方式,它会自动判断是群聊还是私聊,然后调用正确的 API。
## 消息段 (MessageSegment)
除了纯文字QQ 消息还能包含图片、@某人、表情、分享链接等等。在 OneBot 里,这些叫“消息段”。
NEO Bot 用 `MessageSegment` 类来表示消息段。
### 创建消息段
```python
from models.message import MessageSegment
# 文本
text_seg = MessageSegment.text("这是一段文字")
# @某人
at_seg = MessageSegment.at(123456) # @QQ号 123456
at_all = MessageSegment.at("all") # @全体成员
# 图片
image_seg = MessageSegment.image("https://example.com/image.jpg")
# 本地图片
local_image = MessageSegment.image("file:///path/to/image.png")
# 表情 (QQ 表情,不是 emoji)
face_seg = MessageSegment(type="face", data={"id": "123"})
# 分享链接
share_seg = MessageSegment(type="share", data={
"url": "https://example.com",
"title": "示例网站",
"content": "这是一个示例网站",
"image": "https://example.com/thumb.jpg"
})
```
### 组合消息段
你可以把多个消息段组合成一条消息:
```python
# 方法 1: 用列表
message = [
MessageSegment.text("你好,"),
MessageSegment.at(123456),
MessageSegment.text(""),
MessageSegment.image("https://example.com/welcome.jpg")
]
# 方法 2: 用加法运算符(更直观)
message = (
MessageSegment.text("你好,") +
MessageSegment.at(123456) +
MessageSegment.text("")
)
# 发送组合消息
await event.reply(message)
```
### 从 CQ 码转换
如果你熟悉 CQ 码,也可以用 `MessageSegment` 来解析:
```python
# CQ 码字符串转消息段列表(需要手动解析,这里只是示例)
# 实际使用中,框架会自动处理 CQ 码
```
## API 方法详解
### `send_group_msg` - 发送群消息
```python
async def send_group_msg(
self,
group_id: int,
message: Union[str, MessageSegment, List[MessageSegment]],
auto_escape: bool = False
) -> Dict[str, Any]
```
**参数:**
- `group_id`: 群号
- `message`: 消息内容,可以是字符串、单个消息段,或消息段列表
- `auto_escape`: 是否对消息中的 CQ 码特殊字符进行转义(仅当 `message` 是字符串时有效)
**示例:**
```python
# 发文字
await bot.send_group_msg(123456, "大家好!")
# 发图片
await bot.send_group_msg(123456, MessageSegment.image("https://example.com/cat.jpg"))
# 发组合消息
msg = MessageSegment.text("看这只猫:") + MessageSegment.image("https://example.com/cat.jpg")
await bot.send_group_msg(123456, msg)
```
### `send_private_msg` - 发送私聊消息
```python
async def send_private_msg(
self,
user_id: int,
message: Union[str, MessageSegment, List[MessageSegment]],
auto_escape: bool = False
) -> Dict[str, Any]
```
**参数:**
- `user_id`: 对方的 QQ 号
- `message`: 消息内容
- `auto_escape`: 是否转义 CQ 码
**示例:**
```python
await bot.send_private_msg(123456, "你好,这是一条私聊消息")
```
### `send` - 智能发送
```python
async def send(
self,
event: OneBotEvent,
message: Union[str, MessageSegment, List[MessageSegment]],
auto_escape: bool = False
) -> Dict[str, Any]
```
这个方法会根据事件的类型自动选择发群消息还是私聊消息。如果事件是消息事件,它其实会调用 `event.reply()`
**示例:**
```python
# 在事件处理函数中
await bot.send(event, "自动判断是群聊还是私聊")
```
### `delete_msg` - 撤回消息
```python
async def delete_msg(self, message_id: int) -> Dict[str, Any]
```
**参数:**
- `message_id`: 要撤回的消息 ID从消息事件中获取
**示例:**
```python
@matcher.command("recall")
async def handle_recall(event: MessageEvent):
# 撤回上一条消息(假设我们知道 message_id
message_id = event.message_id
await event.bot.delete_msg(message_id)
```
### `get_msg` - 获取消息详情
```python
async def get_msg(self, message_id: int) -> Dict[str, Any]
```
获取一条消息的详细信息,包括发送者、发送时间、内容等。
### `get_forward_msg` - 获取合并转发消息
```python
async def get_forward_msg(self, id: str) -> List[Dict[str, Any]]
```
获取一条合并转发消息(聊天记录)的详细内容。
**参数:**
- `id`: 合并转发消息的 ID从消息中获取
**返回值:**
- 消息节点列表,每个节点包含发送者、时间、内容等信息
## 合并转发
合并转发就是那种“点击展开查看聊天记录”的消息。在 QQ 里很常见。
### 构建转发节点
先用 `bot.build_forward_node()` 创建节点:
```python
# 创建一个转发节点
node = bot.build_forward_node(
user_id=123456, # 发送者的 QQ 号
nickname ="张三", # 显示的名字
message="这是一条测试消息" # 消息内容
)
# 消息内容也可以用消息段
node2 = bot.build_forward_node(
user_id=789012,
nickname="李四",
message=MessageSegment.text("看这个图片:") + MessageSegment.image("https://example.com/img.jpg")
)
```
### 发送合并转发
```python
# 方法 1: 直接发到群聊
nodes = [node1, node2, node3]
await bot.send_group_forward_msg(group_id=123456, messages=nodes)
# 方法 2: 发到私聊
await bot.send_private_forward_msg(user_id=123456, messages=nodes)
# 方法 3: 智能发送(根据事件判断)
await bot.send_forwarded_messages(target=event, nodes=nodes)
```
### 完整示例
```python
@matcher.command("forward")
async def handleforward_(event: MessageEvent):
# 创建几个测试节点
nodes = [
event.bot.build_forward_node(
user_id=10001,
nickname="系统",
message="欢迎使用 NEO Bot"
),
event.bot.build_forward_node(
user_id=event.user_id,
nickname=event.sender.nickname,
message="这个合并转发功能真好用!"
),
event.bot.build_forward_node(
user_id=10002,
nickname="机器人",
message=MessageSegment.text("谢谢夸奖!") + MessageSegment.face(id="123")
)
]
# 发送
await event.bot.send_forwarded_messages(event, nodes)
```
## 消息事件中的快捷方法
在消息事件 (`MessageEvent`) 中,有一些快捷方法:
### `event.reply()`
```python
await event.reply("你好!")
await event.reply(message_segment_list)
```
自动回复到消息来源(群聊或私聊)。
### `event.message`
获取事件中的消息内容(已经是 `MessageSegment` 列表格式)。
```python
# 检查消息是否包含图片
for segment in event.message:
if segment.type == "image":
await event.reply("你发了一张图片!")
break
```
## 注意事项
1. **消息长度限制**: QQ 对单条消息有长度限制,太长的消息会被截断。
2. **频率限制**: 不要疯狂发消息,可能会被腾讯限制。
3. **图片缓存**: 默认情况下,图片会缓存到本地,下次发送同样的图片会更快。
4. **网络错误**: 发消息可能因为网络问题失败,建议做好错误处理。
## 下一步
现在你已经知道怎么发消息了。接下来可以看看:
- [群组 API](./group.md): 管理群聊,比如禁言、踢人
- [好友 API](./friend.md): 处理好友相关操作
- [账号 API](./account.md): 管理机器人自己的状态