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 封装

• 消息相关
  • delete_msg: 撤回消息
  • get_msg: 获取消息
  • get_forward_msg: 获取合并转发消息
  • send_like: 发送点赞
• 群组管理
  • 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: 设置群组专属头衔
• 群组信息
  • get_group_info: 获取群信息
  • get_group_list: 获取群列表
  • get_group_member_info: 获取群成员信息
  • get_group_member_list: 获取群成员列表
  • get_group_honor_info: 获取群荣誉信息
• 用户相关
  • get_login_info: 获取登录号信息
  • get_stranger_info: 获取陌生人信息
  • get_friend_list: 获取好友列表
• 请求处理
  • set_friend_add_request: 处理加好友请求
  • set_group_add_request: 处理加群请求/邀请
• 系统/其他
  • get_version_info: 获取版本信息
  • get_status: 获取状态
  • can_send_image: 检查是否可以发送图片
  • can_send_record: 检查是否可以发送语音
  • clean_cache: 清理缓存

其他改进

• 日志系统优化: 引入更完善的日志记录机制，支持文件输出和日志级别控制。
• 异常处理增强: 增强插件执行过程中的异常捕获，防止单个插件崩溃影响整个 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 或 LLOneBot）

2. 安装依赖

pip install -r requirements.txt

3. 配置文件

修改根目录下的 config.toml，配置 WebSocket 连接信息：

[napcat_ws]
uri = "ws://127.0.0.1:30004"  # OneBot 实现端的 WebSocket 地址
token = "your_token"          # Access Token (如果有)
reconnect_interval = 5        # 断线重连间隔（秒）

[bot]
command = ["/"]               # 指令前缀，支持多个，如 ["/", "#"]

4. 运行

python main.py

🛠️ 开发指南

🔥 热重载调试

项目集成了 watchdog 文件监控。在开发过程中，你只需要：

1. 保持 main.py 运行。
2. 修改或新建 base_plugins 目录下的 .py 插件文件。
3. 保存文件。
4. 控制台会自动提示 [HotReload] 插件重载完成，新的逻辑立即生效。

创建新插件

在 base_plugins 目录下创建一个新的 .py 文件（例如 my_plugin.py），框架会自动加载它。

示例代码

1. 注册消息指令

使用 @matcher.command("指令名") 注册指令。

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("通知类型") 监听通知。

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("请求类型") 监听请求。

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 直接调用。这是通用的备用调用方法。

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