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

Dev (#36)

Browse Source 
* 滚木

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

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

主要变更详情：

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

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

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

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

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

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

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

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

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

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

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

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

* 文档更新

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

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

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

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

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

覆盖率已到达86%（忽略插件）

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

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

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

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

* build: 更新依赖文件

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

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

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

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

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

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

---------

Co-authored-by: baby20162016 <2185823427@qq.com>
This commit is contained in:
镀铬酸钾
2026-01-13 04:49:59 +08:00
committed by GitHub
parent a6464c36b1
commit 95e98dea9c
21 changed files with 981 additions and 724 deletions
Download Patch File Download Diff File Expand all files Collapse all files

257
README.md
View File

@@ -1,214 +1,77 @@
Calglau BOT by NEO Bot Framework # Calglau BOT by NEO Bot Framework
[INTERNAL USE ONLY] > **[INTERNAL USE ONLY]**
>
> 本仓库为 Calglau BOT 的内部开发版本，请遵守相关保密协议。
本仓库为 Calglau BOT 的内部开发版本，请遵守相关保密协议。 **Powered by NEO Bot Framework**
Powered by NEO Bot Framework ## 项目概述
项目概述 **Calglau BOT** 是一个基于 NEO Bot Framework 构建的高性能 QQ 机器人。
Calglau BOT 是一款基于 NEO Bot Framework 构建的功能丰富的 QQ 机器人，采用模块化插件架构设计，支持功能的灵活扩展与独立维护，旨在为社群管理和日常自动化需求提供稳定、高性能且开发体验友好的机器人平台。 简单来说：扣一
核心特性 ### 核心特性
- 模块化插件架构：所有功能均以独立插件形式存放于  plugins/  目录，支持开发、维护与热重载，降低功能迭代成本。 * **模块化插件架构**：所有功能都在 `plugins/` 目录
- 高性能异步核心：基于  asyncio  与  websockets  构建，可高效处理高并发消息，保障机器人在高负载场景下的响应速度。 * **极致性能优化**：
- 开发者友好设计：内置插件热重载机制，修改代码无需重启机器人；配套完整类型提示与清晰 API 设计，提升开发效率。 * **Python 3.14 JIT**：pypy不支持那个浏览器扩展我只能用JIT了。。。
- Redis 缓存集成：自动缓存群信息等高频 API 调用结果，减少重复请求，进一步优化响应性能。 * **Mypyc 编译**：一些核心模块已经编译成机器码了
- 内置帮助系统：通过  /help  指令可自动生成并展示所有已加载插件的功能说明，降低用户使用门槛。 * **Playwright 页面池**：浏览器页面预热池
* **全局连接复用**：HTTP 和 Redis 连接池化管理
* **开发者友好**：完整的类型提示，清晰的 API 设计。
* **集成 Redis 缓存**：能缓存的都缓存了。群信息、用户信息、帮助图片
* **正向 WebSocket 连接**：我只会支持正向WS连接。。。不要提意见，我不会听的。。。
技术栈 ### 技术栈
- 核心框架: Python 3.8+ & NEO Bot Framework * **核心框架**: Python 3.14 JIT & NEO Bot Framework
- 异步库:  asyncio  * **编译器**: Mypyc
- 网络通信:  websockets  (OneBot v11 协议) * **异步核心**: `asyncio` + `uvloop` (Linux) / 原生 Loop (Windows)
- 缓存服务:  Redis  * **网络通信**: `websockets` (OneBot v11), `aiohttp` (Shared Session)
- 日志工具:  Loguru  * **浏览器引擎**: `Playwright` (Chromium) + Page Pool
- 文件监控:  watchdog  (插件热重载依赖) * **数据序列化**: `orjson`
* **缓存**: `Redis`
* **日志**: `Loguru`
项目结构 ---
plaintext ## 项目结构
```
. .
. ├── plugins/ # 插件目录，业务逻辑都在这
├── plugins/ # 功能插件目录 │ ├── admin.py # 管理员指令
│ ├── admin.py # 管理员权限管理插件 │ ├── bili_parser.py # B站解析 (高性能版)
│ ├── bili_parser.py # B站链接解析插件 │ ├── code_py.py # 代码沙箱
│ ├── code_py.py # Python代码执行插件 │ ├── echo.py # 复读机
│ ├── echo.py # 复读与互动插件 │ ├── forward_test.py # 合并转发测试
│ ├── forward_test.py # 合并转发消息演示插件 │ ├── jrcd.py # 今日运势
│ ├── jrcd.py # 今日人品等娱乐功能插件 │ └── thpic.py # 东方图片
│ └── thpic.py # 东方Project图片发送插件 ├── core/ # 框架核心，非请勿动
├── core/ # NEO框架核心代码（无需修改） │ ├── api/ # OneBot API 封装
│ ├── api/ # 框架API模块 │ ├── managers/ # 各种管理器 (指令, 浏览器, 图片, 插件)
│ ├── bot.py # 机器人核心逻辑 │ ├── utils/ # 工具函数
│ └── ws.py # WebSocket通信模块 │ ├── ws.py # WebSocket 通信层 (已编译)
├── data/ # 数据存储目录 │ └── bot.py # Bot 实例
│ ├── admin.json # 管理员列表配置 ├── data/ # 数据存储
│ └── permissions.json # 权限配置文件 │ ├── admin.json # 管理员名单
├── html/ # 静态网页文件目录 │ └── permissions.json # 权限配置
│ ├── 404.html # 404页面 ├── templates/ # Jinja2 模板
│ └── index.html # 主页 ├── setup_mypyc.py # 编译脚本
├── models/ # 数据模型目录（事件、消息段等） └── main.py # 启动入口
├── .gitignore # Git忽略文件配置 ```
├── config.toml # 机器人主配置文件
├── main.py # 项目启动入口
└── requirements.txt # Python依赖清单
## 快速开始
1
想要深入了解框架的工作原理或开发更复杂的插件？
1. **装环境**: Python 3.14，Redis， OneBot 客户端 (推荐 NapCat)。
2. **装依赖**: `pip install -r requirements.txt`
3. **装浏览器**: `playwright install chromium`
4. **编译核心 (可选)**: `python setup_mypyc.py build_ext --inplace`
5. **启动**: `python -X jit main.py`
详细文档去 `docs/` 目录看
- Python 3.12 或更高版本
- 开发调试阶段推荐使用 CPython 解释器，保障第三方库兼容性与调试体验。
- 生产环境部署可考虑 PyPy 解释器，获取潜在性能提升（需注意部分库兼容性）。
- 部署并运行 Redis 服务
- 准备 OneBot v11 协议实现端（推荐 NapCatQQ）
2. 安装依赖
克隆本项目后，在项目根目录执行以下命令安装依赖：
bash
pip install -r requirements.txt
 
3. 配置说明
[内部开发]
项目内置的  config.toml  已预先配置为连接官方 DEV 调试服务器，拉取仓库后通常无需修改即可直接运行。
若需连接本地或其他环境，可参考以下配置模板调整：
toml
# config.toml
[napcat_ws]
# OneBot 实现端的 WebSocket 地址
uri = "ws://127.0.0.1:3001"
# Access Token 鉴权（无则留空）
token = ""
# 断线重连间隔（单位：秒）
reconnect_interval = 5
[bot]
# 机器人指令前缀（支持配置多个）
command_prefixes = ["/", "!", "！"]
[redis]
# Redis 服务连接信息
host = "127.0.0.1"
port = 6379
db = 0
password = ""
 
4. 启动运行
在项目根目录执行以下命令启动机器人：
bash
python main.py
 
启动成功后，控制台将输出已加载插件列表与 WebSocket 连接状态，机器人将自动接入 OneBot 实现端。
插件开发指南
Calglau BOT 的全部功能均通过插件实现，结合热重载机制，开发过程无需频繁重启机器人，极大提升开发效率。
热重载工作流
1. 保持  python main.py  进程持续运行
2. 在  plugins/  目录下创建或修改  .py  插件文件
3. 保存修改后的文件
4. 观察控制台输出  [HotReload] 插件重载完成  提示，修改内容即刻生效
创建新插件
1. 在  plugins/  目录下新建 Python 文件，例如  weather.py 
2. 按照以下步骤编写插件逻辑
1. 定义插件元数据
通过  __plugin_meta__  字典定义插件信息，确保  /help  指令能自动识别并展示插件功能：
python
# plugins/weather.py
__plugin_meta__ = {
"name": "天气查询",
"description": "提供城市天气查询功能",
"usage": "/weather [城市名] - 查询指定城市的实时天气",
}
 
2. 编写指令处理器
使用  @matcher.command()  装饰器注册聊天指令，实现指令响应逻辑：
python
# plugins/weather.py
from core.command_manager import matcher
from models import MessageEvent
@matcher.command("weather")
async def handle_weather_command(event: MessageEvent, args: list[str]):
"""
处理 /weather 指令
:param event: 消息事件对象，用于回复消息
:param args: 用户输入的指令参数列表（按空格分割）
"""
if not args:
await event.reply("请输入要查询的城市名，例如：/weather 北京")
return
city = args[0]
# 此处可接入天气API获取真实数据，以下为示例
weather_data = f"{city}的天气是：晴，25°C"
await event.reply(weather_data)
 
3. 监听系统事件
使用  @matcher.on_notice()  装饰器监听机器人事件（如新成员入群），实现事件响应逻辑：
python
# plugins/weather.py
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):
"""监听群成员增加事件，发送欢迎消息"""
welcome_msg = f"欢迎新成员 @{event.user_id} 加入本群！"
await bot.send_group_msg(event.group_id, welcome_msg)
 
当前功能插件
插件文件 ( plugins/ ) 功能描述
 admin.py  机器人管理员权限管理（添加/移除管理员、权限验证）
 bili_parser.py  自动解析 Bilibili 视频链接，生成精美分享卡片
 code_py.py  执行 Python 代码片段（高危功能，仅限管理员使用）
 echo.py  提供  /echo  复读指令与  /赞我  互动指令
 forward_test.py  演示合并转发消息的发送方法
 jrcd.py  提供今日人品测算、牛牛词典查询等娱乐功能
 thpic.py  随机发送东方 Project 相关图片
路线图 (Roadmap)
Web 仪表盘：开发可视化管理页面，支持查看机器人状态、插件列表与运行日志
权限系统重构：引入精细化权限节点，支持按插件/指令维度配置用户权限

2
config.toml
View File

