K2CrO4/NeoBot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Merge pull request #4 from Fairy-Oracle-Sanctuary/dev

Browse Source 
合并
This commit is contained in:
镀铬酸钾
2026-01-02 15:20:47 +08:00
committed by GitHub
parent 3be6bcdea8 5f80025c73
commit 99bcd3fddb
18 changed files with 1487 additions and 452 deletions
Download Patch File Download Diff File Expand all files Collapse all files

334
README.md
View File

@@ -1,12 +1,40 @@
# NEO Bot Framework
这是一个基于 Python 的轻量级 OneBot 11 协议机器人框架，专为内部团队开发设计。它通过 WebSocket 连接到 OneBot 实现端（如 NapCatQQ），提供了事件分发、指令管理和 API 调用封装。
## 📖 概述
NEO 是一个基于 Python 的现代化 OneBot 11 协议机器人框架，专为需要高性能、可扩展性和开发效率的团队设计。该框架通过 WebSocket 与各种 OneBot 实现端（如 NapCatQQ、LLOneBot 等）通信，提供了一套完整的机器人开发解决方案。
### 设计理念
NEO 框架的设计遵循以下核心理念：
1. **开发者友好**：简洁的 API 设计、完整的类型提示和详细的文档，让开发者能够快速上手和高效开发
2. **架构清晰**：采用模块化设计，分离关注点，使代码易于维护和扩展
3. **高性能异步**：基于 `asyncio` 和 `websockets` 构建，支持高并发消息处理
4. **类型安全**：全面使用 Python 类型系统，提供编译时类型检查，减少运行时错误
5. **热重载支持**：支持插件热重载，开发过程中修改代码无需重启机器人
### 核心价值
- **快速原型开发**：通过简洁的装饰器语法快速定义指令和事件处理器
- **生产环境就绪**：内置断线重连、错误处理和性能监控机制
- **可扩展架构**：支持自定义插件、中间件和权限系统
- **现代化开发体验**：支持热重载、类型提示和完整的 API 文档
### 适用场景
- QQ 群机器人管理
- 自动化客服与问答系统
- 游戏社区管理
- 团队内部工具集成
- 教育与培训辅助
## ✨ 特性
* **OneBot 11 标准支持**：完整支持 OneBot 11 的消息、通知、请求和元事件。
* **类型安全**：基于 `dataclasses` 的强类型事件模型，开发体验更佳。
* **插件系统**：轻量级的装饰器风格插件系统，支持指令 (`@matcher.command`) 和事件监听 (`@matcher.on_notice`, `@matcher.on_request`)。
* **插件元数据与内置帮助**：插件可通过 `__plugin_meta__` 变量进行自我描述。框架核心内置了 `/help` 指令，可自动收集并展示所有插件的帮助信息，无需手动维护。
* **🔥 热重载支持**：内置文件监控，修改 `base_plugins` 下的代码自动重载，无需重启，极大提升调试效率。
* **异步核心**：基于 `asyncio` 和 `websockets` 的高性能异步核心。
* **自动重连**：内置 WebSocket 断线重连机制。
@@ -14,12 +42,12 @@
## 📝 待办事项 (TODO)
### API 封装
- [ ] **消息相关**
- [x] **消息相关**
- `delete_msg`: 撤回消息
- `get_msg`: 获取消息
- `get_forward_msg`: 获取合并转发消息
- `send_like`: 发送点赞
- [ ] **群组管理**
- [x] **群组管理**
- `set_group_kick`: 群组踢人
- `set_group_ban`: 群组单人禁言
- `set_group_anonymous_ban`: 群组匿名禁言
@@ -30,27 +58,42 @@
- `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`: 发送合并转发消息
### 其他改进
- [x] **API 强类型封装**: 将 API 返回值从 `dict` 转换为数据模型对象。
- [ ] **日志系统优化**: 引入更完善的日志记录机制，支持文件输出和日志级别控制。
- [ ] **异常处理增强**: 增强插件执行过程中的异常捕获，防止单个插件崩溃影响整个 Bot。
- [ ] **中间件支持**: 添加消息处理中间件，支持在指令执行前/后进行拦截和处理。
@@ -60,12 +103,20 @@
```
NEO/
├── base_plugins/ # 基础插件目录，新建插件文件即可自动加载（支持热重载）
│ └── echo.py # 示例插件：实现 /echo 指令
├── plugins/ # 插件目录，新建插件文件即可自动加载（支持热重载）
│ └── echo.py # 示例插件：实现 /echo 和 /赞我 指令
├── core/ # 核心框架代码
│ ├── api/ # API 模块抽象层 (MessageAPI, GroupAPI, FriendAPI, AccountAPI)
│ │ ├── __init__.py
│ │ ├── account.py
│ │ ├── base.py
│ │ ├── friend.py
│ │ ├── group.py
│ │ └── message.py
│ ├── bot.py # Bot API 封装，提供 send_group_msg 等方法
│ ├── command_manager.py # 命令与事件分发器
│ ├── config_loader.py # 配置加载器
│ ├── plugin_manager.py # 插件加载与管理
│ └── ws.py # WebSocket 客户端核心
├── models/ # 数据模型
│ ├── events/ # OneBot 事件定义 (Message, Notice, Request, Meta)
@@ -174,8 +225,27 @@ async def auto_approve_friend(bot: Bot, event: FriendRequestEvent):
})
```
#### 4. 通用 API 调用 (call_api)
#### 4. API 调用方式对比
框架提供两种 API 调用方式：**类型化 API**（推荐）和 **通用 API**（备用）。
##### 方式一：类型化 API（推荐）
对于已封装的 API，框架提供了类型化的方法，返回数据模型对象而非原始字典：
```python
from core.command_manager import matcher
from core.bot import Bot
from models import MessageEvent
from models.objects import Group
@matcher.command("info")
async def get_group_info_typed(bot: Bot, event: MessageEvent, args: list[str]):
# 使用类型化 API，返回 Group 对象
group: Group = await bot.get_group_info(event.group_id)
await event.reply(f"群名：{group.group_name}\n成员数：{group.member_count}\n创建时间：{group.create_time}")
```
##### 方式二：通用 API（备用）
如果框架尚未封装某个 OneBot API，你可以使用 `bot.call_api` 直接调用。这是通用的备用调用方法。
```python
@@ -183,8 +253,8 @@ 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]):
@matcher.command("info_legacy")
async def get_group_info_legacy(bot: Bot, event: MessageEvent, args: list[str]):
# 直接调用 get_group_info API
# action: API 名称
# params: API 参数字典
@@ -198,6 +268,161 @@ async def get_group_info(bot: Bot, event: MessageEvent, args: list[str]):
await event.reply(f"当前群名：{group_name}")
```
**建议**：优先使用类型化 API，获得更好的类型安全和代码提示。仅在框架未封装特定 API 时使用通用 API。
## 📖 插件开发指南
### 插件基本结构
一个标准的插件文件应该包含以下部分：
1. **模块文档字符串**：描述插件功能
2. **导入必要的模块**：从 `core` 和 `models` 导入所需类
3. **使用装饰器注册事件处理器**：`@matcher.command()`, `@matcher.on_notice()`, `@matcher.on_request()`
4. **异步函数实现业务逻辑**：使用 `async def` 定义处理函数
### 插件元数据 (`__plugin_meta__`)
为了实现插件的自动发现和帮助信息的自动生成，框架引入了插件元数据机制。你需要在你的插件模块中定义一个名为 `__plugin_meta__` 的字典。
`load_all_plugins` 函数在加载插件时会自动读取这个变量，并将其注册到 `CommandManager` 中。`/help` 指令会遍历所有已注册的元数据，生成格式化的帮助信息。
一个标准的 `__plugin_meta__` 包含以下字段：
- `name` (str): 插件的友好名称，例如 "回声"。
- `description` (str): 对插件功能的简短描述。
- `usage` (str): 插件的使用方法，可以包含多个指令和它们的说明。
**示例：**
```python
# base_plugins/echo.py
__plugin_meta__ = {
"name": "回声与交互",
"description": "提供 echo 和 赞我 功能",
"usage": "/echo [内容] - 复读内容\n/赞我 - 让机器人给你点赞",
}
```
### 使用类型化 API
框架现已提供完整的类型化 API 封装，建议优先使用这些封装方法而非原始的 `call_api`：
| API 方法 | 返回类型 | 说明 |
|---------|---------|------|
| `bot.send_group_msg()` | `Message` | 发送群消息 |
| `bot.get_group_info()` | `Group` | 获取群信息 |
| `bot.get_group_member_info()` | `GroupMember` | 获取群成员信息 |
| `bot.get_friend_list()` | `List[Friend]` | 获取好友列表 |
| `bot.get_login_info()` | `LoginInfo` | 获取登录信息 |
| `bot.get_version_info()` | `VersionInfo` | 获取版本信息 |
#### 示例：使用类型化 API 重构群信息查询
```python
from core.command_manager import matcher
from core.bot import Bot
from models import MessageEvent
from models.objects import Group
@matcher.command("group_info")
async def get_group_info_typed(bot: Bot, event: MessageEvent, args: list[str]):
# 使用类型化 API，返回 Group 对象而非字典
group: Group = await bot.get_group_info(event.group_id)
await event.reply(f"群名：{group.group_name}\n成员数：{group.member_count}\n创建时间：{group.create_time}")
```
### 事件处理模式
除了基本的消息指令，还可以处理多种事件类型：
#### 1. 通知事件处理
```python
from models import GroupCardChangeEvent
@matcher.on_notice("group_card")
async def handle_group_card_change(bot: Bot, event: GroupCardChangeEvent):
# event.card_new 是新名片，event.card_old 是旧名片
await bot.send_group_msg(event.group_id, f"成员 {event.user_id} 的名片从 '{event.card_old}' 改为 '{event.card_new}'")
```
#### 2. 请求事件处理
```python
from models import GroupRequestEvent
@matcher.on_request("group")
async def handle_group_request(bot: Bot, event: GroupRequestEvent):
# 根据请求类型处理
if event.sub_type == "add":
# 自动同意加群请求
await bot.set_group_add_request(event.flag, event.sub_type, approve=True)
await bot.send_group_msg(event.group_id, f"已同意用户 {event.user_id} 的加群请求")
```
### 错误处理
建议在插件中添加适当的错误处理，避免单个插件崩溃影响整个机器人：
```python
@matcher.command("dangerous")
async def dangerous_command(bot: Bot, event: MessageEvent, args: list[str]):
try:
# 可能失败的操作
result = await bot.call_api("some_api", {"param": "value"})
await event.reply(f"成功：{result}")
except Exception as e:
await event.reply(f"执行失败：{str(e)}")
# 记录日志
import logging
logging.error(f"插件执行错误：{e}", exc_info=True)
```
### 插件开发最佳实践
1. **单一职责**：每个插件专注于一个功能领域
2. **错误处理**：妥善处理可能发生的异常
3. **类型提示**：为函数参数和返回值添加类型提示
4. **文档完整**：为每个函数添加文档字符串
5. **性能考虑**：避免在插件中执行耗时同步操作
6. **资源清理**：必要时使用 `try...finally` 确保资源释放
### 示例：完整插件模板
```python
"""
天气查询插件
提供 /weather 指令，查询指定城市的天气信息。
"""
from core.command_manager import matcher
from core.bot import Bot
from models import MessageEvent
# 插件元数据，用于 help 指令
__plugin_meta__ = {
"name": "天气查询",
"description": "查询指定城市的天气信息",
"usage": "/weather [城市名称]",
}
@matcher.command("weather")
async def handle_weather(bot: Bot, event: MessageEvent, args: list[str]):
"""
查询天气信息
:param bot: Bot 实例
:param event: 消息事件对象
:param args: 指令参数列表（城市名称）
"""
if not args:
await event.reply("请输入城市名称，例如：/weather 北京")
return
city = " ".join(args)
try:
# 这里可以调用天气 API
weather_info = f"{city} 的天气：晴，25℃"
await event.reply(weather_info)
except Exception as e:
await event.reply(f"查询天气失败：{str(e)}")
# 可以注册多个事件处理器
@matcher.on_notice("group_increase")
async def welcome_new_member(bot: Bot, event):
await bot.send_group_msg(event.group_id, f"欢迎新成员 {event.user_id} 加入！")
```
## 📚 事件模型说明
项目采用了基于工厂模式的事件处理系统，所有事件定义在 `models/events/` 下：
@@ -209,5 +434,88 @@ async def get_group_info(bot: Bot, event: MessageEvent, args: list[str]):
所有事件均继承自 `OneBotEvent`，并包含 `bot` 属性用于调用 API。
## 🏗️ 技术架构
NEO 框架采用分层架构设计，各层职责明确，便于维护和扩展：
### 架构层次
1. **通信层 (WebSocket Client)**
- 负责与 OneBot 实现端的 Web Socket连接
- 实现断线自动重连机制
- 处理原始消息的收发和协议解析
2. **API 抽象层 (API Mixins)**
- 提供类型安全的 API 封装
- 按功能领域划分：消息、群组、好友、账号
- 所有 API 返回强类型数据模型对象
3. **业务逻辑层 (Bot & Command Manager)**
- Bot 类组合所有 API 功能
- 指令和事件分发器
- 插件加载和管理
4. **插件层 (Plugins)**
- 支持热重载的插件系统
- 装饰器风格的注册方式
- 独立的业务逻辑模块
5. **数据模型层 (Models)**
- 基于 dataclasses 的事件模型
- API 响应数据模型
- 类型安全的序列化/反序列化
### 核心组件交互
```
┌─────────────────────────────────────┐
│ 插件层 (Plugins) │
│ @matcher.command() │
│ @matcher.on_notice() │
└──────────────┬──────────────────────┘
│
┌──────────────▼──────────────────────┐
│ 业务逻辑层 (Command Manager) │
│ • 事件分发与路由 │
│ • 指令参数解析 │
└──────────────┬──────────────────────┘
│
┌──────────────▼──────────────────────┐
│ Bot 组合类 │
│ • 继承所有 API Mixin │
│ • 提供统一接口 │
└──────────────┬──────────────────────┘
│
┌──────────────▼──────────────────────┐
│ API 抽象层 (Mixin) │
│ • MessageAPI │
│ • GroupAPI │
│ • FriendAPI │
│ • AccountAPI │
└──────────────┬──────────────────────┘
│
┌──────────────▼──────────────────────┐
│ 通信层 (WebSocket) │
│ • 连接管理 │
│ • 消息编解码 │
│ • 断线重连 │
└─────────────────────────────────────┘
```
### 设计模式应用
- **工厂模式**：事件对象的创建和管理
- **装饰器模式**：插件和指令的注册
- **组合模式**：Bot 类通过继承组合 API 功能
- **观察者模式**：事件监听和处理
- **策略模式**：不同的消息处理策略
### 性能特点
- **异步非阻塞**：全面基于 asyncio，支持高并发
- **内存高效**：事件和模型对象使用 dataclasses，内存占用小
- **快速响应**：插件热重载和事件分发机制确保快速响应
- **可扩展性**：模块化设计便于功能扩展和定制
---
*Internal Use Only - DOGSOHA ond baby2016*
*Internal Use Only - DOGSOHA ond baby2016 by Fairy-Oracle-Sanctuary*

