From 046dd0860fe6649da5dabf69667e6fe62a5efdf6 Mon Sep 17 00:00:00 2001 From: K2cr2O1 <2221577113@qq.com> Date: Thu, 1 Jan 2026 17:58:17 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8A=BD=E8=B1=A1send=E6=96=B9=E6=B3=95?= =?UTF-8?q?=EF=BC=8C=E6=B7=BB=E5=8A=A0=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- base_plugins/echo.py | 29 ++++++----- core/bot.py | 72 ++++++++++++++++++++++++++++ core/command_manager.py | 65 ++++++++++++++++++++++--- core/config_loader.py | 34 +++++++++++++ core/ws.py | 45 ++++++++++++++--- main.py | 8 +++- models/base.py | 12 ++++- models/event.py | 104 ++++++++++++++++++++++++++++++++++++++-- models/sender.py | 35 ++++++++++++-- 9 files changed, 366 insertions(+), 38 deletions(-) create mode 100644 core/bot.py diff --git a/base_plugins/echo.py b/base_plugins/echo.py index 5a18575..87d4005 100644 --- a/base_plugins/echo.py +++ b/base_plugins/echo.py @@ -1,22 +1,25 @@ +""" +Echo 插件 + +提供 /echo 指令，用于原样回复用户输入的内容。 +""" from core.command_manager import matcher - -from ..core.ws import WS -from ..models.event import Event +from core.bot import Bot +from models.event import Event -# TODO 把该死的这些给抽象化 @matcher.command("echo") -async def handle_echo(bot: WS, event: Event, args: list[str]): +async def handle_echo(bot: Bot, event: Event, args: list[str]): + """ + 处理 echo 指令，原样回复用户输入的内容 + + :param bot: Bot 实例 + :param event: 消息事件对象 + :param args: 指令参数列表 + """ if not args: reply_msg = "请在指令后输入要回复的内容，例如：/echo 你好" else: reply_msg = " ".join(args) - if event.message_type == "group": - await bot.call_api( - "send_group_msg", {"group_id": event.group_id, "message": reply_msg} - ) - else: - await bot.call_api( - "send_private_msg", {"user_id": event.user_id, "message": reply_msg} - ) + await event.reply(reply_msg) diff --git a/core/bot.py b/core/bot.py new file mode 100644 index 0000000..6420d76 --- /dev/null +++ b/core/bot.py @@ -0,0 +1,72 @@ +""" +Bot 抽象模块 + +定义了 Bot 类，封装了 OneBot API 的调用逻辑，提供了便捷的消息发送方法。 +""" +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from .ws import WS + from ..models.event import Event + + +class Bot: + """ + Bot 抽象类，封装 API 调用和常用操作 + """ + + def __init__(self, ws_client: "WS"): + """ + 初始化 Bot 实例 + + :param ws_client: WebSocket 客户端实例，用于底层通信 + """ + self.ws = ws_client + + async def call_api(self, action: str, params: dict = None) -> dict: + """ + 调用 OneBot API + + :param action: API 动作名称 + :param params: API 参数 + :return: API 响应结果 + """ + return await self.ws.call_api(action, params) + + async def send_group_msg(self, group_id: int, message: str) -> dict: + """ + 发送群消息 + + :param group_id: 群号 + :param message: 消息内容 + :return: API 响应结果 + """ + return await self.call_api( + "send_group_msg", {"group_id": group_id, "message": message} + ) + + async def send_private_msg(self, user_id: int, message: str) -> dict: + """ + 发送私聊消息 + + :param user_id: 用户 QQ 号 + :param message: 消息内容 + :return: API 响应结果 + """ + return await self.call_api( + "send_private_msg", {"user_id": user_id, "message": message} + ) + + async def send(self, event: "Event", message: str) -> dict: + """ + 智能发送消息，根据事件类型自动选择发送方式 + + :param event: 触发事件对象 + :param message: 消息内容 + :return: API 响应结果 + """ + if event.message_type == "group" and event.group_id: + return await self.send_group_msg(event.group_id, message) + elif event.user_id: + return await self.send_private_msg(event.user_id, message) + return {"status": "failed", "msg": "Unknown message target"} diff --git a/core/command_manager.py b/core/command_manager.py index ff4bb40..390c51c 100644 --- a/core/command_manager.py +++ b/core/command_manager.py @@ -1,3 +1,8 @@ +""" +命令管理器模块 + +提供装饰器用于注册消息指令、通知处理器和请求处理器，并负责事件的分发。 +""" import inspect from typing import Any, Callable, Dict, List, Tuple @@ -8,7 +13,16 @@ comm_prefixes = global_config.bot.get("command", ("/",)) class CommandManager: + """ + 命令管理器，负责注册和分发指令、通知和请求事件 + """ + def __init__(self, prefixes: Tuple[str, ...] = ("/",)): + """ + 初始化命令管理器 + + :param prefixes: 命令前缀元组 + """ self.prefixes = prefixes self.commands: Dict[str, Callable] = {} # 存储消息指令 self.notice_handlers: List[Dict] = [] # 存储通知处理器 @@ -16,7 +30,12 @@ class CommandManager: # --- 1. 消息指令装饰器 --- def command(self, name: str): - """装饰器：注册消息指令，例如 @matcher.command("echo")""" + """ + 装饰器：注册消息指令 + + :param name: 指令名称（不含前缀） + :return: 装饰器函数 + """ def decorator(func): self.commands[name] = func @@ -26,7 +45,12 @@ class CommandManager: # --- 2. 通知事件装饰器 --- def on_notice(self, notice_type: str = None): - """装饰器：注册通知处理器""" + """ + 装饰器：注册通知处理器 + + :param notice_type: 通知类型，如果为 None 则处理所有通知 + :return: 装饰器函数 + """ def decorator(func): self.notice_handlers.append({"type": notice_type, "func": func}) @@ -36,7 +60,12 @@ class CommandManager: # --- 3. 请求事件装饰器 --- def on_request(self, request_type: str = None): - """装饰器：注册请求处理器""" + """ + 装饰器：注册请求处理器 + + :param request_type: 请求类型，如果为 None 则处理所有请求 + :return: 装饰器函数 + """ def decorator(func): self.request_handlers.append({"type": request_type, "func": func}) @@ -46,7 +75,12 @@ class CommandManager: # --- 消息分发逻辑 --- async def handle_message(self, bot, event): - """解析并分发消息指令""" + """ + 解析并分发消息指令 + + :param bot: Bot 实例 + :param event: 消息事件对象 + """ if not event.raw_message: return @@ -77,21 +111,38 @@ class CommandManager: # --- 通知分发逻辑 --- async def handle_notice(self, bot, event): - """分发通知事件""" + """ + 分发通知事件 + + :param bot: Bot 实例 + :param event: 通知事件对象 + """ for handler in self.notice_handlers: if handler["type"] is None or handler["type"] == event.notice_type: await self._run_handler(handler["func"], bot, event) # --- 请求分发逻辑 --- async def handle_request(self, bot, event): - """分发请求事件""" + """ + 分发请求事件 + + :param bot: Bot 实例 + :param event: 请求事件对象 + """ for handler in self.request_handlers: if handler["type"] is None or handler["type"] == event.request_type: await self._run_handler(handler["func"], bot, event) # --- 通用执行器：自动注入参数 --- async def _run_handler(self, func, bot, event, args=None): - """根据函数签名自动注入 bot, event 或 args""" + """ + 根据函数签名自动注入 bot, event 或 args + + :param func: 目标处理函数 + :param bot: Bot 实例 + :param event: 事件对象 + :param args: 指令参数（仅消息指令有效） + """ sig = inspect.signature(func) params = sig.parameters kwargs = {} diff --git a/core/config_loader.py b/core/config_loader.py index 246ba81..0882266 100644 --- a/core/config_loader.py +++ b/core/config_loader.py @@ -1,3 +1,8 @@ +""" +配置加载模块 + +负责读取和解析 config.toml 配置文件，提供全局配置对象。 +""" from pathlib import Path from typing import Any, Dict @@ -5,12 +10,26 @@ import tomllib class Config: + """ + 配置加载类，负责读取和解析 config.toml 文件 + """ + def __init__(self, file_path: str = "config.toml"): + """ + 初始化配置加载器 + + :param file_path: 配置文件路径，默认为 "config.toml" + """ self.path = Path(file_path) self._data: Dict[str, Any] = {} self.load() def load(self): + """ + 加载配置文件 + + :raises FileNotFoundError: 如果配置文件不存在 + """ if not self.path.exists(): raise FileNotFoundError(f"配置文件 {self.path} 未找到！") @@ -20,14 +39,29 @@ class Config: # 通过属性访问配置 @property def napcat_ws(self) -> dict: + """ + 获取 NapCat WebSocket 配置 + + :return: 配置字典 + """ return self._data.get("napcat_ws", {}) @property def bot(self) -> dict: + """ + 获取 Bot 基础配置 + + :return: 配置字典 + """ return self._data.get("bot", {}) @property def features(self) -> dict: + """ + 获取功能特性配置 + + :return: 配置字典 + """ return self._data.get("features", {}) diff --git a/core/ws.py b/core/ws.py index 1c4a162..655fcc8 100644 --- a/core/ws.py +++ b/core/ws.py @@ -1,3 +1,8 @@ +""" +WebSocket 核心模块 + +负责与 OneBot 实现端建立 WebSocket 连接，处理消息接收、事件分发和 API 调用。 +""" import asyncio import json import traceback @@ -8,12 +13,20 @@ import websockets from models import Event +from .bot import Bot from .command_manager import matcher from .config_loader import global_config class WS: + """ + WebSocket 客户端类，负责与 OneBot 实现端建立连接并处理通信 + """ + def __init__(self): + """ + 初始化 WebSocket 客户端 + """ # 读取参数 cfg = global_config.napcat_ws self.url = cfg.get("uri") @@ -22,9 +35,12 @@ class WS: self.ws = None self._pending_requests = {} + self.bot = Bot(self) async def connect(self): - """主连接循环""" + """ + 主连接循环，负责建立连接和自动重连 + """ headers = {"Authorization": f"Bearer {self.token}"} if self.token else {} while True: @@ -50,7 +66,11 @@ class WS: await asyncio.sleep(self.reconnect_interval) async def _listen_loop(self, websocket): - """核心监听循环""" + """ + 核心监听循环，处理接收到的 WebSocket 消息 + + :param websocket: WebSocket 连接对象 + """ async for message in websocket: try: data = json.loads(message) @@ -72,10 +92,15 @@ class WS: print(f" 解析消息异常: {e}") async def on_event(self, raw_data: dict): - """事件分发层：根据 post_type 调用 matcher 对应的处理器""" + """ + 事件分发层：根据 post_type 调用 matcher 对应的处理器 + + :param raw_data: 原始事件数据字典 + """ try: # 解析为 Event 对象 event = Event.from_dict(raw_data) + event.bot = self.bot # 格式化时间用于打印 t = datetime.fromtimestamp(event.time).strftime("%H:%M:%S") @@ -87,19 +112,19 @@ class WS: print( f" [{t}] [消息] {event.message_type} | {event.user_id}: {event.raw_message}" ) - await matcher.handle_message(self, event) + await matcher.handle_message(self.bot, event) # B. 通知事件 (Notice) elif event.post_type == "notice": print( f" [{t}] [通知] {event.notice_type} | 来自: {event.group_id or '私聊'}" ) - await matcher.handle_notice(self, event) + await matcher.handle_notice(self.bot, event) # C. 请求事件 (Request) elif event.post_type == "request": print(f" [{t}] [请求] {event.request_type} | 内容: {event.comment}") - await matcher.handle_request(self, event) + await matcher.handle_request(self.bot, event) # D. 元事件 (Meta Event) - 通常用来心跳检测，可不处理 elif event.post_type == "meta_event": @@ -110,7 +135,13 @@ class WS: traceback.print_exc() async def call_api(self, action: str, params: dict = None): - """调用 OneBot API""" + """ + 调用 OneBot API + + :param action: API 动作名称 + :param params: API 参数 + :return: API 响应结果 + """ if not self.ws: return {"status": "failed", "msg": "websocket not initialized"} diff --git a/main.py b/main.py index 3fbd9f2..11b79ca 100644 --- a/main.py +++ b/main.py @@ -1,10 +1,16 @@ -# main.py +""" +NEO Bot 主程序入口 +""" import asyncio +import base_plugins # noqa: F401 别动这里是加载插件的 from core import WS async def main(): + """ + 主函数，启动 WebSocket 连接 + """ bot = WS() await bot.connect() diff --git a/models/base.py b/models/base.py index f1d6a0e..d28f641 100644 --- a/models/base.py +++ b/models/base.py @@ -1 +1,11 @@ -# TODO 数据类型 +""" +基础数据模型模块 +""" + + +class BaseModel: + """ + 基础模型类 + """ + + pass diff --git a/models/event.py b/models/event.py index d5d5b60..b48b148 100644 --- a/models/event.py +++ b/models/event.py @@ -1,23 +1,54 @@ +""" +事件模型模块 + +定义了 Event 类和 MessageSegment 类，用于封装 OneBot 11 的上报事件和消息段。 +""" from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, TYPE_CHECKING from .sender import Sender +if TYPE_CHECKING: + from core.bot import Bot + @dataclass class MessageSegment: + """ + 消息段，对应 OneBot 11 标准中的消息段对象 + """ + type: str + """消息段类型，如 text, image, at 等""" + data: Dict[str, Any] + """消息段数据""" @property def text(self) -> str: + """ + 获取文本内容（仅当 type 为 text 时有效） + + :return: 文本内容 + """ return self.data.get("text", "") if self.type == "text" else "" @property def image_url(self) -> str: + """ + 获取图片 URL（仅当 type 为 image 时有效） + + :return: 图片 URL + """ return self.data.get("url", "") if self.type == "image" else "" def is_at(self, user_id: int = None) -> bool: + """ + 判断是否为 @某人 + + :param user_id: 指定的 QQ 号，如果为 None 则只判断是否为 at 类型 + :return: 是否匹配 + """ if self.type != "at": return False if user_id is None: @@ -30,31 +61,93 @@ class MessageSegment: @dataclass class Event: + """ + 事件类，封装了 OneBot 11 的上报事件 + """ + post_type: str + """上报类型: message, notice, request, meta_event""" + self_id: int + """收到消息的机器人 QQ 号""" + time: int + """事件发生的时间戳""" + # --- 消息事件字段 --- message_type: Optional[str] = None + """消息类型: group, private""" + sub_type: Optional[str] = None + """消息子类型""" + message_id: Optional[int] = None + """消息 ID""" + user_id: Optional[int] = None + """发送者 QQ 号""" + raw_message: Optional[str] = None + """原始消息内容""" + message: List[MessageSegment] = field(default_factory=list) + """消息内容列表""" + sender: Optional[Sender] = None + """发送者信息""" + group_id: Optional[int] = None + """群号""" + target_id: Optional[int] = None + """目标 QQ 号""" + # --- 通知事件字段 --- notice_type: Optional[str] = None - operator_id: Optional[int] = None - duration: Optional[int] = None - honor_type: Optional[str] = None + """通知类型""" + operator_id: Optional[int] = None + """操作者 QQ 号""" + + duration: Optional[int] = None + """时长""" + + honor_type: Optional[str] = None + """荣誉类型""" + + # --- 请求事件字段 --- request_type: Optional[str] = None + """请求类型""" + flag: Optional[str] = None + """请求 flag""" + comment: Optional[str] = None + """验证信息""" + + # 注入的 Bot 实例 + bot: Optional["Bot"] = field(default=None, init=False) + """关联的 Bot 实例""" + + async def reply(self, message: str) -> dict: + """ + 快捷回复消息 + + :param message: 回复内容 + :return: API 响应结果 + """ + if not self.bot: + return {"status": "failed", "msg": "Bot instance not attached to event"} + return await self.bot.send(self, message) @classmethod def from_dict(cls, data: dict): + """ + 从字典创建 Event 对象 + + :param data: 原始事件数据字典 + :return: Event 对象 + """ raw_msg_array = data.get("message") segments = [] if isinstance(raw_msg_array, list): @@ -84,12 +177,15 @@ class Event: # --- 快捷判断工具 --- @property def is_message(self) -> bool: + """是否为消息事件""" return self.post_type == "message" @property def is_notice(self) -> bool: + """是否为通知事件""" return self.post_type == "notice" @property def is_request(self) -> bool: + """是否为请求事件""" return self.post_type == "request" diff --git a/models/sender.py b/models/sender.py index b92ecaa..3d63821 100644 --- a/models/sender.py +++ b/models/sender.py @@ -1,17 +1,42 @@ +""" +发送者模型模块 + +定义了 Sender 类，用于封装 OneBot 11 的发送者信息。 +""" from dataclasses import dataclass from typing import Optional @dataclass class Sender: + """ + 发送者信息类，对应 OneBot 11 标准中的 sender 字段 + """ + user_id: int + """发送者 QQ 号""" + nickname: str + """昵称""" + sex: str = "unknown" + """性别，male 或 female 或 unknown""" + age: int = 0 + """年龄""" # 群聊特有字段 - card: Optional[str] = None # 群名片 - area: Optional[str] = None # 地区 - level: Optional[str] = None # 等级 - role: Optional[str] = None # 角色: owner/admin/member - title: Optional[str] = None # 专属头衔 + card: Optional[str] = None + """群名片／备注""" + + area: Optional[str] = None + """地区""" + + level: Optional[str] = None + """成员等级""" + + role: Optional[str] = None + """角色，owner 或 admin 或 member""" + + title: Optional[str] = None + """专属头衔"""