86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
"""
|
|
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"}
|