39
base_plugins/__init__.py
View File

@@ -1,39 +0,0 @@
import importlib
import os
import pkgutil
import sys
def load_all_plugins():
"""
扫描并加载当前包下所有的插件（支持文件和文件夹）
该函数会遍历当前目录下的所有模块：
1. 如果模块已加载，则执行 reload 操作（用于热重载）
2. 如果模块未加载，则执行 import 操作
加载过程中会打印详细的日志信息。
"""
package_name = __package__
package_path = [os.path.dirname(__file__)]
print(f" 正在从 {package_name} 加载插件...")
for loader, module_name, is_pkg in pkgutil.iter_modules(package_path):
full_module_name = f"{package_name}.{module_name}"
try:
if full_module_name in sys.modules:
importlib.reload(sys.modules[full_module_name])
action = "重载"
else:
importlib.import_module(full_module_name)
action = "加载"
type_str = "包" if is_pkg else "文件"
print(f" [{type_str}] 成功{action}: {module_name}")
except Exception as e:
print(f" {action if 'action' in locals() else '加载'}插件 {module_name} 失败: {e}")
load_all_plugins()

4
config.toml
View File

@@ -1,6 +1,6 @@
[napcat_ws]
uri = "ws://127.0.0.1:30004"
token = "12333"
uri = "ws://114.66.58.203:3001"
token = "&d_VTfksE%}ul?_Y"
reconnect_interval = 5
[bot]

