# NEO Bot Framework 这是一个基于 Python 的轻量级 OneBot 11 协议机器人框架,专为内部团队开发设计。它通过 WebSocket 连接到 OneBot 实现端(如 NapCatQQ),提供了事件分发、指令管理和 API 调用封装。 ## ✨ 特性 * **OneBot 11 标准支持**:完整支持 OneBot 11 的消息、通知、请求和元事件。 * **类型安全**:基于 `dataclasses` 的强类型事件模型,开发体验更佳。 * **插件系统**:轻量级的装饰器风格插件系统,支持指令 (`@matcher.command`) 和事件监听 (`@matcher.on_notice`, `@matcher.on_request`)。 * **🔥 热重载支持**:内置文件监控,修改 `base_plugins` 下的代码自动重载,无需重启,极大提升调试效率。 * **异步核心**:基于 `asyncio` 和 `websockets` 的高性能异步核心。 * **自动重连**:内置 WebSocket 断线重连机制。 ## 📝 待办事项 (TODO) ### API 封装 - [x] **消息相关** - `delete_msg`: 撤回消息 - `get_msg`: 获取消息 - `get_forward_msg`: 获取合并转发消息 - `send_like`: 发送点赞 - [x] **群组管理** - `set_group_kick`: 群组踢人 - `set_group_ban`: 群组单人禁言 - `set_group_anonymous_ban`: 群组匿名禁言 - `set_group_whole_ban`: 群组全员禁言 - `set_group_admin`: 群组设置管理员 - `set_group_anonymous`: 群组匿名 - `set_group_card`: 设置群名片(群备注) - `set_group_name`: 设置群名 - `set_group_leave`: 退出群组 - `set_group_special_title`: 设置群组专属头衔 - [x] **群组信息** - `get_group_info`: 获取群信息 - `get_group_list`: 获取群列表 - `get_group_member_info`: 获取群成员信息 - `get_group_member_list`: 获取群成员列表 - `get_group_honor_info`: 获取群荣誉信息 - [x] **用户相关** - `get_login_info`: 获取登录号信息 - `get_stranger_info`: 获取陌生人信息 - `get_friend_list`: 获取好友列表 - [x] **请求处理** - `set_friend_add_request`: 处理加好友请求 - `set_group_add_request`: 处理加群请求/邀请 - [x] **系统/其他** - `get_version_info`: 获取版本信息 - `get_status`: 获取状态 - `can_send_image`: 检查是否可以发送图片 - `can_send_record`: 检查是否可以发送语音 - `clean_cache`: 清理缓存 ### 待实现 API - [ ] **Web 凭证类** - `get_cookies` - `get_csrf_token` - `get_credentials` - [ ] **文件/资源信息** - `get_image` - `get_record` - `get_file` - [ ] **系统控制** - `set_restart` - [ ] **扩展功能** - `send_forward_msg`: 发送合并转发消息 ### 其他改进 - [ ] **API 强类型封装**: 将 API 返回值从 `dict` 转换为数据模型对象。 - [ ] **日志系统优化**: 引入更完善的日志记录机制,支持文件输出和日志级别控制。 - [ ] **异常处理增强**: 增强插件执行过程中的异常捕获,防止单个插件崩溃影响整个 Bot。 - [ ] **中间件支持**: 添加消息处理中间件,支持在指令执行前/后进行拦截和处理。 - [ ] **权限系统**: 实现基础的权限管理(如超级管理员、群管理员等)。 ## 📂 项目结构 ``` NEO/ ├── base_plugins/ # 基础插件目录,新建插件文件即可自动加载(支持热重载) │ └── echo.py # 示例插件:实现 /echo 指令 ├── core/ # 核心框架代码 │ ├── bot.py # Bot API 封装,提供 send_group_msg 等方法 │ ├── command_manager.py # 命令与事件分发器 │ ├── config_loader.py # 配置加载器 │ └── ws.py # WebSocket 客户端核心 ├── models/ # 数据模型 │ ├── events/ # OneBot 事件定义 (Message, Notice, Request, Meta) │ ├── message.py # 消息段定义 (MessageSegment) │ └── sender.py # 发送者定义 (Sender) ├── config.toml # 配置文件 ├── main.py # 启动入口(包含热重载监控) └── requirements.txt # 项目依赖 ``` ## 🚀 快速开始 ### 1. 环境准备 * Python 3.8+ * OneBot 11 实现端(推荐 [NapCatQQ](https://github.com/NapNeko/NapCatQQ) 或 LLOneBot) ### 2. 安装依赖 ```bash pip install -r requirements.txt ``` ### 3. 配置文件 修改根目录下的 `config.toml`,配置 WebSocket 连接信息: ```toml [napcat_ws] uri = "ws://127.0.0.1:30004" # OneBot 实现端的 WebSocket 地址 token = "your_token" # Access Token (如果有) reconnect_interval = 5 # 断线重连间隔(秒) [bot] command = ["/"] # 指令前缀,支持多个,如 ["/", "#"] ``` ### 4. 运行 ```bash python main.py ``` ## 🛠️ 开发指南 ### 🔥 热重载调试 项目集成了 `watchdog` 文件监控。在开发过程中,你只需要: 1. 保持 `main.py` 运行。 2. 修改或新建 `base_plugins` 目录下的 `.py` 插件文件。 3. 保存文件。 4. 控制台会自动提示 `[HotReload] 插件重载完成`,新的逻辑立即生效。 ### 创建新插件 在 `base_plugins` 目录下创建一个新的 `.py` 文件(例如 `my_plugin.py`),框架会自动加载它。 ### 示例代码 #### 1. 注册消息指令 使用 `@matcher.command("指令名")` 注册指令。 ```python from core.command_manager import matcher from core.bot import Bot from models import MessageEvent # 注册 /hello 指令 @matcher.command("hello") async def handle_hello(bot: Bot, event: MessageEvent, args: list[str]): # args 是去除指令后的参数列表 await event.reply("你好!这里是 NEO Bot。") ``` #### 2. 监听通知事件 使用 `@matcher.on_notice("通知类型")` 监听通知。 ```python from core.command_manager import matcher from core.bot import Bot from models import GroupIncreaseNoticeEvent # 监听群成员增加事件 @matcher.on_notice("group_increase") async def welcome_new_member(bot: Bot, event: GroupIncreaseNoticeEvent): await bot.send_group_msg(event.group_id, f"欢迎新成员 {event.user_id} 加入!") ``` #### 3. 监听请求事件 使用 `@matcher.on_request("请求类型")` 监听请求。 ```python from core.command_manager import matcher from core.bot import Bot from models import FriendRequestEvent # 自动同意好友请求 @matcher.on_request("friend") async def auto_approve_friend(bot: Bot, event: FriendRequestEvent): await bot.call_api("set_friend_add_request", { "flag": event.flag, "approve": True }) ``` #### 4. 通用 API 调用 (call_api) 如果框架尚未封装某个 OneBot API,你可以使用 `bot.call_api` 直接调用。这是通用的备用调用方法。 ```python from core.command_manager import matcher from core.bot import Bot from models import MessageEvent @matcher.command("info") async def get_group_info(bot: Bot, event: MessageEvent, args: list[str]): # 直接调用 get_group_info API # action: API 名称 # params: API 参数字典 resp = await bot.call_api("get_group_info", { "group_id": event.group_id, "no_cache": False }) if resp.get("status") == "ok": group_name = resp["data"]["group_name"] await event.reply(f"当前群名:{group_name}") ``` ## 📚 事件模型说明 项目采用了基于工厂模式的事件处理系统,所有事件定义在 `models/events/` 下: * **MessageEvent**: 消息事件,包含 `PrivateMessageEvent` 和 `GroupMessageEvent`。支持 `await event.reply()` 快速回复。 * **NoticeEvent**: 通知事件,如 `FriendAddNoticeEvent`, `GroupRecallNoticeEvent` 等。 * **RequestEvent**: 请求事件,如 `FriendRequestEvent`, `GroupRequestEvent`。 * **MetaEvent**: 元事件,如心跳 `HeartbeatEvent`。 所有事件均继承自 `OneBotEvent`,并包含 `bot` 属性用于调用 API。 --- *Internal Use Only - DOGSOHA ond baby2016*