@@ -23,5 +23,3 @@ tls_verify = true
ca_cert_path = "c:/Users/镀铬酸钾/Documents/NeoBot/ca/ca.crt" ca_cert_path = "c:/Users/镀铬酸钾/Documents/NeoBot/ca/ca.crt"
client_cert_path = "c:/Users/镀铬酸钾/Documents/NeoBot/ca/client-cert.pem" client_cert_path = "c:/Users/镀铬酸钾/Documents/NeoBot/ca/client-cert.pem"
client_key_path = "c:/Users/镀铬酸钾/Documents/NeoBot/ca/client-key.pem" client_key_path = "c:/Users/镀铬酸钾/Documents/NeoBot/ca/client-key.pem"

79
core/managers/browser_manager.py
View File

@@ -15,6 +15,8 @@ class BrowserManager:
_instance = None _instance = None
_playwright: Optional[Playwright] = None _playwright: Optional[Playwright] = None
_browser: Optional[Browser] = None _browser: Optional[Browser] = None
_page_pool: Optional[asyncio.Queue] = None
_pool_size: int = 3
def __new__(cls): def __new__(cls):
if cls._instance is None: if cls._instance is None:
@@ -36,6 +38,73 @@ class BrowserManager:
logger.exception(f"无头浏览器启动失败: {e}") logger.exception(f"无头浏览器启动失败: {e}")
self._browser = None self._browser = None
async def init_pool(self, size: int = 3):
"""
初始化页面池
"""
if not self._browser:
await self.initialize()
if not self._browser:
logger.error("浏览器初始化失败，无法创建页面池")
return
self._pool_size = size
self._page_pool = asyncio.Queue(maxsize=size)
logger.info(f"正在初始化页面池 (大小: {size})...")
for i in range(size):
try:
page = await self._browser.new_page()
await self._page_pool.put(page)
except Exception as e:
logger.error(f"创建页面池页面 {i+1} 失败: {e}")
logger.success(f"页面池初始化完成，当前可用页面: {self._page_pool.qsize()}")
async def get_page(self) -> Optional[Page]:
"""
从池中获取一个页面。如果池未初始化或为空，则尝试创建一个新页面（不入池）。
"""
if self._page_pool and not self._page_pool.empty():
try:
page = self._page_pool.get_nowait()
# 简单的健康检查
if page.is_closed():
logger.warning("检测到池中页面已关闭，重新创建一个...")
if self._browser:
page = await self._browser.new_page()
else:
return None
return page
except asyncio.QueueEmpty:
pass
# 如果池空了或者没初始化，回退到临时创建
logger.debug("页面池为空或未初始化，创建临时页面")
return await self.get_new_page()
async def release_page(self, page: Page):
"""
归还页面到池中。如果池已满或未初始化，则关闭页面。
"""
if not page or page.is_closed():
return
if self._page_pool:
try:
# 重置页面状态 (例如清空内容)，防止数据污染
# 注意: goto('about:blank') 比 close() 快得多
await page.goto("about:blank")
self._page_pool.put_nowait(page)
return
except asyncio.QueueFull:
pass
# 池满或未启用池，直接关闭
await page.close()
async def get_new_page(self) -> Optional[Page]: async def get_new_page(self) -> Optional[Page]:
""" """
获取一个新的页面 (Page) 获取一个新的页面 (Page)
@@ -58,6 +127,16 @@ class BrowserManager:
""" """
关闭浏览器和 Playwright 关闭浏览器和 Playwright
""" """
# 清空页面池
if self._page_pool:
while not self._page_pool.empty():
try:
page = self._page_pool.get_nowait()
await page.close()
except Exception:
pass
self._page_pool = None
if self._browser: if self._browser:
await self._browser.close() await self._browser.close()
self._browser = None self._browser = None

22
core/managers/image_manager.py
View File

@@ -29,6 +29,8 @@ class ImageManager:
# core/managers/image_manager.py -> core/managers -> core -> core/data/temp # core/managers/image_manager.py -> core/managers -> core -> core/data/temp
self.temp_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "temp") self.temp_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "temp")
os.makedirs(self.temp_dir, exist_ok=True) os.makedirs(self.temp_dir, exist_ok=True)
# 模板缓存
self._template_cache: Dict[str, Template] = {}
async def render_template(self, template_name: str, data: Dict[str, Any], output_name: str = "output.png", quality: int = 80, image_type: str = "png") -> Optional[str]: async def render_template(self, template_name: str, data: Dict[str, Any], output_name: str = "output.png", quality: int = 80, image_type: str = "png") -> Optional[str]:
""" """
@@ -50,15 +52,20 @@ class ImageManager:
return None return None
try: try:
# 1. 渲染 HTML # 1. 渲染 HTML (使用缓存)
with open(template_path, "r", encoding="utf-8") as f: if template_name in self._template_cache:
template_str = f.read() template = self._template_cache[template_name]
else:
with open(template_path, "r", encoding="utf-8") as f:
template_str = f.read()
template = Template(template_str)
self._template_cache[template_name] = template
template = Template(template_str)
html_content = template.render(**data) html_content = template.render(**data)
# 2. 使用浏览器截图 # 2. 使用浏览器截图
page = await browser_manager.get_new_page() # 改为从池中获取页面
page = await browser_manager.get_page()
if not page: if not page:
logger.error("无法获取浏览器页面") logger.error("无法获取浏览器页面")
return None return None
@@ -76,10 +83,11 @@ class ImageManager:
if image_type == 'jpeg': if image_type == 'jpeg':
screenshot_args['quality'] = quality screenshot_args['quality'] = quality
screenshot_bytes = await page.screenshot(**screenshot_args) screenshot_bytes = await page.screenshot(**screenshot_args) # type: ignore
finally: finally:
await page.close() # 归还页面到池中，而不是直接关闭
await browser_manager.release_page(page)
# 3. 保存文件 # 3. 保存文件
output_path = os.path.join(self.temp_dir, output_name) output_path = os.path.join(self.temp_dir, output_name)

34
core/utils/json_utils.py Normal file
View File

@@ -0,0 +1,34 @@
"""
JSON 工具模块
统一使用高性能的 orjson 库进行 JSON 序列化和反序列化。
如果 orjson 不可用，则回退到标准库 json。
"""
from typing import Any, Union
import json
# 在模块加载时检查 orjson 是否可用
try:
import orjson
_orjson_available = True
except ImportError:
_orjson_available = False
def dumps(obj: Any) -> str:
"""
将对象序列化为 JSON 字符串。
"""
if _orjson_available:
# orjson.dumps 返回 bytes，需要 decode
return orjson.dumps(obj).decode("utf-8")
else:
return json.dumps(obj, ensure_ascii=False)
def loads(json_str: Union[str, bytes]) -> Any:
"""
将 JSON 字符串反序列化为对象。
"""
if _orjson_available:
return orjson.loads(json_str)
else:
return json.loads(json_str)

55
docs/core-concepts/architecture.md Normal file
View File

@@ -0,0 +1,55 @@
# 骨架
Neobot是面向内部开发者的，我会开源，但是写的很烂。。。
## 1. 动力核心
### Python 3.14 + JIT
镀铬酸钾创项目的时候用的 Python 3.14 3.14兼容JIT，那就这样吧
* **何原理**: 提前编译了源代码，
* **何用途**: 密集CPU运算能提升一些
### Mypyc 编译 (AOT)
光 JIT 还不够。。核心模块（`core/ws.py`, `core/managers/*.py`）我编译成了C扩展
* **何原理**: 因为这个项目有很多类型提示，然后我就编译成C库了。。。
* **何用途**: WS和manager下边的模块都是机器码运行，或许会快一些。。。
### 异步 IO 模型
* **Linux**: uvloop
* **Windows**:IOCP
* *注*: `winloop` 死了，会和面具打架。。。
## 2. 连接模式
### 正向 WebSocket 模式
这是一种简单直接的模式
* **主动出击 (Client)**: Bot 是个客户端
* **好处**: 你电脑能上网就行（实际上是因为没公网ip哈。。。）
```mermaid
graph LR
subgraph Local [你的电脑/服务器]
Bot[NEO Bot]
Browser[Playwright 页面池]
end
subgraph Remote [外部]
NapCat[NapCatQQ]
end
Bot -- "WebSocket (主动连接)" --> NapCat
Bot -- "内部调用" --> Browser
```
## 3. 资源管理
### 单例管理器
所有东西（指令、权限、浏览器、图片）都是全局独一份的。
* **随叫随到**: 在哪都能直接 `import`
* **绝对权威**: 全局就一份数据
### 资源池化
别几把开多个实例。。。
* **Browser Pool**: 浏览器页面提前开好，用完洗干净放回去
* **Connection Pool**: Redis 和 HTTP 请求都用连接池

88
docs/core-concepts/event-flow.md
View File