13
core/api/__init__.py Normal file
View File

@@ -0,0 +1,13 @@
from .base import BaseAPI
from .message import MessageAPI
from .group import GroupAPI
from .friend import FriendAPI
from .account import AccountAPI
__all__ = [
"BaseAPI",
"MessageAPI",
"GroupAPI",
"FriendAPI",
"AccountAPI",
]

124
core/api/account.py Normal file
View File

@@ -0,0 +1,124 @@
"""
账号相关 API 模块
"""
from typing import Dict, Any
from .base import BaseAPI
from models.objects import LoginInfo, VersionInfo, Status
class AccountAPI(BaseAPI):
"""
账号相关 API Mixin
"""
async def get_login_info(self) -> LoginInfo:
"""
获取登录号信息
:return: 登录信息对象
"""
res = await self.call_api("get_login_info")
return LoginInfo(**res)
async def get_version_info(self) -> VersionInfo:
"""
获取版本信息
:return: 版本信息对象
"""
res = await self.call_api("get_version_info")
return VersionInfo(**res)
async def get_status(self) -> Status:
"""
获取状态
:return: 状态对象
"""
res = await self.call_api("get_status")
return Status(**res)
async def bot_exit(self) -> Dict[str, Any]:
"""
退出机器人
:return: API 响应结果
"""
return await self.call_api("bot_exit")
async def set_self_longnick(self, long_nick: str) -> Dict[str, Any]:
"""
设置个性签名
:param long_nick: 个性签名内容
:return: API 响应结果
"""
return await self.call_api("set_self_longnick", {"longNick": long_nick})
async def set_input_status(self, user_id: int, event_type: int) -> Dict[str, Any]:
"""
设置输入状态
:param user_id: 用户 ID
:param event_type: 事件类型
:return: API 响应结果
"""
return await self.call_api("set_input_status", {"user_id": user_id, "event_type": event_type})
async def set_diy_online_status(self, face_id: int, face_type: int, wording: str) -> Dict[str, Any]:
"""
设置自定义在线状态
:param face_id: 状态 ID
:param face_type: 状态类型
:param wording: 状态描述
:return: API 响应结果
"""
return await self.call_api("set_diy_online_status", {
"face_id": face_id,
"face_type": face_type,
"wording": wording
})
async def set_online_status(self, status_code: int) -> Dict[str, Any]:
"""
设置在线状态
:param status_code: 状态码
:return: API 响应结果
"""
return await self.call_api("set_online_status", {"status_code": status_code})
async def set_qq_profile(self, **kwargs) -> Dict[str, Any]:
"""
设置 QQ 资料
:param kwargs: 个人资料相关参数
:return: API 响应结果
"""
return await self.call_api("set_qq_profile", kwargs)
async def set_qq_avatar(self, **kwargs) -> Dict[str, Any]:
"""
设置 QQ 头像
:param kwargs: 头像相关参数
:return: API 响应结果
"""
return await self.call_api("set_qq_avatar", kwargs)
async def get_clientkey(self) -> Dict[str, Any]:
"""
获取客户端密钥
:return: API 响应结果
"""
return await self.call_api("get_clientkey")
async def clean_cache(self) -> Dict[str, Any]:
"""
清理缓存
:return: API 响应结果
"""
return await self.call_api("clean_cache")

24
core/api/base.py Normal file
View File

@@ -0,0 +1,24 @@
"""
API 基础模块
定义了 API 调用的基础接口。
"""
from abc import ABC, abstractmethod
from typing import Any, Dict, Optional
class BaseAPI(ABC):
"""
API 基础抽象类
"""
@abstractmethod
async def call_api(self, action: str, params: Optional[Dict[str, Any]] = None) -> Any:
"""
调用 API
:param action: API 动作名称
:param params: API 参数
:return: API 响应结果
"""
raise NotImplementedError

53
core/api/friend.py Normal file
View File

@@ -0,0 +1,53 @@
"""
好友相关 API 模块
"""
from typing import List, Dict, Any
from .base import BaseAPI
from models.objects import FriendInfo, StrangerInfo
class FriendAPI(BaseAPI):
"""
好友相关 API Mixin
"""
async def send_like(self, user_id: int, times: int = 1) -> Dict[str, Any]:
"""
发送点赞
:param user_id: 对方 QQ 号
:param times: 点赞次数
:return: API 响应结果
"""
return await self.call_api("send_like", {"user_id": user_id, "times": times})
async def get_stranger_info(self, user_id: int, no_cache: bool = False) -> StrangerInfo:
"""
获取陌生人信息
:param user_id: QQ 号
:param no_cache: 是否不使用缓存
:return: 陌生人信息对象
"""
res = await self.call_api("get_stranger_info", {"user_id": user_id, "no_cache": no_cache})
return StrangerInfo(**res)
async def get_friend_list(self) -> List[FriendInfo]:
"""
获取好友列表
:return: 好友信息对象列表
"""
res = await self.call_api("get_friend_list")
return [FriendInfo(**item) for item in res]
async def set_friend_add_request(self, flag: str, approve: bool = True, remark: str = "") -> Dict[str, Any]:
"""
处理加好友请求
:param flag: 加好友请求的 flag（需从上报的数据中获取）
:param approve: 是否同意请求
:param remark: 添加后的好友备注（仅在同意时有效）
:return: API 响应结果
"""
return await self.call_api("set_friend_add_request", {"flag": flag, "approve": approve, "remark": remark})

190
core/api/group.py Normal file
View File

