功能中心
+Dashboard & Command List · {{ plugins|length }} Modules Loaded
+diff --git a/core/data/temp/help_menu.png b/core/data/temp/help_menu.png index 7750044..f5e82c0 100644 Binary files a/core/data/temp/help_menu.png and b/core/data/temp/help_menu.png differ diff --git a/core/managers/command_manager.py b/core/managers/command_manager.py index 43d9091..e79e80f 100644 --- a/core/managers/command_manager.py +++ b/core/managers/command_manager.py @@ -202,17 +202,20 @@ class CommandManager: 内置的 `/help` 命令的实现。 直接从 Redis 获取缓存的图片。 """ - # 1. 尝试从 Redis 获取 - help_pic = await redis_manager.get("neobot:core:help_pic") - - if not help_pic: - await bot.send(event, "帮助图片缓存缺失,正在重新生成...") - await self.sync_help_pic() + try: + # 1. 尝试从 Redis 获取 help_pic = await redis_manager.get("neobot:core:help_pic") - if help_pic: - await bot.send(event, MessageSegment.image(help_pic)) - return + if not help_pic: + await bot.send(event, "帮助图片缓存缺失,正在重新生成...") + await self.sync_help_pic() + help_pic = await redis_manager.get("neobot:core:help_pic") + + if help_pic: + await bot.send(event, MessageSegment.image(help_pic)) + return + except Exception as e: + logger.error(f"获取或生成帮助图片失败: {e}") # 2. 最后的兜底:发送纯文本 help_text = "--- 可用指令列表 ---\n" @@ -225,8 +228,7 @@ class CommandManager: help_text += f" 功能: {description}\n" help_text += f" 用法: {usage}\n" - await bot.send(event, MessageSegment.image(help_pic)) - # await bot.send(event, help_text.strip()) + await bot.send(event, help_text.strip()) # 实例化全局唯一的命令管理器 diff --git a/docs/api/account.md b/docs/api/account.md new file mode 100644 index 0000000..aa3e4e6 --- /dev/null +++ b/docs/api/account.md @@ -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): 怎么发消息、撤回消息 \ No newline at end of file diff --git a/docs/api/base.md b/docs/api/base.md new file mode 100644 index 0000000..8a3a111 --- /dev/null +++ b/docs/api/base.md @@ -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): 图片、语音 \ No newline at end of file diff --git a/docs/api/friend.md b/docs/api/friend.md new file mode 100644 index 0000000..4925d2a --- /dev/null +++ b/docs/api/friend.md @@ -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): 怎么发消息、撤回消息 \ No newline at end of file diff --git a/docs/api/group.md b/docs/api/group.md new file mode 100644 index 0000000..9b8912b --- /dev/null +++ b/docs/api/group.md @@ -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): 怎么发消息、撤回消息 \ No newline at end of file diff --git a/docs/api/index.md b/docs/api/index.md new file mode 100644 index 0000000..791bdef --- /dev/null +++ b/docs/api/index.md @@ -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) 开始,因为发消息是最常用的功能。 \ No newline at end of file diff --git a/docs/api/media.md b/docs/api/media.md new file mode 100644 index 0000000..b59b3b7 --- /dev/null +++ b/docs/api/media.md @@ -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): 管理好友相关功能 \ No newline at end of file diff --git a/docs/api/message.md b/docs/api/message.md new file mode 100644 index 0000000..5363d68 --- /dev/null +++ b/docs/api/message.md @@ -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): 管理机器人自己的状态 \ No newline at end of file diff --git a/docs/core-concepts/architecture.md b/docs/core-concepts/architecture.md index f086ec2..20dcb17 100644 --- a/docs/core-concepts/architecture.md +++ b/docs/core-concepts/architecture.md @@ -6,8 +6,9 @@ Neobot是面向内部开发者的,我会开源,但是写的很烂。。。 ### Python 3.14 + JIT 镀铬酸钾创项目的时候用的 Python 3.14 3.14兼容JIT,那就这样吧 -* **何原理**: 提前编译了源代码, -* **何用途**: 密集CPU运算能提升一些 +* **何原理**: 运行时把热点代码编译成机器码(Just-In-Time) +* **何用途**: 密集CPU运算能提升一些,尤其是插件里的循环和函数调用 +* **怎么开**: 启动时加 `-X jit` 参数 ### Mypyc 编译 (AOT) 光 JIT 还不够。。核心模块(`core/ws.py`, `core/managers/*.py`)我编译成了C扩展 diff --git a/docs/core-concepts/performance.md b/docs/core-concepts/performance.md index b495881..79f588a 100644 --- a/docs/core-concepts/performance.md +++ b/docs/core-concepts/performance.md @@ -60,7 +60,37 @@ Python 自带的 `json` 库性能好像不太好,特别是在处理 OneBot 这 * Rust 编写 * 支持直接返回 `bytes`,减少内存复制。 -## 5. Mypyc 编译 (AOT Compilation) +## 5. Python 3.14 JIT (Just-In-Time Compilation) + +### 痛点 +Python 解释器一边解析一边执行,遇到循环和函数调用就得反复解释。像消息处理这种高频循环,解释开销就特别明显。 + +### 解决方案 +Python 3.14 自带了一个实验性的 JIT 编译器。启动时加上 `-X jit` 参数,它就会在运行时把热点代码编译成机器码。 + +**JIT 怎么工作的?** +1. **监控**: 解释器运行时会统计哪些函数、哪些循环被调最得频繁。 +2. **编译**: 把这些“热点”代码编译成机器码。 +3. **替换**: 下次再执行到这段代码,直接跑机器码,跳过解释步骤。 + +**哪些代码受益最大?** +- `plugins/` 里的业务逻辑(比如 B站解析、代码沙箱)。 +- 循环密集的操作(比如遍历消息段、处理大量群消息)。 +- 频繁调用的工具函数。 + +### 如何启用? +启动机器人时加上 `-X jit` 参数: + +```bash +python -X jit main.py +``` + +### 收益 +* **热点代码加速**: 经常跑的代码能快 2-10 倍(看具体场景)。 +* **零配置**: 不用改代码,加个启动参数就行。 +* **与 Mypyc 互补**: JIT 负责动态、灵活的插件代码;Mypyc 负责静态、类型明确的核心模块。两者结合,全面覆盖。 + +## 6. Mypyc 编译 (AOT Compilation) ### 痛点 Python 作为一种解释型语言,在处理 CPU 密集型任务时性能较差。对于机器人框架的核心部分,如 WebSocket 消息解析、事件分发和插件管理,这些代码被高频调用,其性能直接影响机器人的响应速度和吞吐量。 @@ -84,7 +114,7 @@ python setup_mypyc.py 脚本会自动查找并编译预设的模块列表。 ### 特别注意:关于事件模型的编译 -`Mypyc` 对 Python 的某些动态特性和高级用法支持尚不完善。在实践中,我们发现 `dataclass` 与 `Mypyc` 存在一些兼容性问题,尤其是在使用继承和某些高级特性(如 `slots=True`)时,可能会导致编译失败或运行时错误(例如 `AttributeError: attribute '__dict__' of 'type' objects is not writable`)。 +`Mypyc` 对 Python 某些动态特性和高级用法支持尚不完善。在实践中,我们发现 `dataclass` 与 `Mypyc` 存在一些兼容性问题,尤其是在使用继承和某些高级特性(如 `slots=True`)时,可能会导致编译失败或运行时错误(例如 `AttributeError: attribute '__dict__' of 'type' objects is not writable`)。 - **当前状态**:为了确保稳定性,`setup_mypyc.py` 脚本**默认不编译** `models/events/` 目录下的事件模型文件。这些文件虽然也被频繁使用,但它们的结构相对复杂,与 `Mypyc` 的兼容性问题仍在探索中。 - **未来展望**:我们会持续关注 `Mypyc` 的更新,当其对 `dataclass` 的支持得到改善后,会重新尝试将事件模型加入编译列表,以实现极致的性能。 diff --git a/docs/index.md b/docs/index.md index 881ca84..fb1ecd7 100644 --- a/docs/index.md +++ b/docs/index.md @@ -18,7 +18,15 @@ * [消息流](./core-concepts/event-flow.md): 看看一条消息从被接收到被回复是如何运行的 * [核心](./core-concepts/singleton-managers.md): `matcher`, `browser_manager`... 认识这些核心模块。 -### 3. 插件开发 +### 3. API 参考 +* [API 总览](./api/index.md): 所有 API 的快速导航和调用方式 +* [消息 API](./api/message.md): 发消息、撤回、转发、合并转发 +* [群组 API](./api/group.md): 管群、禁言、踢人、改名片 +* [好友 API](./api/friend.md): 好友列表、点赞、加好友请求 +* [账号 API](./api/account.md): 机器人自己的信息、状态设置 +* [媒体 API](./api/media.md): 图片、语音相关 + +### 4. 插件开发 * [插件开发第一步](./plugin-development/index.md): 带你写第一个插件 * [指南](./plugin-development/command-handling.md): 怎么教你的 Bot 听懂指令和参数。 * [绝对不要做的事情](./plugin-development/best-practices.md): **(必读!)** diff --git a/import sys.py b/import sys.py deleted file mode 100644 index daa4292..0000000 --- a/import sys.py +++ /dev/null @@ -1,16 +0,0 @@ -import sys -import sysconfig - -print(f"Python Version: {sys.version}") - -# 检查 GIL 状态 -try: - # Python 3.13+ free-threading build 才有这个属性 - is_gil_enabled = sys._is_gil_enabled() - print(f"GIL Enabled: {is_gil_enabled}") -except AttributeError: - print("GIL Status: Unknown (sys._is_gil_enabled not found, likely GIL-enabled build)") - -# 检查 JIT 状态 -# 目前没有直接的 API 检查 JIT 是否开启,通常看性能或启动日志 -print("JIT Support: Experimental (Enable with -X jit)") \ No newline at end of file diff --git a/plugins/auto_approve.py b/plugins/auto_approve.py new file mode 100644 index 0000000..f92254e --- /dev/null +++ b/plugins/auto_approve.py @@ -0,0 +1,53 @@ +""" +自动同意请求插件 + +提供自动同意好友请求和群聊邀请的功能。 +""" +from core.managers.command_manager import matcher +from core.bot import Bot +from models.events.request import FriendRequestEvent, GroupRequestEvent + +__plugin_meta__ = { + "name": "自动同意请求", + "description": "自动同意好友请求和群聊邀请", + "usage": "无需手动操作,自动处理请求事件", +} + +@matcher.on_request(request_type="friend") +async def handle_friend_request(bot: Bot, event: FriendRequestEvent): + """ + 处理好友请求事件,自动同意好友申请 + + :param bot: Bot实例 + :param event: 好友请求事件对象 + """ + try: + # 自动同意好友请求 + await bot.call_api( + "set_friend_add_request", + flag=event.flag, + approve=True + ) + print(f"[自动同意] 已同意用户 {event.user_id} 的好友请求") + except Exception as e: + print(f"[自动同意] 同意好友请求失败: {e}") + +@matcher.on_request(request_type="group") +async def handle_group_request(bot: Bot, event: GroupRequestEvent): + """ + 处理群聊邀请事件,自动同意群聊邀请 + + :param bot: Bot实例 + :param event: 群聊邀请事件对象 + """ + try: + # 自动同意群聊邀请 + await bot.call_api( + "set_group_add_request", + flag=event.flag, + sub_type=event.sub_type, + approve=True + ) + print(f"[自动同意] 已同意加入群聊 {event.group_id} (邀请人: {event.user_id})") + except Exception as e: + print(f"[自动同意] 同意群聊邀请失败: {e}") diff --git a/plugins/bili_parser.py b/plugins/bili_parser.py index 3b9030f..af37675 100644 --- a/plugins/bili_parser.py +++ b/plugins/bili_parser.py @@ -13,12 +13,16 @@ from models import MessageEvent, MessageSegment # 创建一个TTL缓存,最大容量100,缓存时间10秒 processed_messages: TTLCache[int, bool] = TTLCache(maxsize=100, ttl=10) +# 插件元数据 __plugin_meta__ = { "name": "bili_parser", "description": "自动解析B站分享卡片,提取视频封面和播放量等信息。", "usage": "(自动触发)当检测到B站小程序分享卡片时,自动发送视频信息。", } +# 常量定义 +BILI_NICKNAME = "B站视频解析" + HEADERS = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' } @@ -29,7 +33,7 @@ _session: Optional[aiohttp.ClientSession] = None async def get_session() -> aiohttp.ClientSession: global _session if _session is None or _session.closed: - _session = aiohttp.ClientSession() + _session = aiohttp.ClientSession(headers=HEADERS) return _session @@ -71,7 +75,7 @@ async def parse_video_info(video_url: str) -> Optional[Dict[str, Any]]: if not script_tag or not script_tag.string: return None - match = re.search(r'window\.__INITIAL_STATE__\s*=\s*(\{.*?\});', script_tag.string) + match = re.search(r'window\.__INITIAL_STATE__\s*=\s*(\{[^\}]*\});', script_tag.string) if not match: return None @@ -126,16 +130,56 @@ async def get_direct_video_url(video_url: str) -> Optional[str]: async with aiohttp.ClientSession() as session: async with session.get(api_url, headers=HEADERS, timeout=10) as response: response.raise_for_status() - data = await response.json() + # 使用 content_type=None 来忽略 Content-Type 检查 + # 因为 API 返回 text/json 而不是标准的 application/json + data = await response.json(content_type=None) if data.get("code") == 200 and data.get("data"): return data["data"][0].get("video_url") except (aiohttp.ClientError, json.JSONDecodeError, KeyError, IndexError) as e: logger.error(f"[bili_parser] 调用第三方API解析视频失败: {e}") return None -BILI_URL_PATTERN = re.compile(r"https?://(?:www\.)?(bilibili\.com/video/[a-zA-Z0-9_]+|b23\.tv/[a-zA-Z0-9]+)") +BILI_URL_PATTERN = re.compile(r"https?://(?:www\.)?(bilibili\.com/video/\w+|b23\.tv/[a-zA-Z0-9]+)") +def extract_url_from_json_segments(segments): + """ + 从消息的JSON段中提取B站链接 + :param segments: 消息段列表 + :return: 提取到的URL或None + """ + for segment in segments: + if segment.type == "json": + logger.info(f"[bili_parser] 检测到JSON CQ码: {segment.data}") + try: + json_data = json.loads(segment.data.get("data", "{}")) + short_url = json_data.get("meta", {}).get("detail_1", {}).get("qqdocurl") + + if short_url and "b23.tv" in short_url: + extracted_url = short_url.split('?')[0] + logger.success(f"[bili_parser] 成功从JSON卡片中提取到B站短链接: {extracted_url}") + return extracted_url + except (json.JSONDecodeError, KeyError) as e: + logger.error(f"[bili_parser] 解析JSON失败: {e}") + continue + return None + +def extract_url_from_text_segments(segments): + """ + 从消息的文本段中提取B站链接 + :param segments: 消息段列表 + :return: 提取到的URL或None + """ + for segment in segments: + if segment.type == "text": + text_content = segment.data.get("text", "") + match = BILI_URL_PATTERN.search(text_content) + if match: + extracted_url = match.group(0) + logger.success(f"[bili_parser] 成功从文本中提取到B站链接: {extracted_url}") + return extracted_url + return None + @matcher.on_message() async def handle_bili_share(event: MessageEvent): """ @@ -151,34 +195,12 @@ async def handle_bili_share(event: MessageEvent): if event.user_id == event.self_id: return - url_to_process = None - # 1. 优先解析JSON卡片中的短链接 - for segment in event.message: - if segment.type == "json": - logger.info(f"[bili_parser] 检测到JSON CQ码: {segment.data}") - try: - json_data = json.loads(segment.data.get("data", "{}")) - short_url = json_data.get("meta", {}).get("detail_1", {}).get("qqdocurl") - - if short_url and "b23.tv" in short_url: - url_to_process = short_url.split('?')[0] - logger.success(f"[bili_parser] 成功从JSON卡片中提取到B站短链接: {url_to_process}") - break # 找到后立即跳出循环 - except (json.JSONDecodeError, KeyError) as e: - logger.error(f"[bili_parser] 解析JSON失败: {e}") - continue + url_to_process = extract_url_from_json_segments(event.message) # 2. 如果未在JSON卡片中找到链接,则在文本消息中查找 if not url_to_process: - for segment in event.message: - if segment.type == "text": - text_content = segment.data.get("text", "") - match = BILI_URL_PATTERN.search(text_content) - if match: - url_to_process = match.group(0) - logger.success(f"[bili_parser] 成功从文本中提取到B站链接: {url_to_process}") - break # 找到后立即跳出循环 + url_to_process = extract_url_from_text_segments(event.message) # 3. 如果找到了任何类型的B站链接,则进行处理 if url_to_process: @@ -246,10 +268,10 @@ async def process_bili_link(event: MessageEvent, url: str): ] nodes = [ - event.bot.build_forward_node(user_id=event.self_id, nickname="B站视频解析", message=text_message), - event.bot.build_forward_node(user_id=event.self_id, nickname="B站视频解析", message=image_message_segment), - event.bot.build_forward_node(user_id=event.self_id, nickname="B站视频解析", message=up_info_segment), - event.bot.build_forward_node(user_id=event.self_id, nickname="B站视频解析", message=video_message) + event.bot.build_forward_node(user_id=event.self_id, nickname=BILI_NICKNAME, message=text_message), + event.bot.build_forward_node(user_id=event.self_id, nickname=BILI_NICKNAME, message=image_message_segment), + event.bot.build_forward_node(user_id=event.self_id, nickname=BILI_NICKNAME, message=up_info_segment), + event.bot.build_forward_node(user_id=event.self_id, nickname=BILI_NICKNAME, message=video_message) ] logger.success(f"[bili_parser] 成功解析视频信息并准备以聊天记录形式回复: {video_info['title']}") diff --git a/scripts/check_python_env.py b/scripts/check_python_env.py new file mode 100644 index 0000000..57b9ec8 --- /dev/null +++ b/scripts/check_python_env.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +""" +Python 环境检查脚本 + +检查当前 Python 环境是否符合 NEO Bot 要求,包括版本、GIL 状态、JIT 支持等。 +""" +import sys +import platform + +def main(): + """主函数""" + print("=" * 60) + print("NEO Bot Python 环境检查") + print("=" * 60) + + # 1. Python 版本信息 + version_info = sys.version_info + print("\n[1] Python 版本:") + print(f" 版本号: {sys.version}") + print(f" 主版本: {version_info.major}.{version_info.minor}.{version_info.micro}") + print(f" 发布日期: {version_info.releaselevel} {version_info.serial}") + + # 检查是否为 Python 3.14 + if version_info.major == 3 and version_info.minor == 14: + print(" ✓ 符合要求: Python 3.14") + else: + print(f" ⚠ 警告: 推荐使用 Python 3.14,当前为 {version_info.major}.{version_info.minor}") + + # 2. 平台信息 + print("\n[2] 平台信息:") + print(f" 操作系统: {platform.system()} {platform.release()}") + print(f" 处理器: {platform.processor()}") + print(f" 架构: {platform.machine()}") + + # 3. GIL 状态 + print("\n[3] GIL (全局解释器锁) 状态:") + try: + # Python 3.13+ free-threading build 才有这个属性 + is_gil_enabled = sys._is_gil_enabled() + if is_gil_enabled: + print(" GIL 已启用 (传统模式)") + else: + print(" GIL 已禁用 (自由线程模式)") + except AttributeError: + print(" GIL 状态: 未知 (sys._is_gil_enabled 未找到,可能是传统 GIL 构建)") + + # 4. JIT 状态 + print("\n[4] JIT (即时编译) 状态:") + + # 检查是否启用了 JIT + jit_enabled = False + jit_details = "未知" + + # 方法1: 检查启动标志 + if hasattr(sys, 'flags'): + # Python 3.14 的 JIT 通过 -X jit 启用 + # 但 sys.flags 中没有直接的 JIT 标志 + pass + + # 方法2: 检查是否有 JIT 相关属性 + try: + # 尝试导入 _jit 模块(如果存在) + import _jit + jit_enabled = True + jit_details = "检测到 _jit 模块" + del _jit # 避免未使用的导入警告 + except ImportError: + # 检查 sys 模块中是否有 JIT 相关属性 + if hasattr(sys, '_jit_enabled'): + jit_enabled = sys._jit_enabled + jit_details = f"sys._jit_enabled = {jit_enabled}" + else: + jit_details = "未检测到 JIT 模块或属性" + + if jit_enabled: + print(" ✓ JIT 已启用") + print(f" 详情: {jit_details}") + else: + print(" ⚠ JIT 未启用或不可用") + print(f" 详情: {jit_details}") + print(" 建议: 启动时使用 -X jit 参数启用 JIT,例如: python -X jit main.py") + + # 5. 其他信息 + print("\n[5] 其他信息:") + print(f" 实现: {platform.python_implementation()}") + print(f" 构建: {platform.python_build()}") + print(f" 编译器: {platform.python_compiler()}") + + # 6. 路径信息 + print("\n[6] 路径信息:") + print(f" 执行文件: {sys.executable}") + print(f" 前缀: {sys.prefix}") + print(" 路径:") + for i, path in enumerate(sys.path[:5], 1): # 只显示前5个 + print(f" {i}. {path}") + if len(sys.path) > 5: + print(f" ... 还有 {len(sys.path) - 5} 个路径") + + print("\n" + "=" * 60) + print("检查完成") + print("=" * 60) + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/scripts/compile_machine_code.py b/scripts/compile_machine_code.py new file mode 100644 index 0000000..70c6992 --- /dev/null +++ b/scripts/compile_machine_code.py @@ -0,0 +1,303 @@ +#!/usr/bin/env python3 +""" +跨平台 Python 模块编译脚本 + +将核心 Python 模块编译为机器码(.pyd 或 .so)以提升性能。 + +支持的平台: +- Windows: 生成 .pyd 文件 +- Linux: 生成 .so 文件 + +使用方法: + python compile_machine_code.py [options] + +选项: + --compile, -c 编译指定的模块(默认) + --list, -l 列出已编译的模块 + --clean, -k 清理编译生成的文件 + --help, -h 显示帮助信息 + +注意: + 1. 需要安装 C 编译器 (Windows 上需要 Visual Studio Build Tools, Linux 上需要 GCC) + 2. 需要安装 mypyc: pip install mypyc + 3. 编译后的文件是平台相关的,不能跨平台复制 + 4. 建议在部署的目标环境上运行此脚本 +""" +import os +import sys +import glob +import subprocess +import shutil +import argparse + +# 检测当前平台和 Python 版本 +PLATFORM = sys.platform +PYTHON_VERSION = f"{sys.version_info.major}{sys.version_info.minor}" # 例如 "314" + +if PLATFORM.startswith('win'): + EXTENSION = '.pyd' + BUILD_PREFIX = f'cp{PYTHON_VERSION}-win_amd64' + BUILD_PATH = os.path.join('build', f'lib.win-amd64-cpython-{PYTHON_VERSION}') +elif PLATFORM.startswith('linux'): + EXTENSION = '.so' + BUILD_PREFIX = f'cp{PYTHON_VERSION}-x86_64-linux-gnu' + BUILD_PATH = os.path.join('build', f'lib.linux-x86_64-cpython-{PYTHON_VERSION}') +else: + print(f"不支持的平台: {PLATFORM}") + sys.exit(1) + +# 要编译的模块列表 +# 注意:Mypyc 对动态特性支持有限,只选择计算密集或类型明确的模块 +MODULES = [ + # 工具模块 + 'core/utils/json_utils.py', # JSON 处理 + 'core/utils/executor.py', # 代码执行引擎 + 'core/utils/singleton.py', # 单例模式基类 + 'core/utils/exceptions.py', # 自定义异常 + 'core/utils/logger.py', # 日志模块 + + # 核心管理模块 + 'core/managers/command_manager.py', # 指令匹配和分发 + 'core/managers/admin_manager.py', # 管理员管理 + 'core/managers/permission_manager.py', # 权限管理 + 'core/managers/plugin_manager.py', # 插件管理器 + 'core/managers/redis_manager.py', # Redis 管理器 + 'core/managers/image_manager.py', # 图片管理器 + + # 核心基础模块 + 'core/ws.py', # WebSocket 核心 + 'core/bot.py', # Bot 核心抽象 + 'core/config_loader.py', # 配置加载 + 'core/config_models.py', # 配置模型 + 'core/permission.py', # 权限枚举 + + # API 模块 - 注意:这些类会被 Bot 类多继承使用 + # 因此不适合编译,否则会导致 "multiple bases have instance lay-out conflict" 错误 + # 'core/api/base.py', # API 基础类 + # 'core/api/account.py', # 账号相关 API + # 'core/api/friend.py', # 好友相关 API + # 'core/api/group.py', # 群组相关 API + # 'core/api/media.py', # 媒体相关 API + # 'core/api/message.py', # 消息相关 API + + # 数据模型(适合编译的高频使用数据类) + 'models/message.py', # 消息段模型 + 'models/sender.py', # 发送者模型 + 'models/objects.py', # API 响应数据模型 + + # 事件处理相关 + 'core/handlers/event_handler.py', # 事件处理器 + + # 注意:以下文件不适合编译 + # - 主程序文件(main.py) + # - 测试文件(tests/目录) + # - 插件文件(plugins/目录) + # - 编译脚本(compile_machine_code.py等) + # - 临时文件(scratch_files/目录) + # - 抽象基类(models/events/base.py) + # - 事件工厂(models/events/factory.py) + # - 包含复杂动态特性的文件 +] + +def list_compiled_modules(): + """列出已编译的模块""" + print(f"\n已编译的 {PLATFORM} 模块:") + print("=" * 50) + + # 查找所有编译后的文件 + compiled_files = [] + for ext in [EXTENSION, f'__mypyc{EXTENSION}']: + compiled_files.extend(glob.glob(f'**/*{ext}', recursive=True)) + + # 过滤掉虚拟环境中的文件 + compiled_files = [f for f in compiled_files if 'venv' not in f] + + if compiled_files: + for f in sorted(compiled_files): + size = os.path.getsize(f) // 1024 # KB + print(f"{f} ({size} KB)") + else: + print(f"未找到已编译的 {EXTENSION} 文件") + + print(f"\n总计: {len(compiled_files)} 个文件") + +def clean_compiled_files(): + """清理编译生成的文件""" + print(f"\n清理编译生成的 {EXTENSION} 文件...") + + # 查找所有编译后的文件 + compiled_files = [] + for ext in [EXTENSION, f'__mypyc{EXTENSION}']: + compiled_files.extend(glob.glob(f'**/*{ext}', recursive=True)) + + # 过滤掉虚拟环境中的文件 + compiled_files = [f for f in compiled_files if 'venv' not in f] + + if compiled_files: + for f in sorted(compiled_files): + try: + os.remove(f) + print(f"已删除: {f}") + except Exception as e: + print(f"删除失败 {f}: {e}") + + # 清理 build 目录 + if os.path.exists('build'): + try: + shutil.rmtree('build') + print("已删除 build 目录") + except Exception as e: + print(f"删除 build 目录失败: {e}") + else: + print(f"没有可清理的 {EXTENSION} 文件") + +def get_platform_specific_module_name(module_path): + """获取平台特定的模块文件名""" + module_name = module_path.replace('.py', '') + return f"{module_name}.{BUILD_PREFIX}{EXTENSION}" + +def compile_module(module_path): + """编译单个模块""" + print(f"\n编译: {module_path}") + + try: + # 直接调用 mypyc 命令行工具 + result = subprocess.run( + [sys.executable, '-m', 'mypyc', module_path], + capture_output=True, + text=True, + check=True + ) + + # 获取平台特定的模块名 + platform_module = get_platform_specific_module_name(module_path) + mypyc_platform_module = platform_module.replace(EXTENSION, f'__mypyc{EXTENSION}') + + # 检查编译产物是否在当前目录 + if os.path.exists(platform_module): + print(f" ✓ 编译成功: {platform_module}") + return True + else: + # 检查 build 目录中是否有编译产物 + build_module_path = os.path.join(BUILD_PATH, platform_module) + build_mypyc_path = os.path.join(BUILD_PATH, mypyc_platform_module) + + if os.path.exists(build_module_path): + # 如果在 build 目录中,复制到正确位置 + os.makedirs(os.path.dirname(platform_module), exist_ok=True) + shutil.copy2(build_module_path, platform_module) + shutil.copy2(build_mypyc_path, mypyc_platform_module) + print(f" ✓ 编译成功(已从 build 目录复制): {platform_module}") + return True + else: + print(f" ✗ 编译失败:找不到编译产物") + if result.stdout: + print(f" 编译输出:{result.stdout[:500]}...") + if result.stderr: + print(f" 错误信息:{result.stderr[:500]}...") + return False + + except subprocess.CalledProcessError as e: + print(f" ✗ 编译失败,退出码: {e.returncode}") + if e.stdout: + print(f" 编译输出:{e.stdout[:500]}...") + if e.stderr: + print(f" 错误信息:{e.stderr[:500]}...") + return False + except Exception as e: + print(f" ✗ 编译失败,意外错误: {e}") + return False + +def should_skip_module(module_path): + """检查模块是否应该被跳过编译""" + try: + with open(module_path, 'r', encoding='utf-8') as f: + content = f.read() + + # 检查是否包含抽象基类相关代码 + if 'from abc import ABC' in content or 'from abc import abstractmethod' in content: + return True, "包含抽象基类,不适合编译" + + # 检查是否包含动态特性 + if 'eval(' in content or 'exec(' in content or 'getattr(' in content or 'setattr(' in content: + return True, "包含动态特性,不适合编译" + + return False, "" + except Exception as e: + return True, f"读取文件时出错: {e}" + +def compile_all_modules(): + """编译所有指定的模块""" + print(f"\n开始编译 {len(MODULES)} 个模块 (平台: {PLATFORM})") + print("=" * 60) + + # 验证模块文件是否存在并检查是否适合编译 + valid_modules = [] + for module_path in MODULES: + if os.path.exists(module_path): + should_skip, reason = should_skip_module(module_path) + if should_skip: + print(f"跳过: {module_path} ({reason})") + else: + valid_modules.append(module_path) + else: + print(f"警告: 模块 {module_path} 不存在,将被跳过") + + if not valid_modules: + print("错误: 没有有效的模块可编译") + return False + + # 编译模块 + success_count = 0 + for module_path in valid_modules: + if compile_module(module_path): + success_count += 1 + + print(f"\n" + "=" * 60) + print(f"编译完成: {success_count}/{len(valid_modules)} 个模块成功") + + if success_count == len(valid_modules): + print("✓ 所有模块编译成功") + return True + else: + print("✗ 部分模块编译失败") + return False + +def main(): + """主函数""" + # 检查 Python 版本 + if not (sys.version_info.major == 3 and sys.version_info.minor == 14): + print("警告: 推荐使用 Python 3.14 以获得最佳性能") + print(f"当前版本: {sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}") + print("继续编译可能导致兼容性问题") + print() + + parser = argparse.ArgumentParser(description='跨平台 Python 模块编译脚本') + + group = parser.add_mutually_exclusive_group() + group.add_argument('--compile', '-c', action='store_true', default=True, + help='编译指定的模块 (默认)') + group.add_argument('--list', '-l', action='store_true', + help='列出已编译的模块') + group.add_argument('--clean', '-k', action='store_true', + help='清理编译生成的文件') + + args = parser.parse_args() + + # 检查是否安装了 mypyc + try: + import mypyc + except ImportError: + print("错误: 未安装 mypyc,请先安装: pip install mypyc") + sys.exit(1) + + if args.list: + list_compiled_modules() + elif args.clean: + clean_compiled_files() + else: + compile_all_modules() + print("\n使用 --list 选项查看已编译的模块") + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/scripts/compile_modules.py b/scripts/compile_modules.py new file mode 100644 index 0000000..f869d9e --- /dev/null +++ b/scripts/compile_modules.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +""" +编译模块脚本 + +这个脚本会单独编译每个Python模块,确保每个模块都在正确位置生成独立的.pyd文件。 +""" +import os +import sys +import glob +from mypyc.build import mypycify +try: + from setuptools import setup +except ImportError: + from distutils.core import setup + +def compile_module(module_path): + """ + 编译单个模块 + + Args: + module_path: 要编译的Python模块路径 + """ + print(f"\nCompiling {module_path}...") + try: + ext_modules = mypycify([module_path]) + setup(name=f'compiled_{os.path.basename(module_path).replace(".py", "")}', + ext_modules=ext_modules) + return True + except Exception as e: + print(f"Error compiling {module_path}: {e}") + return False + +def main(): + """ + 主函数 + """ + # 检查 Python 版本 + if not (sys.version_info.major == 3 and sys.version_info.minor == 14): + print("警告: 推荐使用 Python 3.14 以获得最佳性能") + print(f"当前版本: {sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}") + print("继续编译可能导致兼容性问题") + print() + + # 要编译的模块列表 + modules = [ + 'core/utils/json_utils.py', # JSON 处理 + 'core/utils/executor.py', # 代码执行引擎 + 'core/managers/command_manager.py', # 指令匹配和分发 + 'core/managers/admin_manager.py', # 管理员管理 + 'core/managers/permission_manager.py', # 权限管理 + 'core/ws.py', # WebSocket 核心 + 'core/managers/plugin_manager.py', # 插件管理器 + 'core/bot.py', # Bot 核心抽象 + 'core/config_loader.py', # 配置加载 + ] + + # 自动添加 events 模型 + event_models = glob.glob('models/events/*.py') + event_models = [m for m in event_models if not m.endswith('__init__.py')] + modules.extend(event_models) + + print(f"Found {len(modules)} modules to compile.") + + success_count = 0 + for module in modules: + if compile_module(module): + success_count += 1 + + print(f"\n--- Compilation Summary ---") + print(f"Total modules: {len(modules)}") + print(f"Successfully compiled: {success_count}") + print(f"Failed: {len(modules) - success_count}") + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/scripts/export_requirements.py b/scripts/export_requirements.py new file mode 100644 index 0000000..d0c51b5 --- /dev/null +++ b/scripts/export_requirements.py @@ -0,0 +1,137 @@ +#!/usr/bin/env python3 +""" +导出项目依赖到 requirements.txt 文件 + +支持两种模式: +1. 默认模式:导出当前虚拟环境中的所有包(pip freeze) +2. 本地模式:只导出当前项目的依赖(pip freeze --local) + +使用方法: + python export_requirements.py [options] + +选项: + --local, -l 只导出当前项目的依赖(推荐) + --output, -o 指定输出文件路径(默认为 requirements.txt) + --help, -h 显示帮助信息 +""" +import subprocess +import sys +import argparse + +def run_pip_freeze(local_mode=False): + """ + 运行 pip freeze 命令 + + Args: + local_mode: 是否只导出当前项目依赖 + + Returns: + (success, output): 成功标志和输出内容 + """ + cmd = ['pip', 'freeze'] + if local_mode: + cmd.append('--local') + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + check=True, + encoding='utf-8' + ) + return True, result.stdout + except subprocess.CalledProcessError as e: + error_msg = f"pip freeze 命令失败,退出码: {e.returncode}\n" + if e.stderr: + error_msg += f"错误信息: {e.stderr}" + return False, error_msg + except FileNotFoundError: + return False, "错误: 未找到 pip 命令,请确保 Python 环境已正确安装" + except Exception as e: + return False, f"未知错误: {e}" + +def write_requirements_file(output_path, content): + """ + 将依赖内容写入文件 + + Args: + output_path: 输出文件路径 + content: 依赖内容 + + Returns: + success: 是否成功 + """ + try: + with open(output_path, 'w', encoding='utf-8') as f: + f.write(content) + + # 统计行数(忽略空行) + lines = [line.strip() for line in content.split('\n') if line.strip()] + return True, len(lines) + except IOError as e: + return False, f"写入文件失败: {e}" + except Exception as e: + return False, f"未知错误: {e}" + +def main(): + """主函数""" + parser = argparse.ArgumentParser(description='导出项目依赖到 requirements.txt 文件') + + parser.add_argument('--local', '-l', action='store_true', + help='只导出当前项目的依赖(推荐)') + parser.add_argument('--output', '-o', default='requirements.txt', + help='指定输出文件路径(默认为 requirements.txt)') + + args = parser.parse_args() + + print("=" * 60) + print("NEO Bot 依赖导出工具") + print("=" * 60) + + # 显示模式信息 + if args.local: + print("模式: 本地模式(只导出当前项目依赖)") + else: + print("模式: 全局模式(导出所有已安装包)") + print("提示: 建议使用 --local 选项只导出当前项目依赖") + + print(f"输出文件: {args.output}") + print() + + # 运行 pip freeze + print("正在收集依赖信息...") + success, output = run_pip_freeze(args.local) + + if not success: + print(f"错误: {output}") + sys.exit(1) + + # 写入文件 + print("正在写入文件...") + success, result = write_requirements_file(args.output, output) + + if not success: + print(f"错误: {result}") + sys.exit(1) + + line_count = result + print("✓ 依赖导出完成") + print(f" 文件: {args.output}") + print(f" 依赖数量: {line_count} 个包") + + # 显示前几个依赖(如果有) + lines = [line.strip() for line in output.split('\n') if line.strip()] + if lines: + print("\n前5个依赖:") + for i, line in enumerate(lines[:5], 1): + print(f" {i}. {line}") + if len(lines) > 5: + print(f" ... 还有 {len(lines) - 5} 个依赖") + + print("\n" + "=" * 60) + print("提示: 可以使用 pip install -r requirements.txt 安装这些依赖") + print("=" * 60) + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/templates/help.html b/templates/help.html index 4fdc65e..1388304 100644 --- a/templates/help.html +++ b/templates/help.html @@ -3,132 +3,235 @@
-功能插件列表
+Dashboard & Command List · {{ plugins|length }} Modules Loaded
+