@@ -1,8 +1,8 @@
# 核心概念：事件流转 # 核心概念：事件流转
在 NEO Bot Framework 中，所有交互都由**事件**驱动。理解一个事件从被接收到最终被处理的完整流程，是掌握框架工作原理的关键。 NEO Bot 的核心就是**事件驱动**。搞懂一个事件从哪来、到哪去，你就懂了一大半。
本节将以一个用户发送 `/echo hello` 的群聊消息为例，详细拆解其在框架内部的流转路径。 下面就拿 `/echo hello` 举例
## 事件流转图 ## 事件流转图
@@ -15,40 +15,40 @@ graph TD
classDef plugin fill:#fce4ec,stroke:#c2185b,stroke-width:2px; classDef plugin fill:#fce4ec,stroke:#c2185b,stroke-width:2px;
subgraph External [外部环境] subgraph External [外部环境]
OneBot[OneBot v11 实现端<br/>(如 NapCatQQ)]:::external OneBot["OneBot v11 实现端<br/>(如 NapCatQQ)"]:::external
end end
subgraph NeoBot [NEO Bot Framework] subgraph NeoBot [NEO Bot Framework]
direction TB direction TB
subgraph Network [网络接入层] subgraph Network [网络接入层]
WS[WebSocket 连接<br/>core/ws.py]:::network WS["WebSocket 连接<br/>core/ws.py"]:::network
end end
subgraph Processing [核心处理层] subgraph Processing [核心处理层]
Factory[事件工厂<br/>models/events/factory.py]:::core Factory["事件工厂<br/>models/events/factory.py"]:::core
Dispatcher[命令管理器<br/>core/managers/command_manager.py]:::core Dispatcher["命令管理器<br/>core/managers/command_manager.py"]:::core
Handler[事件处理器<br/>core/handlers/event_handler.py]:::core Handler["事件处理器<br/>core/handlers/event_handler.py"]:::core
BotAPI[Bot API 封装<br/>core/bot.py]:::core BotAPI["Bot API 封装<br/>core/bot.py"]:::core
end end
subgraph Plugins [业务插件层] subgraph Plugins [业务插件层]
UserPlugin[用户插件<br/>plugins/*.py]:::plugin UserPlugin["用户插件<br/>plugins/*.py"]:::plugin
end end
end end
%% 事件上报流程 (实线) %% 事件上报流程 (实线)
OneBot -- 1. WebSocket 消息 --> WS OneBot -- "1. WebSocket 消息" --> WS
WS -- 2. 原始 JSON --> Factory WS -- "2. 原始 JSON" --> Factory
Factory -- 3. Event 对象 --> WS Factory -- "3. Event 对象" --> WS
WS -- 4. 分发事件 --> Dispatcher WS -- "4. 分发事件" --> Dispatcher
Dispatcher -- 5. 匹配指令/事件 --> Handler Dispatcher -- "5. 匹配指令/事件" --> Handler
Handler -- 6. 调用处理函数 --> UserPlugin Handler -- "6. 调用处理函数" --> UserPlugin
%% API 调用流程 (虚线) %% API 调用流程 (虚线)
UserPlugin -. 7. 调用 bot.send() .-> BotAPI UserPlugin -. "7. 调用 bot.send()" .-> BotAPI
BotAPI -. 8. 封装 API 请求 .-> WS BotAPI -. "8. 封装 API 请求" .-> WS
WS -. 9. 发送 JSON .-> OneBot WS -. "9. 发送 JSON" .-> OneBot
%% 链接样式 %% 链接样式
linkStyle 0,1,2,3,4,5 stroke:#333,stroke-width:2px; linkStyle 0,1,2,3,4,5 stroke:#333,stroke-width:2px;
@@ -59,42 +59,42 @@ graph TD
### 1. 接收 WebSocket 消息 (`core/ws.py`) ### 1. 接收 WebSocket 消息 (`core/ws.py`)
* 当用户在 QQ 群里发送消息时，OneBot v11 实现端（如 NapCatQQ）会将其打包成一个 JSON 格式的数据，并通过 WebSocket 连接发送给框架。 * 你在群里发了条消息，OneBot (比如 NapCatQQ) 就会把它打包成一个 JSON，通过 WebSocket 扔给 Bot。
* `core/ws.py` 中的 `_listen_loop` 方法持续监听连接，接收到这个原始的 JSON 字符串。 * `core/ws.py` 里的 `_listen_loop` 一直在那蹲着，收到这个 JSON 字符串。
### 2. 事件对象实例化 (`models/events/factory.py`) ### 2. 变成对象 (`models/events/factory.py`)
* `ws.py` 将接收到的 JSON 数据传递给 `EventFactory.create_event()`。 * `ws.py` 拿到 JSON 后，扔给 `EventFactory.create_event()`。
* `EventFactory` 会根据 JSON 中的 `post_type` 字段（例如 `"message"`）和 `message_type` 字段（例如 `"group"`），智能地将其解析并实例化为对应的 Python 对象，例如 `GroupMessageEvent`。 * 工厂类看一眼 `post_type` 是 `"message"`，`message_type` 是 `"group"`，会包装成 `GroupMessageEvent` 对象。
* 这个 `Event` 对象包含了所有事件信息，并且具有清晰的类型提示，方便后续处理。 * 这时候是python对象了，有属性有方法，感觉很方便。。。
### 3. 事件初步处理与分发 (`core/ws.py`) ### 3. 塞点东西，准备分发 (`core/ws.py`)
* `ws.py` 的 `on_event` 方法接收到 `Event` 对象后，会做两件重要的事： * `ws.py` 拿到这个对象后，干两件事：
1. **注入 `Bot` 实例**：将 `self.bot` 赋值给 `event.bot`。这使得插件开发者可以在事件处理器中直接通过 `event.reply()` 或 `event.bot.send(...)` 来调用 API。 1. **塞 Bot 实例**：把 `self.bot` 塞进 `event.bot` 里。这样你在插件里拿到事件，就能直接 `event.reply()` 回复，不用到处找 Bot 实例。
2. **分发事件**：将 `Event` 对象传递给全局的命令管理器 `matcher.handle_event(bot, event)`。 2. **扔出去**：把事件扔给 `matcher.handle_event(bot, event)`，也就是命令管理器。
### 4. 指令匹配与处理器查找 (`core/managers/command_manager.py`) ### 4. 找找谁来处理 (`core/managers/command_manager.py`)
* `CommandManager` (即 `matcher`) 是事件处理的核心中枢。 * `CommandManager` (就是代码里的 `matcher`)
* 它的 `handle_event` 方法会首先判断事件类型。对于消息事件，它会将其交给内部的 `MessageHandler`。 * 它看了一眼，然后转手交给 `MessageHandler`。
* `MessageHandler` 会检查消息内容是否以已注册的命令前缀（如 `/`）开头。 * `MessageHandler` 看消息内容是以 `/` 开头的吗？”
* 如果匹配成功（例如 `/echo`），它会从已注册的命令字典中查找对应的处理函数（即在 `echo.py` 中被 `@matcher.command("echo")` 装饰的函数）。 * 如果是 `/echo`，已经注册的指令列表，找到了 `plugins/echo.py` 里那个被 `@matcher.command("echo")` 标记的函数。
### 5. 执行插件逻辑 (`plugins/echo.py`) ### 5. 干活 (`plugins/echo.py`)
* `MessageHandler` 找到了匹配的处理器后，会调用它，并将 `Event` 对象和解析出的参数（`args`）传递进去。 * 直接调用它，把 `Event` 对象和参数 `args` 传进去。
* 此时，控制权就完全交给了插件开发者编写的函数，例如 `handle_echo_command(event, args)`。 * 这时候就是你写的代码在跑了。你想干啥都行。。。
* 插件函数可以执行任意逻辑，比如操作数据库、请求外部 API，或者调用 `Bot` 的 API 来回复消息。
### 6. API 调用与响应 (`core/bot.py` -> `core/ws.py`) ### 6. 回复消息 (`core/bot.py` -> `core/ws.py`)
* 当插件调用 `event.reply("hello")` 时，实际上是调用了 `core/bot.py` 中封装的 `send` 方法。 * 你在插件里写了 `await event.reply("hello")`。
* `Bot` 类会将这个调用转换为一个标准的 OneBot v11 API 请求（例如 `{"action": "send_group_msg", "params": {...}}`）。 * 这行代码背后，是 `core/bot.py` 把你的话封装成了一个标准的 OneBot API 请求（`send_group_msg`）。
* 这个请求最终通过 `core/ws.py` 的 `call_api` 方法，被序列化为 JSON 字符串，并通过 WebSocket 发送回 OneBot v11 实现端。 * 然后 `core/ws.py` 把这个请求变成 JSON，通过 WebSocket 扔回给 OneBot。
### 7. 消息发送 ### 7. 发送成功
* OneBot v11 实现端接收到 API 请求后，执行相应的操作——将 "hello" 这条消息发送到原来的 QQ 群。 * OneBot 收到请求，把 "hello" 发到了群里。
* 恩。。。
至此，一个完整的事件流转闭环就完成了。理解这个流程后，您就能明白框架是如何将底层的网络通信与高层的插件逻辑解耦，并为开发者提供便捷接口的。 至此，一个完整的事件流转闭环就完成了。理解这个流程后，您就能明白框架是如何为开发者提供便捷接口的。

73
docs/core-concepts/performance.md Normal file
View File

@@ -0,0 +1,73 @@
# 性能优化详解
NEO Bot 实际上是python，有人说用Java可能更好。。。嗯但是镀铬酸钾不会Java，镀铬酸钾只会python，所以只能用python了
## 1. Playwright 页面池 (Page Pool)
### 痛点
之前 Bot 发图流程：
1. 用户发指令。
2. Bot 启动浏览器。
3. 创建新页面。。
4. 渲染，截图。
5. 关闭浏览器。
这种模式下，发一张图至少要等 1 秒以上。。。
### 解决方案
`BrowserManager` 维护了一个**页面池**。
* **启动时**: 自动预热 3 个页面（可配置），挂在后台待命。
* **运行时**: 需要截图时，直接从池里 `get_page()`
* **结束后**: 截图完成，页面执行 `about:blank` 洗白，然后 `release_page()` 放回池里。
### 收益
我不知道快了多少，也没人测试，嗯
## 2. Jinja2 模板缓存
### 痛点
每次渲染 HTML，都要从硬盘读文件，然后解析模板语法。硬盘 IO 是慢的，解析也是慢的。
### 解决方案
`ImageManager` 引入了内存缓存 `_template_cache`。
* 第一次读取模板后，编译好的 `Template` 对象直接存入字典。
* 后续请求直接从内存拿对象渲染。
### 收益
省了硬盘IO
## 3. 全局 HTTP 连接复用
### 痛点
插件（如 B站解析）每次请求 API 都创建一个新的 `aiohttp.ClientSession`。
这意味着每次都要进行：DNS 解析 -> TCP 握手 -> SSL 握手。这在 HTTPS 下非常慢。
### 解决方案
我们在插件层面实现了 `get_session()`。
* 全局共享一个 `ClientSession`。
* 复用底层的 TCP 连接 (Keep-Alive)。
### 收益
实际上我也不知道，bot没高并发的实验。。。
## 4. orjson 极速序列化
### 痛点
Python 自带的 `json` 库性能好像不太好，特别是在处理 OneBot 这种大量 JSON 通信的场景下。
### 解决方案
全面替换为 `orjson`。
* Rust 编写
* 支持直接返回 `bytes`，减少内存复制。
## 5. Mypyc 编译
### 痛点
Python太慢了。。。
### 解决方案
利用 `setup_mypyc.py` 将核心模块编译为 C 扩展。
* `core/ws.py`: WebSocket 消息处理循环。
* `core/managers/*.py`: 事件分发逻辑。
这些高频调用的代码变成了机器码

109
docs/core-concepts/singleton-managers.md
View File

@@ -1,74 +1,80 @@
# 核心概念：单例管理器 # 核心概念：单例管理器
在 `core/managers/` 目录下，存放着一系列全局唯一的**管理器（Managers）**。它们是 NEO Bot Framework 功能的核心实现，负责处理事件、管理权限、加载插件等关键任务。 `core/managers/` 这地方，放的都是些**管事的**。它们是 NEO Bot 的核心。梨花飘落在你窗前。。。
理解这些管理器的职责，有助于您更好地利用框架提供的能力，并进行更高级的开发。 ## 为啥是单例？
## 设计模式：单例 (Singleton) 就是**全局独一份**。
框架中所有的管理器都采用了**单例设计模式**。这意味着在整个应用程序的生命周期中，每个管理器类只会存在一个实例。 * **到处都能用**: 在插件里 `import` 就行，不用传来传去。
* **数据不打架**: 权限、命令这些东西，全局就一份，改了都认。
* **省资源**: Redis 连接池、浏览器这种东西，开一个就够了，多了浪费。
**为什么使用单例？** 我专门在 `core/utils/singleton.py` 搞了个基类，继承一下就行，你会的，加油。。。
* **全局访问点**: 任何模块（尤其是插件）都可以方便地导入并使用同一个管理器实例，无需手动传递。 ## 认识一下
* **状态共享**: 管理器内部维护的状态（如已注册的命令、用户权限列表）是全局共享和一致的。
* **资源统一管理**: 对于像 Redis 连接这样的资源，单例模式确保了全局只有一个连接池，避免了资源的浪费和冲突。
框架在 `core/utils/singleton.py` 中提供了一个 `Singleton` 基类，所有管理器都继承自它，以轻松实现单例模式。 ### 1. `CommandManager` (`matcher`)
## 核心管理器介绍 * **怎么找**: `from core.managers.command_manager import matcher`
* **管啥**:
* **总调度**: 所有消息都得从它这过一遍
* **发牌的**: 你用的 `@matcher.command()` 这种装饰器，就是它发的。
* **对号入座**: 消息来了，它负责对一下，看是哪个插件的。
### 1. `CommandManager` (全局实例: `matcher`) 写插件天天都得跟它打交道。
* **文件**: `core/managers/command_manager.py` ### 2. `PermissionManager` (`permission_manager`)
* **全局实例**: `from core.managers.command_manager import matcher`
* **核心职责**:
* **事件处理中枢**: 它是事件流转的核心，负责接收所有类型的事件，并将其分发给相应的底层处理器。
* **装饰器提供者**: 为插件提供了 `@matcher.command()`, `@matcher.on_notice()` 等一系列装饰器，用于注册事件处理器。
* **指令匹配**: 内部维护了一个指令注册表，能够根据消息内容匹配到对应的处理函数。
`matcher` 是插件开发者最常打交道的管理器。 * **怎么找**: `from core.managers.permission_manager import permission_manager`
* **管啥**:
* **划分三六九等**: `ADMIN`, `OP`, `USER` 这些等级都是它定的。
* **管理权限**: 谁有啥权限，都记在 `core/data/permissions.json` 里。
* **会自动变通**: 查权限的时候，它会把 `AdminManager` 里的超管也当成 `ADMIN`。
### 2. `PermissionManager` (全局实例: `permission_manager`) ### 3. `AdminManager` (`admin_manager`)
* **文件**: `core/managers/permission_manager.py` * **怎么找**: `from core.managers.admin_manager import admin_manager`
* **全局实例**: `from core.managers.permission_manager import permission_manager` * **管啥**:
* **核心职责**: * **钦差大臣**: 专门管机器人的超级管理员，增删改查都在这。
* **权限定义与检查**: 定义了 `ADMIN`, `OP`, `USER` 等权限等级，并提供了 `check_permission` 方法来验证用户权限。 * **三级缓存**: 内存 -> Redis -> 文件
* **数据持久化**: 负责从 `core/data/permissions.json` 文件中加载和保存用户权限设置。
* **与 `AdminManager` 联动**: 在检查权限和获取所有用户权限时，会自动合并机器人管理员（来自 `AdminManager`）的数据，将其识别为最高权限 `ADMIN`。
### 3. `AdminManager` (全局实例: `admin_manager`)
* **文件**: `core/managers/admin_manager.py`
* **全局实例**: `from core.managers.admin_manager import admin_manager`
* **核心职责**:
* **管理员管理**: 提供 `add_admin`, `remove_admin`, `is_admin` 等接口，用于管理机器人的超级管理员列表。
* **数据同步**: 实现了内存、`core/data/admin.json` 文件以及 Redis 缓存之间的数据同步，确保管理员列表的一致性和高效查询。
### 4. `PluginManager` ### 4. `PluginManager`
* **文件**: `core/managers/plugin_manager.py` * **管啥**:
* **核心职责**: * **拉人头**: 启动时把 `plugins/` 目录下的插件都拉进来。
* **插件加载**: 负责扫描 `plugins/` 目录，导入所有合法的插件模块。 * **热更新**: 你改了插件代码，它负责重载，不用重启机器人。
* **元数据提取**: 读取插件文件中定义的 `__plugin_meta__` 字典，用于 `/help` 指令等功能。
* **热重载支持**: `load_all_plugins` 函数被 `main.py` 中的文件监控服务调用，以实现插件的热重载。
此管理器通常在后台工作，开发者较少直接与其交互。 这一般在幕后，你基本不用找它。
### 5. `RedisManager` (全局实例: `redis_manager`) ### 5. `RedisManager` (`redis_manager`)
* **文件**: `core/managers/redis_manager.py` * **怎么找**: `from core.managers.redis_manager import redis_manager`
* **全局实例**: `from core.managers.redis_manager import redis_manager` * **管啥**:
* **核心职责**: * **接线员**: 管着和 Redis 的连接。
* **连接管理**: 负责初始化和管理与 Redis 服务器的异步连接。 * **提供工具**: 你要用 Redis，就找 `redis_manager.redis`。
* **提供实例**: 通过 `redis_manager.redis` 属性，为其他模块提供一个可用的 `redis` 客户端实例。
## 如何在插件中使用管理器 ### 6. `BrowserManager` (`browser_manager`)
在您的插件中，只需通过 `import` 语句导入相应管理器的全局实例即可使用。 * **怎么找**: `from core.managers.browser_manager import browser_manager`
* **管啥**:
* **浏览器**: 负责启动和关闭 Playwright。
* **页面池**: 提前准备好几个空白页面（默认3个），你要用直接拿
* **循环利用**: 用完记得还回来 (`release_page`)
**示例**: 在插件中检查用户是否为管理员。 ### 7. `ImageManager` (`image_manager`)
* **怎么找**: `from core.managers.image_manager import image_manager`
* **管啥**:
* **美工**: 把数据塞进网页模板
* **记性好**: 模板用一次就记住，下次直接用缓存。
* **自动借还**: 它会自动找 `BrowserManager` 借页面，你只管 `render_template` 就行。
## 咋用？
`import`
**例子**: 查查这人是不是op
```python ```python
# plugins/my_plugin.py # plugins/my_plugin.py
@@ -79,11 +85,10 @@ from models.events.message import MessageEvent
@matcher.command("secret") @matcher.command("secret")
async def secret_command(event: MessageEvent): async def secret_command(event: MessageEvent):
# 使用 permission_manager 检查用户权限 # 只有管理员能看
is_admin = await permission_manager.check_permission(event.user_id, ADMIN) is_admin = await permission_manager.check_permission(event.user_id, ADMIN)
if is_admin: if is_admin:
await event.reply("这是一个只有管理员能看到的秘密。") await event.reply("这是秘密！")
else: else:
await event.reply("抱歉，您没有权限执行此命令。") await event.reply("你没权限看这个。")
``` ```

182
docs/deployment.md
View File

@@ -1,102 +1,102 @@
# 部署指南 # 部署指南
当您的机器人开发完成并准备投入生产环境时，本指南将为您提供部署的最佳实践和建议。 把 Bot 扔到服务器上长期运行，比在自己电脑上跑要多几个步骤。
## 1. 生产环境配置 ## 1. 环境准备
与开发环境不同，生产环境要求更高的稳定性和安全性。 ### a. 安装 Python 3.14
### 创建生产配置文件 用3.14。。。
建议您复制一份 `config.toml` 并重命名为 `config.prod.toml`，专门用于生产环境。 ### b. 安装依赖
**关键修改项**: ```bash
# 切换到项目目录
cd /path/to/your/bot
* **数据库与服务地址**: # 创建虚拟环境 (强烈建议)
* 确保 `napcat_ws` 和 `redis` 部分的地址、端口和密码都指向您的生产服务器，而不是本地开发环境。 python3.14 -m venv venv
source venv/bin/activate
# 安装依赖
pip install -r requirements.txt
## 2. 使用进程守护工具
直接在终端中运行 `python main.py` 适用于开发，但在生产环境中，如果终端关闭或程序意外崩溃，机器人就会下线。
为了确保机器人能够 7x24 小时稳定运行，您应该使用**进程守护工具**。
### 推荐工具
* **PM2 (Node.js)**: 尽管是 Node.js 工具，但 PM2 提供了强大的 Python 进程管理功能，包括崩溃自启、日志管理和性能监控。
* **Supervisor (Python)**: 一个纯 Python 实现的进程控制系统，配置简单，稳定可靠。
* **Systemd (Linux)**: Linux 系统自带的服务管理器，可以创建系统服务来管理机器人进程。
### 使用 PM2 (示例)
1. **安装 PM2**:
```bash
npm install -g pm2
```
2. **创建生态系统文件**:
在项目根目录创建一个 `ecosystem.config.js` 文件：
```javascript
// ecosystem.config.js
module.exports = {
apps: [
{
name: 'neo-bot', // 应用名称
script: 'main.py', // 启动脚本
interpreter: '/path/to/your/venv/bin/python', // 指定虚拟环境的 Python 解释器
env: {
'APP_ENV': 'production', // 设置环境变量
},
},
],
};
```
**注意**: 请务必将 `interpreter` 路径修改为您服务器上虚拟环境的实际路径。
3. **启动应用**:
```bash
pm2 start ecosystem.config.js
```
4. **常用 PM2 命令**:
* `pm2 list`: 查看所有应用状态
* `pm2 logs neo-bot`: 查看日志
* `pm2 restart neo-bot`: 重启应用
* `pm2 stop neo-bot`: 停止应用
* `pm2 startup`: 设置开机自启
## 3. 禁用热重载
热重载功能在开发时非常有用，但在生产环境中会带来不必要的性能开销和潜在的不稳定性。
在部署前，建议您在 `main.py` 中**注释掉**或移除与 `watchdog` 相关的文件监控代码。
**修改 `main.py`**:
```python
# main.py
async def main():
# ...
# 生产环境中禁用文件监控
# loop = asyncio.get_running_loop()
# event_handler = PluginReloadHandler(loop)
# observer = Observer()
# if os.path.exists(plugin_path):
# observer.schedule(event_handler, plugin_path, recursive=True)
# observer.start()
# logger.info(f"已启动插件热重载监控: {plugin_path}")
try:
# ...
finally:
# if observer.is_alive():
# observer.stop()
# observer.join()
``` ```
遵循以上步骤，您就可以将 NEO Bot 机器人稳定、高效地部署在生产服务器上。 ### c. 编译核心模块 (可选，但强烈建议)
为了性能，把核心模块编译成 C 扩展。
```bash
python setup_mypyc.py build_ext --inplace
```
## 2. 使用进程管理器
你想直接 `python main.py` 然后关掉 SSH？那机器人也跟着停了。必须用进程管理器来守护它。
这里推荐用 `pm2`，虽然是 Node.js 的工具，但管 Python 程序一样好用。
### a. 安装 pm2
```bash
# 你需要先装 Node.js 和 npm
npm install pm2 -g
```
### b. 启动 Bot
在项目根目录，创建一个 `ecosystem.config.js` 文件：
```javascript
module.exports = {
apps : [{
name : "neobot",
script : "main.py",
interpreter: "/path/to/your/bot/venv/bin/python", // 指定虚拟环境里的 python
max_memory_restart: "500M", // 内存超过 500M 自动重启
env: {
"PYTHONUNBUFFERED": "1" // 禁用 python 输出缓冲，日志能实时看
}
}]
}
```
然后启动：
```bash
pm2 start ecosystem.config.js
```
### c. 常用 pm2 命令
```bash
pm2 list # 查看所有进程状态
pm2 logs neobot # 查看 neobot 的实时日志
pm2 restart neobot# 重启 neobot
pm2 stop neobot # 停止 neobot
pm2 delete neobot # 删除 neobot
```
## 3. 配置 NapCatQQ
最后一步，修改 NapCatQQ 的配置文件，让它把消息推送到你的服务器上。
找到 NapCatQQ 的 `config/onebot11.json` 文件，修改 `ws_reverse_servers` 部分：
```json
"ws_reverse_servers": [
{
"url": "ws://你的服务器IP:8080/onebot/v11/ws",
"access_token": "你的访问令牌"
}
]
```
* `url`: 改成你服务器的 IP 和 `main.py` 里配置的端口。
* `access_token`: 如果你在 `main.py` 里设置了 `ACCESS_TOKEN`，这里要保持一致。
或者你也可以用napcat的webui，不多赘述了。。。
改完后重启 NapCatQQ，Bot 应该就能收到消息了。

114
docs/getting-started.md
View File

@@ -1,113 +1,95 @@
# 快速上手 # 快速上手
本指南将引导您完成 NEO Bot Framework 的本地开发环境搭建、配置和首次运行。 runit
## 1. 环境准备 ## 1. 你需要准备
在开始之前，请确保您的开发环境中已安装以下软件： * **Python 3.14**: 必须是这个版本别问我为什么。。。
* **Git**: 拉取代码
* **Redis**: 装一个
* **脑子和手**: 这个最重要，或者你去问问镀铬酸钾，会给你一对一教学的。。。
* **OneBot v11 客户端**: 机器人本体，推荐用 [NapCatQQ](https://github.com/NapNeko/NapCatQQ)
* **Python**: 版本要求 `3.12` 或更高。 ## 2. 搭起来
* 我们推荐使用官方的 CPython 解释器。
* 您可以通过在终端运行 `python --version` 来检查您的 Python 版本。
* **Git**: 用于克隆项目仓库。 ### a. 克隆代码
* **Redis**: 一个键值对数据库，用于缓存和数据共享。 找个你喜欢的地方，把代码从 GitHub 上clone下来
* 对于 Windows 用户，可以考虑使用 `memurai` 或通过 WSL2 安装 Redis。
* 对于 macOS 用户，可以使用 `brew install redis`。
* 安装后，请确保 Redis 服务正在运行。
* **OneBot v11 实现端**: 机器人框架需要连接到一个实现了 OneBot v11 协议的客户端。
* **推荐**: [NapCatQQ](https://github.com/NapNeko/NapCatQQ)
## 2. 克隆与安装
### 克隆项目
打开您的终端，并克隆项目仓库到本地：
```bash ```bash
git clone [项目仓库地址] git clone [项目仓库地址]
cd [项目目录] cd [项目目录]
``` ```
### 创建虚拟环境 (推荐) ### b. 创建虚拟环境
为了保持项目依赖的隔离，强烈建议您创建一个 Python 虚拟环境。 别把你的系统环境搞得乱七八糟
```bash ```bash
# 创建虚拟环境
python -m venv venv
# 激活虚拟环境
# Windows # Windows
python -m venv venv
.\venv\Scripts\activate .\venv\Scripts\activate
# macOS / Linux
# Linux / macOS
python3.14 -m venv venv
source venv/bin/activate source venv/bin/activate
``` ```
### 安装依赖 看到命令行前面多了个 `(venv)`，就说明你进来了。
### c. 安装依赖
激活虚拟环境后，使用 `pip` 安装所有必需的第三方库：
```bash ```bash
pip install -r requirements.txt pip install -r requirements.txt
``` ```
## 3. 配置 ### d. 安装 Playwright 依赖
项目的核心配置位于根目录下的 `config.toml` 文件中。 我们用 Playwright 来截图画画，它需要一个浏览器核心。
对于内部开发，该文件通常已预先配置好，可以直接连接到测试服务器。如果您需要连接到自己的环境，请修改以下关键部分： ```bash
playwright install chromium
```
### e. 编译核心 (可选，但强烈建议)
想让你的代码更快？把它的核心代码编译成 C。
```bash
python setup_mypyc.py build_ext --inplace
```
*注：Windows 上可能需要装个 Visual Studio Build Tools，Linux 上需要 GCC。编译失败也别慌，跳过就行，JIT 也能保证不错的速度*
## 3. 第一次
### a. 修改配置
去根目录找 `config.toml`。
```toml ```toml
# config.toml
[napcat_ws] [napcat_ws]
# 您的 OneBot v11 实现端的 WebSocket 地址 # 你的 OneBot 地址
# 格式通常为 ws://<IP地址>:<端口号> # 我们用的是正向连接，也就是 Bot 主动去连 OneBot
uri = "ws://127.0.0.1:3001" uri = "ws://127.0.0.1:3001"
# Access Token (访问令牌)，如果您的 OneBot 端设置了
token = "" token = ""
[redis] [redis]
# Redis 服务的连接信息
host = "127.0.0.1" host = "127.0.0.1"
port = 6379 port = 6379
db = 0 db = 0
password = "" # 如果您的 Redis 设置了密码
``` ```
把 `uri` 改成你自己的 OneBot 地址。
## 4. 首次运行 ### b. 启动！
完成以上所有步骤后，您就可以启动机器人了。在项目根目录运行： 一切就绪
```bash ```bash
python main.py # 推荐开启 JIT 模式启动
python -X jit main.py
``` ```
如果一切顺利，您将在控制台看到类似以下的输出： 如果你看到日志刷出来，最后显示 “连接成功！”，恭喜，你成功了！
``` 现在，试着给你的机器人发个 `/help`看看会返回什么东西
2026-01-07 22:42:41.718 | INFO | ... - 管理员管理器初始化完成
2026-01-07 22:42:41.826 | INFO | ... - 正在从 plugins 加载插件...
2026-01-07 22:42:41.994 | SUCCESS | ... - Redis 连接成功！
...
2026-01-07 22:42:42.618 | SUCCESS | ... - 连接成功！
```
看到 `连接成功！` 的日志，即表示您的机器人已成功连接到 OneBot 客户端并准备好接收消息。
## 5. 常见问题排查 (FAQ)
* **Q: 启动时报错 `redis.exceptions.ConnectionError`**
* **A**: 请检查您的 Redis 服务是否已启动，以及 `config.toml` 中的 `host` 和 `port` 是否正确。
* **Q: 无法连接到 WebSocket，提示 `ConnectionRefusedError`**
* **A**: 请确认您的 OneBot v11 客户端（如 NapCatQQ）是否正在运行，并检查 `config.toml` 中的 `uri` 地址和端口是否匹配。
* **Q: 修改了插件代码但没有生效**
* **A**: 框架默认开启了热重载功能。请检查控制台是否有 `[HotReload]` 相关的日志输出。如果没有，请确认 `watchdog` 库已正确安装。
现在，您的开发环境已经准备就绪。接下来，您可以尝试修改一个现有插件或[创建您的第一个插件](./plugin-development/index.md)！

45
docs/index.md
View File

@@ -1,34 +1,29 @@
# NEO Bot Framework 开发文档 # NEO Bot 开发文档
欢迎来到 NEO Bot Framework 的官方开发文档。 嘿，朋友，欢迎来到 NEO Bot
本文档旨在为开发者提供一个清晰、全面的指南，帮助您理解框架的设计理念、核心功能，并快速上手插件开发。 这里没那么多规矩。这份文档是我写给你——未来的插件开发者、或者单纯好奇想拆开看看的家伙——的一份地图
## 📖 文档结构
本站点的文档分为以下几个主要部分： ## 📖 地图导览
* **基础入门** ### 1. 准备阶段
* [快速上手](./getting-started.md): 从零开始配置和运行您的第一个机器人实例。 * [快速上手](./getting-started.md): 搭环境、装东西、启动。跟着走一遍，能省不少事。
* [项目结构解析](./project-structure.md): 详细介绍框架的目录和文件结构。 * [项目怎么样](./project-structure.md): 看看各个文件夹都是干嘛的，免得迷路。
* [生产环境](./deployment.md): 怎么把你调教好的 Bot 扔服务器上，让它自己 7x24 小时跑。
* **核心概念** ### 2. 核心探秘
* [事件流转](./core-concepts/event-flow.md): 深入理解一个事件从接收到处理的完整生命周期。 * [骨架](./core-concepts/architecture.md): 看看镀铬酸钾和python打架，嗯。。。
* [单例管理器](./core-concepts/singleton-managers.md): 了解框架中核心管理器（如 `CommandManager`, `PermissionManager`）的设计与使用。 * [性能优化](./core-concepts/performance.md): 页面池、JIT、Mypyc...
* [消息流](./core-concepts/event-flow.md): 看看一条消息从被接收到被回复是如何运行的
* [核心](./core-concepts/singleton-managers.md): `matcher`, `browser_manager`... 认识这些核心模块。
* **插件开发** ### 3. 插件开发
* [基础指南](./plugin-development/index.md): 学习如何创建一个插件，包括元数据定义和热重载工作流。 * [插件开发第一步](./plugin-development/index.md): 带你写第一个插件
* [指令处理](./plugin-development/command-handling.md): 掌握如何使用 `@matcher.command()` 装饰器注册和处理聊天指令。 * [指南](./plugin-development/command-handling.md): 怎么教你的 Bot 听懂指令和参数。
* [绝对不要做的事情](./plugin-development/best-practices.md): **（必读！）**
* **部署** ## 贡献
* [部署指南](./deployment.md): 了解如何在生产环境中部署和维护机器人。
## 🤝 如何贡献 发现 Bug 了？觉得文档写得烂？
直接提 Issue 或者 PR。代码质量是第一位的，Talk is cheap, show me the code.
我们欢迎任何形式的贡献，无论是代码提交、文档修正还是功能建议。
* **报告问题**: 如果您在使用中遇到任何问题或 Bug，请通过内部渠道提交 Issue。
* **提交代码**: 请遵循项目的编码规范，并通过 Pull Request 流程提交您的代码。
* **完善文档**: 如果您发现文档中有任何错误或遗漏，可以直接提出修改建议。
我们希望这份文档能让您的开发之旅更加顺畅。如果您有任何疑问，请随时与我们联系。

67
docs/plugin-development/best-practices.md Normal file
View File

@@ -0,0 +1,67 @@
# 插件开发最佳实践
写插件很简单，但写出**高性能、不炸裂**的插件需要遵守规矩。
## 1. 绝对不要阻塞事件循环。。。
这是底线。NEO Bot 是单线程异步架构，如果你在主线程里 `time.sleep(5)`，整个机器人就会卡死 5 秒
* **错误**: `time.sleep(1)`, `requests.get(...)`, 大量 CPU 计算。
* **正确**: `await asyncio.sleep(1)`, `await session.get(...)`。
如果你必须运行同步代码（比如图像处理、复杂计算）：
```python
from core.utils.executor import run_in_thread_pool
# 扔到线程池里去跑，别占着主线程
result = await run_in_thread_pool(heavy_function, arg1, arg2)
```
## 2. 复用资源
别每次都创建新的连接。
* **HTTP 请求**: 使用插件内提供的 `get_session()` 或全局 `aiohttp` session。
* **浏览器**: 必须使用 `browser_manager.get_page()`，严禁自己 `playwright.chromium.launch()`。
## 3. 善用缓存
如果你的插件需要查外部 API（比如查天气、查 B 站），记得加缓存。
Redis 就在那里，不用白不用。
```python
from core.managers.redis_manager import redis_manager
# 存
await redis_manager.set("weather:beijing", "sunny", ex=3600)
# 取
weather = await redis_manager.get("weather:beijing")
```
## 4. 类型提示 (Type Hinting)
我开启了 Mypyc 编译，这意味着你的代码最好有规范的类型提示。
这不仅是为了编译，也是为了让你自己少写 Bug
```python
# 好的写法
async def handle(event: MessageEvent, args: list[str]) -> None:
...
# 不好写法
async def handle(event, args):
...
```
## 5. 异常处理
别让你的插件因为一个报错就崩溃机器人
虽然框架层有捕获机制，但你自己处理好异常是最好的。。。
```python
try:
await do_something()
except Exception as e:
logger.error(f"插件炸了: {e}")
await event.reply("出错了，请稍后再试。")
```

208
docs/plugin-development/command-handling.md
View File

@@ -1,99 +1,137 @@
# 插件开发：指令处理 # 指令处理与参数解析
`@matcher.command()` 是插件开发中使用最频繁的装饰器。本节将深入介绍它的高级用法，帮助您构建功能更强大的指令。 光会 `event.reply()` 只能写小插件。。。认识一下其他的方法吧
## 1. 获取指令参数 ## 1. 获取原始参数
在很多场景下，指令都需要接收用户提供的参数，例如 `/weather 北京`。框架会自动解析这些参数，并通过函数签名注入到您的处理器中。 最简单粗暴的方式，就是直接在处理器函数里声明 `args: str`。
您只需要在处理函数的参数列表中添加一个名为 `args` 的参数，并指定其类型为 `list[str]`。
```python ```python
# plugins/weather.py
from core.managers.command_manager import matcher from core.managers.command_manager import matcher
from models.events.message import MessageEvent from models.events.message import MessageEvent
@matcher.command("weather") @matcher.command("echo")
async def handle_weather_command(event: MessageEvent, args: list[str]): async def handle_echo(event: MessageEvent, args: str):
""" # 如果用户发送 /echo hello world
处理 /weather 指令 # args 的值就是 "hello world"
:param event: 消息事件对象
:param args: 用户发送的参数列表 (已按空格分割)
"""
if not args: if not args:
await event.reply("请输入城市名，例如：/weather 北京") await event.reply("你啥也没说啊")
else:
await event.reply(f"你说了：{args}")
```
`args` 就是去掉命令本身后，后面跟着的**一整坨字符串**。
## 2. 自动解析参数 (推荐)
一整坨字符串用起来太费劲了，还得自己 `split()`。框架提供了更高级的玩法：**参数自动解析**。
你只需要在函数签名里，用类型提示声明你想要的参数，框架会动帮你解析和注入。
### a. 基础用法
```python
from core.managers.command_manager import matcher
from models.events.message import MessageEvent
@matcher.command("add")
async def handle_add(event: MessageEvent, a: int, b: int):
# 如果用户发送 /add 10 20
# 框架会自动把 "10" 转成整数 10，注入给 a
# 把 "20" 转成整数 20，注入给 b
result = a + b
await event.reply(f"计算结果是：{result}")
```
**它是怎么工作的？**
框架会按顺序把 `args` 字符串用空格分割，然后尝试把分割后的每一块，转换成你声明的参数类型。
* `/add 10 20` -> `args` 是 `"10 20"` -> 分割成 `["10", "20"]`
* 第一块 `"10"` -> 尝试转成 `int` -> 成功，`a = 10`
* 第二块 `"20"` -> 尝试转成 `int` -> 成功，`b = 20`
### b. 处理可选参数和默认值
你可以像普通 Python 函数一样，给参数提供默认值。
```python
from typing import Optional
@matcher.command("greet")
async def handle_greet(event: MessageEvent, name: str, title: Optional[str] = "先生"):
# 例 1: /greet 张三
# name = "张三", title = "先生" (默认值)
# 例 2: /greet 李四 女士
# name = "李四", title = "女士"
await event.reply(f"你好，{name} {title}！")
```
### c. 贪婪的最后一个参数
有时候，最后一个参数可能包含空格，比如 `/say hello world`。默认情况下，`hello` 会被解析给第一个参数，`world` 会被解析给第二个。
如果你想让最后一个参数“吃掉”所有剩下的内容，可以用 `...` 作为默认值（这是一个特殊的标记）。
```python
@matcher.command("say")
async def handle_say(event: MessageEvent, target_user: str, content: str = ...):
# 例: /say 张三 早上好，吃了没？
# target_user = "张三"
# content = "早上好，吃了没？"
await event.reply(f"正在对 {target_user} 说：{content}")
```
## 3. 智能的参数注入
除了 `args` 列表，命令处理器还可以自动接收一些非常有用的上下文对象。框架底层使用了 Python 的 `inspect` 模块来分析你函数的参数签名，并自动“注入”你需要的对象。
这是一种轻量级的**依赖注入**，让你的代码更简洁、更易于测试。
### 可用的参数
你可以在命令处理函数的参数中声明以下任意名称，框架会自动为你传入：
| 参数名 | 类型 | 描述 |
| ------------------- | -------------------------------- | ---------------------------------------- |
| `bot` | `Bot` | 当前的 Bot 实例，用于调用 API 发送消息等。 |
| `event` | `MessageEvent` (或其子类) | 触发该命令的完整消息事件对象。 |
| `args` | `List[str]` | 和之前一样，包含命令参数的字符串列表。 |
| `permission_granted`| `bool` | 指示当前用户是否通过了权限检查。 |
### 示例
假设我们想写一个“回声”命令，但只在用户拥有管理员权限时才重复他们的消息。
```python
# plugins/echo_plus.py
from core.bot import Bot
from core.permission import ADMIN
from models.events.message import MessageEvent
from core.managers.command_manager import matcher
@matcher.command("echo_plus", permission=ADMIN)
async def echo_plus(bot: Bot, event: MessageEvent, args: list[str], permission_granted: bool):
"""
一个更强大的回声命令
"""
# 只有当 permission_granted 为 True 时，代码才会执行到这里
# 因为框架会自动处理权限拒绝的情况
if not args:
await bot.send(event, "你想要我复述什么呢？")
return return
# args[0] 就是 "北京" # 我们可以从 event 对象中获取更详细的信息
city = args[0] user_id = event.user_id
message_to_echo = " ".join(args)
response = f"管理员 {user_id} 说：{message_to_echo}"
await bot.send(event, response)
# ...后续逻辑...
await event.reply(f"正在查询 {city} 的天气...")
``` ```
* 如果用户发送 `/weather 北京`，`args` 将是 `['北京']`。 在这个例子中，我们没有手动检查权限。我们只是在 `@matcher.command` 中声明了 `permission=ADMIN`，然后在函数参数中请求了 `permission_granted: bool`。框架会自动完成权限检查，如果失败，甚至不会执行我们的函数，并会发送一条权限不足的消息。这就是依赖注入的强大之处。
* 如果用户发送 `/weather 上海 浦东`，`args` 将是 `['上海', '浦东']`。
* 如果用户只发送 `/weather`，`args` 将是一个空列表 `[]`。
## 2. 设置指令别名
同一个功能，用户可能习惯使用不同的指令名称来触发，例如 `天气` 和 `weather`。`@matcher.command()` 允许您为一个处理器设置多个别名。
只需在装饰器中传入多个名称即可：
```python
@matcher.command("weather", "天气")
async def handle_weather_command(event: MessageEvent, args: list[str]):
# ...
```
现在，用户发送 `/weather 北京` 或 `/天气 北京` 都可以触发这个函数。
## 3. 权限控制
某些敏感指令只希望特定权限的用户才能执行，例如 `/reload` (重载插件) 或 `/ban` (禁言用户)。
`@matcher.command()` 装饰器提供了一个 `permission` 参数，可以轻松实现权限控制。
首先，从 `permission_manager` 导入预设的权限等级：
```python
from core.managers.permission_manager import ADMIN, OP, USER
```
然后，在装饰器中指定所需的权限：
```python
# plugins/admin_tools.py
from core.managers.command_manager import matcher
from core.managers.permission_manager import ADMIN
from models.events.message import MessageEvent
__plugin_meta__ = {
"name": "管理工具",
"description": "提供机器人管理功能",
"usage": "/reload - 重载所有插件 (仅管理员)",
}
@matcher.command("reload", permission=ADMIN)
async def handle_reload_command(event: MessageEvent):
"""
重载所有插件，仅限管理员使用。
"""
# 这里的逻辑只有在权限检查通过后才会执行
await event.reply("正在重载所有插件...")
# ... 执行重载逻辑 ...
```
* **工作原理**: 在调用您的处理函数之前，`CommandManager` 会自动调用 `PermissionManager` 来检查用户的权限。
* **失败响应**: 如果用户权限不足，框架会自动回复一条权限不足的消息（该消息内容可在 `config.toml` 中配置），并且**不会**执行您的处理函数。
可用的权限等级：
* `ADMIN`: 机器人超级管理员。
* `OP`: 管理员（Operator），权限低于 `ADMIN`。
* `USER`: 普通用户，默认权限。
权限关系是 `ADMIN > OP > USER`。设置 `permission=OP` 意味着 `OP` 和 `ADMIN` 都可以使用该指令。
通过组合使用参数处理、别名和权限控制，您可以构建出既灵活又安全的指令来满足各种复杂的需求。

116
docs/plugin-development/index.md
View File

@@ -1,51 +1,10 @@
# 插件开发：基础指南 # 插件开发入门
在 NEO Bot Framework 中，几乎所有的功能都是通过**插件**来实现的。框架提供了一个强大而简单的插件系统，让您可以专注于功能逻辑的实现。 写插件是给 NEO Bot 添加功能的唯一方式，一个 Python 文件就是一个插件。（或者一个文件夹里边有__init__.py）
## 插件是什么？ ## 1. 创建你的第一个插件
一个插件本质上就是一个位于 `plugins/` 目录下的独立 Python 文件 (`.py`)。 在 `plugins/` 目录下，新建一个 `hello.py` 文件。
框架会在启动时自动扫描并加载这个目录下的所有文件作为插件。
## 🔥 热重载工作流
在开始编写插件之前，了解框架的**热重载**机制至关重要，它能极大地提升您的开发效率。
1. **启动机器人**: 首先，在您的终端中运行 `python main.py` 并保持其运行状态。
2. **创建或修改插件**: 在 `plugins/` 目录下创建新的 `.py` 文件，或者修改一个已有的插件文件。
3. **保存文件**: 当您保存文件时，框架会自动检测到文件变更。
4. **自动重载**: 控制台会显示 `插件重载完成` 的日志，这意味着您的新代码已经生效，无需重启整个程序。
## 创建您的第一个插件
让我们来创建一个经典的 "Hello World" 插件。
### 1. 创建文件
在 `plugins/` 目录下创建一个新文件，命名为 `hello.py`。
### 2. 定义插件元数据 (`__plugin_meta__`)
为了让框架能够识别您的插件信息（例如在 `/help` 命令中显示），您需要在文件顶部定义一个名为 `__plugin_meta__` 的特殊字典。
```python
# plugins/hello.py
__plugin_meta__ = {
"name": "你好世界",
"description": "一个简单的插件，用于回复 'Hello, World!'。",
"usage": "/hello - 发送问候。",
}
```
* `name`: 插件的名称。
* `description`: 插件功能的简短描述。
* `usage`: 插件的使用方法说明。
### 3. 编写处理器
现在，让我们来编写一个响应 `/hello` 指令的函数。我们需要从框架中导入 `matcher` 和事件类型。
```python ```python
# plugins/hello.py # plugins/hello.py
@@ -53,36 +12,63 @@ __plugin_meta__ = {
from core.managers.command_manager import matcher from core.managers.command_manager import matcher
from models.events.message import MessageEvent from models.events.message import MessageEvent
# __plugin_meta__ 是插件元信息，会在 /help 指令里显示
__plugin_meta__ = { __plugin_meta__ = {
"name": "你好世界", "name": "你好世界",
"description": "一个简单的插件，用于回复 'Hello, World!'。", "description": "一个简单的示例插件",
"usage": "/hello - 发送问候。", "usage": "/hello - 发送你好"
} }
# 使用 @matcher.command 装饰器来注册一个指令 # @matcher.command() 装饰器注册一个命令
@matcher.command("hello") # "hello" 是命令名，aliases 是别名
async def handle_hello_command(event: MessageEvent): @matcher.command("hello", aliases=["hi", "你好"])
async def handle_hello(event: MessageEvent):
""" """
当用户发送 /hello 时，此函数将被调用。 处理 /hello 命令
""" """
# 使用 event.reply() 方法可以快速回复消息到来源地 # event.reply() 是一个快捷方法，可以直接回复消息
await event.reply("Hello, World!") await event.reply(f"你好，{event.sender.nickname}！")
``` ```
### 4. 测试插件 ## 2. 加载插件
1. 确保 `python main.py` 正在运行。 不用你动手，NEO Bot 启动时会自动加载 `plugins/` 目录下的所有 `.py` 文件。
2. 保存 `plugins/hello.py` 文件。您应该会在控制台看到插件重载的日志。
3. 在任何一个机器人所在的群聊或私聊中，发送 `/hello`。
4. 机器人应该会回复 `Hello, World!`。
恭喜！您已经成功创建并运行了您的第一个插件。 ## 3. 测试插件
## 插件的最佳实践 现在，去群里或者私聊给 Bot 发送：
* **保持独立**: 尽量让每个插件文件只负责一项相关的功能。 * `/hello`
* **清晰命名**: 为您的插件文件和处理函数选择清晰、描述性的名称。 * `/hi`
* **善用模型**: 充分利用 `models` 中定义的事件和消息段类型，以获得完整的类型提示和代码补全支持。 * `/你好`
* **异步优先**: 框架是基于 `asyncio` 构建的。对于任何 I/O 密集型操作（如网络请求、文件读写），请务必使用 `async/await` 语法，以避免阻塞事件循环。
现在您已经掌握了插件的基础，可以继续学习更高级的主题，例如[如何处理带参数的指令](./command-handling.md)。 Bot 应该会回复你：“你好，[你的昵称]！”
## 插件剖析
### `__plugin_meta__`
这个字典不是必须的，但强烈建议写上。它定义了插件的元信息，主要给 `/help` 命令用。
* `name`: 插件叫啥。
* `description`: 这插件是干嘛的。
* `usage`: 怎么用，写上具体的指令和说明。
### `@matcher.command()`
这是最核心的装饰器，用来注册一个命令处理器。
* **第一个参数**: `name` (str)，命令的主名。
* `aliases`: `List[str]`，命令的别名列表。
* `permission`: `int`，执行该命令所需的权限等级，默认为 `USER` (所有人可用)。可以是 `ADMIN`, `OP`。
### 处理器函数
被 `@matcher.command()` 装饰的函数就是处理器。它必须是一个 `async` 异步函数。
* **参数**: 框架会自动往里注入参数，你只需要用类型提示声明你需要什么。
* `event: MessageEvent`: 这是最常用的，包含了消息的所有信息，比如发送者、群号、消息内容等。
* `args: str`: 如果命令有参数（比如 `/echo hello world`），`args` 就是 `hello world` 这部分字符串。
就这么简单，一个最基础的插件就写完了。

84
docs/project-structure.md
View File

@@ -1,70 +1,48 @@
# 项目结构解析 # 项目结构
理解 NEO Bot Framework 的项目结构是高效开发的第一步。本节将详细介绍每个主要目录和文件的用途。 了解项目里每个文件夹是干嘛的，能让你更快找到代码。
``` ```
. .
├── core/ # 框架核心代码 ├── core/ # 核心代码，别乱动
│ ├── api/ # OneBot v11 API 的 Mixin 封装 │ ├── handlers/ # 底层事件处理器
│ ├── data/ # 核心模块的数据存储 (admin, permissions) │ ├── managers/ # 全局单例管理器
│ ├── handlers/ # 底层事件处理器 (message, notice, request) │ ├── utils/ # 工具函数
│ ├── managers/ # 核心单例管理器 (command, permission, etc.) │ └── ws.py # WebSocket 连接实现
│ ├── utils/ # 通用工具 (logger, singleton, etc.) ├── data/ # 存放持久化数据
│ ├── bot.py # Bot 核心类，提供 API 调用接口 │ ├── admin.json # 管理员列表
│ ├── config_loader.py # TOML 配置文件加载器 │ └── permissions.json # 用户权限列表
│ └── ws.py # WebSocket 底层通信模块
├── docs/ # 开发文档 ├── docs/ # 开发文档
├── html/ # 静态网页文件 (用于 Web 仪表盘等) ├── logs/ # 日志文件
├── models/ # 数据模型 (事件, 消息段) ├── models/ # 数据模型
│ ├── events/ # OneBot v11 事件的 Python 对象封装 │ └── events/ # OneBot 事件模型
│ ├── message.py # 消息段 (MessageSegment) 的定义 ├── plugins/ # 你的插件都放这
│ └── ... ├── templates/ # 图片渲染用的网页模板
├── plugins/ # 功能插件目录 ├── venv/ # Python 虚拟环境
├── venv/ # Python 虚拟环境 (推荐) ├── .gitignore # Git 忽略配置
├── .gitignore # Git 忽略文件配置 ├── main.py # 主入口文件
├── config.toml # 主配置文件 ├── requirements.txt # Python 依赖列表
├── main.py # 项目启动入口 └── setup_mypyc.py # Mypyc 编译脚本
└── requirements.txt # Python 依赖列表
``` ```
## 顶层目录 ## 重点目录说明
### `core/` ### `core/`
这是框架的心脏，包含了所有核心逻辑。**通常情况下，您不需要修改此目录下的代码**，只需了解其工作原理即可。 这是框架的心脏。除非你知道自己在干嘛，否则别碰这里面的东西。大部分功能都由 `managers` 里的管理器提供，你只需要 `import` 它们就行。
* `api/`: 将 OneBot v11 的 API 按功能（如 `message`, `group`）拆分为多个 `Mixin` 类，最终由 `bot.py` 继承，提供了清晰的 API 结构。 ### `data/`
* `data/`: 存放核心模块所需的数据文件，例如 `admin.json` 和 `permissions.json`。
* `handlers/`: 定义了最底层的事件处理器，如 `MessageHandler`，负责从 `ws.py` 接收原始事件并进行初步处理和分发。
* `managers/`: 包含一系列全局单例管理器，是框架功能的核心实现。例如，`CommandManager` 负责指令注册与匹配，`PermissionManager` 负责权限控制。
* `utils/`: 提供被广泛使用的工具类，如 `logger` (日志)、`singleton` (单例模式基类)。
* `bot.py`: 定义了 `Bot` 类，这是插件开发者最常与之交互的对象，用于调用所有 OneBot API。
* `config_loader.py`: 负责解析 `config.toml` 文件，并提供一个全局的 `global_config` 对象。
* `config_models.py`: 使用 Pydantic 定义了配置文件的结构和类型验证。
* `ws.py`: 实现了与 OneBot v11 实现端的 WebSocket 连接、心跳、重连和消息收发。
### `docs/` 存放一些 JSON 格式的数据。管理员和用户权限默认存在这里。如果你用 Redis，这些文件会作为备份。
存放项目的所有开发文档。
### `html/`
用于存放未来 Web 仪表盘或其他 Web 功能所需的静态资源（HTML, CSS, JavaScript）。
### `models/`
定义了将 OneBot v11 的 JSON 数据转换为易于使用的 Python 对象。
* `events/`: 将所有上报的事件（如 `MessageEvent`, `GroupIncreaseNoticeEvent`）封装为带有类型提示的类。
* `message.py`: 提供了 `MessageSegment` 类，用于构建复杂的消息内容（如 @某人、发送图片）。
### `plugins/` ### `plugins/`
这是**插件开发者最关心的目录**。所有机器人的功能都以独立的 `.py` 文件形式存放在这里。框架会自动加载此目录下的所有插件，并支持热重载。 **这是你最常待的地方**。你写的所有插件（`.py` 文件）都扔在这个目录里。Bot 启动时会自动加载这里的所有插件。
## 顶层文件 ### `templates/`
* `.gitignore`: 配置 Git 应忽略的文件和目录，如 `__pycache__`、`venv` 等。 如果你要用 `ImageManager` 画图，就需要把 HTML 模板文件放在这里。
* `config.toml`: 项目的主配置文件，用于设置机器人、数据库、API 等所有可变参数。
* `main.py`: 项目的启动入口脚本。它负责初始化日志、加载插件、启动 WebSocket 连接和文件监控（用于热重载）。 ### `main.py`
* `requirements.txt`: 列出了项目运行所需的所有 Python 第三方库及其版本。
程序的入口。负责加载配置、初始化管理器、启动 WebSocket 连接和 FastAPI 服务。

16
import sys.py Normal file
View File

@@ -0,0 +1,16 @@
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)")

19
main.py
View File

@@ -10,6 +10,21 @@ import time
from watchdog.observers import Observer from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler from watchdog.events import FileSystemEventHandler
# 尝试使用高性能事件循环
try:
if sys.platform == 'win32':
# winloop 与 Playwright 存在兼容性问题 (不支持 startupinfo)，暂时禁用
# import winloop
# asyncio.set_event_loop_policy(winloop.EventLoopPolicy())
# print("已启用 winloop 高性能事件循环")
print("Windows 平台检测到 Playwright，已自动禁用 winloop 以确保兼容性")
else:
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
print("已启用 uvloop 高性能事件循环")
except ImportError:
print("未检测到高性能事件循环库 (uvloop/winloop)，将使用默认事件循环")
# 初始化日志系统，必须在其他 core 模块导入之前执行 # 初始化日志系统，必须在其他 core 模块导入之前执行
from core.utils.logger import logger from core.utils.logger import logger
@@ -118,8 +133,8 @@ async def main():
# 初始化管理员管理器 # 初始化管理员管理器
await admin_manager.initialize() await admin_manager.initialize()
# 初始化浏览器管理器 # 初始化浏览器管理器 (使用页面池)
await browser_manager.initialize() await browser_manager.init_pool(size=3)
# 启动文件监控 # 启动文件监控
# 监控 plugins 目录 # 监控 plugins 目录

59
plugins/bili_parser.py
View File

@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import re import re
import json import json
import requests import aiohttp
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from typing import Optional, Dict, Any, Union from typing import Optional, Dict, Any, Union
from cachetools import TTLCache from cachetools import TTLCache
@@ -23,6 +23,15 @@ 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' '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'
} }
# 全局共享的 ClientSession
_session: Optional[aiohttp.ClientSession] = None
async def get_session() -> aiohttp.ClientSession:
global _session
if _session is None or _session.closed:
_session = aiohttp.ClientSession()
return _session
def format_count(num: int) -> str: def format_count(num: int) -> str:
if not isinstance(num, int): if not isinstance(num, int):
@@ -40,20 +49,23 @@ def format_duration(seconds: int) -> str:
return f"{minutes:02d}:{seconds:02d}" return f"{minutes:02d}:{seconds:02d}"
def get_real_url(short_url: str) -> Optional[str]: async def get_real_url(short_url: str) -> Optional[str]:
try: try:
response = requests.head(short_url, headers=HEADERS, allow_redirects=False, timeout=5) session = await get_session()
if response.status_code == 302: async with session.head(short_url, headers=HEADERS, allow_redirects=False, timeout=5) as response:
return response.headers.get('Location') if response.status == 302:
except requests.RequestException as e: return response.headers.get('Location')
print(f"获取真实URL失败: {e}") except Exception as e:
logger.error(f"获取真实URL失败: {e}")
return None return None
def parse_video_info(video_url: str) -> Optional[Dict[str, Any]]: async def parse_video_info(video_url: str) -> Optional[Dict[str, Any]]:
try: try:
response = requests.get(video_url, headers=HEADERS, timeout=5) session = await get_session()
response.raise_for_status() async with session.get(video_url, headers=HEADERS, timeout=5) as response:
soup = BeautifulSoup(response.text, 'html.parser') response.raise_for_status()
text = await response.text()
soup = BeautifulSoup(text, 'html.parser')
script_tag = soup.find('script', text=re.compile('window.__INITIAL_STATE__')) script_tag = soup.find('script', text=re.compile('window.__INITIAL_STATE__'))
if not script_tag or not script_tag.string: if not script_tag or not script_tag.string:
@@ -98,12 +110,12 @@ def parse_video_info(video_url: str) -> Optional[Dict[str, Any]]:
"followers": up_data.get('fans', 0), "followers": up_data.get('fans', 0),
} }
except (requests.RequestException, KeyError, AttributeError, json.JSONDecodeError) as e: except (aiohttp.ClientError, KeyError, AttributeError, json.JSONDecodeError) as e:
print(f"解析视频信息失败: {e}") logger.error(f"解析视频信息失败: {e}")
return None return None
def get_direct_video_url(video_url: str) -> Optional[str]: async def get_direct_video_url(video_url: str) -> Optional[str]:
""" """
调用第三方API解析B站视频直链 调用第三方API解析B站视频直链
:param video_url: B站视频的完整URL :param video_url: B站视频的完整URL
@@ -111,12 +123,13 @@ def get_direct_video_url(video_url: str) -> Optional[str]:
""" """
api_url = f"https://api.mir6.com/api/bzjiexi?url={video_url}&type=json" api_url = f"https://api.mir6.com/api/bzjiexi?url={video_url}&type=json"
try: try:
response = requests.get(api_url, headers=HEADERS, timeout=10) async with aiohttp.ClientSession() as session:
response.raise_for_status() async with session.get(api_url, headers=HEADERS, timeout=10) as response:
data = response.json() response.raise_for_status()
if data.get("code") == 200 and data.get("data"): data = await response.json()
return data["data"][0].get("video_url") if data.get("code") == 200 and data.get("data"):
except (requests.RequestException, json.JSONDecodeError, KeyError, IndexError) as e: return data["data"][0].get("video_url")
except (aiohttp.ClientError, json.JSONDecodeError, KeyError, IndexError) as e:
logger.error(f"[bili_parser] 调用第三方API解析视频失败: {e}") logger.error(f"[bili_parser] 调用第三方API解析视频失败: {e}")
return None return None
@@ -178,7 +191,7 @@ async def process_bili_link(event: MessageEvent, url: str):
:param url: 待处理的B站链接 :param url: 待处理的B站链接
""" """
if "b23.tv" in url: if "b23.tv" in url:
real_url = get_real_url(url) real_url = await get_real_url(url)
if not real_url: if not real_url:
logger.error(f"[bili_parser] 无法从 {url} 获取真实URL。") logger.error(f"[bili_parser] 无法从 {url} 获取真实URL。")
await event.reply("无法解析B站短链接。") await event.reply("无法解析B站短链接。")
@@ -186,7 +199,7 @@ async def process_bili_link(event: MessageEvent, url: str):
else: else:
real_url = url.split('?')[0] real_url = url.split('?')[0]
video_info = parse_video_info(real_url) video_info = await parse_video_info(real_url)
if not video_info: if not video_info:
logger.error(f"[bili_parser] 无法从 {real_url} 解析视频信息。") logger.error(f"[bili_parser] 无法从 {real_url} 解析视频信息。")
await event.reply("无法获取视频信息，可能是B站接口变动或视频不存在。") await event.reply("无法获取视频信息，可能是B站接口变动或视频不存在。")
@@ -197,7 +210,7 @@ async def process_bili_link(event: MessageEvent, url: str):
if video_info['duration'] > 300: # 5分钟 = 300秒 if video_info['duration'] > 300: # 5分钟 = 300秒
video_message = "视频时长超过5分钟，不进行解析。" video_message = "视频时长超过5分钟，不进行解析。"
else: else:
direct_url = get_direct_video_url(real_url) direct_url = await get_direct_video_url(real_url)
if direct_url: if direct_url:
video_message = MessageSegment.video(direct_url) video_message = MessageSegment.video(direct_url)
else: else:

42
setup_mypyc.py Normal file
View File

@@ -0,0 +1,42 @@
"""
Mypyc 编译脚本
用于将核心 Python 模块编译为 C 扩展，以提升性能。
使用方法:
python setup_mypyc.py build_ext --inplace
注意:
1. 需要安装 C 编译器 (Windows 上需要 Visual Studio Build Tools, Linux 上需要 GCC)。
2. 编译后的文件 (.pyd 或 .so) 是平台相关的，不能跨平台复制。
3. 建议在部署的目标环境 (Linux) 上运行此脚本。
"""
from distutils.core import setup
from mypyc.build import mypycify
import os
import sys
# 待编译的模块列表
# 注意：Mypyc 对动态特性支持有限，只选择计算密集或类型明确的模块
modules = [
'core/utils/json_utils.py', # JSON 处理
'core/managers/command_manager.py', # 指令匹配和分发
'core/ws.py', # WebSocket 核心
'core/managers/plugin_manager.py', # 插件管理器
]
# 确保文件存在
valid_modules = []
for m in modules:
if os.path.exists(m):
valid_modules.append(m)
else:
print(f"Warning: Module {m} not found, skipping.")
if not valid_modules:
print("No valid modules found to compile.")
sys.exit(1)
setup(
name='neobot_core_compiled',
ext_modules=mypycify(valid_modules),
)

10
x = 5.py Normal file
View File

@@ -0,0 +1,10 @@
x = 5
# 它有自己的身份
print(id(x))
# 它有自己的类型
print(type(x))
# 它甚至有自己的工具！
print(x.bit_length())