@@ -0,0 +1,190 @@
"""
群组相关 API 模块
"""
from typing import List, Dict, Any, Optional
from .base import BaseAPI
from models.objects import GroupInfo, GroupMemberInfo, GroupHonorInfo
class GroupAPI(BaseAPI):
"""
群组相关 API Mixin
"""
async def set_group_kick(self, group_id: int, user_id: int, reject_add_request: bool = False) -> Dict[str, Any]:
"""
群组踢人
:param group_id: 群号
:param user_id: 要踢的 QQ 号
:param reject_add_request: 拒绝此人的加群请求
:return: API 响应结果
"""
return await self.call_api("set_group_kick", {"group_id": group_id, "user_id": user_id, "reject_add_request": reject_add_request})
async def set_group_ban(self, group_id: int, user_id: int, duration: int = 30 * 60) -> Dict[str, Any]:
"""
群组单人禁言
:param group_id: 群号
:param user_id: 要禁言的 QQ 号
:param duration: 禁言时长（秒），0 表示解除禁言
:return: API 响应结果
"""
return await self.call_api("set_group_ban", {"group_id": group_id, "user_id": user_id, "duration": duration})
async def set_group_anonymous_ban(self, group_id: int, anonymous: Dict[str, Any] = None, duration: int = 30 * 60, flag: str = None) -> Dict[str, Any]:
"""
群组匿名禁言
:param group_id: 群号
:param anonymous: 可选，要禁言的匿名用户对象（群消息事件的 anonymous 字段）
:param duration: 禁言时长（秒）
:param flag: 可选，要禁言的匿名用户的 flag（需从群消息事件的 anonymous 字段中获取）
:return: API 响应结果
"""
params = {"group_id": group_id, "duration": duration}
if anonymous:
params["anonymous"] = anonymous
if flag:
params["flag"] = flag
return await self.call_api("set_group_anonymous_ban", params)
async def set_group_whole_ban(self, group_id: int, enable: bool = True) -> Dict[str, Any]:
"""
群组全员禁言
:param group_id: 群号
:param enable: 是否开启
:return: API 响应结果
"""
return await self.call_api("set_group_whole_ban", {"group_id": group_id, "enable": enable})
async def set_group_admin(self, group_id: int, user_id: int, enable: bool = True) -> Dict[str, Any]:
"""
群组设置管理员
:param group_id: 群号
:param user_id: 要设置的 QQ 号
:param enable: True 为设置，False 为取消
:return: API 响应结果
"""
return await self.call_api("set_group_admin", {"group_id": group_id, "user_id": user_id, "enable": enable})
async def set_group_anonymous(self, group_id: int, enable: bool = True) -> Dict[str, Any]:
"""
群组匿名
:param group_id: 群号
:param enable: 是否开启
:return: API 响应结果
"""
return await self.call_api("set_group_anonymous", {"group_id": group_id, "enable": enable})
async def set_group_card(self, group_id: int, user_id: int, card: str = "") -> Dict[str, Any]:
"""
设置群名片（群备注）
:param group_id: 群号
:param user_id: 要设置的 QQ 号
:param card: 群名片内容，不填或空字符串表示删除群名片
:return: API 响应结果
"""
return await self.call_api("set_group_card", {"group_id": group_id, "user_id": user_id, "card": card})
async def set_group_name(self, group_id: int, group_name: str) -> Dict[str, Any]:
"""
设置群名
:param group_id: 群号
:param group_name: 新群名
:return: API 响应结果
"""
return await self.call_api("set_group_name", {"group_id": group_id, "group_name": group_name})
async def set_group_leave(self, group_id: int, is_dismiss: bool = False) -> Dict[str, Any]:
"""
退出群组
:param group_id: 群号
:param is_dismiss: 是否解散，如果登录号是群主，则仅在此项为 True 时能够解散
:return: API 响应结果
"""
return await self.call_api("set_group_leave", {"group_id": group_id, "is_dismiss": is_dismiss})
async def set_group_special_title(self, group_id: int, user_id: int, special_title: str = "", duration: int = -1) -> Dict[str, Any]:
"""
设置群组专属头衔
:param group_id: 群号
:param user_id: 要设置的 QQ 号
:param special_title: 专属头衔，不填或空字符串表示删除
:param duration: 有效期（秒），-1 表示永久
:return: API 响应结果
"""
return await self.call_api("set_group_special_title", {"group_id": group_id, "user_id": user_id, "special_title": special_title, "duration": duration})
async def get_group_info(self, group_id: int, no_cache: bool = False) -> GroupInfo:
"""
获取群信息
:param group_id: 群号
:param no_cache: 是否不使用缓存
:return: 群信息对象
"""
res = await self.call_api("get_group_info", {"group_id": group_id, "no_cache": no_cache})
return GroupInfo(**res)
async def get_group_list(self) -> List[GroupInfo]:
"""
获取群列表
:return: 群信息对象列表
"""
res = await self.call_api("get_group_list")
return [GroupInfo(**item) for item in res]
async def get_group_member_info(self, group_id: int, user_id: int, no_cache: bool = False) -> GroupMemberInfo:
"""
获取群成员信息
:param group_id: 群号
:param user_id: QQ 号
:param no_cache: 是否不使用缓存
:return: 群成员信息对象
"""
res = await self.call_api("get_group_member_info", {"group_id": group_id, "user_id": user_id, "no_cache": no_cache})
return GroupMemberInfo(**res)
async def get_group_member_list(self, group_id: int) -> List[GroupMemberInfo]:
"""
获取群成员列表
:param group_id: 群号
:return: 群成员信息对象列表
"""
res = await self.call_api("get_group_member_list", {"group_id": group_id})
return [GroupMemberInfo(**item) for item in res]
async def get_group_honor_info(self, group_id: int, type: str) -> GroupHonorInfo:
"""
获取群荣誉信息
:param group_id: 群号
:param type: 要获取的群荣誉类型，可传入 talkative, performer, legend, strong_newbie, emotion 等
:return: 群荣誉信息对象
"""
res = await self.call_api("get_group_honor_info", {"group_id": group_id, "type": type})
return GroupHonorInfo(**res)
async def set_group_add_request(self, flag: str, sub_type: str, approve: bool = True, reason: str = "") -> Dict[str, Any]:
"""
处理加群请求/邀请
:param flag: 加群请求的 flag（需从上报的数据中获取）
:param sub_type: add 或 invite，请求类型（需要与上报消息中的 sub_type 字段相符）
:param approve: 是否同意请求/邀请
:param reason: 拒绝理由（仅在拒绝时有效）
:return: API 响应结果
"""
return await self.call_api("set_group_add_request", {"flag": flag, "sub_type": sub_type, "approve": approve, "reason": reason})

141
core/api/message.py Normal file
View File

@@ -0,0 +1,141 @@
"""
消息相关 API 模块
"""
from typing import Union, List, Dict, Any, TYPE_CHECKING
from .base import BaseAPI
if TYPE_CHECKING:
from models import MessageSegment, OneBotEvent
class MessageAPI(BaseAPI):
"""
消息相关 API Mixin
"""
async def send_group_msg(self, group_id: int, message: Union[str, "MessageSegment", List["MessageSegment"]], auto_escape: bool = False) -> Dict[str, Any]:
"""
发送群消息
:param group_id: 群号
:param message: 消息内容，可以是字符串、MessageSegment 对象或 MessageSegment 列表
:param auto_escape: 是否自动转义（仅当 message 为字符串时有效）
:return: API 响应结果
"""
return await self.call_api(
"send_group_msg", {"group_id": group_id, "message": self._process_message(message), "auto_escape": auto_escape}
)
async def send_private_msg(self, user_id: int, message: Union[str, "MessageSegment", List["MessageSegment"]], auto_escape: bool = False) -> Dict[str, Any]:
"""
发送私聊消息
:param user_id: 用户 QQ 号
:param message: 消息内容，可以是字符串、MessageSegment 对象或 MessageSegment 列表
:param auto_escape: 是否自动转义（仅当 message 为字符串时有效）
:return: API 响应结果
"""
return await self.call_api(
"send_private_msg", {"user_id": user_id, "message": self._process_message(message), "auto_escape": auto_escape}
)
async def send(self, event: "OneBotEvent", message: Union[str, "MessageSegment", List["MessageSegment"]], auto_escape: bool = False) -> Dict[str, Any]:
"""
智能发送消息，根据事件类型自动选择发送方式
:param event: 触发事件对象
:param message: 消息内容
:param auto_escape: 是否自动转义
:return: API 响应结果
"""
# 如果是消息事件，直接调用 reply
if hasattr(event, "reply"):
await event.reply(message, auto_escape)
return {"status": "ok", "msg": "Replied via event.reply()"}
# 尝试从事件中获取 user_id 或 group_id
user_id = getattr(event, "user_id", None)
group_id = getattr(event, "group_id", None)
if group_id:
return await self.send_group_msg(group_id, message, auto_escape)
elif user_id:
return await self.send_private_msg(user_id, message, auto_escape)
return {"status": "failed", "msg": "Unknown message target"}
async def delete_msg(self, message_id: int) -> Dict[str, Any]:
"""
撤回消息
:param message_id: 消息 ID
:return: API 响应结果
"""
return await self.call_api("delete_msg", {"message_id": message_id})
async def get_msg(self, message_id: int) -> Dict[str, Any]:
"""
获取消息
:param message_id: 消息 ID
:return: API 响应结果
"""
return await self.call_api("get_msg", {"message_id": message_id})
async def get_forward_msg(self, id: str) -> Dict[str, Any]:
"""
获取合并转发消息
:param id: 合并转发 ID
:return: API 响应结果
"""
return await self.call_api("get_forward_msg", {"id": id})
async def can_send_image(self) -> Dict[str, Any]:
"""
检查是否可以发送图片
:return: API 响应结果
"""
return await self.call_api("can_send_image")
async def can_send_record(self) -> Dict[str, Any]:
"""
检查是否可以发送语音
:return: API 响应结果
"""
return await self.call_api("can_send_record")
def _process_message(self, message: Union[str, "MessageSegment", List["MessageSegment"]]) -> Union[str, List[Dict[str, Any]]]:
"""
处理消息内容，将其转换为 API 可接受的格式
:param message: 原始消息内容
:return: 处理后的消息内容
"""
if isinstance(message, str):
return message
# 避免循环导入，在运行时导入
from models import MessageSegment
if isinstance(message, MessageSegment):
return [self._segment_to_dict(message)]
if isinstance(message, list):
return [self._segment_to_dict(m) for m in message if isinstance(m, MessageSegment)]
return str(message)
def _segment_to_dict(self, segment: "MessageSegment") -> Dict[str, Any]:
"""
将 MessageSegment 对象转换为字典
:param segment: MessageSegment 对象
:return: 字典格式的消息段
"""
return {
"type": segment.type,
"data": segment.data
}

