""" Bot 抽象模块 定义了 Bot 类,封装了 OneBot API 的调用逻辑,提供了便捷的消息发送方法。 """ from typing import TYPE_CHECKING if TYPE_CHECKING: from .ws import WS from models import OneBotEvent 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, auto_escape: bool = False) -> dict: """ 发送群消息 :param group_id: 群号 :param message: 消息内容 :param auto_escape: 是否自动转义 :return: API 响应结果 """ return await self.call_api( "send_group_msg", {"group_id": group_id, "message": message, "auto_escape": auto_escape} ) async def send_private_msg(self, user_id: int, message: str, auto_escape: bool = False) -> dict: """ 发送私聊消息 :param user_id: 用户 QQ 号 :param message: 消息内容 :param auto_escape: 是否自动转义 :return: API 响应结果 """ return await self.call_api( "send_private_msg", {"user_id": user_id, "message": message, "auto_escape": auto_escape} ) async def send(self, event: "OneBotEvent", message: str, 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"}