383
core/bot.py
View File

@@ -3,16 +3,18 @@ Bot 抽象模块
定义了 Bot 类，封装了 OneBot API 的调用逻辑，提供了便捷的消息发送方法。
"""
from typing import TYPE_CHECKING, Union, List
from typing import TYPE_CHECKING, Dict, Any
if TYPE_CHECKING:
from .ws import WS
from models import OneBotEvent, MessageSegment
from .api import MessageAPI, GroupAPI, FriendAPI, AccountAPI
class Bot:
class Bot(MessageAPI, GroupAPI, FriendAPI, AccountAPI):
"""
Bot 抽象类，封装 API 调用和常用操作
继承各个 API Mixin 以提高代码的可维护性
"""
def __init__(self, ws_client: "WS"):
@@ -23,7 +25,7 @@ class Bot:
"""
self.ws = ws_client
async def call_api(self, action: str, params: dict = None) -> dict:
async def call_api(self, action: str, params: Dict[str, Any] = None) -> Any:
"""
调用 OneBot API
@@ -31,375 +33,4 @@ class Bot:
:param params: API 参数
:return: API 响应结果
"""
return await self.ws.call_api(action, params)
async def send_group_msg(self, group_id: int, message: Union[str, "MessageSegment", List["MessageSegment"]], auto_escape: bool = False) -> dict:
"""
发送群消息
:param group_id: 群号
:param message: 消息内容，可以是字符串、MessageSegment 对象或 MessageSegment 列表
:param auto_escape: 是否自动转义（仅当 message 为字符串时有效）
:return: API 响应结果
"""
return await self.call_api(
"send_group_msg", {"group_id": group_id, "message": self._process_message(message), "auto_escape": auto_escape}
)
async def send_private_msg(self, user_id: int, message: Union[str, "MessageSegment", List["MessageSegment"]], auto_escape: bool = False) -> dict:
"""
发送私聊消息
:param user_id: 用户 QQ 号
:param message: 消息内容，可以是字符串、MessageSegment 对象或 MessageSegment 列表
:param auto_escape: 是否自动转义（仅当 message 为字符串时有效）
:return: API 响应结果
"""
return await self.call_api(
"send_private_msg", {"user_id": user_id, "message": self._process_message(message), "auto_escape": auto_escape}
)
async def send(self, event: "OneBotEvent", message: Union[str, "MessageSegment", List["MessageSegment"]], auto_escape: bool = False) -> dict:
"""
智能发送消息，根据事件类型自动选择发送方式
:param event: 触发事件对象
:param message: 消息内容
:param auto_escape: 是否自动转义
:return: API 响应结果
"""
# 如果是消息事件，直接调用 reply
if hasattr(event, "reply"):
await event.reply(message, auto_escape)
return {"status": "ok", "msg": "Replied via event.reply()"}
# 尝试从事件中获取 user_id 或 group_id
user_id = getattr(event, "user_id", None)
group_id = getattr(event, "group_id", None)
if group_id:
return await self.send_group_msg(group_id, message, auto_escape)
elif user_id:
return await self.send_private_msg(user_id, message, auto_escape)
return {"status": "failed", "msg": "Unknown message target"}
async def delete_msg(self, message_id: int) -> dict:
"""
撤回消息
:param message_id: 消息 ID
:return: API 响应结果
"""
return await self.call_api("delete_msg", {"message_id": message_id})
async def get_msg(self, message_id: int) -> dict:
"""
获取消息
:param message_id: 消息 ID
:return: API 响应结果
"""
return await self.call_api("get_msg", {"message_id": message_id})
async def get_forward_msg(self, id: str) -> dict:
"""
获取合并转发消息
:param id: 合并转发 ID
:return: API 响应结果
"""
return await self.call_api("get_forward_msg", {"id": id})
async def send_like(self, user_id: int, times: int = 1) -> dict:
"""
发送点赞
:param user_id: 对方 QQ 号
:param times: 点赞次数
:return: API 响应结果
"""
return await self.call_api("send_like", {"user_id": user_id, "times": times})
async def set_group_kick(self, group_id: int, user_id: int, reject_add_request: bool = False) -> dict:
"""
群组踢人
:param group_id: 群号
:param user_id: 要踢的 QQ 号
:param reject_add_request: 拒绝此人的加群请求
:return: API 响应结果
"""
return await self.call_api("set_group_kick", {"group_id": group_id, "user_id": user_id, "reject_add_request": reject_add_request})
async def set_group_ban(self, group_id: int, user_id: int, duration: int = 30 * 60) -> dict:
"""
群组单人禁言
:param group_id: 群号
:param user_id: 要禁言的 QQ 号
:param duration: 禁言时长（秒），0 表示解除禁言
:return: API 响应结果
"""
return await self.call_api("set_group_ban", {"group_id": group_id, "user_id": user_id, "duration": duration})
async def set_group_anonymous_ban(self, group_id: int, anonymous: dict = None, duration: int = 30 * 60, flag: str = None) -> dict:
"""
群组匿名禁言
:param group_id: 群号
:param anonymous: 可选，要禁言的匿名用户对象（群消息事件的 anonymous 字段）
:param duration: 禁言时长（秒）
:param flag: 可选，要禁言的匿名用户的 flag（需从群消息事件的 anonymous 字段中获取）
:return: API 响应结果
"""
params = {"group_id": group_id, "duration": duration}
if anonymous:
params["anonymous"] = anonymous
if flag:
params["flag"] = flag
return await self.call_api("set_group_anonymous_ban", params)
async def set_group_whole_ban(self, group_id: int, enable: bool = True) -> dict:
"""
群组全员禁言
:param group_id: 群号
:param enable: 是否开启
:return: API 响应结果
"""
return await self.call_api("set_group_whole_ban", {"group_id": group_id, "enable": enable})
async def set_group_admin(self, group_id: int, user_id: int, enable: bool = True) -> dict:
"""
群组设置管理员
:param group_id: 群号
:param user_id: 要设置的 QQ 号
:param enable: True 为设置，False 为取消
:return: API 响应结果
"""
return await self.call_api("set_group_admin", {"group_id": group_id, "user_id": user_id, "enable": enable})
async def set_group_anonymous(self, group_id: int, enable: bool = True) -> dict:
"""
群组匿名
:param group_id: 群号
:param enable: 是否开启
:return: API 响应结果
"""
return await self.call_api("set_group_anonymous", {"group_id": group_id, "enable": enable})
async def set_group_card(self, group_id: int, user_id: int, card: str = "") -> dict:
"""
设置群名片（群备注）
:param group_id: 群号
:param user_id: 要设置的 QQ 号
:param card: 群名片内容，不填或空字符串表示删除群名片
:return: API 响应结果
"""
return await self.call_api("set_group_card", {"group_id": group_id, "user_id": user_id, "card": card})
async def set_group_name(self, group_id: int, group_name: str) -> dict:
"""
设置群名
:param group_id: 群号
:param group_name: 新群名
:return: API 响应结果
"""
return await self.call_api("set_group_name", {"group_id": group_id, "group_name": group_name})
async def set_group_leave(self, group_id: int, is_dismiss: bool = False) -> dict:
"""
退出群组
:param group_id: 群号
:param is_dismiss: 是否解散，如果登录号是群主，则仅在此项为 True 时能够解散
:return: API 响应结果
"""
return await self.call_api("set_group_leave", {"group_id": group_id, "is_dismiss": is_dismiss})
async def set_group_special_title(self, group_id: int, user_id: int, special_title: str = "", duration: int = -1) -> dict:
"""
设置群组专属头衔
:param group_id: 群号
:param user_id: 要设置的 QQ 号
:param special_title: 专属头衔，不填或空字符串表示删除
:param duration: 有效期（秒），-1 表示永久
:return: API 响应结果
"""
return await self.call_api("set_group_special_title", {"group_id": group_id, "user_id": user_id, "special_title": special_title, "duration": duration})
async def get_group_info(self, group_id: int, no_cache: bool = False) -> dict:
"""
获取群信息
:param group_id: 群号
:param no_cache: 是否不使用缓存
:return: API 响应结果
"""
return await self.call_api("get_group_info", {"group_id": group_id, "no_cache": no_cache})
async def get_group_list(self) -> dict:
"""
获取群列表
:return: API 响应结果
"""
return await self.call_api("get_group_list")
async def get_group_member_info(self, group_id: int, user_id: int, no_cache: bool = False) -> dict:
"""
获取群成员信息
:param group_id: 群号
:param user_id: QQ 号
:param no_cache: 是否不使用缓存
:return: API 响应结果
"""
return await self.call_api("get_group_member_info", {"group_id": group_id, "user_id": user_id, "no_cache": no_cache})
async def get_group_member_list(self, group_id: int) -> dict:
"""
获取群成员列表
:param group_id: 群号
:return: API 响应结果
"""
return await self.call_api("get_group_member_list", {"group_id": group_id})
async def get_group_honor_info(self, group_id: int, type: str) -> dict:
"""
获取群荣誉信息
:param group_id: 群号
:param type: 要获取的群荣誉类型，可传入 talkative, performer, legend, strong_newbie, emotion 等
:return: API 响应结果
"""
return await self.call_api("get_group_honor_info", {"group_id": group_id, "type": type})
async def get_login_info(self) -> dict:
"""
获取登录号信息
:return: API 响应结果
"""
return await self.call_api("get_login_info")
async def get_stranger_info(self, user_id: int, no_cache: bool = False) -> dict:
"""
获取陌生人信息
:param user_id: QQ 号
:param no_cache: 是否不使用缓存
:return: API 响应结果
"""
return await self.call_api("get_stranger_info", {"user_id": user_id, "no_cache": no_cache})
async def get_friend_list(self) -> dict:
"""
获取好友列表
:return: API 响应结果
"""
return await self.call_api("get_friend_list")
async def set_friend_add_request(self, flag: str, approve: bool = True, remark: str = "") -> dict:
"""
处理加好友请求
:param flag: 加好友请求的 flag（需从上报的数据中获取）
:param approve: 是否同意请求
:param remark: 添加后的好友备注（仅在同意时有效）
:return: API 响应结果
"""
return await self.call_api("set_friend_add_request", {"flag": flag, "approve": approve, "remark": remark})
async def set_group_add_request(self, flag: str, sub_type: str, approve: bool = True, reason: str = "") -> dict:
"""
处理加群请求/邀请
:param flag: 加群请求的 flag（需从上报的数据中获取）
:param sub_type: add 或 invite，请求类型（需要与上报消息中的 sub_type 字段相符）
:param approve: 是否同意请求/邀请
:param reason: 拒绝理由（仅在拒绝时有效）
:return: API 响应结果
"""
return await self.call_api("set_group_add_request", {"flag": flag, "sub_type": sub_type, "approve": approve, "reason": reason})
async def get_version_info(self) -> dict:
"""
获取版本信息
:return: API 响应结果
"""
return await self.call_api("get_version_info")
async def get_status(self) -> dict:
"""
获取状态
:return: API 响应结果
"""
return await self.call_api("get_status")
async def can_send_image(self) -> dict:
"""
检查是否可以发送图片
:return: API 响应结果
"""
return await self.call_api("can_send_image")
async def can_send_record(self) -> dict:
"""
检查是否可以发送语音
:return: API 响应结果
"""
return await self.call_api("can_send_record")
async def clean_cache(self) -> dict:
"""
清理缓存
:return: API 响应结果
"""
return await self.call_api("clean_cache")
def _process_message(self, message: Union[str, "MessageSegment", List["MessageSegment"]]) -> Union[str, List[dict]]:
"""
处理消息内容，将其转换为 API 可接受的格式
:param message: 原始消息内容
:return: 处理后的消息内容
"""
if isinstance(message, str):
return message
# 避免循环导入，在运行时导入
from models import MessageSegment
if isinstance(message, MessageSegment):
return [self._segment_to_dict(message)]
if isinstance(message, list):
return [self._segment_to_dict(m) for m in message if isinstance(m, MessageSegment)]
return str(message)
def _segment_to_dict(self, segment: "MessageSegment") -> dict:
"""
将 MessageSegment 对象转换为字典
:param segment: MessageSegment 对象
:return: 字典格式的消息段
"""
return {
"type": segment.type,
"data": segment.data
}
return await self.ws.call_api(action, params)

29
core/command_manager.py
View File

@@ -27,6 +27,35 @@ class CommandManager:
self.commands: Dict[str, Callable] = {} # 存储消息指令
self.notice_handlers: List[Dict] = [] # 存储通知处理器
self.request_handlers: List[Dict] = [] # 存储请求处理器
self.plugins: Dict[str, Dict[str, Any]] = {} # 存储插件元数据
# --- 内置 help 指令 ---
self.commands["help"] = self._help_command
self.plugins["core.help"] = {
"name": "帮助",
"description": "显示所有可用指令的帮助信息",
"usage": "/help",
}
async def _help_command(self, bot, event):
"""
内置的 /help 指令处理器
:param bot: Bot 实例
:param event: 消息事件对象
"""
help_text = "--- 可用指令列表 ---\n"
for plugin_name, meta in self.plugins.items():
name = meta.get("name", "未命名插件")
description = meta.get("description", "暂无描述")
usage = meta.get("usage", "暂无用法说明")
help_text += f"\n{name}:\n"
help_text += f" 功能: {description}\n"
help_text += f" 用法: {usage}\n"
await bot.send(event, help_text.strip())
# --- 1. 消息指令装饰器 ---
def command(self, name: str):

48
core/plugin_manager.py Normal file
View File

@@ -0,0 +1,48 @@
"""
插件管理器模块
负责扫描、加载和管理 `base_plugins` 目录下的所有插件。
"""
import importlib
import os
import pkgutil
import sys
from core.command_manager import matcher
def load_all_plugins():
"""
扫描并加载 `plugins` 目录下的所有插件。
该函数会遍历 `plugins` 目录下的所有模块：
1. 如果模块已加载，则执行 reload 操作（用于热重载）。
2. 如果模块未加载，则执行 import 操作。
加载过程中会提取插件元数据 `__plugin_meta__` 并注册到 CommandManager。
"""
plugin_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "plugins")
package_name = "plugins"
print(f" 正在从 {package_name} 加载插件...")
for loader, module_name, is_pkg in pkgutil.iter_modules([plugin_dir]):
full_module_name = f"{package_name}.{module_name}"
try:
if full_module_name in sys.modules:
module = importlib.reload(sys.modules[full_module_name])
action = "重载"
else:
module = importlib.import_module(full_module_name)
action = "加载"
# 提取插件元数据
if hasattr(module, "__plugin_meta__"):
meta = getattr(module, "__plugin_meta__")
matcher.plugins[full_module_name] = meta
type_str = "包" if is_pkg else "文件"
print(f" [{type_str}] 成功{action}: {module_name}")
except Exception as e:
print(f" {action if 'action' in locals() else '加载'}插件 {module_name} 失败: {e}")

11
main.py
View File

@@ -10,8 +10,8 @@ import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import base_plugins # noqa: F401 别动这里是加载插件的
from core import WS
from core.plugin_manager import load_all_plugins
class PluginReloadHandler(FileSystemEventHandler):
@@ -59,7 +59,7 @@ class PluginReloadHandler(FileSystemEventHandler):
try:
# 重新扫描并加载插件
base_plugins.load_all_plugins()
load_all_plugins()
print("[HotReload] 插件重载完成")
except Exception as e:
print(f"[HotReload] 重载失败: {e}")
@@ -73,9 +73,12 @@ async def main():
2. 初始化 WebSocket 客户端
3. 建立连接并保持运行
"""
# 首次加载插件
load_all_plugins()
# 启动文件监控
# 监控 base_plugins 目录
plugin_path = os.path.join(os.path.dirname(__file__), "base_plugins")
# 监控 plugins 目录
plugin_path = os.path.join(os.path.dirname(__file__), "plugins")
event_handler = PluginReloadHandler()
observer = Observer()

60
models/__init__.py
View File

@@ -1,11 +1,21 @@
from .events.base import OneBotEvent
from .events.message import MessageEvent, PrivateMessageEvent, GroupMessageEvent
from .events.notice import (
NoticeEvent, FriendAddNoticeEvent, FriendRecallNoticeEvent,
GroupRecallNoticeEvent, GroupIncreaseNoticeEvent,
GroupDecreaseNoticeEvent, GroupAdminNoticeEvent, GroupBanNoticeEvent,
GroupUploadNoticeEvent, GroupUploadFile,
NotifyNoticeEvent, PokeNotifyEvent, LuckyKingNotifyEvent, HonorNotifyEvent,
GroupCardNoticeEvent, OfflineFileNoticeEvent, OfflineFile,
ClientStatusNoticeEvent, ClientStatus, EssenceNoticeEvent
)
from .events.request import RequestEvent, FriendRequestEvent, GroupRequestEvent
from .events.meta import MetaEvent, HeartbeatEvent, LifeCycleEvent, HeartbeatStatus
from .events.factory import EventFactory
from .events.message import GroupMessageEvent, MessageEvent, PrivateMessageEvent
from .events.meta import MetaEvent
from .events.notice import NoticeEvent
from .events.request import GroupRequestEvent, RequestEvent
from .message import MessageSegment
from .sender import Sender
from .objects import (
GroupInfo, GroupMemberInfo, FriendInfo, StrangerInfo, LoginInfo,
VersionInfo, Status, EssenceMessage, GroupHonorInfo, CurrentTalkative, HonorInfo
)
# Alias for backward compatibility
Event = OneBotEvent
@@ -19,8 +29,42 @@ __all__ = [
"PrivateMessageEvent",
"GroupMessageEvent",
"NoticeEvent",
"FriendAddNoticeEvent",
"FriendRecallNoticeEvent",
"GroupRecallNoticeEvent",
"GroupIncreaseNoticeEvent",
"GroupDecreaseNoticeEvent",
"GroupAdminNoticeEvent",
"GroupBanNoticeEvent",
"GroupUploadNoticeEvent",
"GroupUploadFile",
"NotifyNoticeEvent",
"PokeNotifyEvent",
"LuckyKingNotifyEvent",
"HonorNotifyEvent",
"GroupCardNoticeEvent",
"OfflineFileNoticeEvent",
"OfflineFile",
"ClientStatusNoticeEvent",
"ClientStatus",
"EssenceNoticeEvent",
"RequestEvent",
"MetaEvent",
"EventFactory",
"FriendRequestEvent",
"GroupRequestEvent",
"MetaEvent",
"HeartbeatEvent",
"LifeCycleEvent",
"HeartbeatStatus",
"EventFactory",
"GroupInfo",
"GroupMemberInfo",
"FriendInfo",
"StrangerInfo",
"LoginInfo",
"VersionInfo",
"Status",
"EssenceMessage",
"GroupHonorInfo",
"CurrentTalkative",
"HonorInfo",
]

84
models/events/factory.py
View File

@@ -13,7 +13,10 @@ from .notice import (
NoticeEvent, FriendAddNoticeEvent, FriendRecallNoticeEvent,
GroupRecallNoticeEvent, GroupIncreaseNoticeEvent,
GroupDecreaseNoticeEvent, GroupAdminNoticeEvent, GroupBanNoticeEvent,
GroupUploadNoticeEvent, GroupUploadFile
GroupUploadNoticeEvent, GroupUploadFile,
NotifyNoticeEvent, PokeNotifyEvent, LuckyKingNotifyEvent, HonorNotifyEvent,
GroupCardNoticeEvent, OfflineFileNoticeEvent, OfflineFile,
ClientStatusNoticeEvent, ClientStatus, EssenceNoticeEvent
)
from .request import RequestEvent, FriendRequestEvent, GroupRequestEvent
from .meta import MetaEvent, HeartbeatEvent, LifeCycleEvent, HeartbeatStatus
@@ -204,6 +207,85 @@ class EventFactory:
user_id=data.get("user_id", 0),
file=file
)
elif notice_type == "notify":
sub_type = data.get("sub_type", "")
if sub_type == "poke":
return PokeNotifyEvent(
**common_args,
notice_type=notice_type,
sub_type=sub_type,
user_id=data.get("user_id", 0),
target_id=data.get("target_id", 0),
group_id=data.get("group_id", 0)
)
elif sub_type == "lucky_king":
return LuckyKingNotifyEvent(
**common_args,
notice_type=notice_type,
sub_type=sub_type,
user_id=data.get("user_id", 0),
group_id=data.get("group_id", 0),
target_id=data.get("target_id", 0)
)
elif sub_type == "honor":
return HonorNotifyEvent(
**common_args,
notice_type=notice_type,
sub_type=sub_type,
user_id=data.get("user_id", 0),
group_id=data.get("group_id", 0),
honor_type=data.get("honor_type", "")
)
else:
return NotifyNoticeEvent(
**common_args,
notice_type=notice_type,
sub_type=sub_type,
user_id=data.get("user_id", 0)
)
elif notice_type == "group_card":
return GroupCardNoticeEvent(
**common_args,
notice_type=notice_type,
group_id=data.get("group_id", 0),
user_id=data.get("user_id", 0),
card_new=data.get("card_new", ""),
card_old=data.get("card_old", "")
)
elif notice_type == "offline_file":
file_data = data.get("file", {})
file = OfflineFile(
name=file_data.get("name", ""),
size=file_data.get("size", 0),
url=file_data.get("url", "")
)
return OfflineFileNoticeEvent(
**common_args,
notice_type=notice_type,
user_id=data.get("user_id", 0),
file=file
)
elif notice_type == "client_status":
client_data = data.get("client", {})
client = ClientStatus(
online=client_data.get("online", False),
status=client_data.get("status", "")
)
return ClientStatusNoticeEvent(
**common_args,
notice_type=notice_type,
client=client
)
elif notice_type == "essence":
return EssenceNoticeEvent(
**common_args,
notice_type=notice_type,
sub_type=data.get("sub_type", ""),
group_id=data.get("group_id", 0),
sender_id=data.get("sender_id", 0),
operator_id=data.get("operator_id", 0),
message_id=data.get("message_id", 0)
)
else:
# 未知通知类型，返回基础通知事件
return NoticeEvent(**common_args, notice_type=notice_type)

140
models/events/notice.py
View File

@@ -157,3 +157,143 @@ class GroupUploadNoticeEvent(GroupNoticeEvent):
"""
file: GroupUploadFile = field(default_factory=GroupUploadFile)
"""文件信息"""
@dataclass
class NotifyNoticeEvent(NoticeEvent):
"""
系统通知事件基类 (notify)
"""
sub_type: str = ""
"""
子类型
poke: 戳一戳
lucky_king: 运气王
honor: 群荣誉变更
"""
user_id: int = 0
"""发送者 QQ 号"""
@dataclass
class PokeNotifyEvent(NotifyNoticeEvent):
"""
戳一戳通知
"""
target_id: int = 0
"""被戳者 QQ 号"""
group_id: int = 0
"""群号 (如果是群内戳一戳)"""
@dataclass
class LuckyKingNotifyEvent(NotifyNoticeEvent):
"""
群红包运气王通知
"""
group_id: int = 0
"""群号"""
target_id: int = 0
"""运气王 QQ 号"""
@dataclass
class HonorNotifyEvent(NotifyNoticeEvent):
"""
群荣誉变更通知
"""
group_id: int = 0
"""群号"""
honor_type: str = ""
"""
荣誉类型
talkative: 龙王
performer: 群聊之火
emotion: 快乐源泉
"""
@dataclass
class GroupCardNoticeEvent(GroupNoticeEvent):
"""
群成员名片更新通知
"""
card_new: str = ""
"""新名片"""
card_old: str = ""
"""旧名片"""
@dataclass
class OfflineFile:
"""
离线文件信息
"""
name: str = ""
"""文件名"""
size: int = 0
"""文件大小"""
url: str = ""
"""下载链接"""
@dataclass
class OfflineFileNoticeEvent(NoticeEvent):
"""
接收离线文件通知
"""
user_id: int = 0
"""发送者 QQ 号"""
file: OfflineFile = field(default_factory=OfflineFile)
"""文件数据"""
@dataclass
class ClientStatus:
"""
客户端状态
"""
online: bool = False
"""是否在线"""
status: str = ""
"""状态描述"""
@dataclass
class ClientStatusNoticeEvent(NoticeEvent):
"""
其他客户端在线状态变更通知
"""
client: ClientStatus = field(default_factory=ClientStatus)
"""客户端信息"""
@dataclass
class EssenceNoticeEvent(GroupNoticeEvent):
"""
精华消息变动通知
"""
sub_type: str = ""
"""
子类型
add: 添加
delete: 删除
"""
sender_id: int = 0
"""消息发送者 ID"""
operator_id: int = 0
"""操作者 ID"""
message_id: int = 0
"""消息 ID"""

238
models/objects.py Normal file
View File

@@ -0,0 +1,238 @@
"""
API 响应数据模型模块
定义了 API 返回的数据结构。
"""
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class GroupInfo:
"""
群信息
"""
group_id: int = 0
"""群号"""
group_name: str = ""
"""群名称"""
member_count: int = 0
"""成员数"""
max_member_count: int = 0
"""最大成员数"""
@dataclass
class GroupMemberInfo:
"""
群成员信息
"""
group_id: int = 0
"""群号"""
user_id: int = 0
"""QQ 号"""
nickname: str = ""
"""昵称"""
card: str = ""
"""群名片/备注"""
sex: str = "unknown"
"""性别, male 或 female 或 unknown"""
age: int = 0
"""年龄"""
area: str = ""
"""地区"""
join_time: int = 0
"""加群时间戳"""
last_sent_time: int = 0
"""最后发言时间戳"""
level: str = ""
"""成员等级"""
role: str = "member"
"""角色, owner 或 admin 或 member"""
unfriendly: bool = False
"""是否不良记录成员"""
title: str = ""
"""专属头衔"""
title_expire_time: int = 0
"""专属头衔过期时间戳"""
card_changeable: bool = False
"""是否允许修改群名片"""
@dataclass
class FriendInfo:
"""
好友信息
"""
user_id: int = 0
"""QQ 号"""
nickname: str = ""
"""昵称"""
remark: str = ""
"""备注"""
@dataclass
class StrangerInfo:
"""
陌生人信息
"""
user_id: int = 0
"""QQ 号"""
nickname: str = ""
"""昵称"""
sex: str = "unknown"
"""性别, male 或 female 或 unknown"""
age: int = 0
"""年龄"""
@dataclass
class LoginInfo:
"""
登录号信息
"""
user_id: int = 0
"""QQ 号"""
nickname: str = ""
"""昵称"""
@dataclass
class VersionInfo:
"""
版本信息
"""
app_name: str = ""
"""应用名称"""
app_version: str = ""
"""应用版本"""
protocol_version: str = ""
"""OneBot 标准版本"""
@dataclass
class Status:
"""
运行状态
"""
online: bool = False
"""是否在线"""
good: bool = True
"""运行状态是否良好"""
@dataclass
class EssenceMessage:
"""
精华消息
"""
sender_id: int = 0
"""发送者 QQ 号"""
sender_nick: str = ""
"""发送者昵称"""
sender_time: int = 0
"""发送时间"""
operator_id: int = 0
"""操作者 QQ 号"""
operator_nick: str = ""
"""操作者昵称"""
operator_time: int = 0
"""操作时间"""
message_id: int = 0
"""消息 ID"""
@dataclass
class CurrentTalkative:
"""
龙王信息
"""
user_id: int = 0
"""QQ 号"""
nickname: str = ""
"""昵称"""
avatar: str = ""
"""头像 URL"""
day_count: int = 0
"""持续天数"""
@dataclass
class HonorInfo:
"""
荣誉信息
"""
user_id: int = 0
"""QQ 号"""
nickname: str = ""
"""昵称"""
avatar: str = ""
"""头像 URL"""
description: str = ""
"""荣誉描述"""
@dataclass
class GroupHonorInfo:
"""
群荣誉信息
"""
group_id: int = 0
"""群号"""
current_talkative: Optional[CurrentTalkative] = None
"""当前龙王"""
talkative_list: List[HonorInfo] = field(default_factory=list)
"""历史龙王"""
performer_list: List[HonorInfo] = field(default_factory=list)
"""群聊之火"""
legend_list: List[HonorInfo] = field(default_factory=list)
"""群聊炽焰"""
strong_newbie_list: List[HonorInfo] = field(default_factory=list)
"""冒尖小春笋"""
emotion_list: List[HonorInfo] = field(default_factory=list)
"""快乐源泉"""

24
base_plugins/echo.py → plugins/echo.py
View File

@@ -1,12 +1,17 @@
"""
Echo 插件
Echo 与交互插件
提供 /echo 指令，用于原样回复用户输入的内容。
提供 /echo 和 /赞我 指令。
"""
from core.command_manager import matcher
from core.bot import Bot
from models import MessageEvent
__plugin_meta__ = {
"name": "echo",
"description": "提供 echo 和 赞我 功能",
"usage": "/echo [内容] - 复读内容\n/赞我 - 让机器人给你点赞",
}
@matcher.command("echo")
async def handle_echo(bot: Bot, event: MessageEvent, args: list[str]):
@@ -24,17 +29,18 @@ async def handle_echo(bot: Bot, event: MessageEvent, args: list[str]):
await event.reply(reply_msg)
@matcher.command("poke")
@matcher.command("赞我")
async def handle_poke(bot: Bot, event: MessageEvent, args: list[str]):
"""
处理 poke 指令，发送群戳一戳
处理 赞我 指令，发送点赞
:param bot: Bot 实例
:param event: 消息事件对象
:param args: 指令参数列表（本指令不使用参数）
"""
await bot.call_api("group_poke", {
"group_id": event.group_id,
"user_id": event.user_id
})
try:
# 尝试发送赞
await bot.send_like(event.user_id, times=10)
await event.reply("戳一戳发送成功！")
except Exception as e:
await event.reply(f"戳一戳发送失败：{str(e)}")